Chapter 1. On Concurrent Programming

Programming with concurrency is hard. Concurrency can make programs faster than sequential ones, but having multiple threads read and update shared variables concurrently and synchronize with one another makes programs more complicated than programs where only one thing happens at a time. Why are concurrent programs more complicated than sequential ones? There are, at least, two reasons:

• The execution of a sequential program is usually deterministic: if you run the program twice with the same input, the same output will be produced. Bugs are reproducible and thus easy to track down, for example by instrumenting the program. They are called Bohrbugs. However, the output of running concurrent programs depends on how the execution of the various threads are interleaved. Some bugs may occur only occasionally and may never occur when the program is instrumented to find them. We call these Heisenbugs---overhead caused by instrumentation leads to timing changes that can make such bugs less likely to cause havoc.
• In a sequential program, each statement and each function can be thought of as happening atomically (indivisibly) because there is no other activity interfering with their execution. Even though a statement or function may be compiled into multiple machine instructions, they are executed back-to-back until completion. Not so with a concurrent program, where other threads may update memory locations while a statement or function is being executed.

This book is intended to help people with understanding and developing concurrent code, which includes programs for distributed systems. In particular, it comes with a tool called Harmony that helps with testing concurrent code. The approach is based on model checking [8]: instead of relying on luck, Harmony will run all possible executions of a particular test program. So, even if a bug is unlikely to occur, if the test can expose the bug it will. Moreover, if the bug is found, the model checker precisely shows how to trigger the bug and will try to minimize the number of steps.

Model checking is not a replacement for formal verification. Formal verification proves that a program is correct. Model checking only verifies that a program is correct for some model. Think of a model as a test program. Because model checking tries every possible execution, the test program needs to be simple---otherwise it may take longer than we care to wait for or it may run out of memory. In particular, the model needs to have a relatively small number of reachable states.

If model checking does not prove a program correct, why is it useful? To answer that question, consider a sorting algorithm. Suppose we create a test program, a model, that tries sorting all lists of up to five numbers chosen from the set { 1, 2, 3, 4, 5 }. Model checking proves that for those particular scenarios the sorting algorithm works: the output is a sorted permutation of the input. In some sense it is an excellent test: it will have considered all corner cases, including lists where all numbers are the same, lists that are already sorted or reversely sorted, etc. If there is a bug in the sorting algorithm, most likely it will be triggered and the model checker will produce a scenario that makes it easy to find the source of the bug.

However, if the model checker does not find any bugs, we do not know for sure that the algorithm works for lists of more than five numbers or for lists that have values other than the numbers 1 through 5. Still, we would expect that the likelihood that there are bugs remaining in the sorting algorithm is small. That said, it would be easy to write a program that sorts all lists of up to five numbers correctly but fails to do so for a list of 6 numbers. (Hint: simply use an if statement.) The model checker would not find a bug in this adversarial program.

While model checking does not in general prove an algorithm correct, it can help with proving an algorithm correct. The reason is that many correctness properties can be proved using invariants: predicates that must hold for every state in the execution of a program. A model checker can find violations of proposed invariants when evaluating a model and provide valuable early feedback to somebody who is trying to construct a proof, even an informal one. We will include examples of such invariants as they often provide excellent insight into why a particular algorithm works.

So, what is Harmony? Harmony is a concurrent programming and specification language with a model checker. It was designed to teach the basics of concurrent and distributed programming, but it is also useful for testing new concurrent algorithms or even sequential and distributed algorithms. Harmony programs are not intended to be "run" like programs in most other programming languages---instead Harmony programs are model checked to test that the program has certain desirable properties and does not suffer from bugs.

The syntax and semantics of the Harmony programming language are similar to that of Python. Python is familiar to many programmers and is easy to learn and use. We will assume that the reader is familiar with the basics of Python programming. We also will assume that the reader understands some basics of machine architecture and how programs are executed. For example, we assume that the reader is familiar with the concepts of CPU, memory, register, stack, and machine instructions.

Harmony is heavily influenced by Leslie Lamport's work on TLA+, TLC, and PlusCal [29, 30], Gerard Holzmann's work on Promela and SPIN [25], and University of Washington's DSLabs system [35]. Some of the examples in this book are derived from those sources. Harmony is designed to have a lower learning curve than those systems, but may not be as powerful for specific problems. When you finish this book and want to learn more, we strongly encourage checking those out. Another excellent resource is Fred Schneider's book ``On Concurrent Programming'' [40]. (This chapter is named after that book.)

The book proceeds as follows:

• Chapter 2 introduces the Harmony programming language, as it provides the language for presenting synchronization problems and solutions.
• Chapter 3 illustrates the problem of concurrent programming through a simple example in which two threads are concurrently incrementing a shared variable.
• Chapter 4 presents the Harmony virtual machine to understand the problem underlying concurrency better.
• Chapter 5 introduces the concept of a critical section and presents various flawed implementations of critical sections to demonstrate that implementing a critical section is not trivial. The chapter also introduces Peterson's Algorithm, an elegant (although not very efficient or practical) solution to implementating a critical section.
• Chapter 6 gives some more details on the Harmony language needed for the rest of the book.
• Chapter 7 talks about how Harmony can be used as a specification language. It introduces how to specify atomic constructs such as locks.
• Chapter 8 looks at various ways in which a lock specification can be implemented.
• Chapter 9 gives an introduction to building concurrent data structures.
• Chapter 10 discusses approaches to testing concurrent code in Harmony.
• Chapter 11 instead goes into how to find a bug in concurrent code using the Harmony output.
• Chapter 12 talks about threads having to wait for certain conditions. As examples, it presents the reader/writer lock problem and the bounded buffer problem.
• Chapter 13 presents condition variables, a synchronization construct to simplify efficient conditional waiting.
• Chapter 14 talks about starvation: the problem that in some synchronization approaches threads may not be able to get access to a resource they need.
• Chapter 15 describes deadlock where a set of threads are indefinitely waiting for one another to release a resource.
• Chapter 16 presents the actor model and message passing as an approach to synchronization.
• Chapter 17 describes barrier synchronization, useful in high-performance computing applications such as parallel simulations.
• Chapter 18 delves deeper into barrier synchronization for situations when there are more threads than the barrier requires.
• Chapter 19 presents a concurrent file system as as larger example of a concurrent program.
• Chapter 20 discusses how to handle interrupts, a problem closely related to---but not the same as---synchronizing threads.
• Chapter 21 introduces non-blocking or wait-free synchronization algorithms, which prevent threads waiting for one another more than a bounded number of steps.
• Chapter 22 presents a problem and a solution to the distributed systems problem of having two threads communicate reliably over an unreliable network.
• Chapter 23 presents a protocol for electing a leader on a ring of processors, where each processor is uniquely identified and only knows its successor on the ring.
• Chapter 24 describes atomic database transactions and the two-phase commit protocol used to implement them.
• Chapter 25 describes state machine replication and the chain replication protocol to support replication.
• Chapter 26 describes an alternative way to write concurrent and distributed specifications in Harmony, using chain replication as an example.
• Chapter 27 presents a protocol for a fault-tolerant replicated object that supports only read and write operations.
• Chapter 28 demonstrates a fault-tolerant distributed consensus algorithm (aka protocol) expressed in Harmony.
• Chapter 29 shows how one can specify and check the well-known Paxos consensus protocol.
• Chapter 30 demonstrates using Harmony to find a (known) bug in the original Needham-Schroeder authentication protocol.

Chapter 2. Hello World!

print "hello world"

Figure 2.1. [code/hello1.hny] Hello World!

Harmony, too, has a Hello World program. Figure 2.1 shows the program and the corresponding output. After installation (see https://harmony.cs.cornell.edu), you can run it as follows from the command line:

But programs can usually have more than one execution and produce multiple different outputs as a result. This is usually as a result of different inputs, but Harmony programs do not have inputs. Instead, Figure 2.2 demonstrates nondetermistic choice in Harmony programs. In this case, the program chooses to print either "hello" or "world". The corresponding DFA captures both possibilities. You can think of the choose operator as enumerating all possible inputs to the program.

Figure 2.3 shows a program that has an infinite number of possible outputs by using a loop with a non-deterministic stopping condition. Harmony usually requires that any program must be able to terminate, so the loop is conditioned on a nondeterministic choice between False and True. The possible outputs consist of zero or more copies of the string "hello world". Note that this single state DFA (where the initial state and the final state happen to be the same) captures an infinite number of possible executions of the program.

print choose { "hello", "world" }

Figure 2.2. [code/hello3.hny] Harmony program with two possible outputs

while choose { False, True }:
    print "hello world"

Figure 2.3. [code/hello4.hny] Harmony program with an infinite number of possible outputs

def p(s): print s p("hello") p("world")	def p(s): print s spawn p("hello") spawn p("world")
(a) [code/hello5.hny]	[code/hello6.hny]

Figure 2.4. Demonstrating Harmony methods and threads

def hello(name):
    print "hello"
    print name

spawn hello("Lesley")
spawn hello("Robbert")

Figure 2.5. [code/hello7.hny] Various interleavings of threads

def hello(name):
    atomically:
        print "hello"
        print name

spawn hello("Lesley")
spawn hello("Robbert")

Figure 2.6. [code/hello8.hny] Making groups of operations atomic reduces interleaving

Harmony is a programming language that borrows much of Python's syntax. Like Python, Harmony is an imperative, dynamically typed programming language. There are also some important differences:

• Harmony purposely only supports basic operator precedence or associativity. Use parentheses liberally to remove ambiguity. For example, write (2 * 3) + 4 instead of 2 * 3 + 4.
• Harmony does not support floating point;
• Python is object-oriented, supporting classes with methods and inheritance; Harmony has objects but does not support classes. Harmony supports pointers, allowing construction of complicated data structures.
• In Python, lists, dictionaries, and sets are (mutable) objects. In Harmony, they are (immutable) values.

const N = 10

def triangle(n) returns result:   # computes the n'th triangle number
    result = 0
    for i in {1..n}:     # for each integer from 1 to n inclusive
        result += i      # add i to result

x = choose {0..N}        # select an x between 0 and N inclusive
assert triangle(x) == ((x * (x + 1)) / 2)

$ harmony --noweb code/triangle.hny

Running this Harmony program (Figure 2.8) will try all possible executions, which includes all possible values for x. The --noweb flag tells Harmony not to automatically pop up the web browser window. The text output from running Harmony is in Markdown format.

The assert statement checks that the output is correct. If the program is correct, Harmony reports the size of the "state graph" (13 states in this case). If not, Harmony also reports what went wrong, typically by displaying a summary of an execution in which something went wrong.

In Harmony, constants have a default specified value, but those can be overridden on the command line using the -c option. Figure 2.9 shows how to test the code for N = 100.

Exercises

• [Ex. 2.1]: Write a Harmony program that uses choose instead of spawn to create the same output DFA as Figure 2.4(b).
• [Ex. 2.2]: Add the line print(x, triangle(x)) to the end of the program and create an output png file. Before you look at it, what do you think it should look like?
• [Ex. 2.3]: See what happens if, instead of initializing result to 0, you initialize it to 1. (You do not need to understand the error report at this time. They will be explained in more detail in Chapter 4.)
• [Ex. 2.4]: Write a Harmony program that computes squares by repeated adding. So, the program should compute the square of x by adding x to an initial value of 0 x times.

Chapter 3. The Problem of Concurrent Programming

shared = True

def f(): assert shared
def g(): shared = False

f()
g()

shared = True

def f(): assert shared
def g(): shared = False

spawn f()
spawn g()

Concurrent programming, aka multithreaded programming, involves multiple threads running in parallel while sharing variables. Figure 3.1 shows two programs. Program (a) is sequential. It sets shared to True, asserts that shared = True and finally sets shared to False. If you run the program through Harmony, it will not find any problems because there is only one execution possible and 1) in that execution the assertion does not fail and 2) the execution terminates. Program (b) is concurrent---it executes methods f() and g() in parallel. If method g() runs and completes before f(), then the assertion in f() will fail when f() runs. This problem is an example of non-determinism: methods f() and g() can run in either order. In one order, the assertion fails, while in the other it does not. But since Harmony checks all possible executions, it will always find the problematic one.

Figure 3.2 shows the output of running Figure 3.1(b) through Harmony. Underneath the line, there is a summary of what happened in one of the executions. First, the initialization thread (T0) runs and sets the global variable shared to True. Then, the thread running g() runs to completion, setting shared to False. Finally, the thread running f() runs, and the assertion fails.

Figure 3.3 presents a more subtle example that illustrates non-atomicity. The program initializes two shared variables: an integer count and an array done with two booleans. The program then spawns two threads. The first runs incrementer(0); the second runs incrementer(1).

Method incrementer takes a parameter called self. It increments count and sets done[self] to True. It then waits until the other thread is done. (await c is shorthand for while not c: pass.) After that, method incrementer verifies that the value of count equals 2.

Note that although the threads are spawned one at a time, they will execute concurrently. It is, for example, quite possible that incrementer(1) finishes before incrementer(0) even gets going. And because Harmony tries every possible execution, it will consider that particular execution as well. What would the value of count be at the end of that execution?

count = 0
done = [ False, False ]

def incrementer(self):
    count = count + 1
    done[self] = True
    await done[1 - self]
    assert count == 2

spawn incrementer(0)
spawn incrementer(1)

• Before you run the program, what do you think will happen? Is the program correct in that count will always end up being 2? (You may assume that load and store instructions of the underlying virtual machine architecture are atomic (indivisible)---in fact they are.)

count = 0
done = [ False, False ]

def incrementer(self):
    var register = count    # load shared variable count into a private register
    register += 1           # increment the register
    count = register        # store its value into variable count
    done[self] = True
    await done[1 - self]
    assert count == 2

spawn incrementer(0)
spawn incrementer(1)

What is going on is that the Harmony program is compiled to machine instructions, and it is the machine instructions that are executed by the underlying Harmony machine. The details of this appear in Chapter 4, but suffice it to say that the machine has instructions that load values from memory and store values into memory. Importantly, it does not have instructions to atomically increment or decrement values in shared memory locations. So, to increment a value in memory, the machine must do at least three machine instructions. Figure 3.4 illustrates this. (The var statement declares a new local variable register.) Conceptually, the machine

1. loads the value of count from its memory location into a register;
2. adds 1 to the register;
3. stores the new value into the memory location of count.

If the threads run one at a time to completion, then count will be incremented twice and ends up being 2. However, the following is also a possible interleaving of incrementer(0) and incrementer(1):

1. incrementer(1) loads the value of count, which is 0;
2. incrementer(0) loads the value of count, which is still 0;
3. incrementer(0) adds 1 to the value that it loaded (0), and stores 1 into count;
4. incrementer(1) adds 1 to the value that it loaded (0), and stores 1 into count;
5. incrementer(1) sets done[1] to True;
6. incrementer(0) sets done[0] to True.

count = 0

finally count == 2

def incrementer():
    count = count + 1

spawn incrementer()
spawn incrementer()

The exercises below have you try the same thing (having threads concurrently increment an integer variable) in Python. As you will see, the bug is not easily triggered when you run a Python version of the program. But in Harmony Murphy's Law applies: if something can go wrong, it will. Usually that is not a good thing, but in Harmony it is. It allows you to find bugs in your concurrent programs much more easily than using a conventional programming environment.

Exercises

• [Ex. 3.1]: Harmony programs can usually be easily translated into Python by hand using minor modifications. For example, Figure 3.7 is a Python version of Figure 3.3.
  1. Run Figure 3.7 using Python. Does the assertion fail?
  2. Using a script, run Figure 3.7 1000 times. For example, if you are using the bash shell (in Linux or Mac OS X, say), you can do the following:

     for i in {1..1000}
     do
         python Up.py
     done
     

     If you're using Windows, the following batch script does the trick:

     FOR /L %%i IN (1, 1, 1000) DO python Up.py
     PAUSE
     

     How many times does the assertion fail (if any)?
• [Ex. 3.2]: Figure 3.8 is a version of Figure 3.7 that has each incrementer thread increment count N times. Run Figure 3.8 10 times (using Python). Report how many times the assertion fails and what the value of count was for each of the failed runs. Also experiment with lower values of N. How large does N need to be for assertions to fail? (Try powers of 10 for N.)
• [Ex. 3.3]: Can you think of a fix to Figure 3.3? Try one or two different fixes and run them through Harmony. Do not worry about having to come up with a correct fix at this time---the important thing is to develop an understanding of concurrency. (Also, you do not get to use the atomically keyword or a lock, yet.)

import threading

count = 0
done = [ False, False ]

def incrementer(self):
    global count
    count = count + 1
    done[self] = True
    while not done[1 - self]:
        pass
    assert count == 2

threading.Thread(target=incrementer, args=(0,)).start()
threading.Thread(target=incrementer, args=(1,)).start()

import threading

N = 1000000
count = 0
done = [ False, False ]

def incrementer(self):
    global count
    for i in range(N):
        count = count + 1
    done[self] = True
    while not done[1 - self]:
        pass
    assert count == 2*N, count

threading.Thread(target=incrementer, args=(0,)).start()
threading.Thread(target=incrementer, args=(1,)).start()

Chapter 4. The Harmony Virtual Machine

Harmony programs are compiled to Harmony bytecode (a list of machine instructions for a virtual machine), which in turn is executed by the Harmony virtual machine (HVM). The Harmony compiler, harmony, places the bytecode for file x.hny in file x.hvm. The model checker (called Charm) executes the code in x.hvm and places its output in a file called x.hco. From the x.hco file, harmony creates a detailed human-readable output file in x.hvb and an interactive HTML file called x.htm. The x.htm file is automatically opened in your default web browser unless you specify the --noweb flag to harmony.

To understand the problem of concurrent computing, it is important to have a basic understanding of machine instructions, and in our case those of the HVM.

   0 Frame __init__ ()
code/Up.hny:1 count = 0
   1 Push 0
   2 Store count
code/Up.hny:2 done = [ False, False ]
   3 Push [False, False]
   4 Store done
code/Up.hny:4 def incrementer(self):
   5 Jump 35
   6 Frame incrementer self
code/Up.hny:5     count = count + 1
   7 Load count
   8 Push 1
   9 2-ary +
   10 Store count

harmony -a Up.hny

• Safety violation: This means something went wrong with at least one of the executions of the program that it tried. This can include a failing assertion, behavior violations, divide by zero, using an uninitialized or non-existent variable, dividing a set by an integer, and so on. Harmony will print a trace of the bad execution that it found (prefering short traces over longer ones).
• Non-terminating State: Harmony found one or more states from which there does not exist an execution such that all threads terminate. Harmony will not only print the non-terminating state with a corresponding trace, but also the list of threads at that state, along with their program counters.
• Behavior Violation: The program can terminate in a state not allowed by the behavioral specification (Chapter 10).
• Active Busy Waiting: There are states in which some thread cannot make progress without the help of another thread, but does not block (Chapter 12).
• Data Race: There are states in which two or more threads concurrently access a shared variable, at least one of which is a store operation (Chapter 8).

Figure 4.2. The HTML output of running Harmony on Figure 3.3

In the top right, the HTML file contains the reported issue in red. Underneath it, a table shows the five turns in the execution, with the current turn highlighted in yellow. Each turn displays a list of blocks for each instruction executed in that turn. If it is white, the instruction has not yet been executed. We call this the timeline. You can click on such a block to see the state of the Harmony virtual machine just after executing the corresponding instruction. If a thread has finished its turn, there is also information on the status of that thread. For example, at the end of turn 2, incrementer(1) is about to store the value 1 in variable count, but at that point is preempted by incrementer(0). The table also lists the program counter of the thread at each turn, the values of the shared variables, and any values the thread may have printed (none in this case). Underneath the table it shows the line of Harmony source code that is being executed in blue (with the specific part of the line that is being evaluated in green), and the HVM instruction that is about to be executed in green (along with an explanation in parentheses).

The bottom left shows the bytecode of the program being executed. The instruction that is being or about to be executed, if any, is highlighted in red. If you hover the mouse over a machine instruction, it provides a brief explanation of what the instruction does.

The bottom right contains a table with the state of each thread. The thread that is executing is highlighted in yellow. Status information for a thread can include:

• runnable: the thread is runnable but not currently running. In Harmony, threads are interleaved and so at most one thread is actually running;
• running: the thread is currently executing instructions;
• terminated: the thread has completed all its instructions;
• failed: the thread has encountered an error, such as violating an assertion or divide by zero;
• blocked: the thread cannot make progress until another thread has updated the shared state. For example, this occurs when one of the implementers is waiting for the other to set its done flag;
• atomic: the thread is in atomic mode, not allowing other threads to be scheduled. This is, for example, the case when an assertion is being checked;
• read-only: the thread is in read-only mode, not able to modify shared state. Assertions can execute arbitrary code including methods, but they are not allowed to modify the shared state.

When you load the HTML file, it shows the final state. As mentioned above, you can go to any point in the execution by clicking on one of the blocks in the timeline. When you do so, the current turn and thread will be highlighted in yellow. There are also various handy keyboard shortcuts:

Right arrow:	go to the next instruction;
Left arrow:	go back to the previous instruction;
Down arrow:	finish executing the current method;
Up arrow:	go back to the beginning of the current method;
Enter (aka Return):	go to the next line of Harmony code;
0:	go to the initial state.

If you want to see an animation of the entire execution, one instruction at a time, you can first hit 0 and then hold down the right arrow. If you want to see it one line of Harmony code at a time, hold down the enter (aka return) key instead. If you hold down the down arrow key, the movie will go by very quickly.

Exercises

• [Ex. 4.1]: Figure 4.3 shows an attempt at trying to fix the code of Figure 3.3. Run it through Harmony and see what happens. Based on the error output, describe in English what is wrong with the code by describing, in broad steps, how running the program can get into a bad state.
• [Ex. 4.2]: What if we moved Line 5 of Figure 4.3 to after the if statement (between Lines 7 and 8)? Do you think that would work? Run it through Harmony and describe either why it works or why it does not work.

count = 0

entered = done = [ False, False ]

def incrementer(self):
    entered[self] = True
    if entered[1 - self]:        # if the other thread has already started
        await done[1 - self]     # wait until it is done
    count = count + 1
    done[self] = True
    await done[1 - self]
    assert count == 2

spawn incrementer(0)
spawn incrementer(1)

Chapter 5. Critical Sections

def thread():
    while True:
        # Critical section is here
        pass
    
spawn thread()
spawn thread()

Critical sections are useful when accessing a shared data structure, particularly when that access requires multiple underlying machine instructions. A counter is a very simple example of a data structure (it is a list of bits), but---as we have seen---incrementing it requires multiple instructions. A more involved one would be accessing a binary tree. Adding a node to a binary tree, or re-balancing a tree, often requires multiple operations. Maintaining "consistency" is certainly much easier if during this time no other thread also tries to access the binary tree. Typically, you want some property of the data structure to hold at the beginning and at the end of the critical section, but in the middle the property may be temporarily invalid---this is not a problem as critical sections guarantee that no other thread will be able to see the intermediate state of the data structure. An implementation of a data structure that can be safely accessed by multiple threads and is free of race conditions is called thread-safe.

A critical section is often modeled as threads in an infinite loop entering and exiting the critical section. Figure 5.1 shows the Harmony code. We need to ensure is that there can never be two or more threads in the critical section. This property is called mutual exclusion. Mutual exclusion by itself is easy to ensure. For example, we could insert the following code to enter the critical section:

# number of threads in the critical section
in_cs = 0
invariant in_cs in { 0, 1 }

def thread():
    while choose { False, True }:
        # Enter critical section
        atomically in_cs += 1

        # Critical section is here
        pass

        # Exit critical section
        atomically in_cs -= 1
    
spawn thread()
spawn thread()

In order to detect violations of progress, and other liveness problems in algorithms in general, Harmony requires that every execution must be able to reach a state in which all threads have terminated. Clearly, even if mutual exclusion holds in Figure 5.1, the spawned threads never terminate.

We will instead model threads in critical sections using the framework in Figure 5.2: a thread can choose to enter a critical section more than once, but it can also choose to terminate, even without entering the critical section ever. (Recall that Harmony will try every possible execution, and so it will evaluate both choices.) As it turns out, there is an advantage to doing it this way: we also test if a thread can enter when there is no other thread trying to enter the critical section. As we will see below, this is not always obvious.

This code specifies that at most one thread can be executing in the critical section by using a counter in_cs that a thread atomically increments when entering the critical section and atomically decrements when it leaves. The code specifies the invariant that in_cs must be either 0 or 1. You can think of this as the type of in_cs.

We will now consider various approaches toward implementing this specification.

in_cs = 0
invariant in_cs in { 0, 1 }

lockTaken = False

def thread(self):
    while choose({ False, True }):
        # Enter critical section
        await not lockTaken
        lockTaken = True

        atomically in_cs += 1
        # Critical section
        atomically in_cs -= 1

        # Leave critical section
        lockTaken = False
    
spawn thread(0)
spawn thread(1)

in_cs = 0
invariant in_cs in { 0, 1 }

flags = [ False, False ]

def thread(self):
    while choose({ False, True }):
        # Enter critical section
        flags[self] = True
        await not flags[1 - self]

        atomically in_cs += 1
        # Critical section
        atomically in_cs -= 1

        # Leave critical section
        flags[self] = False

spawn thread(0)
spawn thread(1)

in_cs = 0
invariant in_cs in { 0, 1 }

turn = 0

def thread(self):
    while choose({ False, True }):
        # Enter critical section
        turn = 1 - self
        await turn == self

        atomically in_cs += 1
        # Critical section
        atomically in_cs -= 1

        # Leave critical section

spawn thread(0)
spawn thread(1)

You may already have heard of the concept of a lock and have realized that it could be used to implement a critical section. The idea is that the lock is like a baton that at most one thread can own (or hold) at a time. A thread that wants to enter the critical section at a time must obtain the lock first and release it upon exiting the critical section. Using a lock is a good thought, but how does one implement one? Figure 5.3 presents an attempt at mutual exclusion based on a naïve (and, as it turns out, incorrect) implementation of a lock. Initially the lock is not owned, indicated by lockTaken being False. To enter the critical section, a thread waits until lockTaken is False and then sets it to True to indicate that the lock has been taken. The thread then executes the critical section. Finally, the thread releases the lock by setting lockTaken back to False.

Unfortunately, if we run the program through Harmony, we find that the assertion fails. Figure 5.3 also shows the Harmony output. thread(1) finds that the lock is available, but just before it stores True in lockTaken, thread(0) gets to run. Because lockTaken is still False, it too believes it can acquire the lock, and stores True in lockTaken and moves on to the critical section. Finally, thread(1) moves on, also stores True into lockTaken and also moves into the critical section. The lockTaken variable suffers from the same sort of race condition as the count variable in Figure 3.3: testing and setting the lock consists of several instructions. It is thus possible for both threads to believe the lock is available and to obtain the lock at the same time.

Preventing multiple threads from updating the same variable, Figure 5.4 presents a solution based on each thread having a flag indicating that it is trying to enter the critical section. A thread can write its own flag and read the flag of its peer. After setting its flag, the thread waits until the other thread ( 1 - self ) is not trying to enter the critical section. If we run this program, the assertion does not fail. In fact, this solution does prevent both threads being in the critical section at the same time.

To see why, first note the following invariant: if thread i is in the critical section, then flags[i] = True. Without loss of generality, suppose that thread 0 sets flags[0] at time t₀ . Thread 0 can only reach the critical section if at some time t₁ , t₁ > t₀ , it finds that flags[1] = False. Because of the invariant, flags[1] = False implies that thread 1 is not in the critical section at time t₁ . Let t₂ be the time at which thread 0 sets flags[0] to False. Thread 0 is in the critical section sometime between t₁ and t₂ . It is easy to see that thread 1 cannot enter the critical section between t₁ and t₂ , because flags[1] = False at time t₁ . To reach the critical section between t₁ and t₂ , it would first have to set flags[1] to True and then wait until flags[0] = False. But that does not happen until time t₂ .

However, if you run the program through Harmony, it turns out the solution does have a problem: if both try to enter the critical section at the same time, they may end up waiting for one another indefinitely. (This is a form of deadlock, which will be discussed in Chapter 15.) Thus the solution violates progress.

The final naïve solution that we propose is based on a variable called turn. Each thread politely lets the other thread have a turn first. When turn = i, thread i can enter the critical section, while thread 1 - i has to wait. An invariant of this solution is that while thread i is in the critical section, turn = i. Since turn cannot be 0 and 1 at the same time, mutual exclusion is satisfied. The solution also has the nice property that the thread that has been waiting the longest to enter the critical section can go next.

Run the program through Harmony. It turns out that this solution also violates progress, albeit for a different reason: if thread i terminates instead of entering the critical section, thread 1 - i , politely, ends up waiting indefinitely for its turn. Too bad, because it would have been a great solution if both threads try to enter the critical section ad infinitum.

in_cs = 0
invariant in_cs in { 0, 1 }

sequential flags, turn
flags = [ False, False ]
turn = choose({0, 1})

def thread(self):
    while choose({ False, True }):
        # Enter critical section
        flags[self] = True
        turn = 1 - self
        await (not flags[1 - self]) or (turn == self)

        atomically in_cs += 1
        # Critical section
        atomically in_cs -= 1

        # Leave critical section
        flags[self] = False

spawn thread(0)
spawn thread(1)

A thread first indicates its interest in entering the critical section by setting its flag. It then politely gives way to the other thread should it also want to enter the critical section---if both do so at the same time one will win because writes to memory in Harmony are atomic. The thread continues to be polite, waiting until either the other thread is nowhere near the critical section (flag[1 - self] = False) or has given way (turn = self).

Running the algorithm with Harmony shows that it satisfies both mutual exclusion and progress. Why does it work? To answer that we need to prove the correctness of Peterson's algorithm. Unfortunately, proving the correctness of Peterson's algorithm is not trivial. You can find an informal proof in Appendix H.

Peterson's algorithm implements a critical section, but it is not efficient, especially if generalized to more than two threads. Worse, Peterson relies on load and store operations to be executed atomically, but this may not be the case. There are a variety of possible reasons for this.

• Variables may have more bits than the processor's data bus. For example, variables may have 32 bits, but the data bus may only have 16 bits. Thus to store or load a variable takes two 16-bit operations each. Take, for example, a variable that has value 0xFFFFFFFF, and consider a concurrent load and store operation on the variable. The store operation wants to clear the variable, but because it takes two store operations on the bus, the load operation may return either 0xFFFF0000 or 0x0000FFFF, a value that the variable never was supposed to have. This is the case even if the processor supports a 32-bit load or store machine instruction: on the data bus it is still two operations. And even with a 32-bit data bus, the problem could arise if the variable is not aligned on a 4-byte word boundary.
• Modern processors sometimes re-orders load and store operations (out-of-order execution) for improved performance. On a sequential processor, the re-ordering is not a problem as the processor only re-orders operations on independent memory locations. However, as Exercise 5.4 shows, Peterson's algorithm breaks down if such seemingly independent operations are re-ordered. Some memory caches can also cause non-atomic behavior of memory when shared among multiple cores.
• Even compilers, in their code generation, may make optimizations that can reorder operations, or even eliminate operations, on variables. For example, a compiler may decide that it is unnecessary to read the same variable more than once, because how could it possibly change if there are no store operations in between?

Exercises

• [Ex. 5.1]: Run Figure 5.2 using Harmony. As there is no protection of the critical section, mutual exclusion is violated, the assertion should fail, and a trace should be reported. Now insert await False just before entering the critical section in Figure 5.2 and run Harmony again. Mutual exclusion is guaranteed but progress is violated. Harmony should print a trace to a state from which a terminating state cannot be reached. Describe in English the difference in the failure reports before and after inserting the code.
• [Ex. 5.2]: See if you can come up with some different approaches that satisfy both mutual exclusion and progress. Try them with Harmony and see if they work or not. If they don't, try to understand why.
• [Ex. 5.3]: Figure 5.7 presents another solution to the mutual exclusion problem. It is similar to the one in Figure 5.4, but has a thread back out and try again if it finds that the other thread is either trying to enter the critical section or already has. Compare this algorithm with Peterson's. Why does Harmony complain about active busy waiting? Does the algorithm guarantee that at least one thread can enter the critical section?
• [Ex. 5.4]: Consider if the first two assignments in Peterson's algorithm (setting flags[self] to True and turn to 1 - self can be reversed. After all, they are different variables assigned independent values---in a sequential program one could surely swap the two assignments. Then run the program in Figure 5.6 after reversing the two assignments and describe in English what happens.
• [Ex. 5.5]: Bonus question: Can you generalize Peterson's algorithm to more than two threads?
• [Ex. 5.6]: Bonus question: Implement Dekker's Algorithm, Dijkstra's Algorthm [15], Eisenstein and McGuire's Algorithm, Szymański's Algorithm, or Lamport's Bakery Algorithm. Note that the last one uses unbounded state, so you should modify the threads so they only try to enter the critical section a bounded number of times.

in_cs = 0
invariant in_cs in { 0, 1 }

sequential flags
flags = [ False, False ]

def thread(self):
    while choose({ False, True }):
        # Enter critical section
        flags[self] = True
        while flags[1 - self]:
            flags[self] = False
            flags[self] = True
        
        atomically in_cs += 1
        # Critical section
        atomically in_cs -= 1

        # Leave critical section
        flags[self] = False

spawn thread(0)
spawn thread(1)

Chapter 6. Harmony Methods and Pointers

A method m with argument a is invoked in its most basic form as follows (assigning the result to r).

You may note that all this looks familiar. Indeed, the syntax is the same as that for dictionaries and lists (see Chapter 4). Dictionaries, lists, and methods all map Harmony values to Harmony values, and their syntax is indistinguishable. If f is a method, list, or dictionary, and x is some Harmony value, then f x, f(x), and f[x] are all the same expression in Harmony.

def Peterson_enter(pm, pid):
    pm->flags[pid] = True
    pm->turn = 1 - pid
    await (not pm->flags[1 - pid]) or (pm->turn == pid)

def Peterson_exit(pm, pid):
    pm->flags[pid] = False

def Peterson_mutex() returns result:
    result = { .turn: choose({0, 1}), .flags: [ False, False ] }

#### The code above can go into its own Harmony module ####

in_cs = 0
invariant in_cs in { 0, 1 }

sequential mutex
mutex = Peterson_mutex()

def thread(self):
    while choose({ False, True }):
        Peterson_enter(?mutex, self)

        atomically in_cs += 1
        # Critical section
        atomically in_cs -= 1

        Peterson_exit(?mutex, self)

spawn thread(0)
spawn thread(1)

Harmony is not an object-oriented language like Python is. In Python, you can pass an object reference to a method, and that method can then update the object. In Harmony, it is also sometimes convenient to have a method update a shared variable specified as an argument. For this, as mentioned in Chapter 4, each shared variable has an address, itself a Harmony value. If x is a shared variable, then the expression ?x is the address of x. If a variable contains an address, we call that variable a pointer. If p is a pointer to a shared variable, then the expression !p is the value of the shared variable. In particular, !?x = x. This is similar to how C pointers work (*&x = x).

Often, pointers point to dictionaries, and so if p is such a pointer, then (!p).field would evaluate to the specified field in the dictionary. Note that the parentheses in this expression are needed, as !p.field would wrongly evaluate !(p.field). (!p).field is such a common expression that, like C, Harmony supports the shorthand p->field (written p->field), which greatly improves readability.

Figure 6.1 again shows Peterson's algorithm, but this time with methods defined to enter and exit the critical section. The name mutex is often used to denote a variable or value that is used for mutual exclusion. Peterson_mutex is a method that returns a "mutex," which, in this case, is a dictionary that contains Peterson's Algorithm's shared memory state: a turn variable and two flags. Both methods Peterson_enter and Peterson_exit take two arguments: a pointer to a mutex and the thread identifier (0 or 1). pm->turn is the value of the .turn key in the dictionary that pm points to.

You can put the first three methods in its own Harmony source file and include it using the Harmony import statement. This would make the code usable by multiple applications.

Finally, methods can have local variables. Method variables are either mutable (writable) or immutable (read-only). The arguments to a method and the bound variable (or variables) within a for statement are immutable; the result variable is mutable. Using the var statement, new mutable local variables can be declared. For example, var x = 3 declares a new mutable local variable x. The let statement allows declaring new immutable local variables. For example: let x = 3: y += x adds 3 to the global variable y.

current = [ [1, 2, 3], [], [] ]

while current[2] != [1, 2, 3]:
    let moves = { (s, d) for s in {0..2} for d in {0..2}
        where current[s] != []
        where (current[d] == []) or (current[s][0] < current[d][0]) }
    let (src,dst) = choose moves:
        print str(src) + " -> " + str(dst)
        current[dst] = [current[src][0],] + current[dst]
        del current[src][0]

The program also contains an example of set comprehension (Lines 4 to 6). This is similar to set comprehension in Python, except that Harmony uses the keyword where instead of if (simplifying parsing).

const FIFO = False

def CLOCK(n) returns result:
    result = { .entries: [None,] * n, .recent: {}, .hand: 0, .misses: 0 }

def ref(clock, x):
    if x not in clock->entries:
        while clock->entries[clock->hand] in clock->recent:
            clock->recent -= {clock->entries[clock->hand]}
            clock->hand = (clock->hand + 1) % len(clock->entries)
        clock->entries[clock->hand] = x
        clock->hand = (clock->hand + 1) % len(clock->entries)
        clock->misses += 1
    if not FIFO:
        clock->recent |= {x}

clock3, clock4, refs = CLOCK(3), CLOCK(4), []

const VALUES = { 1..5 }

var last = {}
for i in {1..100}:
    let x = i if i < 5 else choose(VALUES - last):
        refs = refs + [x,]
        ref(?clock3, x); ref(?clock4, x)
        assert(clock4.misses <= clock3.misses)
        last = {x}

In this program, CLOCK maintains the state of a cache (in practice, typically pages in memory). The set recent maintains whether an access to the cache for a particular reference was recent or not. (It is not used if FIFO is True.) The integer misses maintains the number of cache misses. Method ref(ck, x) is invoked when x is referenced and checked against the cache ck.

The program declares two caches: one with 3 entries (clock3) and one with 4 entries (clock4). The interesting part is in the last block of code. It runs every sequence of references of up to 100 entries, using references in the range 1 through 5. Note that all the constants chosen in this program (3, 4, 5, 100) are the result of some experimentation---you can try other ones. To reduce the search space, the first four references are pinned to 1, 2, 3, and 4. Further reducing the search space, the program never repeats the same reference twice in a row (using the local variable last).

The two things to note here is the use of the choose expression and the assert statement. Using choose, we are able to express searching through all possible strings of references without a complicated nested iteration. Using assert, we are able to express the anomaly we are looking for.

In case you want to check if you get the right results. For FIFO, the program finds the same anomaly that Bélády et al. found: 1 2 3 4 1 2 5 1 2 3 4 5. For the CLOCK algorithm the program actually finds a shorter reference string: 1 2 3 4 2 1 2 5 1 2.

Exercises

• [Ex. 6.1]: (This is just for fun or exercise as it is not a concurrent programming problem.) Implement a Harmony program that finds solutions to the "cabbage, goat, and wolf" problem. In this problem, a person accompanied by these three items has to cross a stream in a small boat, but can only take one item at a time. So, the person has to cross back and forth several times, leaving two items on one or the other shore by themselves. Unfortunately, if left to themselves, the goat would eat the cabbage and the wolf would eat the goat. What crossings does the person need to make in order not to lose any items?

Chapter 7. Specifying a Lock

def Lock() returns result:
    result = False

def acquire(lk):
    atomically when not !lk:
        !lk = True

def release(lk):
    assert !lk
    atomically !lk = False

from synch import Lock, acquire, release

const NTHREADS = 5

thelock = Lock()

def thread():
    acquire(?thelock)
    pass             # critical section is here
    release(?thelock)

for i in {1..NTHREADS}:
    spawn thread()

from synch import Lock, acquire, release

sequential done

count = 0
countlock = Lock()
done = [ False, False ]

def thread(self):
    acquire(?countlock)
    count = count + 1
    release(?countlock)
    done[self] = True
    await done[1 - self]
    assert count == 2

spawn thread(0)
spawn thread(1)

The code in Figure 7.1 is also in Harmony's synch module. Figure 7.2 shows how locks may be used to implement a critical section. Figure 7.3 gives an example of how locks may be used to fix the program of Figure 3.3.

Note that the code of Figure 7.1 is executable in Harmony. However, the atomically keyword is not available in general programming languages and should not be used for implementation. Peterson's algorithm is an implementation of a lock, although only for two processes. In the following chapters, we will look at more general ways of implementing locks using atomic constructions that are usually available in the underlying hardware.

In Harmony, any statement can be preceded by the atomically keyword. It means that the statement as a whole is to be executed atomically. (Harmony can accomplish this by controlling how threads are scheduled.) The atomically keyword can be used to specify the behavior of methods such as acquire and release. But an actual executable program should not use the atomically keyword because---on a normal machine---it cannot be directly compiled into machine code. If we want to make the program executable on hardware, we have to show how Lock, acquire, and release are implemented, not just how they are specified. Chapter 8 presents such implementations.

The code in Figure 7.1 also uses the Harmony when statement. A when statement waits until a time in which condition holds (not necessarily the first time) and then executes the statement block. The "await condition" statement is the same as "when condition: pass". Combined with the atomically keyword, the entire statement is executed atomically at a time that the condition holds.

It is important to appreciate the difference between an atomic section (the statements executed within an atomic block of statements) and a critical section (protected by a lock of some sort). The former ensures that while the statements in the block are executing no other thread can execute. The latter allows multiple threads to run concurrently, just not within the critical section. The former is rarely available to a programmer (e.g., none of Python, C, or Java support it), while the latter is very common.

Atomic statements are not intended to replace locks or other synchonization primitives. When implementing synchronization solutions you should not directly use atomic statements but use the synchronization primitives that are available to you. But if you want to specify a synchronization primitive, then use atomically by all means. You can also use atomic statements in your test code. In fact, assert statements are executed atomically and specify that some condition must hold in every execution of the program. Note two important differences with other programming languages: in most programming languages, an assert statement only checks the current execution, and the condition is not evaluated atomically.

Chapter 8. Lock Implementations

const N = 5

in_cs = 0
invariant in_cs in { 0, 1 }

shared = False
private = [ True, ] * N
invariant [x for x in [shared,] + private where not x] == [False,]

def swap(s, p):
    atomically !p, !s = !s, !p

def thread(self):
    while choose({ False, True }):
        # Enter critical section
        while private[self]:
            swap(?shared, ?private[self])

        atomically in_cs += 1
        assert not private[self]
        atomically in_cs -= 1

        # Leave critical section
        swap(?shared, ?private[self])

for i in {0..N-1}:
    spawn thread(i)

When there is a bug in a program because some code omitted obtaining a lock before accessing a shared data structure, that is known as a data race. More precisely, a data race happens when there is a state in which multiple threads are trying to access the same variable, at least one of those accesses updates the variable, and at least one of those accesses is non-atomic. In many environments, including C and Java programs, the behavior of concurrent load and store operations have tricky or even undefined semantics. One should therefore avoid data races, which is why Harmony reports them even though Harmony has sequentially consistent memory.

Because atomic operations cannot overlap, data races can be avoided by making operations atomic. In Harmony, this can be done in two ways. First, using the sequential statement, you can specify that concurrent load and store operations to the specified variables are considered atomic. Second, you can make accesses atomic using the atomically keyword. However, these are Harmony specification constructs. In a practical implementation, a programmer will often use one or more locks to prevent overlapping access to shared variables.

To implement locks, multi-core processors provide so-called atomic instructions: special machine instructions that can read memory and then write it in an indivisible step. While the HVM does not have any specific built-in atomic instructions besides loading and storing variables, it does have support for executing multiple instructions atomically. We can use the atomically keyword to specify a wide variety of atomic operations. For example, we could fix the program in Figure 3.3 by constructing an atomic increment operation for a counter, like so:

def atomic_inc(ptr):
    atomically !ptr += 1
    
count = 0
atomic_inc(?count)

Figure 8.1 goes on to implement mutual exclusion for a set of N threads. The approach is called spinlock, because each thread is "spinning" (executing a tight loop) until it can acquire the lock. The program uses N + 1 boolean variables. Variable shared (the lock) is initialized to False while private[i] for each thread i is initialized to True.

An important invariant, ₁ , of the program is that at any time exactly one of these variables is False (expressed in Harmony in Line 8). Another invariant, ₂(i) , is that if thread i is in the critical section, then private[i] = False (expressed in Line 20). Between the two (i.e., ₁ ∧∀i : ₂(i) ), it is clear that only one thread can be in the critical section at the same time. To see that invariant ₁ is maintained, we can use induction. Note that !p = True upon entry of swap (because of the condition on the while loop that the swap method is invoked in). There are two cases:

1. !s is False upon entry to swap. Then upon exit !p = False and !s = True, maintaining the invariant.
2. !s is True upon entry to swap. Then upon exit nothing has changed, maintaining the invariant.

Harmony can check these invariants. ₁(i) is specified in Line 8, while ₂(i) is expressed in Line 20 using an assert statement in the critical section. The expression in Line 8 counts the number of False values in the list consisting of the shared variable and the private variables and checks that the result is 1.

def test_and_set(s) returns oldvalue:
    atomically:
        oldvalue = !s
        !s = True

def atomic_store(p, v):
    atomically !p = v

def Lock() returns initvalue:
    initvalue = False

def acquire(lk):
    while test_and_set(lk):
        pass

def release(lk):
    atomic_store(lk, False)

const MAX_THREADS = 8

def fetch_and_increment(p) returns oldvalue:
    atomically:
        oldvalue = !p
        !p = (!p + 1) % MAX_THREADS

def atomic_load(p) returns value:
    atomically value = !p

def Lock():
    result = { .counter: 0, .dispenser: 0 }

def acquire(lk):
    let my_ticket = fetch_and_increment(?lk->dispenser):
        while atomic_load(?lk->counter) != my_ticket:
            pass

def release(lk):
    fetch_and_increment(?lk->counter)

import lock

const N = 5

in_cs = 0
invariant in_cs in { 0, 1 }

thelock = lock.Lock()

def thread():
    while choose({ False, True }):
        lock.acquire(?thelock)

        atomically in_cs += 1
        # Critical section
        atomically in_cs -= 1

        lock.release(?thelock)

for i in {1..N}:
    spawn thread()

A more common solution to implementing a lock uses an atomic test-and-set instruction provided by most CPUs. Figure 8.2 shows the specification of test_and_set and the implementation of a spinlock based on it. Method test_and_set involves only a single variable: the method sets the variable to True and returns the old value of the variable. The variable represents the lock: if free, the value is False; if taken, the value is True. Method Lock() returns the initial value of a lock, which is False. Method acquire(lk) invokes test_and_set(lk) until it returns False, meaning that the lock variable pointed to by lk was available (but now taken). Method release(lk) releases the lock by setting !lk back to False. The lock is cleared in an atomic statement to prevent a data race. (CPUs typically provide an atomic store operation.) Besides this solution being more efficient than Figure 8.1, the solution is general for any number of threads.

You can test the spinlock with the program in Figure 8.4 using the command harmony -m synch=lock_tas code/lock_test1.hny. The -m flag tells Harmony to use the lock_tas.hny file for the synch module rather than the standard synch module (which contains only a specification of the lock methods). The test program has a collection of threads repeatedly enter a critical section and testing that there is at most one thread in the critical section at any time.

The spinlock implementation suffers potentially from starvation: an unlucky thread may never be able to get the lock while other threads successfully acquire the lock one after another. It could even happen with just two threads: one thread might successfully acquire the lock repeatedly in a loop, while another thread is never lucky enough to acquire the lock in between. A ticket lock (Figure 8.3 is an implementation of a lock that prevents starvation using an atomic fetch-and-increment operator. It is inspired by European bakeries. A European bakery often has a clearly displayed counter (usually just two digits) and a ticket dispenser. Tickets are numbered 0 through 99 and repeat over and over again (in the case of a two digit counter). When a customer walks into the bakery, they draw a number from the dispenser and wait until their number comes up. Every time a customer has been helped, the counter is incremented. (Note that this only works if there can be no more than 100 customers in the bakery at a time.)

Figure 8.3 uses two variables for a ticket lock, counter and dispenser. When a thread tries to acquire the lock, it first fetches the current value of the dispenser variable (i.e., the ticket) and increments the variable modulo MAX_THREADS, all in one atomic operation. In practice, MAX_THREADS would be a number like 2³² or 2⁶⁴ , but since the Harmony model checker checks every possible state, limiting MAX_THREADS to a small number helps to keep the time and memory needed to model check a Harmony program within reason. Moreover, it is easier to check that it fails when you run it with more than MAX_THREADS threads. After obtaining the ticket, the thread waits until its ticket value equals the counter. Note that loading the counter must also be done atomically in order to avoid a data race. The release the lock, it suffices to atomically increment the counter. You can test the implementation using the command harmony -m synch=lock_ticket code/lock_test1.hny. To see it fail, try harmony -c N=10 -m synch=lock_ticket code/lock_test1.hny.

We now turn to a radically different way of implementing locks, one that is commonly provided by operating systems to user processes. We call a thread blocked if a thread cannot change the state or terminate unless another thread changes the state first. A thread trying to acquire a lock held by another thread is a good example of a thread being blocked. The only way forward is if the other thread releases the lock.

def Lock() returns result:
    result = { .acquired: False, .suspended: [] }

def acquire(lk):
    atomically:
        if lk->acquired:
            stop ?lk->suspended[len lk->suspended]
            assert lk->acquired
        else:
            lk->acquired = True

def release(lk):
    atomically:
        assert lk->acquired
        if lk->suspended == []:
            lk->acquired = False
        else:
            go (lk->suspended[0]) ()
            del lk->suspended[0]

Figure 8.5 shows the lock interface using suspension. It is implemented as follows:

• A lock maintains both a boolean indicating whether the lock is currently acquired and a list of contexts of threads that want to acquire the lock.
• acquire() acquires the lock if available and suspends the invoking thread if not. In the latter case, the context of the thread is added to the end of the list of contexts. Note that stop is called within an atomic statement block---this is the only exception to such an atomic statement block running to completion. While the thread is running no other threads can run but, after the thread suspends itself, other threads can run.
• release() checks to see if any threads are waiting to acquire the lock. If so, it selects the first context on the list and resumes it.

You will find that using the implementation of a lock instead of the specification of a lock (in the synch module) often leads to the model checker searching a significantly larger state space. Thus it makes sense to model check larger programs in a modular fashion: model check one module implementation at a time, using specifications for the other modules.

Exercises

• [Ex. 8.1]: Run the program in Figure 7.3 using (i) synch and (ii) synchS. Report how many states were explored by Harmony for each module.
• [Ex. 8.2]: Figure 8.6 shows a Harmony program with two variables x (initially 0) and y (initially 100) that can be accessed through methods setX and getXY. An application invariant is that getXY should return a pair that sums to 100. Add the necessary synchronization code.
• [Ex. 8.3]: Implement tryAcquire(b) as an additional interface for both the synch and synchS modules. This interface is like acquire(b) but never blocks. It returns True if the lock was available (and now acquired) or False if the lock was already acquired. Hint: you do not have to change the existing code.
• [Ex. 8.4]: People who use an ATM often first check their balance and then withdraw a certain amount of money not exceeding their balance. A negative balance is not allowed. Figure 8.7 shows two operations on bank accounts: one to check the balance and one to withdraw money. Note that all operations on accounts are carefully protected by a lock (i.e., there are no data races). The customer method models going to a particular ATM and withdrawing money not exceeding the balance. Running the code through Harmony reveals that there is a bug. It is a common type of concurrency bug known as Time Of Check Time Of Execution (TOCTOE). In this case, by the time the withdraw operation is performed, the balance can have changed. Fix the code in Figure 8.7. Note, you should leave the customer code the same. You are only allowed to change the atm_ methods, and you cannot use the atomically keyword.

x, y = 0, 100

def setX(a):
    x = a
    y = 100 - a

def getXY() returns xy:
    xy = [x, y]

def checker():
    let xy = getXY():
        assert (xy[0] + xy[1]) == 100, xy
    
spawn checker()
spawn setX(50)

from synch import Lock, acquire, release

const N_ACCOUNTS = 2
const N_CUSTOMERS = 2
const N_ATMS = 2
const MAX_BALANCE = 1

accounts = [ { .lock: Lock(), .balance: choose({0..MAX_BALANCE})}
                            for i in {1..N_ACCOUNTS} ]

invariant min(accounts[acct].balance for acct in {0..N_ACCOUNTS-1}) >= 0

def atm_check_balance(acct) returns balance:  # return the balance on acct
    acquire(?accounts[acct].lock)
    balance = accounts[acct].balance
    release(?accounts[acct].lock)

def atm_withdraw(acct, amount) returns success: # withdraw amount from acct
    assert amount >= 0
    acquire(?accounts[acct].lock)
    accounts[acct].balance -= amount
    release(?accounts[acct].lock)
    success = True

def customer(atm, acct, amount):
    assert amount >= 0
    let bal = atm_check_balance(acct):
        if amount <= bal:
            atm_withdraw(acct, amount)
        
for i in {1..N_ATMS}:
    spawn customer(i, choose({0..N_ACCOUNTS-1}),
                      choose({0..MAX_BALANCE}))

Chapter 9. Concurrent Data Structures

def Queue() returns empty:
    empty = []

def put(q, v):
    !q += [v,]

def get(q) returns next:
    if !q == []:
        next = None
    else:
        next = (!q)[0]
        del (!q)[0]

def Queue() returns empty:
    empty = []

def put(q, v):
    atomically !q += [v,]

def get(q) returns next:
    atomically:
        if !q == []:
            next = None
        else:
            next = (!q)[0]
            del (!q)[0]

import queue

def sender(q, v):
    queue.put(q, v)

def receiver(q):
    let v = queue.get(q):
        assert v in { None, 1, 2 }

demoq = queue.Queue()
spawn sender(?demoq, 1)
spawn sender(?demoq, 2)
spawn receiver(?demoq)
spawn receiver(?demoq)

The queue module can be defined as follows:

• x = Queue(): initialize a new queue x;
• put(?x, v): add v to the tail of x;
• r = get(?x): returns r = None if x is empty or r = v if v was at the head of x.

Unfortunately, this sequential specification does not work well with threads concurrently accessing this queue. This is because operations can overlap in execution, and so one operation can witness the intermediate state of another operation. As a compiler would be free to implement the sequential specification in any way it wants to, it gets to decide what data structures to use to store the list, and what machine instructions to use to implement the operations. For a sequential specification that is not a problem because we only need to understand the state in between operations, but in a concurrent setting Figure 9.1(a) is underspecified; it does not say what happens when two or more operations overlap.

Figure 9.1(b) shows the corresponding concurrent specification. It is almost identical to the sequential specification, but it states that the actual execution of an operation of each method has to happen atomically sometime between invoking the method and the method completing. It cannot be used as an implementation for a queue, as processors generally do not have atomic operations on lists, but it will work well as a specification. See Figure 9.2 for a simple demonstration program that uses a concurrent queue; the specification can be used with any concurrent program that uses a queue. But this concurrent specification is too much for compilers to handle; we have to write down how to implement it.

While in the concurrent specification operations cannot overlap execution, the same need not be true for an implementation. Indeed, we will show two implementations of a concurrent queue. The first uses critical sections, ensuring that operations cannot overlap. The second allows a get and a put operation on a non-empty queue to execute simultaneously. Both are correct implementations of the concurrent queue specification.

In both implementations, the queue data structure is maintained as a linked list. They use the alloc module for dynamic allocation of nodes in the list, which provides methods malloc() and free(). malloc(v) returns a new memory location initialized to v, which should be released with free() when it is no longer in use.

The implementation in Figure 9.3 uses a lock-based critical section to operate on the data structure. The implementation maintains a head pointer to the first element in the list and a tail pointer to the last element in the list. The head pointer is None if and only if the queue is empty. (None is a special address value that is not the address of any memory location.)

Queue() returns the initial value for a queue object consisting of a None head and tail pointer and a lock. The put(q, v) and get(q) methods both take a pointer q to the queue object. It has to be a pointer rather than the value of the queue object because both methods must be able to modify the queue object. Before a method accesses the value of the head or tail of the queue, it first obtains the lock. When the method is done, it releases the lock. Doing so prevents concurrently executing methods from witnessing an intermediate, inconsistent state of the linked list.

Importantly, note Lines 7 and 8 in Figure 9.2. It would be incorrect to replace these by:

The implementation in Figure 9.3 uses a single lock to protect the queue data structure. A thread acquires the lock before any access and releases it afterwards. This is possibly the simplest way to make a data structure concurrent. However, it can cause significant contention over the lock between a producer and a consumer, reducing performance by serializing operations.

Figure 9.4 shows another concurrent queue implementation that addresses this problem [34]. It is based on the insight that a put method operates on the tail of a queue, while a get method operates on the head, and so in theory they should be able to execute concurrently. Although the algorithm is well-known, what is not often realized (because it is not stated explicitly in the paper) is that it requires sequentially consistent memory. The algorithm must be coded carefully to work correctly with modern programming languages and computer hardware that generally lack sequentially consistent memory.

The implementation uses separate locks for the head and the tail, allowing a put and a get operation to proceed concurrently. The implementation uses a dummy node (aka sentinel node) at the head of the linked list. Except initially, the dummy node is the last node that was dequeued. Using dummy nodes can simplify implementations because they can reduce the number of exceptional conditions. In this case, note that neither the head nor tail pointer are ever None, reducing the need for if statements in the code.

The problem with the original specification of the algorithm is when the queue is empty and there are concurrent get and put operations. They obtain separate locks and then concurrently access the next field in the dummy node---a data race with undefined semantics in most modern environments. To get around this problem, the implementation in Figure 9.4 uses atomic_load and atomic_store from the synch module.

from synch import Lock, acquire, release
from alloc import malloc, free

def Queue() returns empty:
    empty = { .head: None, .tail: None, .lock: Lock() }

def put(q, v):
    let node = malloc({ .value: v, .next: None }):
        acquire(?q->lock)
        if q->tail == None:
            q->tail = q->head = node
        else:
            q->tail->next = node
            q->tail = node
        release(?q->lock)
    
def get(q) returns next:
    acquire(?q->lock)
    let node = q->head:
        if node == None:
            next = None
        else:
            next = node->value
            q->head = node->next
            if q->head == None:
                q->tail = None
            free(node)
    release(?q->lock)

Figure 9.3. [code/queue_lock.hny] An implementation of a concurrent queue data structure and a depiction of a queue with three elements

from synch import Lock, acquire, release, atomic_load, atomic_store
from alloc import malloc, free

def Queue() returns empty:
    let dummy = malloc({ .value: (), .next: None }):
        empty = { .head: dummy, .tail: dummy, .hdlock: Lock(), .tllock: Lock() }

def put(q, v):
    let node = malloc({ .value: v, .next: None }):
        acquire(?q->tllock)
        atomic_store(?q->tail->next,  node)
        q->tail = node
        release(?q->tllock)

def get(q) returns next:
    acquire(?q->hdlock)
    let dummy = q->head
    let node = atomic_load(?dummy->next):
        if node == None:
            next = None
            release(?q->hdlock)
        else:
            next = node->value
            q->head = node
            release(?q->hdlock)
            free(dummy)

Figure 9.4. [code/queue_MS.hny] A queue with separate locks for enqueuing and dequeuing items and a depiction of a queue with two elements

from alloc import malloc

def SetObject() returns object:
    object = malloc({})

def insert(s, v):
    atomically !s |= {v}

def remove(s, v):
    atomically !s -= {v}

def contains(s, v) returns present:
    atomically present = v in !s

from setobj import *

myset = SetObject()

def thread1():
    insert(myset, 1)
    let x = contains(myset, 1):
        assert x

def thread2(v):
    insert(myset, v)
    remove(myset, v)

spawn thread1()
spawn thread2(0)
spawn thread2(2)

from synch import Lock, acquire, release
from alloc import malloc, free

def _node(v, n) returns node:   # allocate and initialize a new list node
    node = malloc({ .lock: Lock(), .value: v, .next: n })

def _find(lst, v) returns pair:
    var before = lst
    acquire(?before->lock)
    var after = before->next
    acquire(?after->lock)
    while after->value < (0, v):
        release(?before->lock)
        before = after
        after = before->next
        acquire(?after->lock)
    pair = (before, after)

def SetObject() returns object:
    object = _node((-1, None), _node((1, None), None))

def insert(lst, v):
    let before, after = _find(lst, v):
        if after->value != (0, v):
            before->next = _node((0, v), after)
        release(?after->lock)
        release(?before->lock)

def remove(lst, v):
    let before, after = _find(lst, v):
        if after->value == (0, v):
            before->next = after->next
            free(after)
        release(?before->lock)

def contains(lst, v) returns present:
    let before, after = _find(lst, v):
        present = after->value == (0, v)
        release(?after->lock)
        release(?before->lock)

Figure 9.5 gives the specification of a concurrent set object. SetObject() returns a pointer to a variable that contains an empty set, rather than returning an empty set value. As such, it is more like an object in an object-oriented language than like a value in its own right. Values can be added to the set object using insert() or deleted using remove(). Method contains() checks if a particular value is in the list. Figure 9.6 contains a simple (although not very thorough) test program to demonstrate the use of set objects.

Figure 9.7 implements a concurrent set object using an ordered linked list without duplicates. The list has two dummy "book-end" nodes with values (-1, None) and (1, None). A value v is stored as (0, v)---note that for any value v, (-1, None) < (0, v) < (1, None) (because of lexicographical ordering of tuples). An invariant of the algorithm is that at any point in time the list is "valid," starting with a (-1, None) node and ending with an (1, None) node.

Each node has a lock, a value, and next, a pointer to the next node (which is None for the (1, None) node to mark the end of the list). The _find(lst, v) helper method first finds and locks two consecutive nodes before and after such that before->data.value < (0, v) <= after->data.value. It does so by performing something called hand-over-hand locking. It first locks the first node, which is the (-1, None) node. Then, iteratively, it obtains a lock on the next node and release the lock on the last one, and so on, similar to climbing a rope hand-over-hand. Using _find, the insert, remove, and contains methods are fairly straightforward to implement.

Exercises

• [Ex. 9.1]: Add a method head(q) that returns the value at the head of the queue, or None if the queue is empty.
• [Ex. 9.2]: Add a method contains(q, v) to Figure 9.1(b) that checks to see if v is in queue q.
• [Ex. 9.3]: Add a method length(q) to Figure 9.3 that returns the length of the given queue. The complexity of the method should be O(1) , which is to say that you should maintain the length of the queue as a field member and update it in put and get.
• [Ex. 9.4]: Write a method check(q) that checks the integrity of the queue in Figure 9.3. In particular, it should check the following integrity properties:
  • If the list is empty, q->tail should be None. Otherwise, the last element in the linked list starting from q->head should equal q->head. Moreover, q->tail->next should be None;
  • The length field that you added in Exercise 9.4 should equal the length of the list.
  Method check(q) should not obtain a lock; instead add the following line just before releasing the lock in put and get: assert check()
• [Ex. 9.5]: Add a method remove(q, v) to Figure 9.3 that removes all occurrences of v, if any, from queue q.
• [Ex. 9.6]: The test program in Figure 9.2 is not a thorough test program. Design and implement a test program for Figure 9.2. Make sure you test the test program by trying it out against some buggy queue implementations. (You will learn more about testing concurrent programs in Chapter 10.)
• [Ex. 9.7]: Add methods to the data structure in Figure 9.7 that report the size of the list, the minimum value in the list, the maximum value in the list, and the sum of the values in the list. (All these should ignore the two end nodes.)
• [Ex. 9.8]: Create a thread-safe sorted binary tree. Implement a module bintree with methods BinTree() to create a new binary tree, insert(t,v) that inserts v into tree t, and contains(t, v) that checks if v is in tree t. Use a single lock per binary tree.
• [Ex. 9.9]: Create a binary tree that uses, instead of a single lock per tree, a lock for each node in the tree.

Chapter 10. Testing: Checking Behaviors

import queue, queueconc

const NOPS = 4
const VALUES = { 1..NOPS }

specq = queue.Queue()
implq = queueconc.Queue()

for i in {1..NOPS}:
    let op = choose({ "get", "put" }):
        if op == "put":
            let v = choose(VALUES):
                queueconc.put(?implq, v)
                queue.put(?specq, v)
        else:
            let v = queueconc.get(?implq)
            let w = queue.get(?specq):
                assert v == w

Checking the behavior of a concurrent queue can be tricky. For example, suppose that, in some execution of Figure 9.2, thread 1 invokes put(?demoq, 1) and then thread 2 invokes put(?demoq, 2) and then thread 3 invokes get(?demoq). What value should thread 3 obtain? The answer is: it depends, or in other words, not enough information is provided. All of 1, 2, and None are possible outcomes. Certainly, if put(?demoq, 1) completes before put(?demoq, 2) is invoked and put(?demoq, 2) completes before get(?demoq) is invoked, then 1 is the only correct result. However, if put(?demoq, 1) is invoked first, and then put(?demoq, 2) is invoked before put(?demoq, 1) has completed, then those two operations are executing concurrently, and it might come to pass that 2 is enqueued before 1. In that case, if get(?demoq) is invoked after both those operations complete, both 1 and 2 are possible results, but None cannot be. However, if get(?demoq) is invoked before the other two operations complete, then None is a valid outcome as well, even if the other two operations were invoked before get(?demoq) was invoked.

The issue is that operations take time, and therefore operations can overlap. While locks reduce the amount to which operations can overlap, they do not completely eliminate it, and obtaining a lock takes time as well. Unlike critical sections, it turns out that checking the correct behavior of a queue is difficult to do with assertions alone because it is difficult to characterize the set of correct behaviors of a concurrent queue.

Behaviors say something about how we got to a state. The same state can be reached by multiple behaviors, and the behaviors are often an integral part of whether a program is correct or not. Just because a state satisfies some invariant---however important---does not mean that the state is valid given the sequence of operations. For example, a state in which the queue is empty is certainly a valid state in its own right, but if the last operation to get there was an enqueue operation, there must be a bug in the program. It can therefore be important to capture the behaviors. We could store behaviors in the state itself by adding what is known as a history variable that keeps track of all the operations. While this can be useful for correctness proofs, for model checking this approach presents a problem: introducing this additional state can lead to state explosion or even turn a finite model (a model with a finite number of states) into an infinite one.

Fortunately, if we have a specification of a queue, we can generate the set of correct behaviors of a queue. For simplicity, we will first check if the queue implementation in Figure 9.3 meets the sequential queue specification in Figure 9.1(a). To check if the queue implementation meets the specification, we need to see if any sequence of queue operations in the implementation matches a corresponding sequence in the specification. We say that the implementation and the specification have the same behaviors or are behaviorally equivalent. This is called differential testing.

Figure 10.1 presents a differential test program for sequences of up to NOPS queue operations. It maintains two queues:

• specq: the queue of the specification;
• implq: the queue of the implementation.

Test programs themselves should be tested. Just because a test program works with a particular implementation does not mean the implementation is correct---it may be that the implementation is incorrect but the test program does not have enough coverage to find any bugs in the implementation. So, run a test program like this with a variety of queue implementations that have known bugs in them and make sure that the test program finds them. Conversely, a test program may be broken in that it finds bugs that do not exist. In my experience, it is often harder to implement the test program than the algorithm that the test program tests.

As with any other test program, Figure 10.1 may not trigger extant bugs, but it nonetheless inspires reasonable confidence that the queue implementation is correct, at least sequentially. The higher NOPS, the higher the confidence. It is possible to write similar programs in other languages such as Python, but the choose expression in Harmony makes it relatively easy to explore all corner cases. For example, a common programming mistake is to forget to update the tail pointer in get() in case the queue becomes empty. Normally, it is a surprisingly tricky bug to find. You can comment out those lines in Figure 9.3 and run the test program---it should easily find the bug and explain exactly how the bug manifests itself, adding confidence that the test program is reasonably thorough.

The test program also finds some common mistakes in using locks, such as forgetting to release a lock when the queue is empty, but it is not designed to find concurrency bugs in general. If you remove all acquire() and release() calls from Figure 9.3, the test program will not (and should not) find any errors, but it would be an incorrect implementation of a concurrent queue.

import queue

const N_PUT = 2
const N_GET = 2
q = queue.Queue()

def put_test(self):
    print("call put", self)
    queue.put(?q, self)
    print("done put", self)

def get_test(self):
    print("call get", self)
    let v = queue.get(?q):
        print("done get", self, v)

for i in {1..N_PUT}:
    spawn put_test(i)
for i in {1..N_GET}:
    spawn get_test(i)

Figure 10.2. [code/queue_btest2.hny] Concurrent queue test. The behavior DFA is for N_PUT = N_GET = 1.

import queue, threading, random

N_PUT = 3
N_GET = 3
q = queue.Queue()

def put_test(self):
    print("call put", self)
    q.put(self)
    print("done put", self)

def get_test(self):
    print("call get", self)
    try:
        v = q.get(block=False)
        print("done get", self, v)
    except queue.Empty:
        print("done get empty", self)

for i in range(N_PUT):
    threading.Thread(target=put_test, args=(i,)).start()
for i in range(N_GET):
    threading.Thread(target=get_test, args=(i,)).start()

from synch import Lock, acquire, release

def Queue() returns empty:
    empty = { .list: [], .lock: Lock() }

def put(q, v):
    acquire(?q->lock)
    q->list += [v,]
    release(?q->lock)

def get(q) returns next:
    acquire(?q->lock)
    if q->list == []:
        next = None
    else:
        next = q->list[0]
    release(?q->lock)
    acquire(?q->lock)
    if q->list != []:
        del q->list[0]
    release(?q->lock)

We would like a way that---similar to the sequential test---systematically compares behaviors of the concurrent queue implementation with behaviors of the concurrent queue specification. But we cannot do this by composing the specification and the implementation and simply run the same test operations on both as we did before---concurrency make the operations non-determistic and thus the specification and implementation of a single execution might produce different results, even if both are correct. Instead, we will create a test program that tries various concurrent combinations of queue operations, and run it twice: once against the specification of the concurrent queue and once against the implementation. In the second phase, we check if the behaviors obtained from running the implementation are also behaviors obtained from the specification.

Figure 10.2 shows the test program. It starts N_PUT threads doing a put operation and N_GET threads doing a get operation. In case of a put operation, the thread enqueues its own name (which is provided as an argument to the thread). In order to capture the behaviors, each thread prints what operation it is about to perform, and afterwards it prints that the operation has completed (including the return value if any). The figure also shows the \emph{behavior DFA} that captures the generated behaviors in case N_PUT = N_GET = 1. You can see that if get() completes before put(1) is invoked, then get() must return None. Vice versa, if put(1) completes before get() is invoked, then get() must return 1. Otherwise get() can return either None or 1 depending on how the two operations are serialized.

This is probably much like you would do if you were trying to find a bug in a program. Figure 10.3 shows a Python implementation of the same test program. You can run it a bunch of times and manually check the output. There are, however, two problems with this approach. First, it is often difficult to check if the behaviors you find are correct ones, and it is easy to overlook problems in the output. Second, the test program may not check all possible behaviors.

Using Harmony, these problems can be avoided. One approach is to compare the two behavior DFAs by manual inspection:

$ harmony -c N_PUT=1 -c N_GET=1 code/queue_btest2.hny
$ harmony -c N_PUT=1 -c N_GET=1 -m queue=queue_lock code/queue_btest2.hny

$ harmony -c N_GET=2 -o x.png code/queue_btest2.hny
$ harmony -c N_GET=2 -o y.png -m queue=queue_broken1 code/queue_btest2.hny

Harmony does have a way to check the behaviors of one program against the behaviors of another. In particular, we want to check if the behaviors of the an implementation matches the behaviors of the specification. The following shows, for example, how to check the queue_lock.hny implementation on the command line:

$ harmony -o queue.hfa code/queue_btest2.hny
$ harmony -B queue.hfa -m queue=queue_lock code/queue_btest2.hny

Now try the same using the queue_broken1.hny implementation:

$ harmony -o queue.hfa code/queue_btest2.hny
$ harmony -B queue.hfa -m queue=queue_broken1 code/queue_btest2.hny

Exercises

• [Ex. 10.1]: Figure 7.1 shows a specification of a lock. Write a program that checks the behaviors of lock implementations such as Figure 8.2 and Figure 8.3. That is, it should not rely on assertions such as in Figure 5.2.
• [Ex. 10.2]: Write a Harmony program that checks if Figure 9.7 satisfies the specification of Figure 9.5 sequentially.
• [Ex. 10.3]: Write a Harmony program that checks if Figure 9.7 satisfies the specification of Figure 9.5 concurrently.

Chapter 11. Debugging

from synch import Lock, acquire, release
from alloc import malloc, free

def Queue() returns empty:
    empty = { .next: None, .value: None, .lock: Lock() }

def put(q, v):
    let node = malloc({ .next: None, .value: v, .lock: Lock() }):
        var nq = q
        while nq != None:
            acquire(?nq->lock)
            let n = nq->next:
                if n == None:
                    nq->next = node
                release(?nq->lock)
                nq = n

def get(q) returns next:
    acquire(?q->lock)
    if q->next == None:
        next = None
    else:
        let node = q->next:
            q->next = node->next
            next = node->value
            free(node)
    release(?q->lock)

Figure 11.3. HTML output of Figure 11.2 but for N_PUT=2 and N_GET=1

.

Figure 11.1 contains an attempt at a queue implementation where the queue is implemented by a linked list, with the first node being a dummy node to prevent data races. Each node in the list contains a lock. The put() method walks the list until it gets to the last node, each time acquiring the lock to access the node's fields. When put() gets to the last node in the list, it appends a new one. The get() method locks the first (dummy) node, removes the second from the list and frees it. The method returns the value from the removed node.

Let us run the code through the test programs in the last chapter. Harmony does not detect any issues with the sequential test in Figure 10.1. (Run this using the -m flag like this: harmony -m queue=queue_broken2 code/qtestseq.hny) However, when we run the new queue code through the test in Figure 10.2, Harmony reports a safety violation (even without specifying a behavior). Figure 11.2 shows the command line to reproduce this and the first few lines of markdown output.

Before we go look at the details of what went wrong, we want to make sure that we generate the simplest scenario. So, first we want to explore what the smallest N_PUT and N_GET parameters are. With some experimentation, we find that N_PUT = 2 and N_GET = 1 generates a problem (harmony -m queue=queue_broken2 -c N_PUT=2 -c N_GET=1 code/queue_btest2.hny)). Figure 11.3 shows the HTML output.

There is quite a bit of information in the HTML output. Let's start with looking at the red text. Harmony found a safety violation (something bad happened during one of the possible executions), and in particular put_test(1) (thread T1) was trying to dereference the address ?alloc$pool[0]["lock"]. The alloc module maintains a shared array pool that it uses for dynamic allocation. Apparently T1 tried to access pool[0], but it does not exist, meaning that either it was not yet allocated, or it had been freed since it was allocated. When we look at the top half of the figure under "Shared Variables", we see that in fact thread T2 allocated pool[0] in turn 2 (during put_test(2)), but T3 freed it in turn 4 (during get_test(1)). Looking at the stack traces in the bottom table, we can see that T3 was in the process of executing release(?q.lock) within get(?q). T1 is currently executing acquire(?alloc$pool[0].lock) within put(?q, 1), but alloc$pool[0] does not exist.

So, how did we get there? In the top we can see that the order of events was the following:

1. initialization completed, with q being { .lock: False, .next: None, .value: None };
2. thread T2 (put_test(2)) ran and finished executing put(2) (and is about to print ["done put", 2]). We can see that q.next now points to alloc$pool[0], which the thread must have allocated. The contents is { .lock: False, .next: None, .value: 2 }, as expected;
3. thread T1 (put_test(1)) started running, calling put(?q, 1). We can see it got as far as allocating a node, but it has not yet added the node to the end of the queue. It then tries to acquire alloc$pool[0].lock;
4. thread T3 (get_test(1)) started running, calling get(?q). We can also see that it freed pool[0], and is now releasing q.lock;
5. thread T1 resumes and tries to access alloc$pool[0], which no longer exists (because T3 just freed it).

It is time to start stepping through the code that has been executed before this happened. This is sometimes known as reverse debugging. In fact, Harmony allows you to step through an execution forwards and backwards. In this case, we first want to see what T2 is doing. You can click on its first (top-left) orange box to time-travel to that part in the execution. The boxes are color-coded: each method has its own color. Now by hitting the ⟨return⟩ key repeatedly, we can quickly skip through the code, line by line. T1 first calls put(?q, 1) and then allocates a new node initialized with a lock. Keep stepping until it executes nq = q. Hit ⟨return⟩ once more and inspect the state of T1 in the lower-right corner. You can see that variable nq is initialized to ?q. T1 then enters into the while loop and tries to acquire nq->lock. This succeeds, and next T1 executes let n = nq->next. Now n = ?alloc$pool[0], which is not None. It then releases nq->lock (nq points to q). It then sets nq to n, which is still alloc$pool[0]. Finally, it calls acquire(?nq->lock). But before it can complete that operation, T3 runs next.

T3 chooses "get" and then goes on to invoke get(?q). This first successfully acquires q->lock. T3 then finds out that q->next points to alloc$pool[0]. T3 sets node to alloc$pool[0] as well and sets q->next to node->next. T3 sets the method result next to node->value (which is 2) and then frees node. This is where the problem is---T1 is about to acquire the lock in that same node.

To fix the code without changing the data structure, we can use hand-over-hand locking (Chapter 9). Figure 11.4 shows an implementation that uses hand-over-hand locking both for put() and for get(). It passes all tests.

from synch import Lock, acquire, release
from alloc import malloc, free

def Queue() returns empty:
    empty = { .next: None, .value: None, .lock: Lock() }

def put(q, v):
    var nq = q
    let node = malloc({ .next: None, .value: v, .lock: Lock() }):
        acquire(?nq->lock)
        var n = nq->next
        while n != None:
            acquire(?n->lock)
            release(?nq->lock)
            nq = n
            n = n->next
        nq->next = node
        release(?nq->lock)

def get(q) returns next:
    acquire(?q->lock)
    if q->next == None:
        next = None
    else:
        let node = q->next:
            acquire(?node->lock)
            q->next = node->next
            next = node->value
            release(?node->lock)
            free(node)
    release(?q->lock)

Chapter 12. Conditional Waiting

Critical sections enable multiple threads to easily share data structures whose modification requires multiple steps. A critical section only allows one thread to execute the code of the critical section at a time. Therefore, when a thread arrives at a critical section, the thread waits until there is no other thread in the critical section.

A thread that is waiting for a condition without changing shared variables is considered blocked. In other words, a blocked thread can only change shared variables if some other thread changes them first. A spinlock to protect a critical section is a good example: a thread that is waiting for a spinlock does not change the state of the lock variable until it acquires the lock.

Sometimes it is useful for a thread to wait for additional conditions besides the critical section being unoccupied. For example, when dequeuing from an empty shared queue, the thread may want to block until the queue is non-empty. The alternative to blocking would be active busy waiting (or also just busy waiting) where the thread repeatedly tries to dequeue an item until it is successful. A thread that is active busy waiting until the queue is non-empty cannot make progress until another thread enqueues an item. However, the thread is not considered blocked because it is changing the shared state by repeatedly acquiring and releasing the lock. Doing so wastes CPU cycles and adds contention to queue access.

We would like to find a solution to conditional waiting so that a thread blocks until the condition holds---or at least most of the time. Before we do so, we will give two classic examples of synchronization problems that involve conditional waiting: reader/writer locks and bounded buffers.

def RWlock() returns lock:
    lock = { .nreaders: 0, .nwriters: 0 }

def read_acquire(rw):
    atomically when rw->nwriters == 0:
        rw->nreaders += 1

def read_release(rw):
    atomically rw->nreaders -= 1

def write_acquire(rw):
    atomically when ((rw->nreaders == 0) and (rw->nwriters == 0)):
        rw->nwriters = 1

def write_release(rw):
    atomically rw->nwriters = 0

import rwlock

nreaders = nwriters = 0
invariant ((nreaders >= 0) and (nwriters == 0)) or
            ((nreaders == 0) and (nwriters == 1))

const NOPS = 4

rw = rwlock.RWlock()

def thread():
    while choose({ False, True }):
        if choose({ "read", "write" }) == "read":
            rwlock.read_acquire(?rw)
            atomically nreaders += 1
            atomically nreaders -= 1
            rwlock.read_release(?rw)
        else:                       # write
            rwlock.write_acquire(?rw)
            atomically nwriters += 1
            atomically nwriters -= 1
            rwlock.write_release(?rw)

for i in {1..NOPS}:
    spawn thread()

import synch

def RWlock() returns lock:
    lock = synch.Lock()

def read_acquire(rw):
    synch.acquire(rw)

def read_release(rw):
    synch.release(rw)

def write_acquire(rw):
    synch.acquire(rw)

def write_release(rw):
    synch.release(rw)

import rwlock

const NOPS = 3

rw = rwlock.RWlock()

def thread(self):
    while choose({ False, True }):
        if choose({ "read", "write" }) == "read":
            print(self, "enter ra")
            rwlock.read_acquire(?rw)
            print(self, "exit ra")

            print(self, "enter rr")
            rwlock.read_release(?rw)
            print(self, "exit rr")
        else:                       # write
            print(self, "enter wa")
            rwlock.write_acquire(?rw)
            print(self, "exit wa")

            print(self, "enter wr")
            rwlock.write_release(?rw)
            print(self, "enter wr")

for i in {1..NOPS}:
    spawn thread(i)

from synch import Lock, acquire, release

def RWlock() returns lock:
    lock = { .lock: Lock(), .nreaders: 0, .nwriters: 0 }

def read_acquire(rw):
    acquire(?rw->lock)
    while rw->nwriters > 0:
        release(?rw->lock)
        acquire(?rw->lock)
    rw->nreaders += 1
    release(?rw->lock)

def read_release(rw):
    acquire(?rw->lock)
    rw->nreaders -= 1
    release(?rw->lock)

def write_acquire(rw):
    acquire(?rw->lock)
    while (rw->nreaders > 0) or (rw->nwriters) > 0):
        release(?rw->lock)
        acquire(?rw->lock)
    rw->nwriters = 1
    release(?rw->lock)

def write_release(rw):
    acquire(?rw->lock)
    rw->nwriters = 0
    release(?rw->lock)

$ harmony -o rw.hfa -cNOPS=2 code/rwlock_btest.hny
$ harmony -B rw.hfa -cNOPS=2 -m rwlock=rwlock_cheat code/rwlock_btest.hny

def BoundedBuffer(size) returns buffer:
    buffer = { .buffer: [], .size: size }

def put(bb, v):
    atomically when len(bb->buffer) < bb->size:
        bb->buffer += [v,]

def get(bb) returns next:
    atomically when bb->buffer != []:
        next = bb->buffer[0]
        del bb->buffer[0]

Exercises

• [Ex. 12.1]: Optimize your solutions to Exercise 9.2 to use reader/writer locks.

Chapter 13. Condition Variables

The last chapter introduced conditional waiting, but the implementations provided were unsatisfactory, either because they unnecessarily restricted behaviors or because they were inefficient. Condition variables are a synchronization primitive designed to allow for general and efficient solutions to conditional waiting problems.

In the late 70s, researchers at Xerox PARC, where among others the desktop and Ethernet were invented, developed a new programming language called Mesa [31]. Mesa introduced various important concepts to programming languages, including software exceptions and incremental compilation. The Mesa language also incorporated a version of condition variables. There are two main classes of condition variable semantics. Appendix I describes Hoare condition variables. However, most programming languages today support the semantics of Mesa's condition variables. Below, we will use the semantics that Mesa provides.

A condition variable works in conjunction with a lock, often called mutex (for mutual exclusion) in this context. A thread p that holds the mutex can invoke the wait operation on a condition variable. This temporarily releases the mutex and places thread p on a queue associated with the condition variable so p no longer executes. Another thread can notify the condition variable. If the condition variable's queue is non-empty, this operation removes a thread q from the queue (according to some scheduling policy) and makes q runnable again. Thread q first re-acquires the mutex before resuming from the wait operation it invoked. If, on the other hand, the condition variable's queue is empty, the notify operation is a no-op.

Condition variables also allow notifying multiple threads. For example, a thread can invoke notify twice---if there are two or more threads waiting on the condition variable, two will be resumed. Operation notifyAll (aka broadcast)) notifies all threads that are waiting on a condition.

from synch import *

def RWlock() returns lock:
    lock = {
            .nreaders: 0, .nwriters: 0, .mutex: Lock(),
            .r_cond: Condition(), .w_cond: Condition()
        }
    
def read_acquire(rw):
    acquire(?rw->mutex)
    while rw->nwriters > 0:
        wait(?rw->r_cond, ?rw->mutex)
    rw->nreaders += 1
    release(?rw->mutex)

def read_release(rw):
    acquire(?rw->mutex)
    rw->nreaders -= 1
    if rw->nreaders == 0:
        notify(?rw->w_cond)
    release(?rw->mutex)

def write_acquire(rw):
    acquire(?rw->mutex)
    while (rw->nreaders > 0) or (rw->nwriters > 0):
        wait(?rw->w_cond, ?rw->mutex)
    rw->nwriters = 1
    release(?rw->mutex)

def write_release(rw):
    acquire(?rw->mutex)
    rw->nwriters = 0
    notifyAll(?rw->r_cond)
    notify(?rw->w_cond)
    release(?rw->mutex)

Note that wait is always invoked within a while loop that checks for the condition that the thread is waiting for. It is imperative that there is always a while loop around any invocation of wait containing the negation of the condition that the thread is waiting for. Many implementation of Mesa condition variables depend on this, and optimized implementations of condition variables often allow so-called "spurious wakeups," where wait may sometimes return even if the conditon variable has not been notified. As a rule of thumb, one should always be able to replace wait by release followed by acquire. This turns the solution into a busy-waiting one, inefficient but still correct.

In read_release, notify(?w_cond) is invoked when there are no readers left, without checking if there are writers waiting to enter. This is ok, because calling notify is a no-op if no thread is waiting.

write_release executes notifyAll(?r_cond) as well as notify(?w_cond). Because we do not keep track of the number of waiting readers or writers, we have to conservatively assume that all waiting readers can enter, or, alternatively, up to one waiting writer can enter. Awakening both readers and writers is ok because both execute wait within a while loop, re-checking the condition that they are waiting for. So, if both types of threads are waiting, either all the readers get to enter next or one of the writers gets to enter next. (If you want to prevent waking up both readers and a writer, then you can keep track of how many threads are waiting for each condition.)

When using condition variables, you have to be careful to invoke notify or notifyAll in the right places. Much of the complexity of programming with condition variables is in figuring out when to invoke notify and when to invoke notifyAll. As a rule of thumb: be conservative---it is better to wake up too many threads than too few. In case of doubt, use notifyAll. For example, in Figure 13.1 it is ok, if inefficient, to replace the notify operations with notifyAll, but it would be incorrect to replace the notifyAll operation by notify. Waking up too many threads may lead to some inefficiency, but waking up too few may cause the application to get stuck. Harmony can be particularly helpful here, as it examines each and every corner case. You can try to replace each notifyAll with notify and see if every possible execution of the application still terminates.

Andrew Birrell's paper on Programming with Threads gives an excellent introduction to working with Mesa-style condition variables [5]. They have been adopted by all major programming languages. In Java, each object has a hidden lock and a hidden condition variable associated with it. Methods declared with the synchronized keyword automatically obtain the lock. Java objects also support wait, notify, and notifyAll. In addition, Java supports explicit allocations of locks and condition variables. In Python, locks and condition variables must be explicitly declared. The with statement makes it easy to acquire and release a lock for a section of code. In C and C++, support for locks and condition variables is entirely through libraries.

def Condition() returns condition:
    condition = bag.empty()

def wait(c, lk):
    var cnt = 0
    let _, ctx = save():
        atomically:
            cnt = bag.multiplicity(!c, ctx)
            !c = bag.add(!c, ctx)
            !lk = False
        atomically when (not !lk) and (bag.multiplicity(!c, ctx) <= cnt):
            !lk = True

def notify(c):
    atomically if !c != bag.empty():
        !c = bag.remove(!c, bag.bchoose(!c))
        
def notifyAll(c):
    !c = bag.empty()

Method wait adds the context of the thread---used as a unique identifier for the thread---to the bag, incrementing the number of threads in the bag with the same context. The Harmony save expression (Section C.3) returns a tuple containing a value (in this case `()') and the context of the thread. wait then loops until that count is restored to the value that it had upon entry to wait. Method notify removes an arbitrary context from the bag, allowing one of the threads with that context to resume and re-acquire the lock associated with the monitor. notifyAll empties out the entire bag, allowing all threads in the bag to resume.

Exercises

• [Ex. 13.1]: Implement a solution to the bounded buffer problem using Mesa condition variables.
• [Ex. 13.2]: Implement a "try lock" module using Mesa condition variables (see also Exercise 8.3). It should have the following API:
  1. tl = TryLock() # create a try lock
  2. acquire(?tl) # acquire a try lock
  3. tryAcquire(?tl) # attempt to acquire a try lock
  4. release(?tl) # release a try lock
  tryAcquire should not wait. Instead it should return True if the lock was successfully acquired and False if the lock was not available.
• [Ex. 13.3]: Implement a thread-safe GPU allocator by modifying Figure 13.3. There are N GPUs identified by the numbers 1 through N. Method gpuAlloc() returns the identifier of an available GPU, blocking if there is currently no GPU available. Method gpuRelease(gpu) releases the given GPU. It never needs to block.
• [Ex. 13.4]: Bonus problem: Figure 13.4 shows an iterative implementation of the quicksort algorithm, and Figure 13.5 an accompanying test program. Shared variable testqs.arr keeps track of both the array to be sorted and a todo list containing the ranges of the array that need to be sorted (initially the entire array). Re-using as much of this code as you can, implement a parallel version of this. You should not have to change the methods swap, partition, or sortrange for this. Create NWORKERS "worker threads" that should replace the qsort code. Each worker loops until todo is empty and sorts the ranges that it finds until then. The main thread needs to wait until all workers are done.
• [Ex. 13.5]: Cornell's campus features some one-lane bridges. On a one-lane bridge, cars can only go in one direction at a time. Consider northbound and southbound cars wanting to cross a one-lane bridge. The bridge allows arbitrary many cars, as long as they're going in the same direction. Implement a lock that observes this requirement. Write methods OLBlock() to create a new "one lane bridge" lock, nb_enter() that a car must invoke before going northbound on the bridge and nb_leave() that the car must invoke after leaving the bridge. Similarly write sb_enter() and sb_leave() for southbound cars.
• [Ex. 13.6]: Extend the solution to Exercise 13.5 by implementing the requirement that at most n cars are allowed on the bridge. Add n as an argument to OLBlock.

const N = 10

availGPUs = {1..N}

def gpuAlloc() returns gpu:
    gpu = choose(availGPUs)
    availGPUs -= { result }

def gpuRelease(gpu):
    availGPUs |= { gpu }

def Qsort(arr) returns state:
    state = { .arr: arr, .todo: { (0, len(arr) - 1) } }

def swap(p, q):               # swap !p and !q
    !p, !q = !q, !p

def partition(qs, lo, hi) returns pivot:
    pivot = lo
    for i in {lo..hi - 1}:
        if qs->arr[i] <= qs->arr[hi]:
            swap(?qs->arr[pivot], ?qs->arr[i])
            pivot += 1
    swap(?qs->arr[pivot], ?qs->arr[hi])

def sortrange(qs, range):
    let lo, hi = range let pivot = partition(qs, lo, hi):
        if (pivot - 1) > lo:
            qs->todo |= { (lo, pivot - 1) }
        if (pivot + 1) < hi:
            qs->todo |= { (pivot + 1, hi) }

def sort(qs) returns sorted:
    while qs->todo != {}:
        let range = choose(qs->todo):
            qs->todo -= { range }
            sortrange(qs, range)
    sorted = qs->arr

import qsort, bag

const NITEMS = 4

a = [ choose({1..NITEMS}) for i in {1..choose({1..NITEMS})} ]
testqs = qsort.Qsort(a)
sa = qsort.sort(?testqs)
assert all(sa[i - 1] <= sa[i] for i in {1..len(sa)-1}) # sorted?
assert bag.fromList(a) == bag.fromList(sa) # is it a permutation?

Chapter 14. Starvation

from synch import *

def RWlock() returns lock:
    lock = {
            .nreaders: 0, .nwriters: 0, .mutex: Lock(),
            .r_waiting: 0, .r_cleared: 0, .w_waiting: 0, .w_cleared: 0,
            .r_cond: Condition(), .w_cond: Condition()
        }
    
def read_acquire(rw):
    acquire(?rw->mutex)
    if (rw->nwriters == 0) and (rw->w_waiting == 0):
        assert rw->r_waiting == 0
        rw->nreaders += 1
    else:
        rw->r_waiting += 1
        while rw->r_cleared == 0:
            wait(?rw->r_cond, ?rw->mutex)
        rw->r_cleared -= 1
    assert rw->nreaders > 0
    assert rw->nwriters == 0
    release(?rw->mutex)

def read_release(rw):
    acquire(?rw->mutex)
    assert rw->nreaders > 0
    assert rw->nwriters == 0
    rw->nreaders -= 1
    if (rw->nreaders == 0) and (rw->w_waiting > 0):
        rw->w_cleared = rw->nwriters = 1
        rw->w_waiting -= 1
        notify(?rw->w_cond)
    release(?rw->mutex)

def write_acquire(rw):
    acquire(?rw->mutex)
    if (rw->nreaders == 0) and (rw->nwriters == 0):
        assert rw->r_waiting == rw->w_waiting == 0
        rw->nwriters = 1
    else:
        rw->w_waiting += 1
        while rw->w_cleared == 0:
            wait(?rw->w_cond, ?rw->mutex)
        rw->w_cleared = 0
    assert rw->nreaders == 0
    assert rw->nwriters == 1
    release(?rw->mutex)

def write_release(rw):
    acquire(?rw->mutex)
    assert rw->nreaders == 0
    assert rw->nwriters == 1
    if rw->r_waiting > 0:
        rw->nwriters = 0
        rw->r_cleared = rw->nreaders = rw->r_waiting
        rw->r_waiting = 0
        notifyAll(?rw->r_cond)
    elif rw->w_waiting > 0:
        rw->w_waiting -= 1
        rw->w_cleared = 1
        notify(?rw->w_cond)
    else:
        rw->nwriters = 0
    release(?rw->mutex)

A property is a set of traces. If a program has a certain property, that means that the traces that that program allows are a subset of the traces in the property. So far, we have pursued two properties: mutual exclusion and progress. The former is an example of a safety property---it prevents something "bad" from happening, like a reader and writer thread both acquiring a reader/writer lock. The progress property is an example of a liveness property---guaranteeing that something good eventually happens. Informally (and inexactly), progress states that if no threads are in the critical section, then some thread that wants to enter can.

Progress says that some thread eventually can enter, but it does not prevent a scenario such as the following in which some thread never is able to enter. There are three threads repeatedly trying to enter a critical section using a spinlock. Two of the threads successfully keep entering, alternating, but the third thread never gets a turn. This is an example of starvation. With a spinlock, this scenario could even happen with two threads. Initially both threads try to acquire the spinlock. One of the threads is successful and enters. After the thread leaves, it immediately tries to re-enter. This state is identical to the initial state, and there is nothing that prevents the same thread from acquiring the lock yet again.

Peterson's Algorithm (Figure 5.6) does not suffer from starvation, thanks to the turn variable that alternates between 0 and 1 when two threads are contending for the critical section. Ticket locks (Figure 8.3) are also free from starvation.

While spinlocks suffer from starvation, it is a uniform random process and each thread has an equal chance of entering the critical section. Thus the probability of starvation is exponentially vanishing. We shall call such a solution fair (although it does not quite match the usual formal nor vernacular concepts of fairness).

Unfortunately, such is not the case for the reader/writer solution that we presented in Figure 13.1. Consider this scenario: there are two readers and one writer. One reader is in the critical section while the writer is waiting. Now the second reader tries to enter and is able to. The first reader leaves. We are now in a similar situation as the initial state with one reader in the critical section and the writer waiting, but it is not the same reader. Unfortunately for the writer, this scenario can repeat itself indefinitely. So, even if neither reader was in the critical section all of the time, and the second reader arrived well after the writer, the writer never had a chance.

Figure 14.1 and Figure 14.2 present a fair implementation of a read/writer lock. When there is contention between readers and writers, readers and writers end up alternating entering the critical section. While readers can still starve other readers and writers can still starve other writers, readers can no longer starve writers nor vice versa. Other fairness is based on the fairness of scheduling the threads themselves.

Besides the number of readers and writers in the critical section, the implementation keeps track of the number of readers and writers that are waiting to enter the critical section, and the number of readers and writers that are cleared to enter the critical section.

Starting with read_acquire in Figure 14.1, the thread first checks to see if there are no writers in the critical section and there are no writers waiting to enter the critical section (Line 14). If so, the thread can enter the critical section. Otherwise it waits until it is cleared to enter the critical section. read_release checks to see if the thread is the last one to leave the critical section and there are writers waiting to enter. If so, it clears one of the writers.

write_acquire in Figure 14.2 first checks to see that there are no other threads in the critical section. If so, the thread enters the critical section. Otherwise, the thread waits until it is cleared to enter. write_release first checks to see if there are readers waiting to enter. If so, it clears all the readers. If not, it checks to see if there are writers waiting to enter. If so, it clears one of the writers.

Exercises

• [Ex. 14.1]: Write a fair solution to the one-lane bridge problem of Exercise 13.5.

Chapter 15. Deadlock

from synch import Lock, acquire, release

const N = 5

forks = [Lock(),] * N

def diner(which):
    let left, right = (which, (which + 1) % N):
        while choose({ False, True }):
            acquire(?forks[left])
            acquire(?forks[right])
            # dine
            release(?forks[left])
            release(?forks[right])
            # think

for i in {0..N-1}:
    spawn diner(i)

Imagine five philosopers sitting around a table, each with a plate of food in front of them and a fork between every two plates. Each philosopher requires two forks to eat. To start eating, a philosopher first picks up the fork on the left, then the fork on the right. Each philosopher likes to take breaks from eating to think for a while. To do so, the philosopher puts down both forks. Each philosopher repeats this procedure. Dijkstra had them repeating this for ever, but for the purposes of this book, philosophers can---if they wish---leave the table when they are not using any forks.

Figure 15.1 implements the dining philosophers in Harmony, using a thread for each philosopher and a lock for each fork. If you run it, Harmony complains that the execution may not be able to terminate, with all five threads being blocked trying to acquire the lock.

• Do you see what the problem is?
• Does it depend on N, the number of philosophers?

import synch

const N = 5

mutex = synch.Lock()
forks = [False,] * N
conds = [synch.Condition(),] * N

def diner(which):
    let left, right = (which, (which + 1) % N):
        while choose({ False, True }):
            synch.acquire(?mutex)
            while forks[left] or forks[right]:
                if forks[left]:
                    synch.wait(?conds[left], ?mutex)
                if forks[right]:
                    synch.wait(?conds[right], ?mutex)
            assert not (forks[left] or forks[right])
            forks[left] = forks[right] = True
            synch.release(?mutex)
            # dine
            synch.acquire(?mutex)
            forks[left] = forks[right] = False
            synch.notify(?conds[left])
            synch.notify(?conds[right])
            synch.release(?mutex)
            # think

for i in {0..N-1}:
    spawn diner(i)

1. Mutual Exclusion: each resource can only be used by one thread at a time:
2. Hold and Wait: each thread holds resources it already allocated while it waits for other resources that it needs;
3. No Preemption: resources cannot be forcibly taken away from threads that allocated them;
4. Circular Wait: there exists a directed circular chain of threads, each waiting to allocate a resource held by the next.

• No Hold and Wait: a thread must request all resources it is going to need at the same time;
• Preemption: if a thread is denied a request for a resource, it must release all resources that it has already acquired and start over;
• No Circular Wait: define an ordering on all resources and allocate resources in a particular order.

A common mistake is to write the following code instead:

while forks[left]:
    synch.wait(?conds[left], ?mutex)
while forks[right]:
    synch.wait(?conds[right], ?mutex)

• Can you see why this does not work? What can go wrong?
• Run it through Harmony in case you are not sure!

The No Circular Waiting approach is to prevent a cycle from forming, with each thread waiting for the next thread on the cycle. We can do this by establishing an ordering among the resources (in this case the forks) and, when needing more than one resource, always acquiring them in order. In the case of the philosopers, they could prevent deadlock by always picking up the lower numbered fork before the higher numbered fork, like so:

if left < right:
    synch.acquire(?forks[left])
    synch.acquire(?forks[right])
else:
    synch.acquire(?forks[right])
    synch.acquire(?forks[left])

synch.acquire(?forks[min(left, right)])
synch.acquire(?forks[max(left, right)])

This avoidance technique can be generalized using something called the Banker's Algorithm [14], but it is outside the scope of this book. The problem with these kinds of schemes is that one needs to know ahead of time the set of threads and what the maximum number of resources is that each thread wants to allocate, making them usually impractical.

from synch import *

const N = 5

forks = [Lock(),] * N
sema = Semaphore(N - 1)     # can be procured up to N-1 times

def diner(which):
    let left, right = (which, (which + 1) % N):
        while choose({ False, True }):
            P(?sema)                # procure counting semaphore
            acquire(?forks[left])
            acquire(?forks[right])
            # dine
            release(?forks[left])
            release(?forks[right])
            V(?sema)                # vacate counting semaphore
            # think

for i in {0..N-1}:
    spawn diner(i)

Exercises

• [Ex. 15.1]: The solution in Figure 15.2 can be simplified by, instead of having a condition variable per fork, having a condition variable per diner. It uses the same number of condition variables, but you will not need to have if statements nested inside the while loop waiting for the forks. See if you can figure it out.
• [Ex. 15.2]: Figure 15.4 shows an implementation of a bank with various accounts and transfers between those accounts. Unfortunately, running the test reveals that it sometimes leaves unterminated threads. Can you fix the problem?
• [Ex. 15.3]: Add a method total() to the solution of the previous question that computes the total over all balances. It needs to obtain a lock on all accounts. Make sure that it cannot cause deadlock.
• [Ex. 15.4]: Add an invariant that checks that the total of the balances never changes. Note that this predicate only holds if none of the locks are held. You can use the held(lk) method in the synch module to check if a lock is held or not.

from synch import Lock, acquire, release

const MAX_BALANCE = 2
const N_ACCOUNTS = 2
const N_THREADS = 2

accounts = [ { .lock: Lock(), .balance: choose({0..MAX_BALANCE})}
                            for i in {1..N_ACCOUNTS} ]

def transfer(a1, a2, amount) returns success:
    acquire(?accounts[a1].lock)
    if amount <= accounts[a1].balance:
        accounts[a1].balance -= amount 
        acquire(?accounts[a2].lock)
        accounts[a2].balance += amount 
        release(?accounts[a2].lock)
        success = True
    else:
        success = False
    release(?accounts[a1].lock)

def thread():
    let a1 = choose({0..N_ACCOUNTS-1})
    let a2 = choose({0..N_ACCOUNTS-1} - { a1 }):
        transfer(a1, a2, choose({1..MAX_BALANCE}))

for i in {1..N_THREADS}:
    spawn thread()

Chapter 16. Actors and Message Passing

Figure 16.1. Depiction of three actors. The producer does not receive messages.

import synch

const NCLIENTS = 3

server_queue = synch.Queue()

def server():
    var counter = 0
    while True:
        let q = synch.get(?server_queue):   # await request
            synch.put(q, counter)           # send response
            counter += 1

def client(client_queue):
    synch.put(?server_queue, client_queue)      # send request
    let response = synch.get(client_queue):     # await response
        print(response)

spawn eternal server()

alice_queue = synch.Queue()
spawn client(?alice_queue)
bob_queue = synch.Queue()
spawn client(?bob_queue)
charlie_queue = synch.Queue()
spawn client(?charlie_queue)

The actor synchronization model is popular in a variety of programming languages, including Erlang and Scala. Actor support is also available through popular libraries such as Akka, which is available for various programming languages. In Python, Java, and C/C++, actors can be easily emulated using threads and synchronized queues (aka blocking queues) for messaging. Each thread would have one such queue for receiving messages. Dequeuing from an empty synchronized queue blocks the thread until another thread enqueues a message on the queue.

The synch library supports a synchronized message queue, similar to the Queue object in Python. Its interface is as follows:

• Queue() returns an empty queue;
• put(q, item) adds item to the queue pointed to by q;
• get(q) waits for and returns an item on the queue pointed to by q.

Figure 16.2 illustrates the actor approach. There are three client threads that each want to be assigned a unique identifier from the set { 0, 1, 2 }. Normally one would use a shared 0-initialized counter and a lock. Each client would acquire the lock, get the value of the counter and increment it, and release the lock. Instead, in the actor approach the counter is managed by a separate server thread. The server never terminates, so it is spawned with the keyword eternal to suppress non-terminating state warnings. Each client sends a request to the server, consisting in this case of simply the queue to which the server must send the response. The server maintains a local, zero-initialized counter variable. Upon receiving a request, it returns a response with the value of the counter and increments the counter. No lock is required.

This illustration is an example of the client/server model. Here a single actor implements some service, and clients send request messages and receive response messages. The model is particularly popular in distributed systems, where each actor runs on a separate machine and the queues are message channels. For example, the server can be a web server, and its clients are web browsers.

Exercises

• [Ex. 16.1]: Actors and message queues are good for building pipelines. Develop a pipeline that computes Mersenne primes (primes that are one less than a power of two). Write four actors:
  1. an actor that generates a sequence of integers 1 through N;
  2. an actor that receives integers and forwards only those that are prime;
  3. an actor that receives integers and forwards only those that are one less than a power of two;
  4. an actor that receives integers but otherwise ignores them.
  Configure two versions of the pipeline, one that first checks if a number is prime and then if it is one less than a power of two, the other in the opposite order. Which do you think is better?

Chapter 17. Barrier Synchronization

import barrier

const NTHREADS = 3
const NROUNDS = 4

barr = barrier.Barrier(NTHREADS)
round = [0,] * NTHREADS
in_b = [False,] * NTHREADS

invariant all((in_b[i] and in_b[j]) => (round[i] == round[j])
        for i in { 0 .. NTHREADS - 1 } for j in { 0 .. NTHREADS - 1 })

def thread(self):
    for r in {0..NROUNDS-1}:
        round[self] += 1
        barrier.bwait(?barr)
        in_b[self] = True
        pass
        in_b[self] = False

for i in {0..NTHREADS-1}:
    spawn thread(i)

from synch import *

def Barrier(required) returns barrier:
    barrier = {
        .mutex: Lock(), .cond: Condition(),
        .required: required, .left: required, .cycle: 0
    }

def bwait(b):
    acquire(?b->mutex)
    b->left -= 1
    if b->left == 0:
        b->cycle = (b->cycle + 1) % 2
        b->left = b->required
        notifyAll(?b->cond)
    else:
        let cycle = b->cycle:
            while b->cycle == cycle:
                wait(?b->cond, ?b->mutex)
    release(?b->mutex)

import barrier

const NTHREADS = 3
const NROUNDS = 4

barr = barrier.Barrier(NTHREADS)
round = [0,] * NTHREADS
phase = 0

def thread(self):
    for r in {0..NROUNDS-1}:
        round[self] += 1
        if self == 0:                # coordinator prepares
            phase += 1
        barrier.bwait(?barr)         # enter parallel work
        assert round[self] == phase
        barrier.bwait(?barr)         # exit parallel work

for i in {0..NTHREADS-1}:
    spawn thread(i)

from barrier import *

const N = 5     # size of list to be sorted

list = [ choose({ 1 .. N }) for i in { 1 .. N } ]

finally all(list[i-1] <= list[i] for i in { 1 .. N - 1 })

const NTHREADS = N / 2
bar = Barrier(NTHREADS)
count = 0                   # to detect termination

def fetch_and_increment(p): # atomic increment
    atomically !p += 1

def sorter(i):
    var sorted = False
    var oldcount = 0
    while not sorted:
        # Even phase
        if list[i - 1] > list[i]:
            list[i - 1], list[i] = list[i], list[i - 1]
            fetch_and_increment(?count)

        bwait(?bar)

        # Odd phase
        if (i < (N - 1)) and (list[i] > list[i + 1]):
            list[i], list[i + 1] = list[i + 1], list[i]
            fetch_and_increment(?count)

        bwait(?bar)

        # Sorted if nobody swapped anything
        sorted = count == oldcount
        oldcount = count

        bwait(?bar)

for k in { 0 .. NTHREADS - 1 }:
    spawn sorter((2*k) + 1)

Barrier synchronization is a problem that comes up in high-performance parallel computing. The Harmony model checker uses it internally to synchronize its worker threads. A barrier is almost the opposite of a critical section: the intention is to get a group of threads to run some code at the same time, instead of having them execute it one at a time. More precisely, with barrier synchronization, the threads execute in rounds. Between each round, there is a so-called barrier where threads wait until all threads have completed the previous round and reached the barrier---before they start the next round. For example, in an iterative matrix algorithm, the matrix may be cut up into fragments. During a round, the threads run concurrently, one for each fragment. The next round is not allowed to start until all threads have completed processing their fragment.

A barrier is used as follows:

• b = Barrier(n): initialize a barrier b for a collection of n threads;
• bwait(?b): wait until all threads have reached the barrier

When implementing a barrier, a complication to worry about is that a barrier can be used over and over again. If this were not the case, then a solution based on a lock, a condition variable, and a counter initialized to the number of threads could be used. The threads would decrement the counter and wait on the condition variable until the counter reaches 0.

Figure 17.2 shows how one might implement a reusable barrier. Besides a counter .left that counts how many threads still have to reach the barrier, it uses a counter .cycle that is incremented after each use of the barrier---to deal with the complication above. The last thread that reaches the barrier restores .left to the number of threads (.required) and increments the cycle counter. The other threads are waiting for the cycle counter to be incremented. The cycle counter is allowed to wrap around---in fact, a single bit suffices for the counter.

A common design pattern with barriers in parallel programs, demonstrated in Figure 17.3, is to use the barrier twice in each round. Before a round starts, one of the threads---let's call it the coordinator---sets up the work that needs to be done while the other threads wait. Then all threads do the work and go on until they reach a second barrier. The second barrier is used so the coordinator can wait for all threads to be done before setting up the work for the next round.

Figure 17.4 shows an implementation of a parallel sorting algorithm based on bubblesort. The threads (one for every two elements) go through three phases. In the first phase, the threads swap entries 0 and 1, 2 and 3, ... as needed. In the second phase, they swap entries 1 and 2, 3 and 4, ... as needed. Finally, they check if any elements were swapped. If so, they repeat the phases.

Chapter 18. Advanced Barrier Synchronization

from synch import *

def RollerCoaster(nseats): result = {
    .mutex: Lock(), .nseats: nseats, .entered: 0, .left: nseats,
    .empty: Condition(), .full: Condition()
}

def enter(b):
    acquire(?b->mutex)
    while b->entered == b->nseats:  # wait for car to empty out
        wait(?b->empty, ?b->mutex)
    b->entered += 1
    if b->entered != b->nseats:     # wait for car to fill up
        while b->entered < b->nseats:
            wait(?b->full, ?b->mutex)
    else:                           # car is ready to go
        b->left = 0
        notifyAll(?b->full)         # wake up others waiting in car
    release(?b->mutex)

def exit(b):
    acquire(?b->mutex)
    b->left += 1
    if b->left == b->nseats:    # car is empty
        b->entered = 0
        notifyAll(?b->empty)    # wake up riders wanting to go
    release(?b->mutex)

Given is a roller coaster with a single car. Safety precautions require the following: each ride requires that all seats are filled. That is, partially filled cars are not allowed to ride. After a ride, the car must completely empty out before new riders are allowed to enter the car. Write code to model this.

Entering a section consists of two phases. A thread should first wait until any threads that already entered the section have exited. Then, it should wait until conditions are met, in this case until there are enough users (threads) to fill up the car. Not following this recipe might lead to the following problem: as the car is emptying out, new riders may try to enter the car because it is no longer full.

In the first phase, a thread needs to be able to detect that all threads that have entered previously have now left. In the second phase, a thread should check to see if it is the last thread to enter the second phase. If not, it should wait. If so, it should notify the ones that are waiting. When leaving the section, a thread should check if it is the last thread to leave. If so, it should notify the threads that are waiting in the first phase.

Figure 18.1 presents a solution. Method RollerCoaster(nseats) returns the initial state of a roller coaster with nseats seats to a car. Method enter(b) takes a pointer b to a roller coaster variable. Following the recipe sketched above, a thread that calls enter(b) blocks until

1. all previous riders have left the car, and
2. the car has filled up again.

An important part of solving the problem is figuring out what state must be kept about the resource. The code keeps track of the number of threads that have entered the car (including the threads that have not yet entered the section and the threads that have left the section) and the number of threads that have left the section. The first number is only reset after that last thread has left the section. The second number is reset after the last thread has entered the car.

In this example exam question there was just a single instance of a resource, and all users were in some sense the same in that they are interchangeable. In the problem set below we include a problem in which there are multiple resources and a problem in which not all users are the same. Nonetheless, a similar approach can be used as illustrated above.

Exercises

• [Ex. 18.1]: Imagine a pool hall with a certain number of tables. A table is full from the time there are two players until both players have left. When someone arrives, they can join a table that is not full, preferably one that has a player ready to start playing. Implement a simulation of this.
• [Ex. 18.2]: The following is a classical problem. There is a ferry that is used by both Linux hackers and Microsoft employees (serfs) to cross a river. The ferry can hold exactly four people; it won’t leave the shore with more or fewer. To guarantee the safety of the passengers, it is not permissible to put one hacker in the boat with three serfs, or to put one serf with three hackers. Any other combination is safe. Implement a simulation of this.

Chapter 19. Example: A Concurrent File Service

This chapter presents a concurrent file service to illustrate many of the techniques we have discussed inside a single example. We will cover the specification of such a service as well as that of a disk, and show how the specification can be implemented on top of the disk. The file service implementation will use a collection of worker threads synchronizing using both ordinary locks and reader/writer locks. Clients of the file service implementation (threads themselves) use blocking synchronized queues to communicate with the workers. The example will also illustrate modular model checking, as the disk, the locks, and the queues are only specified.

from alloc import malloc

def File(n_files) returns fs:
    fs = malloc([ [], ] * n_files)

def read(fs, ino, offset) returns result:
    atomically result = (!fs)[ino][offset] if 0 <= offset < len (!fs)[ino] else None

def write(fs, ino, offset, data) returns result:
    atomically:
        let n = len (!fs)[ino]:
            if 0 <= offset <= n:
                (!fs)[ino][offset] = data
            else:
                (!fs)[ino] += ([ None, ] * (offset - n)) + [data,]
        result = "ok"
            

def delete(fs, ino) returns result:
    atomically:
        (!fs)[ino] = []
        result = "ok"

Figure 19.1 shows the file system interface. Just like in Unix-like file systems, you have to specify the (maximum) number of files when you initialize the file system. File(n) returns a handle that must be passed to file operations, where n is the maximum number of files. For our example, we have only included three operations on files. read(fs, ino, offset) returns the block of file ino at the given offset, or None if nothing has been stored at that offset. write(fs, ino, offset, data) stores data at the given offset in file ino. If needed, the file is grown to include the given offset. "Holes" (unwritten blocks) are plugged with None values. delete(fs, ino) deletes a file.

from file import *

const N_FILES = 2
const MAX_FILE_SIZE = 2

const N_READ = 1
const N_WRITE = 1
const N_DELETE = 1

system = File(N_FILES)

def read_test(i):
    let ino = choose { 0 .. N_FILES - 1 }
    let offset = choose { 0 .. MAX_FILE_SIZE - 1 }:
        print(i, "read", ino, offset)
        let data = read(system, ino, offset):
            print(i, "read done", ino, offset, data)

def write_test(i):
    let ino = choose { 0 .. N_FILES - 1 }
    let offset = choose { 0 .. MAX_FILE_SIZE - 1 }:
        print(i, "write", ino, offset)
        write(system, ino, offset, i)
        print(i, "write done", ino, offset)

def delete_test(i):
    let ino = choose { 0 .. N_FILES - 1 }:
        print(i, "delete", ino)
        delete(system, ino)
        print(i, "delete done", ino)

for i in { 1 .. N_READ }:
    spawn read_test(i)
for i in { 1 .. N_WRITE }:
    spawn write_test(i)
for i in { 1 .. N_DELETE }:
    spawn delete_test(i)

from alloc import malloc

def new(n_blocks) returns disk:
    disk = malloc([ None, ] * n_blocks)

def getsize(disk) returns size:
    size = len !disk

def read(disk, bno) returns contents:
    contents = (!disk)[bno]

def write(disk, bno, block):
    (!disk)[bno] = block

For the implementation, we will use a simplified Unix file system. In a Unix file system, the disk is subdivided into three parts (Figure 19.4(a)):

1. The superblock, at offset 0.
2. An array of fixed-sized inodes, stored in a range of m disk blocks starting at block 1. An inode number indexes into this array. Each inode maintains some information about a file, including the size and where to find the data.
3. The remaining blocks, starting at offset 1 + m , which will either store data, metadata, or are free. Metadata blocks contain a list of block numbers.

Figure 19.4. The file system data structure: (a) disk layout (1 superblock, n blocks, m inode blocks, 4 inodes per block); (b) inode for a file with 3 data blocks

Free blocks are maintained in a simple linked list. The superblock contains the block number of the first free block, which every block on the free list contains the block number of the next free block, or None for the end of the list. (In a more realistic Unix file system, each block on the free list would maintain pointers to additional free blocks.)

Note that the entire file system data structure is essentially a tree of blocks, with the superblock acting as the root of the tree. The superblock points to the free list and the inode blocks. The inode blocks point to the blocks that are allocated. An invariant of the data structure is that all blocks are in the tree and each block (except for the superblock) is pointed to exactly once. The invariant may not hold while the data structure is being updated. For example, temporarily a block may be both on the free list but also be part of an inode, or a block may not be referenced at all.

Figure 19.5 shows the modules that the file system implementation will use and some constants. The implementation uses the actor model (Chapter 16)---the synch module provides blocking multi-reader/multi-writer queues that the actors will use for messaging. The file server itself is implemented as a multithreaded actor. The threads synchronize using a plain lock for the free list and reader/writer locks for each of the inode blocks. N_BLOCKS specifies the size of the disk to be used in blocks. INODES_PER_BLOCK specifies how many inodes fit in an inode block. INDIR_PER_BLOCK specifies how many block numbers fit in a metadata block. Note that the maximum file size is this simplified file system is 1 + INDIR_PER_BLOCK blocks. In a more realistic Unix file system, indirect blocks can point to other indirect blocks, allowing for much larger files. N_WORKERS specifies the number of worker threads or actors.

from synch import *             # shared queue for file server and lock for superblock
from rwlock import *            # read/write locks for inode blocks
from alloc import *             # malloc/free
from list import subseq         # list slicing
import disk                     # reading and writing blocks

const N_BLOCKS = 10             # total number of disk blocks
const INODES_PER_BLOCK = 2      # number of inodes that fit in a block
const INDIR_PER_BLOCK  = 4      # number of block pointers per block

def File(n_files) returns req_q:
    req_q = malloc(Queue())
    let n_inode_blocks = (n_files + (INODES_PER_BLOCK - 1)) / INODES_PER_BLOCK
    let n_workers = 2
    let d = disk.new(N_BLOCKS):
        # Initialize the file system on disk
        fs_init(d, n_inode_blocks)

        # Allocate the in-memory shared state of the file server
        let fs_state = malloc({ .next: None,
                .disk: d, .req_q: req_q, .free_lock: Lock(),
                .n_inode_blocks: n_inode_blocks,
                .ib_locks: [ RWlock(), ] * n_inode_blocks }):

            # Start worker threads to handle client requests
            for i in { 1 .. n_workers }:
                spawn eternal fs_worker(fs_state, i)

def read(req_q, ino, offset) returns result:
    let res_q = malloc(Queue()):
        put(req_q, { .type: "read", .ino: ino, .offset: offset, .q: res_q })
        result = get(res_q)
        free(res_q)

def write(req_q, ino, offset, data) returns result:
    let res_q = malloc(Queue()):
        put(req_q, { .type: "write", .ino: ino, .offset: offset, .data: data, .q: res_q })
        result = get(res_q)
        free(res_q)

def delete(req_q, ino) returns result:
    let res_q = malloc(Queue()):
        put(req_q, { .type: "delete", .ino: ino, .q: res_q })
        result = get(res_q)
        free(res_q)

• .disk: points to the disk object;
• .req_q: the shared queue on which requests from clients arrive;
• .free_lock: a lock on the free list;
• .ib_locks: an array of reader/writer locks, one for each inode block.

The remaining interfaces simply put a request on the request queue and wait for a response on another queue res_q that is allocated just for this purpose. Note that the request queue has concurrent producers (the clients) and concurrent consumers (the worker threads). The response queues are single use only and have a single producer (a worker thread) and a single consumer (the client).