Thread pool that binds tasks for a given ID to the same thread

Question

Are there any implementations of a thread pool (in Java) that ensures all tasks for the same logical ID are executed on the same thread?

The logic I'm after is if there is already a task being executed on a specific thread for a given logical ID, then new tasks with the same ID are scheduled on the same thread. If there are no threads executing a task for the same ID then any thread can be used.

This would allow tasks for unrelated IDs to be executed in parallel, but tasks for the same ID to be executed in serial and in the order submitted.

If not, are there any suggestions on how I might extend ThreadPoolExecutor to get this behaviour (if that's even possible)?

UPDATE

Having spent longer thinking about this, I don't actually require that tasks for the same logical ID get executed on the same thread, just that they don't get executed at the same time.

An example for this would be a system that processed orders for customers, where it was OK to process multiple orders at the same time, but not for the same customer (and all orders for the same customer had to be processed in order).

The approach I'm taking at the moment is to use a standard ThreadPoolExecutor, with a customised BlockingQueue and also wrapping the Runnable with a custom wrapper. The Runnable wrapper logic is:

1. Atomically attempt to add ID to concurrent 'running' set (ConcurrentHashMap) to see if a task for the same ID is currently running
   • if add fails, push the task back on to the front of the queue and return immediately
   • if succeeeds, carry on
2. Run the task
3. Remove the task's associated ID from the 'running' set

The queue's poll() methods then only return tasks that have an ID that is not currently in the 'running' set.

The trouble with this is that I'm sure there are going to be a lot of corner cases that I haven't thought about, so it's going to require a lot of testing.

Answer 1

Create an array of executor services running one thread each and assign your queue entries to them by the hash code of your item id. The array can be of any size, depending on how many threads at most do you want to use.

This will restrict that we can use from the executor service but still allows to use its capability to shut down the only thread when no longer needed (with allowCoreThreadTimeOut(true)) and restart it as required. Also, all queuing stuff will work without rewriting it.

Answer 2

The simplest idea could be this:

Have a fixed map of BlockingQueues. Use hash mechanism to pick a queue based on task id. The hash algorithm should pick the same queue for the same ids. Start one single thread for every queue. every thread will pick one task from it's own dedicated queue and execute it.

p.s. the appropriate solution is strongly depends on the type of work you assign to threads

UPDATE

Ok, how about this crazy idea, please bear with me :)

Say, we have a ConcurrentHashMap which holds references id -> OrderQueue

ID1->Q1, ID2->Q2, ID3->Q3, ...

Meaning that now every id is associated with it's own queue. OrderQueue is a custom blocking-queue with an additional boolean flag - isAssociatedWithWorkingThread.

There is also a regular BlockingQueue which we will call amortizationQueue for now, you'll see it's use later.

Next, we have N working threads. Every working thread has it's own working queue which is a BlockingQueue containing ids associated with this thread.

When a new id comes, we do the following:

create a new OrderQueue(isAssociatedWithWorkingThread=false)
put the task to the queue
put id->OrderQueue to the map
put this OrderQueue to amortizationQueue

When an update for existing id comes we do the following:

pick OrderQueue from the map
put the task to the queue
if isAssociatedWithWorkingThread == false
    put this OrderQueue to amortizationQueue

Every working thread does the following:

take next id from the working queue
take the OrderQueue associated with this id from the map
take all tasks from this queue
execute them
mark isAssociatedWithWorkingThread=false for this OrderQueue
put this OrderQueue to amortizationQueue

Pretty straightforward. Now to the fun part - work stealing :)

If at some point of time some working thread finds itself with empty working queue, then it does the following:

go to the pool of all working threads
pick one (say, one with the longest working queue)
steal id from *the tail* of that thread's working queue
put this id to it's own working queue
continue with regular execution

And there also +1 additional thread which provides amortization work:

while (true)
    take next OrderQueue from amortizationQueue
    if queue is not empty and isAssociatedWithWorkingThread == false
         set isAssociatedWithWorkingThread=true
         pick any working thread and add the id to it's working queue

Will have to spend more time thinking if you can get away with AtomicBoolean for isAssociatedWithWorkingThread flag or there is a need to make it blocking operation to check/change this flag.