CS4231 Notes

Peterson's Algorithm

Mutual exclusion for $2$ processes. Can be extended to $n$ processes by using a tournament tree of Peterson's Algorithm: must acquire from leaf to root and release from root to leaf!

bool wantCS[0] = false;
bool wantCS[1] = false;
int turn = 0;

// Process 0
void RequestCS() {
    wantCS[0] = true;
    turn = 1; // let it be `other`'s turn to prevent starvation
    // wait if `other` wants and it's their turn
    while (wantCS[1] == true && turn == 1);
}
void ReleaseCS() {
    wantCS[0] = false; // stop wanting it
}

// Process 1 is mirrored

Lamport's Bakery Algorithm

Mutual exclusion for $n$ processes.

for $n$ processes:

• get a number
• get served when all processes with a lower number has been served

bool choosing[N]; // choosing[i] = true <=> process i is getting a number
int number[N]; // number[i] => number of process i (if number is 0, process doesn't wanna be served)

void RequestCS(int id) {
    choosing[id] = true; // "weak mutex"
    for (int j = 0; j < n; j++) {
        temp = number[j];
        if (temp > number[id]) {
            number[id] = temp;
        }
    }
    number[id]++; // max(everyone else) + 1
    // won't be correct because everyone is racing
    // but it's fine

    // wait for processes with a smaller number OR *might* have a smaller number
    for (int j = 0; j < n; j++) {
        while (choosing[j] == true);
        // compare by (number, id)
        while (number[j] != 0 && Smaller(number[j], j, number[id], id));
    }
}
void ReleaseCS(int id) {
    number[id] = 0;
}

Dekker's Algorithm

bool wantCS[0] = false;
bool wantCS[1] = false;
int turn = 1;

void RequestCS(int i) {
    // turn doesn't change in RequestCS

    int j = 1 - i;
    wantCS[i] = true;
    while (wantCS[j]) {
        if (turn == j) {
            // temporarily release so the other guy can enter
            wantCS[i] = false;
            while (turn == j);
            wantCS[i] = true;
        }
    }
}

void ReleaseCS(int i) {
    turn = 1 - i;
    wantCS[i] = false;
}

Semaphore Semantics

Semaphores can be used to implement monitors, and vice versa.

// both of these operations are done ATOMICALLY!!
void P() {
    if (value == false) {
        add myself to queue;
        block;
    }
    value = false;
}
void V() {
    value = true;
    if (queue is not empty) {
        // exactly **ONE** process is woken
        wake up one *arbitrary process* on the queue;
    }
}

void RequestCS() { P(); }
void ReleaseCS() { V(); }

The queue is not FIFO, it has an arbitrary ordering. *unless otherwise specified by the API.

Dining Philosopher Problem

Need to avoid cycles or have a total ordering of the chopsticks to prevent deadlocks

Monitor Semantics

Semaphores can be used to implement monitors, and vice versa.

Every object in Java is a monitor.

synchronized (object) { // enters monitor
    object.wait();
    object.notify();
    object.notifyAll();
} // exits monitor

• enter monitor
  • if no one is in the monitor: I will enter
  • otherwise: I enter the monitor-queue and block
• exit monitor
  • if monitor-queue is non-empty: unblock one arbitrary process
• object.wait() -> this is one atomic operation
  • add to wait-queue
  • and block
• object.notify()
  • if wait-queue is empty, pick one arbitrary process from wait-queue and unblock it
• object.notifyAll()
  • unblock all processes on the wait-queue
  • this is only for Java-style, not Hoare-style

assume that the monitor-queue is starvation-free for CS4231 because we have no control over this.

NOTE: the wait-queue is not FIFO, so, we must maintain our own FIFO queue if we need FIFO

• Hoare-style: when you notify, someone else will takeover
  • possibly can use if (x == 1) object.wait();
• Java-style: when you notify, you will continue running
  • the other guy wakes up and immediately re-enters the queue (nothing will be done!!)
  • should probably use while (x == 1) object.wait();

// P0
synchronized (x) {
    synchronized (y) {
        y.wait();
        // only monitor-lock on y is released
        // still holds monitor-lock on x
    }
}

// P1
synchronized (x) { // P1 cannot acquire monitor-lock on x
    synchronized (y) {
        // never runs
        y.notify();
    }
}

Producer-Consumer Problem

void producer() {
    synchronized (buffer) {
        if (buffer.isFull()) {
            buffer.wait();
        }
        add to buffer;
        if (buffer WAS empty) {
            buffer.notify();
        }
    }
}

void consumer() {
    synchronized (buffer) {
        if (buffer.isEmpty()) {
            buffer.wait();
        }
        remove from buffer;
        if (buffer WAS full) {
            buffer.notify();
        }
    }
}

Note that notification will be lost if no one is waiting! Doesn't matter for this Producer-Consumer solution tho.

Reader-Writer Problem

// this solution will starve writers!

void writeFile() {
    synchronized (object) {
        while (numReader > 0 || numWriter > 0) {
            object.wait();
        }
        numWriter = 1;
    }
    write to file;
    synchronized (object) {
        numWriter = 0;
        object.notifyAll(); // wake up all readers
    }
}

void readFile() {
    synchronized (object) {
        while (numWriter > 0) {
            object.wait();
        }
        numReader++;
    } // must leave the monitor, so other readers can enter!
    write to file;
    synchronized (object) {
        numReader--;
        object.notify(); // wake up a writer
        // can be proven that only writers are waiting!
        // proof by contradiction: suppose a reader gets notified, then it is blocked, but it shouldn't be blocked because `numWriter == 0`
    }
}

To be starvation-free: maintain an explicit queue and let each thread wait on its own monitor. Avoid waiting on a common object (because there will be no guarantee of FIFO). Use a flag to handle the case where you notify before the other guy calls wait.

// this solution is starvation-free!!

Queue queue;

void writeFile() {
    Writer w = new Writer();

    synchronized (queue) {
        if (numReader > 0 || numWriter > 0) {
            w.okToGo = false;
            queue.add(w);
        } else { // no writers and no readers
            w.okToGo = true;
            numWriter = 1;
        }
    }
    synchronized (w) {
        if (!w.isOkToGo) {
            w.wait();
        }
    }

    write to file;

    synchronized (queue) {
        numWriter = 0;
        if (!queue.isEmpty()) {
            // remove a single writer OR a batch of readers from the queue
            // in a FIFO way
            for (auto request : objects) {
                numWriter++ OR numReader++;
                synchronized (request) {
                    // important because this can be called **between** line 16 and line 17
                    request.okToGo = true;
                    request.notify();
                }
            }
        }
    }
}

void readFile() {
    Reader r = new Reader();

    synchronized (queue) {
        if (numWriter > 0 || !queue.isEmpty()) {
            // queue is only non-empty if there is at least
            // one writer waiting
            r.okToGo = false;
            queue.add(r);
        } else { // no writers waiting or running
            r.okToGo = true;
            numReader++;
        }
    }
    synchronized (r) {
        if (!r.isOkToGo) {
            r.wait();
        }
    }

    write to file;

    synchronized (queue) {
        numReader--;
        if (numReader > 0) return; // only last reader runs the following code

        if (!queue.isEmpty()) {
            // remove a single writer OR a batch of readers from the queue
            // in a FIFO way
            for (auto request : objects) {
                numWriter++ OR numReader++;
                synchronized (request) {
                    // important because this can be called **between** line 16 and line 17
                    request.okToGo = true;
                    request.notify();
                }
            }
        }
    }
}

Sequential Consistency

Results should be the same as if all operations are executed in some sequential order (as if it was ran on a simple single-core system).

Operation: a single invocation-response pair of a single method of a single shared object by a process.

• Two invocation events are the same if invoker, invokee, parameters are the same.
• Two response events are the same if invoker, invokee, response are the same.

A history H is a sequence of invocations and responses ordered by wall clock time. All invocations in H must have their responses in H too.

A history H is sequential if

• any invocation is immediately followed by its response
  • no interleaving
• otherwise: it is concurrent

A history H is legal if

• all responses satisfy the sequential semantics (acts like there's one process)

Sequential $\subseteq$ Legal.

• (H | p) is process p's subhistory of H.
  • this is always sequential (because of process order)
• (H | o) is object o's subhistory of H.

Two histories are equivalent if they have exactly the same set of events

• all responses are the same (because responses are part of the event)
• ordering of events can differ

A history H is sequentially consistent if it is equivalent to (i.e. same result as) some legal sequential history S that preserves process order (partial ordering among all events).

• same result as some single-core version (invocation-response pair happens immediately)
• preserve program order

Sequential consistency is NOT a local property.

Linearizability

A history H is linearizabile if

1. execution is equivalent to (i.e. same result as) some execution such that each operation happens instantaneously (called linearization point).
2. execution is equivalent to (i.e. same result as) some legal sequential history S AND S preserves the external order in H
   • aka sequentially consistent + preserves external order
   • external order (occurred-before order): o1 < o2 if response of o1 happens before invocation of o2
     • aka we can only reorder things that are concurrent

(^ two equivalent definitions)

Sequential Consistency $\subseteq$ Linearizability

Linearizability is a local property: H is linearizable <=> for any object x, (H | x) is linearizable. Proof:

• $\implies$
  • construct a graph, then it is easy to show
• $\impliedby$
  • construct graphs for all (H | x)
  • show that we can join them with the cross edges
    • Lemma: resulting directed graph is acyclic (prove using contradiction)
  • any topological sorting gives us a linearizable H

Consistency for Registers

A register is atomic if the implementation always ensures linearizability of the history.

A register is called regular if

• when a read does not overlap with any write, the read returns the value written by one of the most recent writes
• when a read overlaps with one or more writes, the read returns the value written by one of the most recent writes OR the value written by one of the overlapping writes

Regular DOES NOT imply Sequential Consistency.
Sequential Consistency DOES NOT imply Regular.

A register is safe if

• when a read does not overlap with any write, the read returns the value written by one of the most recent writes
• when a read overlaps with one or more writes, it can return anything

Safe DOES NOT imply Sequential Consistency.
Sequential Consistency DOES NOT imply Safe.

To find "most recent writes":

1. find the write with the latest response time
2. any write that is concurrent with (1) is "most recent"

Happened-before

Happened-before relationship (a partial order)

• process order
• send-receive order
• transitivity

Note that concurrent with is not transitive!

(Integer) Logical Clock

Each process maintains a single integer.

• increment local_time at each local computation and send event
• when sending a message, attach local_time
• when receiving a message, local_time = max(local_time, sender.time) + 1
  • any f works as long as f(x, y) > x and f(x, y) > y

Event $s$ happen before $t$ $\implies$ logical clock of $s$ $<$ logical clock of $t$

^ Use the definition to prove

logical clock of $s$ $<$ logical clock of $t$ $\implies$ $s$ happened before $t$ OR they are concurrent

Vector Clock

Each process maintains a vector of size $n$, where $n$ is the number of processes.

local_vector[i] is known as the principle entry.

• increment local_vector[i] at each local computation and send event
• when sending a message, attach local_vector
• when receiving a message:
  • local_vector = pairwise_max(local_vector, sender.time)
  • local_vector[i]++

Event $s$ happen before $t$ $\iff$ vector clock of $s$ $<$ vector clock of $t$

^ ($\implies$) Use the definition to prove
^ ($\impliedby$) Consider two cases: (1) on same process; (2) there was a sequence of events to propagate the clock

• vectors $v1 < v2 \implies$ all fields in $v1$ are $\le$ all fields in $v2$ AND at least one field in $v1$ is $<$ than the corresponding field in $v2$
• this $<$ relationship is not a total order. Example: $(1,0)$ and $(0,1)$
• this $<$ relationship is transitive (because happens-before is transitive)

To convert Vector Clock to Logical Clock: it is sufficient to take the summation of all the entries in the vector clock. (taking maximum is not correct)

Matrix Clock

You are able to know what other processes (definitely) have seen.

Each process maintains $n$ vector clocks: one for each process.

local_matrix[i] is the principle vector.

• for the principle vector, we do exactly the same thing as in Vector Clock
• for other vectors in the matrix:
  • perform pairwise-max to update local copy

Total Ordering Mechanism

Use the partial order created by one of the clock systems. Then, tie-break with one of the following ways:

1. Tie-break using the process ID.
   1. unfair because it would favour those with a lower process ID
2. Tie-break randomly.
   1. Tie-break by using the time to seed a random number generator.
      1. You have to re-seed it every time, or you will run into the same problem as solution (1).
   2. Generate a random ordering of the process IDs.
   3. Use this random ordering to tie-break.

Chandy & Lamport's Snapshot Protocol

NOTE: messages are ordered and guaranteed to be FIFO through message numbers!

Each process is either

• red (has taken local snapshot)
• OR white (has not taken local snapshot)

Trigger the protocol on one process $p$:

1. $j$ turns from white -> red
2. it immediately sends out $n-1$ Marker messages to all other processes
3. upon receiving first Marker message, a process will turn from white -> red and propagate the Marker message

Need to capture "on-the-fly" application messages.

There are only four possible cases of a message M in relation to capturing snapshot

1. case 1 (M is captured by both)
   • sent before local snapshot on sender
   • received before local snapshot on receiver
2. case 2 (M is not captured by both)
   • sent after local snapshot on sender
   • received after local snapshot on receiver
3. case 3 (not CGS)
   • sent after local snapshot on sender
   • received before local snapshot on receiver
     • impossible because of FIFO
4. case 4 (need to handle this case separately *)
   • sent before local snapshot on sender
   • received after local snapshot on receiver
     • possible because receiver turned red from someone else before receiving this sender's marker

Case 4 is handled by appending these on-the-fly messages to the local snapshot.

Lamport's Logical Clock to compute CGS

• cut(e) = { f | f has smaller logical clock value than e }
• for any event e, cut(e) is a consistent global snapshot
  • just use definition of GS and CGS
    • because of process order & send-receive order
    • this will be maintained by the logical clock algorithm

Causal Order

Causal order: if $s1$ happened before $s2$, and $r1$ and $r2$ are on the same process, then $r1$ must be before $r2$.
(pessimistically assume that $s1$ caused $s2$)

Each process maintains a $n$ by $n$ matrix $M$ (NOT the matrix clock).

$M[i, j]$ is the number of messages sent from $i$ to $j$ as known by the local process

• if $i$ sends a message to $j$:
  • on $i$: $M[i,j]++$
  • piggyback $M$
• upon receiving the message with $T$, set M = pairwise-max(M,T) if
  • $T[k,j] \le M[k,j]$ for all $k \ne i$
  • AND $T[i,j] = M[i,j] + 1$
• Intuitively, $M[i,j]$ must be sent sequentially and I have seen whatever stuff that I'm supposed to have seen

Proof:

• show that if s1 happens before s2, r2 will not happen before r1 (show using properties of matrix)
  • case 1: s1 and s2 on same process
  • case 2: s1 and s2 on different processes
• show that at any given time (when all messages have been received, but not delivered), at least one message can be delivered, induction will handle everything else
  • define set of successor messages:
    • the next-to-deliver message from each sender (if the sender has an undelivered message)
  • define top successor message:
    • at least one of the successor messages has no other send events that happened-before it
  • any of these top successor messages can be sent

Causal Ordering of Broadcast Messages

Exactly the same as point-to-point.

Total Ordering of Broadcast Messages

Total ordering only applies to broadcast messages.

Atomic broadcast: all messages delivered to all processes in exactly the same order. (impossible in asynchronous system)

Total does NOT imply Causal
Causal does NOT imply Total

| | | | -------------------------------------------------------------------------- | -------------------------------------------------------------------------- |

Leader Election on Anonymous Ring

• Ring topology
  • simplest topology is line topology
  • but ring topology doesn't break from a single edge failure
  • so, we consider ring topology
• No unique identifiers
  • the nodes are completely indistinguishable

Leader Election on anonymous ring is impossible with deterministic algorithms.

• same initial state
• same algorithm on each node
• same steps are taken
  • consider worst case of all nodes executing the same step at the same time and speed
  • we only need to consider the algorithm failing for one case
• same final state
  • either the algorithm fails because no leader is found
  • or everyone claims they're the leader.

Leader Election on a Ring (Chang-Roberts Algorithm)

Each node has a unique identifier.

Nodes only send messages clockwise.

Each node acts independently on their own.

• a node sends election message with its own ID clockwise
• election message is forwarded if message.ID > own.ID
  • else ignore it
• a node becomes the leader if it sees its own election message

Performance:

For distributed systems, network communication is the bottleneck! Performance is described by message complexity: total number of messages sent by all the nodes.

• in the best case, with $n$ rings: $n+n - 1 = \Theta(n)$ messages
  • consider sorted clockwise
  • $n$ messages sent / forwarded (of the winner's election message)
  • $n-1$ messages, one each by the other nodes
• in the worst case, with $n$ rings: $(n*(n+1)) / 2 = \Theta(n^2)$ messages
  • consider sorted anti-clockwise
• in the average case, $O(n \log n)$ (probability MATH)

Leader Election on General Graph (n is known)

• Complete Graph
  • send own ID to all other nodes
  • wait for all $n-1$ other IDs
  • if you're the biggest, you win
• Any connected graph
  • flood your ID to all other nodes
    • with forwarding of message
    • won't relay if relayed before
  • wait for all $n - 1$ other IDs
  • if you're the biggest, you win

Leader Election on General Graph (n unknown)

• Complete Graph
  • no such case because you're connected
• Any connected graph
  • calculate the number of nodes (request-response spanning tree construction)
    • construct a spanning tree, rooted at whoever wants to count the nodes
    • goal: each node will know its parent and children (such that it's a tree!!)
      • node X will "child request" all its neighbours
      • the neighbour will return "YES" or "NO"
        • a node can only say "YES" to one parent
        • a node must say "NO" to all other requests
    • then we just do the good old BFS / DFS on the spanning tree
      • don't send messages to non-tree edges (if the node replied "NO")
  • resolve to previous case of known $n$

Distributed Consensus Concepts

Version 1: No failures

Trivial. Just broadcast to every other node, take the majority of the results (with tie breaking).

Version 2: Node Crash; Reliable Channels; Synchronous

• Termination: all nodes (that have not failed) eventually decide
  • proof: obvious since no waiting (well-defined rounds)
• Agreement: all nodes that decide should decide on the same value
  • including the nodes that would crash after deciding (because its decision can be used)
  • proof: below
• Validity: if all nodes have the same initial input: that value is the only possible decision
  • to avoid a trivial algorithm
  • proof: obvious since len(S) == 1

To tolerate $f$ failures: requires $f+1$ rounds of broadcasting!

Theorem: With $f$ crash failures, any consensus protocol will take $\Omega(f)$ rounds.

• Any deterministic consensus protocol will take at least $f+1$ rounds. (proof is beyond CS4231)

Consensus(my_input) {
    S = {my_input}

    // f+1 rounds
    for int i = 1; i <= f+1; i++ {
        send S to all other nodes
        receive n-1 sets
        for each received set T {
            S = S union T
        }
    }

    decide on min(S)
    return decision
}

Version 3: No Node Crash; Unreliable Channels

(aka Coordinated Attack Problem)

Impossible to achieve the goals with a deterministic algorithm.

Proof:

• For a deterministic algorithm: two executions are indistinguishable if the nodes see the same things (messages and inputs)
• Consider the case where all messages are dropped, if the input is the same, the executions are indistinguishable
• Consider this:
  • A (input = 0, decision = 0), B (input = 0, decision = 0) by validity
  • A (input = 1, decision = ?), B (input = 0, decision = 0) by indistinguishability, B must decide on 0 if its input is 0
  • A (input = 1, decision = 0), B (input = 0, decision = 0) by agreement, A must also decide on 0 since B decides on 0
  • A (input = 1, decision = 0), B (input = 1, decision = ?) by indistinguishability, A must decide on 0 if its input is 1
  • A (input = 1, decision = 0), B (input = 1, decision = 0) by agreement, B must also decide on 0 since A decides on 0
  • however, this contradicts with validity

Goals:

• Termination: all nodes (that have not failed) eventually decide
  • proof: obvious since no waiting (well-defined rounds)
• Weakened Agreement: all nodes that decide should decide on the same value with probability of (1 - error_prob)
  • including the nodes that would crash after deciding (because its decision can be used)
  • proof: below
• Weakened Validity: (assume input of 0 or 1)
  • if all nodes start with 0: decision should be 0
  • if all nodes start with 1 and no message is lost throughout the execution, decision should be 1
  • otherwise, nodes can decide on anything
  • weakened validity is not sufficient by itself, consider: (requires weakened agreement)
    • A (input = 1, decision = 1), B (input = 1, decision = 1) by weakened validity with no messages dropped
    • drop the last message (defined using real clock time)
      • sender won't know it's dropped, so it must still decide on 1 by indistinguishable
      • receiver must also decide on 1 by agreement
    • inductively keep dropping the messages until all messages are dropped
    • A and B start with 1 and all messages are dropped, and both decide on 1
    • use the previous argument (with un-weakened validity), but flip 0 and 1
    • then, A and B start with 0 and all messages are dropped, and both decide on 1
      • this is not valid by weakened validity

Version 4: Node Crash; Reliable Channels; Asynchronous

Impossible to solve. Provable using the FLP theorem.

Version 5: Byzantine Failures; Reliable Channels; Synchronous

• Termination: all non-faulty nodes eventually decide
  • proof: trivial because $f+1$ rounds
• Agreement: all non-faulty nodes should decide on the same value
  • proof:
    • at least one phase is deciding
    • after that deciding phase, all non-faulty processes have the same $V[i]$ (by Lemma 1)
    • in the following phases, $V[i]$ never changes on the non-faulty processes (by Lemma 2)
• Validity: if all non-faulty nodes have the same initial input: that value is the only possible decision
  • to avoid a trivial algorithm
  • must ignore Byzantine nodes' inputs because they can ignore / "change" their input
  • proof: follows from Lemma 1

Main problem: even if you can detect that some node is faulty, hard to decide which node is faulty

Theorem: If $n \le 3f$, Byzantine consensus problem cannot be solved. (proof outside of CS4231)

// process i, with my_input as input
Consensus(i, my_input) {
    V[1..n] = 0
    V[i] = my_input

    // f+1 phases
    for k = 1; k <= f+1; k++ {
        // all-to-all broadcast
        send V[i] to all processes
        set V[1..n] to the n values received

        if X occurs (> n/2) times in V {
            // majority
            proposal = X
        } else {
            proposal = 0
        }

        // coordinator
        if (k == i) {
            // i am coordinator
            send proposal to all
        } else {
            receive the coordinator's proposal
        }

        // should I listen to coordinator?
        if value y occurs (> n/2 + f) times in V {
            // *overwhelming* majority -> ignore coordinator
            V[i] = y
        } else {
            V[i] = coordinator's proposal
        }
    }

    decide on V[i]
}

Self Stabilising Spanning Tree Algorithm

Specs:

• constructs a spanning tree, rooted at a special node
• can be used to compute shortest path
• runs in the background, constantly updating the variables
  • will eventually be correct (might be wrong an any point in time)
  • can be overwhelmed if there are too many failures

Algorithm:

• each process maintains two variables: parent, dist (distance to root)
• on root node (executed periodically):
  • sets dist = 0; parent = null
• on all other nodes (executed periodically):
  • retrieve dist from neighbours into neighbour_dists
  • set dist = min(neighbour_dist) + 1
  • set parent = neighbour (with smallest dist) - tie break arbitrarily
• ^these steps do not need to be executed atomically (arbitrary interleaving is allowed)