ThreadPool, don't execute two jobs simultaneously if certain conditions are met

[Sebastian Janisch]

hey guys,

let's say we create a thread pool with a pool size of 2. Now we add 4 jobs, named job1, ..., job4.

is there a way to tell the thread pool that job1 and job2 must not be executed simultaneously, but get added into a queue?


Thanks

Sebastian

[Stephan van Hulst]

Add one job to the pool that handles both jobs 1 and 2, one after the other.

This smells a bit fishy though. Why do you want to do this?

[Sebastian Janisch]

hey,

thanks for your reply ...

why do we want to do this ...

imagine there are 200 sources that pump in tasks ... the requirement is that tasks that come from the same source must not get executed at the same time, tasks from different sources can, though ...

so there should be one threadpool with, say, 40 threads that those tasks from 200 sources get pumped in to but it needs to watch out not to execute two tasks from the same source at once.

we could create one threadpool for each of those 200 sources, but that doesn't make any sense as 200 * 40 = a lot of threads ...

[Stephan van Hulst]

No, you would create one thread per source, because each source should be handled by only one thread anyway. If this is not acceptable, and you want to use a fixed pool regardless of how many sources you have, then you can just make your own construction using the ExecutorService.submit() method. I don't readily know if there's a better option available, but this shouldn't be too hard to implement yourself.

It depends on how you get notified of a new job though. How do you retrieve jobs from sources?

[Sebastian Janisch]

hey,

it's a client server app and tasks come in from various sources ... and it's important that tasks from the same source are processed synchronously ...

[Stephan van Hulst]

No what I mean is, how do get the jobs, on a more concrete level? Do you use a ServerSocket and block until something comes in? Do you actively retrieve jobs or do you passively wait until you get notified of one, etc?

[Sebastian Janisch]

the whole thing uses the jboss netty server socket framework and as messages come in, tasks are created that belong to a certain group ...

[Stephan van Hulst]

I guess what you can do is make some sort of multiplexer class.

Make a queue for every source you have and register it with your multiplexer. Now what the multiplexer does is it polls every queue associated with a source that had its last task handled, to see if it has a new task. If it does, then it simply submits that task to your thread pool.

If you give me a minute, I'll have an example for you.

[Sebastian Janisch]

hey,

I have created such class, but i'm facing synchronization issues ... here's the code... it's pretty long but most of it are comments. .. would be great if you could help finding the error ..

/**
 * This class is a specialization of a standard thread-pool that allows for synchronous execution of selected tasks.
 * <p>
 * To understand the functioning, consider the following example:
 * <p>
 * Tasks are received from 200 sources. Tasks from the same source must <u>not</u> get executed at the same time, but one after the other in order of arrival.
 * <p>
 * If you opened a thread-pool with, say, a pool-size of 40 for each of these sources, you would end up with 200*40 active threads at maximum which would grind the application to a halt.
 * <p>
 * This class exactly addresses the above issue. It groups tasks by a given identifier and within the same group it executes only one task at a time.
 * 
 * @author Sebastian Janisch
 * 
 */
public class SynchronousThreadPool
{
    /**
     * This interface is an extension of {@link Runnable} and adds the functionality to obtain a key that identifies a given task within it's task-group.
     * 
     * @author Sebastian Janisch
     * 
     */
    public static interface SynchronousRunnable extends Runnable
    {
        /**
         * 
         * @return an identifier that must be equal for every task that is executed within the same group of tasks. Never <code>null</code>.
         */
        Object getIdentifier();
    }

    private final ExecutorService                              _underlyingThreadPool;

    /**
     * This map groups together tasks that belong to the same group.
     */
    private final Map<Object, ConcurrentLinkedQueue<Runnable>> _taskMap = new ConcurrentHashMap<Object, ConcurrentLinkedQueue<Runnable>>();

    /**
     * Creates a new SynchronousThreadPool.
     * 
     * @require <code>underlyingThreadPool != null</code>
     * 
     * @param underlyingThreadPool
     *            the thread-pool that is used for all tasks executed by this SynchronousThreadExecutor.
     *            <p>
     *            For example, invoke the constructor like this: <code>new SynchronousThreadPool(Executors.newCachedThreadPool())</code>
     */
    public SynchronousThreadPool(ExecutorService underlyingThreadPool)
    {
        assert (underlyingThreadPool != null) : "Precondition violated: underlyingThreadPool != null";

        _underlyingThreadPool = underlyingThreadPool;
    }

    /**
     * Executes the given task once it has it's turn.
     * <p>
     * This task will not get executed before all tasks of the same group ( {@link SynchronousRunnable#getIdentifier()} is the same) that have been passed to this method have finished execution.
     * 
     * @require <code>task != null</code>
     * @require <code>task.getIdentifier() != null</code>
     * 
     * @param task
     *            the task to get executed
     */
    public void execute(SynchronousRunnable task)
    {
        assert (task != null) : "Precondition violated: task != null";
        assert (task.getIdentifier() != null) : "Precondition violated: task.getIdentifier() != null";

        ConcurrentLinkedQueue<Runnable> taskQueue = _taskMap.get(task.getIdentifier());

        if (taskQueue == null)
        {
            taskQueue = new ConcurrentLinkedQueue<Runnable>();
            _taskMap.put(task.getIdentifier(), taskQueue);
        }

        /*
         * if there are no task prior to this one, we pass this task directly to the thread-pool, otherwise we enqueue it.
         */
        synchronized (taskQueue)
        {
            taskQueue.add(new WrappedSynchronousRunnable(task));

            if (taskQueue.size() == 1)
            {
                _underlyingThreadPool.execute(taskQueue.peek());
            }
        }
    }

    /**
     * This class wraps a {@link SynchronousRunnable} to integrate it's execution into the synchronous environment.
     * 
     * @author Sebastian Janisch
     * 
     */
    private class WrappedSynchronousRunnable implements Runnable
    {
        private SynchronousRunnable _wrapped;

        public WrappedSynchronousRunnable(SynchronousRunnable wrapped)
        {
            _wrapped = wrapped;
        }

        @Override
        public void run()
        {
            // the actual execution of the task
            _wrapped.run();

            ConcurrentLinkedQueue<Runnable> taskQueue = _taskMap.get(_wrapped.getIdentifier());

            synchronized (taskQueue)
            {
                Runnable executedJob = taskQueue.poll();

                assert executedJob == this;

                // if there are tasks left for execution, go ahead and execute
                // the next in line
                if (!taskQueue.isEmpty())
                {
                    _underlyingThreadPool.execute(taskQueue.peek());
                }
                else
                {
                    _taskMap.remove(taskQueue);
                }
            }
        }
    }

}

[Nitesh Kant]

I feel it is better to put the throttling logic in the queue implementation itself. The reason being that the implementation is cleaner than maintaining the buffer in the executor.

The following is a way to implemented such a throttled queue:

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

public class MyBlockingQueue extends LinkedBlockingQueue<Runnable> implements TaskCompletionListener {

    private ConcurrentHashMap<String, OverflowTaskHolder> overflowTasks = new ConcurrentHashMap<String, OverflowTaskHolder>();

    @Override
    public boolean offer(Runnable task) {
        if (null == task) {
            return false;
        }
        IdentifiableTask identifiableTask = (IdentifiableTask) task;
        OverflowTaskHolder holder = new OverflowTaskHolder(identifiableTask);
        OverflowTaskHolder existingHolder = overflowTasks.putIfAbsent(identifiableTask.getIdentifier(), holder);
        if (null != existingHolder) {
            existingHolder.tasks.add(identifiableTask);
            boolean wasEmpty = existingHolder.isEmpty.compareAndSet(true, false);
            if(wasEmpty) {
                System.out.println("Offering since empty. Task: " + identifiableTask);
                return super.offer(identifiableTask);
            }
            return true;
        } else {
            System.out.println("First time offer. Task: " + identifiableTask);
            return super.offer(identifiableTask); // This indicates a first task submission for this source.
        }
    }

    @Override
    public void onTaskComplete(String source, IdentifiableTask completedTask) {
        System.out.println("Completed task: " + completedTask);
        OverflowTaskHolder holder = overflowTasks.get(source);// You will always get this if there was a task which was ever completed for this source.
        holder.tasks.remove(completedTask);
        IdentifiableTask task = holder.tasks.peek();
        if (null != task) {
            super.offer(task);
        } else {
            boolean wasNotEmpty = holder.isEmpty.compareAndSet(false, true);
            if(wasNotEmpty) {
                offer(holder.tasks.poll());
            }
        }
    }

    private class OverflowTaskHolder {

        private ConcurrentLinkedQueue<IdentifiableTask> tasks;
        private AtomicBoolean isEmpty;

        public OverflowTaskHolder(IdentifiableTask identifiableTask) {
            tasks = new ConcurrentLinkedQueue<IdentifiableTask>();
            tasks.add(identifiableTask);
            isEmpty = new AtomicBoolean(false);
        }
    }
}

This depends on the following callback interface:

public interface TaskCompletionListener {

    void onTaskComplete(String source, IdentifiableTask completedTask);
}

and the following task implementation (Although you can use a different task definition as per your convenience):

public class IdentifiableTask implements Runnable {

    private String identifier;
    private TaskCompletionListener listener;
    private Runnable runnable;

    public IdentifiableTask(String identifier, TaskCompletionListener listener, Runnable runnable) {
        this.identifier = identifier;
        this.listener = listener;
        this.runnable = runnable;
    }

    public String getIdentifier() {
        return identifier;
    }

    @Override
    public void run() {
        System.out.println("Running task: " + this);
        runnable.run();
        listener.onTaskComplete(identifier, this);
    }

    @Override
    public String toString() {
        return super.toString() + "---:---" + identifier;
    }
}

In order to use the above queue, it is mandatory to use a thread pool executor where the core pool size and the max pool size is the same and all the core threads a re pre-started. The reason for this is that the executor does not enqueue it to the queue if there are spare threads that can be created. The following will be one helper class to do that:

import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ThrottledSubmitter {

    private MyBlockingQueue workQueue;
    private ThreadPoolExecutor executor;

    public ThrottledSubmitter(int noOfWorkes) {
        workQueue = new MyBlockingQueue();
        executor = new ThreadPoolExecutor(noOfWorkes, noOfWorkes, 100/*Does not matter*/, TimeUnit.MINUTES, workQueue);
        executor.prestartAllCoreThreads();
    }

    public void submit(Runnable runnable, String identifier) {
        executor.execute(new IdentifiableTask(identifier, workQueue, runnable));
    }
}

I have just created a sample loader class to test the above:

public class WorkLoader {

    private static final ConcurrentLinkedQueue<MyRunnable> tasks = new ConcurrentLinkedQueue();
    private static ThrottledSubmitter ts;

    public static void main(String[] args) throws InterruptedException {
        ts = new ThrottledSubmitter(2);
        load();
    }

    private static void load() throws InterruptedException {
        System.out.println("Loading new work....");
        submit("Source - 1");
        submit("Source - 1");
        submit("Source - 2");
        submit("Source - 2");
        submit("Source - 1");
        synchronized (tasks) {
            tasks.wait();
            load();
        }
    }

    private static void submit(String id) {
        MyRunnable myRunnable = new MyRunnable(id);
        tasks.add(myRunnable);
        ts.submit(myRunnable, myRunnable.id);
    }

    private static class MyRunnable implements Runnable {

        private String id;

        private MyRunnable(String id) {
            this.id = id;
        }

        @Override
        public void run() {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            tasks.remove(this);
            if(tasks.isEmpty()) {
                synchronized (tasks) {
                    tasks.notifyAll();
                }
            }
        }
    }
}

[Stephan van Hulst]

Here is a way you could do it from the other end, by having one dedicated thread pumping the tasks into the thread pool. I have not tested it, I don't know if it works, but it should be an interesting alternative.

Nitesh' way is much nicer though, having a custom BlockingQueue. I just thought it was nice to see if things worked on the other end.

import java.util.*;
import java.util.concurrent.*;

public final class JobQueue extends Observable {

  private final Queue<Runnable> jobs;
  private Future<?> lastExecution;

  public JobQueue() {
    jobs = new ConcurrentLinkedQueue<Runnable>();
  }

  public void addJob(Runnable newJob) {
    if (newJob == null)
      throw new NullPointerException();

    jobs.add(newJob);
    setChanged();
    notifyObservers();
  }

  boolean ready() {
    return (lastExecution == null || lastExecution.isDone());
  }

  boolean executeJob(ExecutorService pool) {
    if (!ready())
      throw new IllegalStateException("Still executing last job");

    Runnable job = jobs.poll();
    if (job != null) {
      lastExecution = pool.submit(job);
      return true;
    }

    return false;
  }
}

import java.util.*;
import java.util.concurrent.*;

public final class JobMultiplexer implements Observer {

  private final List<JobQueue> queues;
  private final ExecutorService pool;
  private volatile boolean locked, processing, again;

  public JobMultiplexer(ExecutorService pool) {
    if (pool == null)
      throw new NullPointerException();

    queues = new ArrayList<JobQueue>();
    this.pool = pool;
  }

  public synchronized void register(JobQueue queue) {
    if (queue == null)
      throw new NullPointerException();

    queue.addObserver(this);
    queues.add(queue);
  }

  public void process() {
    synchronized (this) {
      if (locked)
        throw new IllegalStateException("Already processing");

      locked = true;
    }

    processing = true;
    while (processing) {
      synchronized (this) {
        again = false;

        for (JobQueue queue: queues) {
          if (queue.ready() && queue.executeJob(pool))
            again = true;
        }

        try {
          while (processing && !again)
            wait();
        } catch (InterruptedException ex) {
          ex.printStackTrace();
        }
      }

      Thread.yield();
    }

    locked = false;
  }

  public synchronized void interrupt() {
    processing = false;
    notifyAll();
  }

  public synchronized void update(Observable o, Object arg) {
    if (queues.contains(o)) {
      again = true;
      notifyAll();
    }
  }
}

The idea is that you keep a JobQueue for every source, and register them with a JobMultiplexer. The multiplexer will submit jobs to a thread pool. You need one dedicated thread running the process method in order to make it work. This is why Nitesh' solution is nicer.

[Stephan van Hulst]

Actually, after a simple test, my solution is almost sure to deadlock because the processing thread doesn't get notified when jobs are completed.