Difficulty: Advanced | Prerequisites: Parts 1 and 2 study notes, shell scripting basics
Workers can die at any time. The Manager detects death via missed heartbeats or connection errors, marks the Worker dead without removing it, and reassigns its tasks. The project also requires an init script (bin/mapreduce) to start, stop, restart, and check the status of the server. Good design means shared networking code, proper logging, no busy-waiting, and understanding how the test suite uses mocking.
Fault tolerance
The system's ability to keep operating correctly when a Worker process crashes. The Manager detects the failure and reassigns the dead Worker's task to another Worker.
Busy-waiting
A loop that repeatedly checks a condition without sleeping or blocking, consuming 100% of a CPU core. The autograder's restricted environment will time out if your code busy-waits. Use blocking socket operations and threading events instead.
Mocking
The test suite uses Python's Mock class to simulate the other side of the system. Worker tests use a fake Manager; Manager tests use fake Workers. The mock replaces socket.recv() to return pre-defined messages. When the mock's message list is exhausted, a StopIteration exception is raised.
Init script
A shell script at bin/mapreduce that manages the MapReduce server as a background service. It supports four commands: start, stop, status, restart.
ThreadSafeOrderedDict
A provided utility class in mapreduce.utils that wraps an OrderedDict with locking. Use it for any dictionary accessed by multiple threads to avoid concurrency bugs.
Each Worker sends a UDP heartbeat to the Manager every 2 seconds after receiving register_ack. The heartbeat message contains message_type, worker_host, and worker_port.
The Manager tracks the time of each Worker's last heartbeat. If more than 5 consecutive heartbeats are missed (10 seconds without contact), the Manager marks that Worker as dead.
There is a second detection path: if the Manager tries to send a TCP message to a Worker and gets a ConnectionRefusedError, it immediately marks that Worker as dead, regardless of the heartbeat count.
What happens when a Worker dies mid-task:
The Manager marks the Worker as dead (does not remove it from data structures)
The unfinished task goes back into the pool of tasks to assign
The Manager finishes assigning all normal (not-yet-assigned) tasks in the current stage first
Then it reassigns tasks from dead Workers to free Workers
If all Workers are dead, the Manager waits until a new Worker registers and resumes
Edge case: Worker dies and re-registers. A Worker may crash, restart, and send a new register message before the Manager detects the death. All tasks from the running job must still be completed. The Manager's handling of task assignment order when a Worker re-registers is left to the implementer.
Concurrency principle: avoid duplication. Do not send the same task to two different Workers at the same time. Only reassign a task once you are certain the previous Worker is dead.
The init script is a Bash script that manages the MapReduce server as a background service. It supports four subcommands.
start: Exit 1 if the Manager or any Worker is already running (check with pgrep). Otherwise, create var/log/, remove old log files, then launch the Manager and two Workers as background processes (with &), writing logs to var/log/manager.log, var/log/worker-6001.log, and var/log/worker-6002.log. Sleep 2 seconds between the Manager and the Workers so the Manager has time to bind its port.
stop: Send a shutdown message using mapreduce-submit --shutdown. Sleep 2 seconds to let the shutdown propagate. Then check whether the Manager and Workers are still running with pgrep. If they are, kill them with pkill. Use "|| true" after pkill to suppress non-zero exit codes.
status: Print "manager running" or "manager not running" and "workers running" or "workers not running". Exit 0 if both the Manager and at least one Worker are running; exit 1 otherwise.
restart: Run the stop logic, then the start logic.
The spec defines the protocol but leaves the implementation to you. A few patterns from the instructor solution are worth knowing:
Shared networking code (mapreduce/utils/). Write reusable functions: tcp_server (listens on a socket, calls a handler callback for each message), udp_server (same for UDP), tcp_client (sends a JSON message over a new TCP connection), udp_client (sends a JSON message over UDP). Both the Manager and Worker import these.
Manager abstractions. Consider a Job class to manage pending tasks, current stage, and task_id assignment. Consider a RemoteWorker class to track each Worker's host, port, state (ready/busy/dead), and current task. Consider a collection class to manage the pool of RemoteWorkers.
Thread safety. Use ThreadSafeOrderedDict from mapreduce.utils for any dictionary shared across threads. Use threading.Event or similar for signalling between threads (e.g. shutdown).
Logging. Use Python's logging module (LOGGER.info, LOGGER.debug) rather than print(). The logging module shows which process and thread produced each message, which is essential for debugging a multi-process system.
repr() methods. Implement repr on your custom classes so log output is human-readable. For example, a MapTask should print its input_files, executable, and output_directory rather than a memory address.
Paths. Use pathlib throughout. Never assume paths are absolute or relative. Use the provided PathJSONEncoder class when serialising pathlib.Path objects to JSON.
Running tests. Use pytest -vvsx --log-cli-level=INFO to run with verbose output and stop at the first failure. Run individual test files to isolate problems, e.g. pytest -vvsx tests/test_manager_02.py.
Understanding the mocks. Manager tests mock the Workers: when your Manager sends a TCP message, the test intercepts it and feeds back pre-defined responses. Worker tests mock the Manager in the same way. When the mock runs out of messages, socket.recv() raises StopIteration. Make sure your code does not call recv() after the final shutdown message.
Testing busy-waiting. Run "time mapreduce-manager" without Workers, wait a few seconds, then Ctrl-C. If "user" time is close to "real" time, your code is busy-waiting. Good implementations show user time near zero because they block on socket operations.
Testing fault tolerance manually. Start the server, submit a slow job (using wc_map_slow.sh / wc_reduce_slow.sh), then kill one Worker with "pgrep -f mapreduce-worker | head -n1 | xargs kill". Check the logs to confirm the Manager reassigned the dead Worker's task.
Code style. All code must pass pycodestyle, pydocstyle, and pylint with no errors or warnings. Use context managers for sockets (with socket.socket(...) as sock). Use contextlib.ExitStack for opening multiple files.
Students sometimes try to reassign a dead Worker's task immediately, before finishing the normal task queue. The spec says: assign all normal tasks in the current stage first, then reassign tasks from dead Workers.
Students forget to handle ConnectionRefusedError when sending TCP messages. If the Manager cannot reach a Worker, it must mark that Worker as dead instantly and treat the task as unassigned.
Students confuse the heartbeat thread's behaviour: the Worker's heartbeat thread runs independently and does not need to check for shutdown before each send. The heartbeat thread can simply exit when the main thread closes (use daemon threads or a shutdown event).
Students sometimes run the init script without checking for shell script pitfalls. Remember: use set -Eeuxo pipefail (or at minimum set -e), quote your variables, and use the correct pgrep flags for your OS (macOS uses -lf, Linux/WSL uses -af).
Heartbeat-based failure detection with a configurable threshold (5 missed pings) is a standard distributed systems technique. Know the trade-off: a shorter timeout detects failures faster but increases false positives.
The pattern of mark-dead-but-do-not-remove is specific to Python's dictionary iteration constraint, but the broader lesson (do not mutate data structures while iterating) is universal.
Busy-waiting is a common performance bug in concurrent programs. The distinction between blocking I/O (good) and spinning in a loop (bad) is fundamental.
The init script covers process management basics: starting daemons in the background, checking process existence with pgrep, and graceful vs forced shutdown.
Q: Worker A is assigned map task 3. Worker A dies. Worker B finishes its current task and becomes free. Describe the Manager's behaviour.
A: The Manager detects Worker A's death (via missed heartbeats or ConnectionRefusedError). It marks Worker A as dead but does not remove it. If there are still normal unassigned tasks in the current stage, the Manager assigns those to Worker B first. Once all normal tasks have been assigned, the Manager reassigns task 3 (from dead Worker A) to Worker B.
Q: What does the Manager do if all Workers die?
A: It waits. The job stays in progress. When a new Worker registers, the Manager resumes by assigning it the next available task (including any tasks that need reassignment from dead Workers).
Q: How do you detect busy-waiting in your implementation?
A: Run "time mapreduce-manager" without any Workers. Let it run for a few seconds and Ctrl-C. If the "user" CPU time is close to the "real" (wall clock) time, the process is busy-waiting. A correct implementation blocks on socket operations, so user time should be near zero.
Q: Why must the Worker's TCP listener be open before it sends the register message?
A: The Manager may send a task immediately after receiving the register message (if a job is already queued). If the Worker's TCP listener is not yet open, the Manager's TCP connection will be refused, and the Manager will mark the Worker as dead.
fault tolerance, heartbeat, UDP heartbeat, dead Worker detection, ConnectionRefusedError, task reassignment, init script, daemon, background process, pgrep, pkill, busy-waiting, blocking I/O, threading.Event, ThreadSafeOrderedDict, mocking, Mock class, StopIteration, pycodestyle, pydocstyle, pylint, context manager, pathlib, PathJSONEncoder, EECS 485 P4 implementation, MapReduce fault tolerance