Video

For vector clocks, Mattern's original paper "Virtual Time and Global States of Distributed Systems" (vs.inf.ethz.ch/publ/papers/VirtTimeGlobStates.pdf) is a good read. Even better, in my opinion, is Schwarz and Mattern's "Detecting Causal Relationships in Distributed Computations: In Search of the Holy Grail", from a few years later (vs.inf.ethz.ch/publ/papers/holygrail.pdf).

Thanks for watching! Martin Kleppmann's videos are great; I hope you'll give them another try after watching mine!

I discuss state machine replication a few lectures later in the course: czcams.com/video/wHpB44jtS4g/video.html

Do you have any hot gossip about it?

The issue is that, if a physical node dies, and that physical node is mapped to many points on the ring represented by virtual nodes, then *many* points on the ring are affected. We no longer have the property that the immediate neighbor takes over the downed node's entire key range. Instead, if there are, say, 80 virtual nodes, then they're all gone at once and then 80 neighbors are affected. Of course, those 80 neighbors can be virtual, too, and mapping to various physical nodes. The Dynamo paper (in section 4.2) actually describes this as a feature of using virtual nodes, not a bug. If a physical node dies or is removed, the responsibility for its part of the key range is then spread out over the remaining physical nodes, instead of falling to just one neighbor. Likewise, if a physical node is added, instead of just taking a large burden off one neighbor, the relief will be spread out over the system. I think that's a great point, but it does makes it a little harder to reason about exactly which nodes will be affected by node addition/removal, whereas if virtual nodes aren't in the picture, it's clear that one neighbor will be affected and other nodes won't be.

If the updates are concurrent, are a few options that Dynamo-like systems support. You could store both concurrent updates and let the client sort it out. Or you could break the tie with physical timestamps, meaning that someone will lose their write. If neither of these options is workable for a given application, then a Dynamo-like system may be the wrong choice for that application.

The dynamo paper mentions the utilization of vector clocks and maintaining all values that are causally unrelated - to be resolved manually. For the DynamoDB offering: "If applications update the same item in different Regions at about the same time, conflicts can arise. To help ensure eventual consistency, DynamoDB global tables use a last writer wins reconciliation between concurrent updates, in which DynamoDB makes a best effort to determine the last writer. This is performed at the item level. With this conflict resolution mechanism, all the replicas will agree on the latest update and converge toward a state in which they all have identical data." Note that DynamoDB, undoubtedly named after the original Dynamo service, is not the exact same service - but I thought it would be helpful to include how DynamoDB (what the public gets to use) resolves this.

You're right: when we record on a channel, we're recording messages received on that channel. However, the recorded *channel* state (which is a sequence of messages) is separate from the recorded *process* state (which is a sequence of events). I go into some more detail about this scenario in the last part of this blog post: decomposition.al/blog/2019/04/26/an-example-run-of-the-chandy-lamport-snapshot-algorithm/#discussion

What message delivery property are you trying to enforce? It's a good idea to get clear on that before you start designing mechanisms to enforce the property. But let's suppose the property you want is causal message delivery. In that case, if you're in a setting where unicast messages are allowed, then it's *not* the case that a deliverable message is one that would only increment the recipient's clock by 1 in a single position (which sounds like it's what you're proposing, unless I've misunderstood). For instance, suppose we have Alice, Bob, and Carol. Alice sends a message m1 to Bob with VC [1,0,0]. Bob delivers m1, then sends m2 to Carol with VC [1,1,0]. Both m1 and m2 are unicast messages here, and m2 *should* be deliverable at Carol -- but Carol's VC is currently [0,0,0].

@@lindseykuperwithasharpie Thank You Prof! I believe that an ideal system would enforce all three properties, - totally ordered delivery - casual message delivery - fifo message delivery. The example of "unicast messages" between Alice, Bob & Carol where deliverable messages will **not** increment by 1 in a single position totally makes sense. If I may Prof., Can I also say that "broadcast only" systems ("self" messages are excluded) also do not follow "n+1" just like "unicast included" systems? I'm unsure of above because I can consider a system with Alice, Bob & Carol. Alice (now 0,0,0) sends a message m1 to both Bob (now 0,0,0) & Carol (now 0,0,0). Then Bob (now 1,0,0) sends a message m2 to both Alice (now 1,0,0) and Carol (now 1,0,0). Let's say this system didn't behave as expected & Carol receives message m2 from Bob before message m1 from Alice. Message m2 from Bob holds the vector clock (1,1,0) & Message m1 from Alice holds (1,0,0). Here Carol can compute all possible next values from her own vector clock (0,0,0) which would be a set of {(1,0,0), (0,1,0)}. This way, she can know in advance which message it would accept next from her current vector clock regardless of if she has received a message or not. If a received message isn't within the set, She can put it into a buffer & wait until another message is received. Would this work Prof.?

As for "all three properties": causal message delivery gives you FIFO message delivery too, so if you have causal delivery, you don't have to separately do anything to enforce FIFO delivery. Totally-ordered delivery, though, is another can of worms, and can't be enforced using vector clocks. For causal delivery specifically, and for the broadcast case specifically, it sounds like you're getting closer to proposing the causal broadcast protocol I described in the lecture (which, if you're curious, is the CBCAST protocol described in this Birman et al. paper: dl.acm.org/doi/10.1145/128738.128742 ). Any message that has a VC that (a) is one bigger than the receiver's VC in the sender's position, and (b) is less than or equal to the receiver's VC in every other position, is deliverable. And yes, if you wanted to, you could pre-compute the set of all VCs that meet those two criteria and then check whether any arriving message has a VC that's in the set. (It could be expensive to search the set, though, especially as it gets large.)

@@lindseykuperwithasharpie That is very cool & it clears my confusion. I had realised the inefficiency of the "pre-compute" approach later in the lecture professor when you answered your approach on "how to decide if one process should deliver a message or put in a reserve buffer for later delivery". I still find it quite interesting. I'll definately give that paper a read. Thank you so much Prof. Lindey! 💙

I'm not sure I understand your question, but there's no assumption here that the application code is something that you can actually access or inspect. All you have is the single execution that's being snapshot.

The "leftover" message is part of the global snapshot, because it's part of the recorded channel state for channel C_21, and the global snapshot consists of all the recorded process states and all the recorded channel states. What to do with any particular part of a snapshot is naturally going to depend on your reason for having taken the snapshot.

By "has seen" here, I mean "has sent or received". A marker message has no interesting payload; it only says, "Hi, I'm a marker message."

Great question! Yes, at any given time the delay queue may have multiple delayed messages in it, and when a message gets delivered, it could unblock multiple queued messages. Importantly, though, messages in the delay queue are in vector clock order (or, more specifically, in a total order that's consistent with the vector clock order), with the smallest first. So, if a message getting delivered unblocks *any* other message(s), it'll unblock the first one in the queue (and possibly the one after that, and so on). The name "delay queue" comes from the paper "Lightweight causal and atomic group multicast" by Birman, Schiper, and Stevenson ( dl.acm.org/doi/10.1145/128738.128742 ).

Yay!

That's right. I just didn't put any channel states in the diagram.

Why not give it a try and see how it goes?

We could have sent Accept(6, foo) to A1 also. Paxos only requires that we send it to a majority of acceptors. It's interesting that the algorithm still works even if we don't communicate with all the acceptors.

The idea is that d is the time it takes a message to get from one node to another, and r is the time it takes for the message to be processed on the remote end. So, it's the time it takes for a request to go across the network to a remote node (d), be processed there (r), and a response to come back (d again). This is all a bit silly, though, since in an asynchronous network, there is no fixed d, since a sent message can take arbitrarily long to be received.

I don't make the assignments for this course public, but as an alternative, I recommend checking out the excellent DSLabs: github.com/emichael/dslabs

I think the confusion here comes from the fact that RYW is what's known as a "client-centric" consistency model, whereas FIFO (and all the other consistency models I discussed in this lecture) are "data-centric". The data-centric consistency models put restrictions on how data is updated on individual replicas. They have to do with how those replicas are in sync (or not) with each other. The client-centric consistency models, on the other hand, take into account the history of interactions between a client and the replicas. Client-centric consistency models include RYW and the other "session guarantees" identified by Terry et al. ( ieeexplore.ieee.org/document/331722 ). I think it was a mistake of me to try to put a client-centric consistency model into a hierarchy with data-centric ones, since that's confusing. That said, Brzezinski et al. do have a paper ( link.springer.com/chapter/10.1007/978-3-540-24669-5_1 ) where they establish that FIFO is stronger than RYW. In particular, Theorem 1 of that paper says: "At the server side, PRAM consistency is fulfilled if and only if RYW and MW conditions are preserved." (Here, PRAM is a synonym for FIFO, and MW stands for "monotonic writes", which is another one of Terry et al.'s session guarantees.) In other words, we have FIFO if we have RYW *and* something else (MW), and we have RYW *and* something else (MW) if we have FIFO. Hence, FIFO is stronger than RYW. But the Brzezinski et al. paper is quite technical, and I have no intuitive explanation of why this is the case, sorry. (I came across this result secondhand, in Viotti and Vukolić's big scary survey of consistency models ( arxiv.org/abs/1512.00168 ) -- notice that they too put RYW below PRAM (FIFO) in their hierarchy.)

How'd your exam go?

@@lindseykuperwithasharpie excellent thank you so much you life saver

Thanks for watching.

I'm not sure if the slides for this specific talk are available, but Chris's slides for his own distributed systems course are available at www.distributedsystemscourse.com, and some of the lectures are very similar in content to this talk.

Oh, nvm, in later point, it's mentioned in lecture that map workers data is stored in the files

The causal broadcast algorithm I show here is from the classic paper "Lightweight Causal and Atomic Group Multicast" by Birman, Schiper, and Stephenson ( dl.acm.org/doi/10.1145/128738.128742 ). That paper gives informal safety and liveness proofs for the algorithm. Recently, my students and I mechanically verified the safety property ( dl.acm.org/doi/10.1145/3587216.3587222 ) using Liquid Haskell. Here's a blog post about a talk I gave describing that work: decomposition.al/blog/2022/09/07/verified-causal-broadcast-with-liquid-haskell/

@@lindseykuperwithasharpie thanks a lot 🙏 Amazing lectures.