Problem Overview

As with the previous chapter, I will be showing the development process and the dead ends we can run into. That way, if you encounter these issues in your own specifications, you’ll have some resources to address them.

Part One: Basics

I automatically extend TLC, Sequences, and Integers because they’re almost always useful. My assumption is that we’ll have a single reducer process and multiple worker processes, so I make them constants. Workers will be a model set, Reducer a model constant. That way we can tweak the number of workers in the spec. I also add NULL because it comes in handy all of the time. I’ll take it out if the spec doesn’t need it, but my base prior is that it will.

What input are we putting in? I think we can assume that each “book” can be represented as a single number (its wordcount), the list of all “books” represented by a sequence of numbers. I want a sequence instead of a set because multiple books might have the same wordcount.

As I write this, though, I immediately see a question: We’re assuming there are four books. What if there is one, or zero? Will the algorithm still work? I decide to leave that out until later. Checking a wider range may be safer, but it will also take longer to model check, and we want to get quick feedback at the start. That way we can remove the obvious bugs before looking for the subtler ones. Later, as we decide to explore a wider state space, we’ll replace the hard-coded numbers with CONSTANTS.

Uncheck Deadlock, we don’t need it for now. Run the model. If everything is set up properly, you should see a model failure. This is to be expected, as we haven’t actually implemented anything.

The reducer should wait until a worker has changed its result to something other than NULL and then add the new value to its own total. While writing what this looks like, it occurs to me that we don’t want the reducer reading a non-null result from the same worker twice. We could prevent this by setting its result back to NULL. But that doesn’t seem right to me: if we set a result back to NULL, there’s no way of knowing whether that worker had finished and been consumed, or if it’s still working. A better idea would be for the reducer to privately track which worker results it has consumed.

Try this again and confirm it fails. But it should fail with final = 10. Inspecting the error trace, you should see that final only increments after a worker has run. That suggests to us ReduceResult is successfully summing and retrieving the worker values.

Better. We have to put the await in a separate step from the while, because while loops must come directly after a label. One thing I notice is that result[self] will never be updated if we never send any items to this worker. That’s not a problem just yet but will be a problem if we have fewer items than nodes. It’s something to keep an eye on.

We now have everything except the Schedule step . This requires us to divvy up the inputs among all of the workers. Up above I decided that we’d assign each one based on their order. However, there’s no predefined order to the workers. By using PT!OrderSet on them, we can arbitrarily pick one as the first worker, one as the second, etc. What’s important, however, is that TLC will not try to break the spec by reordering the workers. This means that our spec works only if we assume that ordering of workers does not matter. Whenever we write specs, we should carefully keep track of our assumptions and recheck them regularly to confirm they’re still safe.

The code is a little complex, but all we are doing is assigning a number to each worker and cyclically assigning the items to each worker. We run into some friction because sequences have domain 1..n, while x % n has range 0..(n-1). We have to subtract 1 from our offsets to keep them in sync.

Part Two: Liveness

Add this as a temporal property and rerun the model. You should see it fail, eventually reaching a stuttering step. If the workers never complete, then we will never finish reducing.

We could solve this by making the workers fair processes. But they aren’t. Workers crash all the time in the field, and MapReduce should assume they can fail. We want our algorithm to work not only in the happy path, but also be fault tolerant. This makes up part two of this example: ensuring that MapReduce continues to work if some (but not all) of the workers stutter.

Now we are guaranteed that at least one worker will finish its assigned queue. Rerun the model and it should still fail, but it may fail after more steps complete. While one worker completes, the rest may not, and the reducer still waits forever.

The other thing it can move a failing worker’s queue to a valid worker. How does it know if a worker is failing? Again, this is abstraction specific, but for now we can think of it like this: if a worker result is consumed, then it definitely didn’t fail. So any worker we haven’t consumed might have failed. We’ll pick one of those.

* Reduce and * Reassign are ‘real’ comments in the spec, not just teaching annotations. I could have made them labels. This would make the concurrency more fine-grained at the cost of a slower model. I don’t think they’re necessary yet. I may choose to add them later, once I have the coarser concurrency model working properly. So, what goes in the * REASSIGN LOGIC block? Here’s how I thought through it.

First of all, this is handling a failed worker, right? But the while loop keeps us going until we’ve consumed all of the workers. If a worker truly crashed, we’d be waiting on it forever. If the worker didn’t crash and later gets a result, it will mess up our calculations. Either way, we can prevent the problem by declaring the failed worker consumed. We might also want to record that we thought it failed, but that’s not (yet) useful to us. I’ll burn that bridge when we get to it.

Add this and rerun the model. It will still fail, of course, but it should fail because of the assert statement, not the Liveness property. While we still can’t guarantee it gets the correct answer, at least we can guarantee it gets some answer.

Now for part two: What does it actually mean to “reassign”? Ideally, that we dump anything that was in from_worker‘s queue into to_worker’s queue. But we can’t get that ’anything’ from queue! It represents the data we sent directly between the reducer and the fair worker, so using it would violate our abstraction. Also, we’re destructively updating it, so we can’t guarantee it’s correct data.

I’m tempted to merge consumed and assignments into a single structure. But since I’m reassigning to the entire queue at once in Schedule, nesting it in a structure would make that mutation considerably more complicated. I also update assignments in the Reassign block. It doesn’t yet change the current behavior of the spec, but it is more comprehensive.

Change return to goto WaitForQueue, recompile, and rerun. This still fails, because we didn’t null out the relative value. Add result[to_worker] := NULL; to the with statement and try again. This fails because the reducer can reassign after the worker finishes its queue but before it can run Result. Every small tweak we make leads to a different concurrency error.

Let’s take a step back. While blind guessing can work for tests or typecheckers, it won’t help with specification. We need to think about what we’re doing. Our problem here is this: the reducer has no way of knowing whether a given result includes every item assigned to the worker. We originally knew that because the worker would only write a result once it had completed the entire queue. But we can no longer rely on that. What can we do instead?

This, finally, is successful (32,238 states)! Try adding a third or even fourth worker and confirm that the spec is still successful. We now have a working, fault-tolerant version of MapReduce.

Then again, we assumed that the reducer would only reassign from unfair workers and to fair ones. How could it know, though? It’s not like it can tell which workers are stuttering from the outside. In the next section, we will account for exactly that.

Part Three: Statuses

In theory, we don’t have a way of distinguishing failing nodes from passing ones. In practice, we can do things that give us a reasonable amount of confidence. For example, we can ping all the servers every N seconds and assume that the ones that don’t answer in time are failing. Of course, the node might not be failing, and it could be that our reducer is acting up.

If you now run this, you will see … the model never stops, ever. It turns out we accidentally created an unbounded model. TLC was able to find an infinite number of unique states. This suggests something is growing without limit, such as always being able to increment a number. That, obviously, is an error, too.

In the last chapter we mentioned that it’s good practice to define a TypeInvariant to constrain the values of our variables. We didn’t do that here and it came back to bite us. One of the type invariants of our system is that there’s only a fixed number of items, so no worker should have more than that number of items enqueued. If we had written a type invariant, TLC would have failed that state with an error instead of running forever. Let’s define it now.

We could be more elaborate, but the important thing here is that we check all of the queues have fewer than ItemCount items in them. We could also check the types of the private variables, in which case TypeInvariant would have to go after the translation.

If you set INVARIANT TypeInvariant and rerun the model, you should get a definite failure. What happens is we can reassign away from a worker, mark it consumed, and then reassign back to the worker and unconsume it. This effectively duplicates its queue, leading to an incorrect count.

For the actual fix, I considered “wiping” the old assignments from the internal queue. This might work, but we track when workers are done by the number of assignments they complete. Wiping the assignments might conflict with that logic. A simpler solution is to recognize that “consumed” is tracking two separate responsibilities: a node that’s finished working, and a node that we consider bad. There’s actually three states, though: “active,” “inactive,” and “broken.” We’re done when there are no “active” nodes. When reassigning, we only reassign to “active” or “inactive” nodes. Let’s make those changes.

A quick back-of-the-envelope suggests that the fixed model is going to have a very high number of states. To sanity-check my fixes, I decide to run them on a smaller model first. I clone the first one and set Workers <- {w1, w2}, ItemCount <- 2, ItemRange <- 0..2. One fewer worker and half the items assigned. I use the same invariants and properties and run the new model, getting a pass with 2,664 states.

Exercise

But I don't want to leave such a large failure mode in the spec, so we should try to fix it. Try running the model with Workers <- {w1, w2, w3}, ItemCount <- 2, PROPERTY Liveness. Surprisingly, it passes! This is because Liveness only checks we reach the correct answer, not that the reducer terminates with the correct answer. It can still correctly sum up all of the values but stay trapped in the ReduceResult loop, never reaching Finish.

Add PROPERTY ReducerTerminates and rerun. You should see it fail.

I’ll leave this last change to you as an exercise: How can you modify our MapReduce algorithm to satisfy ReducerTerminates? Make sure you also ensure that Liveness and TypeInvariant remain satisfied, and that your fix works for both ItemCount <- 2 and ItemCount <- 4. You might want to make two separate models so that switching between the two is easier. Good luck!

Summary

We fully specified an example of MapReduce. Obviously not every single aspect is covered, but we’ve provided enough to understand how we deal with partial failures. While our model was barely over 100 lines of PlusCal, it was able to find complex errors and liveness issues.