msackman

The software industry seems to fetishize complexity. It's extremely harmful, yet often appealing, and I don't know how we can avoid it.

Paxos: I've been trying to look for that. Having cloned the code and grepped for paxos I'm getting no hits. Where is the paxos implementation?

So GoshawkDB doesn't have an SQL engine currently, so in that way it's probably not comparable with F1. GoshawkDB stores and accesses an object graph. Hopefully the howtos on the website will help people get into the mindset.

I'm not sure if it's worth trying to compare anything to Spanner or F1 because unless I'm mistaken, no one outside of Google can use F1 or Spanner - they're not released. So who knows what the performance of these things actually is? There's no way to verify the claims made in any Google paper.

Heh, definitely! I could ramble for hours about how there are so many disincentives to share anything until launch - the whole industry really isn't set up to actually make progress. However, multiple independent co-invention of stuff is still valuable from a validation point of view. The worst bits I see are when something from 20 years ago gets reinvented. I have emailed Irene just to see if there's anything or anyway to collaborate.

I'm the author of GoshawkDB.

I've just been watching the talk on this at https://www.youtube.com/watch?v=yE3eMxYJDiE. GoshawkDB has a very similar design wrt the messaging and replication design. In fact, in some places, it appears that GoshawkDB's design is a little simpler.

There are obviously many differences too: for example GoshawkDB runs transactions on the client, GoshawkDB uses Paxos Synod instead of 2PC, and GoshawkDB clients only connect to one server so there are 2 extra hops, but that's a constant so from a scaling pov, it should behave the same.

One of the biggest differences is GoshawkDB uses Vector Clocks (that can grow and shrink) rather than loosely synchronized clocks.

This TAPIR work does look great - I had no idea that it was ongoing. I'll read through the paper too, but it's great that GoshawkDB has so many design ideas in common.

Not quite. Only F+1 are required per object to vote. As I explain in the blog post (https://goshawkdb.io/blog.html#20151224), the remaining F can be sent after the outcome is known.

Well I have no idea of how Cassandra works, so I can't comment on that.

GoshawkDB gets every object touched in the transaction to vote on the outcome of the transaction, in parallel, and guarantees that if two transactions touch the same object, there must be at least one copy of that object that gets to vote on both transactions, thus can enforce any dependencies as necessary.

Edit: further clarity: the use of consistent hashing in GoshawkDB is to ensure that each node has an equal number of object copies. It is the Paxos Synod algorithm that ensures a majority of those copies of any object get to vote on every transaction that touches that object, thus ensures that for any two transactions touching the same object, there is at least one copy of the object that gets to vote on both.

Yes, you're right. Now where that paper finishes is pretty much where GoshawkDB starts. Because with GoshawkDB, the server controls the object UUId, you can push as much entropy in there as necessary so the problem identified in the paper is gone. However, ultimately what GoshawkDB ends up generating is basically a set of numbers that you interpret as a way to construct an order of the different nodes (it's the path down the tree, as described in the paper). This set of numbers you could intentionally change in order to change the load balancing to exploit locality.

So yes, currently it'll rigorously enforce a uniform balancing. However, it should be possible to move objects around if/when necessary. I doubt that'll be in for 0.2 though!

For those interested in a bit more detail how GoshawkDB actually works, I've added a (slightly rushed) blog post on the topic, which covers transaction lifecycle. https://goshawkdb.io/blog.html#20151224

I don't think that's the problem though. The problem that I'm thinking of is that when a cluster grows in size, due to the use of consistent hashing, there'll be a set of objects that need to move between nodes. Calculating and achieving that movement is what concerns me. The exact properties are explained early on in http://arxiv.org/abs/1503.04988

I'm not expecting to ever need to model a global property of "these are the set of nodes that are up and running". I always worry about the effect of weird and wacky network issues on such systems.

Excellent points well made :) I think I may well change the description then as you suggest, and once I have clients that support the same API model as things like ZODB then I'd describe it as something like "document store that can also be accessed like an object store" (well, hopefully not that long winded...).

Thanks for the link. Whilst I'm not arguing with your point, I believe I've never used the term "object database" to describe GoshawkDB, only "object store". I guess I'm struggling to find a more accurate term than "object store" to describe GoshawkDB. I don't like "document store", as to me "document" tends to be something human readable.

Having just had a quick look at both, the only point I'm willing to make is that ignite's distributed transactions use 2PC. https://apacheignite.readme.io/docs/transactions

2PC, at least in its usual forms, is not fool-proof: if failures happen in the "right" order then the whole txn gets blocked: it's actually proven to be equivalent to Paxos with F=0 (i.e. no ability to tolerate failure). On this point alone, GoshawkDB stands apart from anything that uses 2PC.

So that I understand, what is the definition of object store?

> As I understand it, in this hypothetical case, the majority of the system would work fine if 3 nodes fail, but there could also be some percentage of transactions which are unfulfillable due to this scenario while there should be many that can still be answered with confidence. ( if I'm understanding correctly that data does not go into every node with a configuration like this and is consistently hashed across )

Yes, you understand it perfectly. :)

> Ultimately it sounds like it's headed in a good path and embracing this partial-fail-but-that-doesn't-mean-show-is-over with specifics is a good thing to get ironed out and would certainly be a key thing to get right early, even if it's largely semantics and client behavior.

Indeed. Semantics are very important and I want to make sure GoshawkDB is easy to understand and reason about at all times.

Ok, so F=2, so 2F+1 is 5. So client 1 creates a txn that writes to object x. x has replicas on A-E, but not F and G. So client 1's txn gets sent to F+1 drawn from A-E. In your scenario A and B are down so that only leaves C, D and E. So they vote on the txn and vote to Commit. This vote forms a Paxos round as per Gray and Lamport's paper "Consensus on transaction commit". Assuming that all these votes get received and written to disk by at least F+1 acceptors (which will almost certainly also be C, D, and E), and that C, D and E all form the exact same overall outcome then consensus has been achieved: the outcome is fixed and stable.

A and B come up at the same time that C and D go down. C, D and E all knew (in their role as acceptors, not voters) that the result of that txn has to go to _all_ of A-E. So when A and B come up, E will see this and send them the txn outcome. Now A and B can't progress at this point yet because they've only seen the outcome from one acceptor (E), not the minimum F+1. So as they spot C and D are down, they will start additional paxos rounds to try and get the txn aborted. Those rounds will _have_ to contact E, and they will find that actually the txn votes led to a commit. Now A and B will also play the role of acceptor, taking over from the failed C and D. Thus we'll finally have A B and E as acceptors all with the original commit votes, they will form the same outcome, and all the voters and learners will get the same result. I realise this is probably not the cleanest explanation, but this is the chief purpose of consensus algorithms in general: once a majority (F+1) of nodes form consensus about something, that consensus can never change, and can only be propagated further.

> Client 2 tries to read, contacting nodes A, B, C, F, G. 4 out of the 5 respond successfully, so it calls it good. But it can't possibly read the value that client 1 wrote.

Client 2 is trying to read object x and object x only exists on A-E, not F and G. So client 2 cannot ask F and G to take part. I assume you mistyped and meant A, B, and E, not C as I thought C and D were down at this point.

Client 2 will get A B and E to vote on this next transaction and that will occur as normal as we have only have 2 failures currently (C and D).

> Whatever way you do it, the number of nodes contacted for a write + number of nodes contacted for a read has to add up to more than the cluster size.

No, that's not true. It has to add up to more than the number of replicas of an object. That can be much less than the cluster size in GoshawkDB. This property is not true of several other data stores.

> If your cluster is somehow sharded then you can work around this with e.g. consistent hashing (so that client 2 knows which 5 of the 7 nodes a particular key must be stored on, and contacts the correct subset), but that only works if client 2 can know what the key was

Yes, this is exactly what GoshawkDB does (though my consistent hashing algorithms are quite unlike the norm).

> and I think Cassandra already supports that

Maybe - I'm certainly not a cassandra expert. I'm pretty sure cassandra does not support full transactions. For example: https://docs.datastax.com/en/cassandra/2.0/cassandra/dml/dml...

"Cassandra supports atomicity and isolation at the row-level, but trades transactional isolation and atomicity for high availability and fast write performance".

Which is fair enough - there's definitely a use case for the properties that Cassandra offers - as I've written on the GoshawkDB website, GoshawkDB is certainly not a competitor of Cassandra. GoshawkDB does support full transactions.

So ZODB has really great integration with Python: the fact that you can just subclass Persistent and get persistent objects is a really nice model.

That type of model is certainly the goal for GoshawkDB but I certainly accept your point that currently the Go client API offers something that is more similar to a document store. With GoshawkDB it really is up to the client as to what API you offer. For languages which allow you full proxies of all object field gettors and settors then you should be able to match ZODB, and that would certainly be the goal. But without that sort of support in the programming language, the only other way to get there is to define your object types externally, and then compile special classes that do the magic for you. That too is something I'll look at for GoshawkDB: it's likely to ultimately be faster than relying on reflection and other techniques.

An awful lot of problems with these sorts of products are caused by the same terminology being used for different meanings. I'm sorry if you feel I've made this worse. Hopefully it is the case that in future releases there will be both more clients and they will offer a model which is closer to what you're expecting with the term "object store".

> Considering more than one exact txn I imagine will hit a single specific node often, at large scale with a single mode down even if that means 5% of transactions block, you are basically growing a queue of waiting work indefinitely with the only upper bound being how much ram you have. Meanwhile 5% of clients will be waiting and this node may take awhile to come back if it needs something.

Ahh, no! For each obj, there are 2F+1 replicas, each replica on a different node. For each txn that hits an obj, you only need F+1 of those replicas to vote on the txn. So, provided F > 0, a single failure will never cause anything to block.

> I would suggest that you have at minimum strict timeouts for the the transactions, in conjunction with an immediate fail if the node is answer is not available right now. So a client would never wait more than X seconds or the transaction is aborted and if necessary rolled back.

I agree. I think it would be very difficult for a client to do anything sensible with such information, but even if all I'm doing is getting the client to resubmit the txn verbatim, at least it clears up the resource usage on the server, which is the most important thing.

> Furthermore , what if a node never comes back. Somehow there seems to need a transaction failure that is handed back to the client whether it's node is down, node is gone forever , or node simply timed out.

Well, this only becomes a problem if > F nodes fail and never come back - the whole design of consensus systems is to cope with failures up to a certain threshold. Provided <= F nodes fail, the failures are detected and any txns that are in-flight are safely aborted (or, if it actually committed, that information is propogated) - this is all just usual Paxos stuff. But yes, again, I completely agree: if you have a massive failure and you lose data, then you are going to have to recover from that. For goshawkDB, that's going to require changing topology which is not supported in 0.1, but is the main goal for 0.2.

> At the end of the day even if your system is able to handle N thousands of transactions pleasantly waiting and can still answer other requests indefinitely that is a great accomplishment, but in practice may not be ideal for many workloads. People and computers tend to both retry interactions with data that are slow or failed and the combination of just taking on more and more work and hoping everything flushes out when things become healthy is a recipe for a thundering herd, and better served by something like an async work queue type of pattern.

Oh absolutely. In a previous life I did much of the core engineering on RabbitMQ. It was there that I slowly learnt the chief problem tends to be that under heavy load, you end up spending more CPU per event than under light load, so as soon as you go past a certain tipping point, it's very difficult to come back. I certainly appreciate that human interaction with a data store is going to require consistent and predictable behaviour.

Thanks for your input.

Not quite: Paxos Synod is merely used as the replacement for 2PC as 2PC does not work. So that's just achieving consensus on the txn votes for each node. It's actually vector clocks (and improvements thereof) that manage the dependencies between txns. The novel bits of vector clocks in GoshawkDB is getting them to expand an contract as necessary, rather than just becoming enormous, which happens with some systems.

> With your situation #1, what if this is very common transaction and therefore you have 100 of these all waiting. What about 1000, 5000 etc. what system resources are used to let these transactions wait indefinitely ( if I understand your semantics with specific regard to blocking )?

So, as each client can only have 1 outstanding txn at a time, this requires there are 100, 1000, 5000 etc clients all submitting the same txn at the same time, and presumably they're all connected to the same server node, and for each of those txns, the set of server nodes that need to be contacted has been calculated, and the necessary messages sent. At that point failures occur so it's not known who has received which messages, so all these txns block.

The only system resources that are held at this point is RAM. Yes, it could get so bad that you eat up a lot of RAM - this is release 0.1 after all. There are a few areas where I have tickets open to make sure that goshawkdb can spill this sort of state to disk as necessary, though it may not be too horrible just relying on swap to start with.

> Some systems handle this as a failure that is communicated to the client rapidly. Other systems let N clients actually wait indefinitely but at the cost of taking up a thread / file descriptor, etc. in systems that have finite amounts of threads for example this would then be communicated in his paradigm as a such of upper bounds as to how many requests one could have waiting.

The problem is that in this particular case, the txn could have committed, it's just the server who initially received the txn from the client and then forwarded it to the necessary server nodes can't learn that it's committed due to failures. Now certainly I could make the client informed that the outcome is delayed, but the client may not be able to do anything with that information: it can't know for sure if the txn has been committed or aborted.

The entire codebase is written using actors and finite state machines. Because Go supports green threads and go-routines can be pre-emptively suspended, there is no problem with eating OS threads. In general, the design is such that the only thing that should block is the client and on the server the actor/state-machine that is talking to that client.

Hi, I'm the author of GoshawkDB. Thanks for your questions - they're certainly not answered on the website so I'll do my best to answer them here and incorporate them into the website later.

> Curious how this works at scale. For example if a node is down and N requests start blocking, is there an upper bound to N? What happens when this is reached? Does N map to a goroutine? A file descriptor? Is there timeouts?

The blocking actually occurs on a per-client basis: when a client submits a txn to the server to which it's connected, that server will calculate which other server nodes need to be contacted in order for the txn to be voted on. If it cannot find enough servers due to failures then it will outright reject the txn and tell the client that there are currently too many failures going on.

However, the server could find the right set of server nodes and send off the txn to be voted on and send of the txn for verification and validation. Right at that point there could be further failures and so it's not known how far those messages got. It is in this case that the txn could block as until some node recover it's impossible to safely abort the txn. But it's important to note that:

1. If a txn starts blocking, that'll block just that txn and that client (as the client is blocked waiting on the outcome of the txn). No other part of the server or any other server is affected.

2. Other clients can be submitting other txns at the same time that need different un-failed nodes to proceed, and they progress just fine.

3. There is no global determination (yet) of whether a node is failed or not. Thus this approach can cope with really weird network issues - eg random connections between different nodes failing. There is no need for the whole remaining cluster to agree that nodes X, Y and Z are "out" - none of the algorithms depend on that sort of thing.

I hope this answers your questions.

Hi, I'm the author of GoshawkDB.

I'm no Cassandra expert, but I'm certainly under the impression that Cassandra does not offer full transactions - lightweight txns only that are atomic only for a single row.

Yup, on the whole, in order to increase performance, you don't want more and more nodes to be contacted. Remember that every txn has to have a minimum of F+1 replicas vote on each txn, so if F is really large, although they vote in parallel, you have greater chance of network delay and greater load on all machines. So GoshawkDB allows you to increase cluster size without increasing F, so getting you greater performance.

Hi. I'm the author of GoshawkDB. Thanks for your questions.

> It isn't true. There're plenty of storages with same strong consistency model (tolerates F failures of 2F+1 nodes), among them are Cassandra with lightweight transactions, Riak with consistent buckets, CockroachDB and ZooKeeper.

The point I was trying to make is the decoupling between cluster size and F. ZK, for example, requires that the cluster size is exactly 2F+1, not is minimum 2F+1. That means that as the cluster size increases, performance will decrease as more nodes have to be contacted.

> Can you please describe the algorithm that you use for implementing distributed transactions?

Yes. In detail, there will be a blog post forthcoming. However the two main points is that 2PC is replaced by Gray and Lamport's paper "Consensus on transaction Commit", and dependencies between transactions are modelled through vector clocks. As I'm visiting lots of family over xmas, I'm afraid I don't have the 12+ hours necessary to document the entire algorithm just at the moment. However, I will certainly get to this asap.

> Do you use the same approach?

I don't know. I don't use the term round-trip because that doesn't make it clear if we're talking individual messages or message-hops, and these sorts of conversation and discussion can often go wrong if we use different terminology. The 3 network-delays that I mention comes directly from the Gray and Lamport paper I mention above.

> There is no requirement for weave that the password is human readable. It can be supplied through a file, so you can happily dd if=/dev/random of=/my/weave/passwd bs=1k count=1 to create a suitable weave password

Turns out that feature I was thinking about has been removed, so the above is not true.

DH is used to ensure that every connection between every pair of weave router nodes that ever gets established uses a unique session key. Yes, the public key is exchanged in the clear, and yes, that is MITMable. But as you say, the subsequent combination of the result of the DH with the non-exchanged password solves that.

What is the weakness with this approach? Is your point that there's nothing gained from doing the DH - you may as well just generate a secret key randomly, exchange that between the peers in the clear, and still have them combine that with the password secret, and the level of information exposed is the same?

I think there is a difference. Say you have a captured stream of traffic and you want to decrypt it. With the simpler non-DH scheme, you already have the basic key, so all you need to do is guess the password, and run that through sha256 and then you can decrypt the entire stream. Now you have the password, you can decrypt every stream you capture from now on. But with the DH scheme, you have to both guess the password, and the ephemeral private keys which are never exchanged (either that, or break Curve25519). Even once you've done that, you only get access to that one captured stream - sure, you now know the password, but every connection between weave routers will use different random private keys to their DH, so you'll still have to brute force those for every stream you capture (or, again, break Curve25519). So ISTM there is a substantial difference there. That, to me, is the point of using the DH. But maybe I'm missing something... I'm slightly wondering whether people here are considering the use of DH and Public Key crypto by weave in the context of the usual "generate keys once and save them". This is just not the case in weave - weave generates fresh public and private keys for EVERY connection between routers.

There is no requirement for weave that the password is human readable. It can be supplied through a file, so you can happily dd if=/dev/random of=/my/weave/passwd bs=1k count=1 to create a suitable weave password