Task Queue

Track: Protocols and sockets. 5 stages. Languages: Go, Python, JavaScript.

Many programs have work that should happen later, or somewhere else: send an email, resize a photo, charge a card. They put each job on a task queue, and workers take the jobs off it one by one. Build the client side of one: the code that adds tasks, and the worker loop that takes tasks, does them, and tells the queue they are done.

The queue server is given. It speaks a tiny text protocol over a network connection. You send one line, and it answers with one line. Every line ends with \n.

you send     server answers        meaning
ENQ t7       OK                    add task t7 to the queue
DEQ          TASK t7  or  EMPTY    give me the next task, or there is none right now
ACK t7       OK  or  NOACK         t7 is done, remove it for good

When the server gives a task to a worker, it keeps that task aside for a short time. If no ACK arrives in that time, it puts the task back so a worker can take it again. This is what keeps work safe when a worker crashes. It also means one rule matters more than any other: send ACK only after the work is really finished. A task you mark done too early is gone if the work then fails.

Implement these, using your language's own names for them:

New(dial) creates a client. dial is a function we give you. Each call opens one new connection to the server, or fails. Always use it, never connect to the server yourself.
Enqueue(ids) opens one connection, sends ENQ <id> for each id and waits for OK each time, then closes the connection.
Consume(process) is the worker loop. Send DEQ. On TASK <id>, call process(id), then send ACK <id>. On EMPTY, wait about 2 milliseconds and ask again. Keep going until the tester tells you to stop. If the connection breaks, open a new one with dial and carry on.
Metrics() reports two numbers you count yourself: how many ENQ lines the server answered with OK, and how many ACK lines it answered with OK.

Several workers call Consume on the same client at the same time, so give each call its own connection, and keep your counts correct when calls overlap. Standard library only.

The tester puts a small relay between your client and the server, so it can break a connection on purpose. It also reads the server's own record of every task it handed out and every ACK it got, so it knows exactly what happened, whatever your code reports. For help with the ideas, read the notes. New to code that many callers run at the same time? Read Many callers at once first.

Where it is used

Amazon SQS hides a message from other workers while one worker handles it. If the worker does not delete it in time, the message comes back for another worker.
RabbitMQ puts a message back on the queue if the worker's connection closes before it acknowledged the message.
Celery, a Python task queue, acknowledges a task before running it by default. With task_acks_late, it acknowledges after the task has run instead. If the worker process dies in the middle, Celery still acknowledges the task by default, unless task_reject_on_worker_lost is set.

Marking a task done before the work is finished is how a photo never gets resized or an email never gets sent, with nothing in the logs to show it.

Stages

tq1: do every task, and mark it done only after the work (medium)

The tester adds 20 tasks and runs 4 workers for up to 800 milliseconds. A task is put back if no ACK comes within 1 second. Then a second run: 10 tasks, 1 worker, and the 5th process call fails on purpose, as if the worker crashed. The tester starts your Consume again, and the server hands the unfinished task back after 100 milliseconds.

Enqueue(t0, t1)   -> ENQ t0 / OK, ENQ t1 / OK
worker:  DEQ      -> TASK t0
         process(t0), then ACK t0 -> OK      (t0 is done for good)
worker:  DEQ      -> TASK t1
         process(t1) fails, no ACK           (t1 comes back later)
worker:  DEQ      -> EMPTY                   (wait 2 ms, ask again)

The tester will verify: all 20 tasks are marked done, no task is with two workers at once, and in the second run all 10 tasks get processed, including the one that failed the first time.

tq4: add tasks quickly (L2, hard)

Keep Enqueue fast. A loop adds one task at a time, 30,000 times, with no real network: dial returns a fake connection that answers OK at once. Every task must still be sent as a real ENQ line. Example: if our solution takes 100 ms, the limit is 150 ms. Your solution takes 140 ms, so it passes. In Go, your code may also make at most twice as many memory allocations as ours.

The tester will verify: your time is at most 1.5 times our solution's time, and every task really reached the fake server.

tq2: keep going when the connection breaks (L3, hard)

20 tasks, 1 worker, up to 800 milliseconds. A task is put back if no ACK comes within 100 milliseconds. Enqueue uses connection #1. The tester breaks connection #2, the worker's first one, as soon as it is used.

Enqueue          -> connection #1, 20 x ENQ / OK
worker:  dial    -> connection #2
         DEQ     -> error, the connection was cut
         dial    -> connection #3              (open a new one, don't give up)
         DEQ     -> TASK t0 ... and so on until all 20 are done

The tester will verify: all 20 tasks are marked done even though a connection broke, and the tester broke exactly 1 connection.

tq5: keep going when the server stops answering (L3, hard)

A connection can stay open and go silent. 20 tasks, 1 worker, up to 2.5 seconds. A task is put back if no ACK comes within 100 milliseconds. On your worker's first connection, the tester cuts the server's first answer short and then sends nothing more.

worker:  DEQ      -> T...                (the rest never arrives)
         wait about one second, close the connection, dial again
worker:  DEQ      -> TASK t1             (and later the task the server put back)

Wait about one second for each answer, not much longer: with a 2 second limit there is not enough time left to finish.

The tester will verify: all 20 tasks are marked done, including the one the server handed out on the silent connection and later put back, and no task is with two workers at once.

tq3: report your own numbers correctly (L4, medium)

The same work as tq1: 20 tasks, 4 workers. At the end, the tester compares your Metrics() with the server's own record. Count a task as added only when ENQ got OK. Count a task as done only when ACK got OK, not when you send it.

server saw: 20 added, 20 done
Metrics() -> enqueued 20, acked 20      (correct)
Metrics() -> enqueued 20, acked 0       (wrong: never counted the ACKs)

The tester will verify: your added count and your done count are exactly the server's numbers.

Rules

Standard library only: sockets, locks, timers, threads or async, whatever your language offers.
Use the dial function you are given. Never connect to the server address yourself.
Send ACK after process finishes, never before.
A broken connection is a reason to open a new one, not to stop.
An answer that does not arrive in about one second means the connection is dead. Treat it like a broken one.

Read the Task Queue notes in the wiki