How can you ensure order of execution in concurrent tasks?

Question

Here is what I am specifically doing:

1. I have a thread-safe queue
2. One 'write' thread constantly writes to the queue with data that comes from another service
3. Multiple 'read' threads take from the queue, each thread taking several items at once and doing some processing with them
4. Once each 'read' thread processes its current batch of items taken from the queue, it is written to a database

The problem is that items need to be written to the database in step 4 in the same order that they come in step 1.

So for instance if I am processing some events, this rare case could happen:

1. 'Entity 1 Created' event is added to the queue.
2. Some other events are added to the queue.
3. 'Entity 1 Deleted' event is added to the queue.
4. Read thread #1 takes its next batch which includes 'Entity 1 Created' and a few others
5. Read thread #2 takes its next batch which includes 'Entity 1 Deleted' and a few others
6. If for any reason read thread #2 reaches the database request sooner than Read thread #1, then the database will get the request for 'Entity 1 Deleted' first and since it doesn't exist yet, it will do nothing, then it will get the 'Entity 1 Created' request and it will add an entity and it will remain undeleted.

Is there any way to prevent the problem at step 6. without completely destroying performance. I could limit the 'read' threads to 1, but I am looking for ways to have multiple read threads.

Answer 1

Since your example only contains "Create" and "Delete" events for single entities, here is a simple solution which works as long as all events affect a full entity (like creation, full update, and deletion), without using the former entities state.

• give each event a strictly increasing ordering number when it is inserted into the queue

• for each entity, store the last operation's ordering number (for deletions, don't delete the data really, instead set a "deleted" flag inside the entity, so you still have a place where the ordering number is stored even after deletion)

When an event hits the database, by comparing its ordering number to number of the latest operation, you can now decide which operation shall have precedence. "Earlier" operations / events can be ignored if the last operation was newer.

This solution has the benefit it does not require any complicated synchronizing mechanics, not even a timeout mechanics, and should be simple to implement. The drawback is, it will not work if you have events with "partial updates" or more complex events with calculations based on the current state of the entity. For partial updates, you could extend the approach by storing the ordering number for each individual attribute. But beware, this may require a lot of additional boilerplate data in your DB.

Answer 2

You need some for of queue marshaling at step 1. Essentially batching the messages by entity to ensure the correct order of processing. You seem to be doing the whole thing in a single application, but it would be more common to have an external MQ system and multiple workers. eg

• MQ allMessages
• Routing Worker pulls from allMessages
• is this a new entity ? create queue for entity, remember id and push message to that queue : push to existing queue for that entity
• Processing Worker contacts Router (prob though another queue)
• Do you have any entities that need processing ? receive entity queue to work on : wait 10 and loop
• Processing Worker: finishes a message
• send I am alive message to router
• Processing Worker: finishes entityQueue
• contact Router, I have finished that queue
• Router : remove entityId from known list
• Worker crashes : no "i am alive" message in X amount of time, assign entityQueue to new worker

Now you can spin up as many Workers as you like and each entity will only have a single worker processing its messages ensuring that they are in the correct order.

Workers naturally switch entities as the messages are processed so you don't end up waiting for entity X while the messages for other entities pile up.

The Router is a single point of failure, but everything is talking to each other so you should be able to scale it out with some extra messages

Answer 3

You could for do something like the following

1. The source thread creates a 'task' for each data item and puts it in two queues, a processing queue and an output queue.
2. 1..n worker threads picks tasks from the processing queue, does the processing, and sets the task as complete.
3. A single output threads picks tasks from the output queue and waits for it to complete. Once the task completes it writes the result to the database and repeats the process for the next item.

This could be implemented with various types of synchronization primitives. For example TaskCompletionSource and data-object pair. Or use a LimitedConcurrencyTaskScheduler instead of multiple explicit worker threads.

This should ensure the results are processed in-order, while allowing the processing to be done out of order. You might want to consider if you need some timeout or other mechanisms to handle objects that take a long time to process.

It might also be possible to use existing libraries like DataFlow for this. But I'm not familiar enough with it to provide any specific recommendations.

Answer 4

I feel this is similar to Ewan's answer in terms of batching and single-point of failure; the queues in this example could also be in-memory queues or MQ.

Consider separating the database updates away from the worker threads into its own single thread or process, leaving the workers just to perform the expensive processing actions, with each yielding a 'result' object after they finish.

Secondly, introduce a new orchestration step which dispatches tasks to workers, keeping a list of workers/tasks, which can waits/join those worker threads based on their original order.

Finally, the list and wait/join can be used to separately pass those completed tasks to a database-update queue, guaranteeing the original order.

To illustrate an orchestrator using pseudo-code grabbing 10 tasks at a time (for 10 separate workers), possibly using constructs such as Task, Future or Promise (depending on language):

CONST num_workers = 10

BEGIN FUNCTION orchestratorThread()
    REPEAT WHILE input_queue HAS tasks:
        DEFINE worker_threads[num_workers]

        /* Trigger the workers, tasks may complete in any order */
        FOR n <- 0 TO num_workers:
            /* Read but don't dequeue. Database updates must respect this order */
            item <- READ(input_queue, n)  
            worker_threads[n] <- START_THREAD(item)
        NEXT FOR

        /* wait/join each thread in series to ensure DB updates respect the order */
        FOR EACH t IN worker_threads:
            JOIN_THREAD(t) ELSE TIMEOUT

            IF SUCCESSFUL(t) THEN
                ENQUEUE(completed_tasks_for_database, RESULT_OF(t))

                /* The task successfully completed.
                 * Remove from the front of the input queue.
                 */
                DEQUEUE(input_queue)
            ELSE
                /* Abort to prevent timeout or failure interfering with order. */
                BREAK FOR EACH
            END IF
        NEXT FOR EACH

    NEXT WHILE
END FUNCTION

The idea of enqueueing completed tasks to a separate queue is that this guarantees the order is maintained for all successful tasks after the long-running processing work is done, and also does not hold up the orchestrator with database updates -- a separate single process or thread should handle the database by itself.

Answer 5

Every task is given an increasing ID, like 1, 2, 3, 4. You also remember the ID of the last task that went into the database, in this case 0.

At the point where a task is finished, you put it into a queue. Then as long as the queue contains the next task to be added to the database, you add it.

If you have say four worker threads, and each worker thread grabs 10 tasks at a time, you might give it not ten consecutive tasks, but say task 1, 5, 9, 13 etc. So when four threads perform tasks, they will tend to be completed in the right order.