11
\$\begingroup\$

I can have a maximum of three concurrent threads processing tasks. Each of these threads can process 1 to 100 tasks simultaneously (by connecting to an external system). The time taken to process 100 tasks at once in one thread is the same as it takes to process 1 task in one thread because the overhead of connecting to the external system is what takes 95% of the time. Tasks come in at random intervals from other threads in the application, these threads need to block until the task is done, or a timeout is hit. Responding quickly to the threads submitting tasks is the primary goal here, and making use of the ability to process tasks in batch is secondary (but useful and important when we have a big queue of tasks as it saves time).

This code works (or appears to anyway), but before I build it into my larger application I want to check that this can't be done better/faster etc. It has the potential to create a major bottleneck in the application so I want to make sure it is done in the most efficient way possible. I'm not very experienced with the concurrent package. Any feedback would be greatly appreciated.

Here is the concept code that I have written to test with. System outs will be replaced with logging when it is built into the main application.

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;

public class TaskProcessor implements Runnable
{
    private final LinkedBlockingQueue<Task> queue = new LinkedBlockingQueue<>();

    private final Semaphore semaphore = new Semaphore(3);

    private final Executor executor = Executors.newCachedThreadPool();

    public static void main(final String[] args)
    {
        final TaskProcessor taskProcessor = new TaskProcessor();
        new Thread(taskProcessor).start();
        for (int i = 0; i < 20; i++)
        {
            new Thread(new Runnable()
            {
                @Override
                public void run()
                {
                    try
                    {
                        taskProcessor.submitTask(new Task());
                    }
                    catch (final InterruptedException e)
                    {
                        throw new RuntimeException(e); // unreachable
                    }
                }
            }).start();
        }
    }

    private static class Task
    {
        public void setResult() {}
        public void getResult() {}
    }

    public void submitTask(final Task task) throws InterruptedException
    {
        final long taskStartTime = System.currentTimeMillis();
        this.queue.add(task);
        synchronized (task)
        {
            task.wait();
            System.out.println("task complete after " + (System.currentTimeMillis() - taskStartTime) + "ms");
        }
        task.getResult();
    }

    @Override
    public void run()
    {
        try
        {
            while (true)
            {
                this.semaphore.acquire(); // blocks until we can get a permit

                final List<Task> tasks = new ArrayList<>();

                System.out.println("waiting for task");
                tasks.add(this.queue.take()); // will block if the queue is empty

                // get all the other tasks in the queue
                Task nextTask = this.queue.poll();
                while (nextTask != null && tasks.size() < 100)
                {
                    tasks.add(nextTask);
                    nextTask = this.queue.poll();
                }

                this.processTasks(tasks);
            }
        }
        catch (final InterruptedException e)
        {
            throw new RuntimeException(e); // unreachable
        }
    }

    private void processTasks(final List<Task> tasks)
    {
        final Runnable runnable = new Runnable()
        {
            @Override
            public void run()
            {
                try
                {
                    TaskProcessor.this.executeTasks(tasks);
                    for (final Task task : tasks)
                    {
                        task.setResult(); // once fully implemented will set the result to be the result of exeuteTasks()
                        synchronized (task)
                        {
                            task.notify();
                        }
                    }
                }
                finally
                {
                    TaskProcessor.this.semaphore.release();
                }
            }
        };
        this.executor.execute(runnable);
    }

    private void executeTasks(final List<Task> tasks)
    {
        try
        {
            System.out.println("executing " + tasks.size() + " tasks.");
            Thread.sleep(1000);
        }
        catch (final InterruptedException e)
        {

            e.printStackTrace();
        }
    }
}
\$\endgroup\$
3
\$\begingroup\$

I am not a Java expert, but

  • In your description you talk about tasks and jobs, however your code is only using task, which makes your code harder to follow.
  • If you are going to log "waiting for task" then you should also log "waiting for semaphore"
  • Related to that, I have a suspicion that your semaphore approach does not work, you should try to run more than 3 tasks.
  • I would allow the caller to pass 3 and 100 as parameters or read this from a config file or environment variables
\$\endgroup\$
  • \$\begingroup\$ Thanks, to respond to your bullet points: 1. I removed all references to 'jobs', you're quite right that was confusing. Hopefully the updated first paragraph makes the problem clearer. 2. Added extra logging as suggested. 3. This code runs and appears to work, so I think the semaphore is working, could you elaborate on that point? 4. Good point, when this code is integrated into the larger application I will make use of the existing config system. \$\endgroup\$ – jimjim Apr 16 '14 at 15:38
  • \$\begingroup\$ Now your code will be even more confusing, because you have tasks running tasks ? I would have suggested to re-introduce jobs. As for point 3, if you see the metaphor work and never have more than 3 main tasks, then I was probably wrong. \$\endgroup\$ – konijn Apr 16 '14 at 15:47
  • \$\begingroup\$ Well, the code as it stands uses a TaskProcessor class to process Tasks or Task groups in anonymous threads, so I think it was the description that was confusing the issue. Also as Marc-Andre pointed out I can't modify the code posted, so hopefully changing the description helps people understand what I'm trying to do here. \$\endgroup\$ – jimjim Apr 16 '14 at 15:54
3
\$\begingroup\$

I can have a maximum of three concurrent threads processing tasks. Each of these threads can process 1 to 100 tasks simultaneously (by connecting to an external system).

Q: How many threads should you have? A: 4 - the management thread, and 3 worker threads.

How many threads are in your code? At least 24 .... potentially hundreds .... let's count them...

  1. The current thread
  2. 20 threads created in the main method
  3. The CachedThreadPool has no upperbound on the number of threads, and will create a new thread each time you call this.executor.execute(runnable); but, that is gated by the Semaphore

About that Semaphore.... Your code is asymmetrical in the sense that the semaphore is acquired in one class, and released in another (the Runnable)., This makes it hard to follow.

Fundamentally, though, your system is a poor example of using the java.util.concurrent.* package, and there are more 'idiomatic' ways to do this:

// no interruptedException handling, not validated... just to give you the idea.
public class TaskProcessor {

    private static final class RemoteTask implements Callable<List<Task>> {
        private final List<Task> tasks;
        public RemoteTask(List<Task> torun) {
            this.tasks = torun;
        }

        public List<Task> call() {
            // submit all torun tasks to the remote system
            // get the results.... and populate the return value:
            .....
            return tasks;
        }
    }

    private final class TaskManager implements Runnable {
        public void run () {


            boolean done = false;
            while (!done) {
                List<Task> tosubmit = new ArrayList<>(CHUNKSIZE);
                tosubmit.add(queue.take());
                queue.drainTo(tosubmit, CHUNKSIZE - 1);
                Task last = tosubmit.get(tosubmit.size() - 1);
                if (last == null) {
                    done = true;
                    tosubmit.remove(tosubmit.size() -1);
                }
                if (!tosubmit.isEmpty()) {
                    service.submit(new RemoteTask(tosubmit));
                }
            }
            service.shutDown();
            service.awaitTermination();   
        }
    }


    private static final int THREADS = 3;
    private static final int CHUNKSIZE = 100;

    private final ExecutorService service;

    private final LinkedBlockingQueue<Task> queue = new LinkedBlockingQueue<>();

    public TaskProcessor() {
        service = Executors.getFixedThreadPool(THREADS);
        Thread managerthread = new Thread(new TaskManager(), "Task Manager Thread");
        managerthread.setDaemon(true);
        managerthread.start();
    }

    public void submitTask(Task task) {
        queue.put(task);
    }

    .....

}
\$\endgroup\$
  • \$\begingroup\$ Thanks, this is very helpful. I have a couple of questions... how should the thread calling submitTask get told that the task is finished? Should I use wait/notify like I did in my example? (notify at the end of your call(..) method? or is there a better way? \$\endgroup\$ – jimjim Apr 17 '14 at 11:52
  • \$\begingroup\$ On the threads point, I was trying to say in the description that we can only have 3 simultaneous connections to the external system. So three concurrent task processing threads. The actual number of threads running in the system is not an issue (there will be up to 50 submitting tasks to this TaskProcessor, and then waiting for the result before resuming). Having a manager thread seems unavoidable because of the requirement to sometimes process groups of tasks instead of single tasks? \$\endgroup\$ – jimjim Apr 17 '14 at 11:53

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.