Concurrency Utilities

    class MyCallable implements Callable<Integer> {
        public Integer call() { return 2+2; }
    }

    // or anonymously
    Callable<Integer> callable = new Callable<Integer>() {
        public Integer call() { return 2+2; }
    };

There is also a convenience method for bridging Runnables to Callables in the Executors class. It takes a Runnable and a fixed value to return as a value when it completes:

    Callable<Integer> callable = Executors.callable( runnable, 
        42 /*return value*/ );

The new Future class is used with Callable and serves as a handle to wait for and retrieve the result of the task or cancel the task before it is executed. A Future is returned by the submit() methods of an ExecutorService, which is essentially a beefed-up Executor. We’ll discuss ExecutorServices in the next section.

    Future<Integer> result = executorService.submit( callable );
    int val = result.get();  // blocks until ready

    public class ExecutorHttpd
    {
      ExecutorService executor = Executors.newFixedThreadPool(3);

      public void start( int port ) throws IOException
      {
        final ServerSocket ss = new ServerSocket( port );
        while ( !executor.isShutdown() )
          executor.submit( new TinyHttpdConnection( ss.accept() ) );
      }

      public void shutdown() throws InterruptedException {
        executor.shutdown();
        executor.awaitTermination( 30, TimeUnit.SECONDS );
        executor.shutdownNow();
      }

      public static void main( String argv[] ) throws Exception
      {
        new ExecutorHttpd().start( Integer.parseInt(argv[0]) );
      }
    }

    List<Callable<Integer>> taskList = ...;
    ExecutorService execService = Executors.newFixedThreadPool(3);
    List<Future<Integer>> resultList = execService.invokeAll( taskList );

By contrast, the invokeAny() method returns just the first successfully completed task’s result (cancelling all the remaining unexecuted tasks):

    int result = execService.invokeAny( taskList );

    Executors.newScheduledThreadPool(int);
    Executors.newSingleThreadScheduledExecutor();

These are exactly like the similarly named methods for regular executor services, with the exception of returning a scheduled executor type.

To execute a task in the future, you specify a delay from the current time. For example:

    ScheduledExecutorService exec = Executors.newScheduledThreadPool(3);

    exec.schedule( runnable, 60, TimeUnit.SECONDS ); // run one minute in the 
                                                     // future

    // run at specified date and time
    Calendar futureDate = ...; // convertfrom calendar
    Date date = futureDate.getTime(); // to Date
    long delay  = date.getTime() - System.currentTimeMillis(); // to relative 
                                                               // millis
    exec.schedule( runnable, delay, TimeUnit.MILLISECONDS ); // run at specified 
                                                             // date

    Runnable cleanup = new Runnable() {
        public void run() { cleanUpLogFiles(); }
    };

    long period = 12*60*60, delay = period; // seconds

    Future<?> logService = executionService.scheduleAtFixedRate(
        cleanup, delay, period, TimeUnit.SECONDS );

Because the task for periodic schedules is a Runnable, the Future object does not return a useful value (it returns null) so we don’t specify a parameter type in its generic type instantiation. The Future is still useful for cancelling the task at a later time if we wish:

    logService.cancel();

    Executor executor = Executors.newFixedThreadPool(3);
    CompletionService<Integer> completionService =
        new ExecutorCompletionService<Integer>( executor );

    completionService.submit( callable );
    completionService.submit( runnable, resultValue );

    // poll for result
    Future<Integer> result = completionService.poll();
    if ( result != null )
        // use value...

    // block, waiting for result
    Future<Integer> result = completionService.take();

        ExecutorService executorService = new ThreadPoolExecutor(
        corePoolSize, maximumPoolSize, keepAliveTime, timeUnit, taskQueue );

The ThreadPoolExecutor implementation also has methods that allow you to change the core and maximum pool size while the service is active or to “prestart” the core threads before the service is used.

Actually, these last features bring up an interesting issue. If we know that our executor service is an implementation of ThreadPoolExecutor, we can cast it at runtime to get access to these extra methods and do things like change the pool size. This may not be what the designers of some services had in mind; in fact, it could be downright dangerous in the wrong hands. For this reason, Executors offers a number of “unconfigurable” wrapper methods that act something like the “unmodifiable” collection methods we’ll see in the Java Collections API. These methods wrap an executor service in a delegator object that does not expose the implementation to the caller:

    ExecutorService tweakable = Executors.newFixedThreadPool();
    ExecutorService safe = Executors.unconfigurableExecutorService( tweakable );

public class Mandelbrot extends JFrame
{
    @Override public void paint( Graphics g ) {
        BufferedImage image = new BufferedImage( getWidth(), getHeight(), 
            BufferedImage.TYPE_INT_RGB );
        ForkJoinPool pool = new ForkJoinPool(); // defaults thread per processor
        pool.invoke( new MandelbrotTask( image, 0, image.getWidth()-1, 0, 
            image.getHeight()-1 ) );
        g.drawImage( image, 0, 0, null );
    }

    public static void main( String[] args ) {
        Mandelbrot mandy = new Mandelbrot();
        mandy.setSize( 768, 768 );
        mandy.setVisible( true );
    }
}

class MandelbrotTask extends RecursiveAction
{
    private static double size = 3.0, offsetX = -0.7, thresholdSq = 100;
    private static int maxIterations = 30;
    private BufferedImage image;
    private int xStart, xEnd, yStart, yEnd;
    private static int taskSplitThreshold = 1024;

    MandelbrotTask( BufferedImage image, int xStart, int xEnd, int yStart, 
        int yEnd ) {

        this.image = image;

        this.xStart = xStart;
        this.xEnd = xEnd;
        this.yStart = yStart;
        this.yEnd = yEnd;
    }

    public void render()
    {
        for ( int x = xStart; x <= xEnd; x++ ) {
            for ( int y = yStart; y <= yEnd; y++ )  {
                double r = x * size / image.getWidth() -size/2 + offsetX;
                double i = y * size / image.getHeight() -size/2;
                double zr=0, zi=0;
                int iter;
                for ( iter = 0; iter < maxIterations; iter++ ) {
                    double nzr = zr*zr - zi*zi + r;
                    double nzi = 2*zr*zi + i;
                    if ( nzr*nzr + nzi*nzi > thresholdSq ) { break; }
                    zr = nzr; zi=nzi;
                }
                image.setRGB( x, y, Color.HSBtoRGB( 0.5f * iter / maxIterations,
                    1.0f, 1.0f) );
            }
        }
    }

    @Override protected void compute()
    {
        int width = xEnd-xStart,  height = yEnd-yStart;
        if ( width*height < taskSplitThreshold ) {
            render();
        } else {
            invokeAll(
                new MandelbrotTask( image, xStart, xStart+width/2, yStart,
                    yStart+height/2 ),
                new MandelbrotTask( image, xStart+width/2+1, xEnd, yStart,
                    yStart+height/2 ),
                new MandelbrotTask( image, xStart, xStart+width/2, 
                    yStart+height/2+1, yEnd ),
                new MandelbrotTask( image, xStart+width/2+1, xEnd, 
                    yStart+height/2+1, yEnd )
            );
        }
    }
}

Try running the example and then dragging the window out to different sizes. Watch how it redraws as the window is dragged out to a large size. The fractal is generated by treating each point in the image as a complex number (a two-dimensional number) and applying a simple formula to it repeatedly: Z=Z²+C, where Z is initially zero and C is related to the coordinate of the point. Then we color-code the point based on how fast that value grows. In some areas of the image, the values will grow quickly and we’ll stop iterating on them; in other areas, we’ll go on until we reach a number (maxIterations) of iterations. This means that some regions will take longer than others to generate and some threads will therefore steal work from others.

    ReadWriteLock rwLock = new ReentrantReadWriteLock( fair );

    // reader thread 1
    rwLock.readLock().lock();
    // reader thread 2
    rwLock.readLock().lock();

    // writer thread
    rwLock.writeLock().lock(); // blocks on threads 1 & 2

In this code snippet, two readers hold read locks while a writer blocks waiting on the write lock. When both readers have unlock()ed their read locks, the writer gains exclusive access to the lock and any subsequent readers block until the writer is finished.

    Lock lock = ...
    Condition condition = lock.newCondition();
    lock.lock();
    condition.await(); // block, waiting for signal()
    lock.unlock();

    // meanwhile, in another thread...
    lock.lock();
    condition.signal();
    lock.unlock();

    CountDownLatch latch = new CountDownLatch( 2 ); // count from 2

    // thread 1
    latch.await(); // blocks thread 1

    // thread 2
    latch.countDown(); // count is 1
    latch.countDown(); // count is 0, thread 1 proceeds

    int concurrentReaders = 5;
    boolean fair = true;
    Semaphore sem = new Semaphore( concurrentReaders, fair );

    Data readData() throws InterruptedException {
        sem.acquire();
        // read data ...
        sem.release();

        return data;
    }

In this code snippet, readData() effectively limits itself to five concurrent reading threads at any given time. Additional threads are blocked in the acquire() method until a permit is free. In this sense, a semaphore is vaguely like a lock with multiple owners. This is where the similarity ends, however.

In actuality, a semaphore differs from a lock in several ways. First, the “pool of permits” is really just a number. No actual value is returned by acquire() and no association is made with the acquirer of the lock. This means that any actual locking behavior is strictly cooperative (by convention in the application). It also means that “permits” can be acquired and released by different callers without respect to who actually “acquired” them. It’s really just incrementing or decrementing the number. Also, because there is no real association with an “owner,” semaphores are not reentrant in the way that real locks are. That is, if a thread calls acquire() multiple times, it simply decrements the counter multiple times. This behavior could be useful in some cases to count levels of recursion for security APIs, for example, but is not like a lock, in which one caller “owns” multiple permits. Finally, because the permits pool is really just a number, calling acquire() and release() out of sync can increase the permit pool beyond its starting point or decrement it below zero. It can even be initialized with a negative number if you wish to require releases before anyone acquires a permit.

    import java.util.*;
    import java.util.concurrent.*;
    import java.net.*;
    import java.io.IOException;

    public class SiteTimer
    {
       CyclicBarrier barrier;
       List<Result> results = new ArrayList<Result>();

       private class Result implements Comparable<Result>
       {
          Long time;
          String site;
          Result( Long time, String site ) {
             this.time = time;
             this.site = site;
          }
          public int compareTo( Result r ) { return time.compareTo( r.time ); }
       }

       static long timeConnect( String site ) {
          long start = System.currentTimeMillis();
          try {
             new URL( site ).openConnection().connect();
          } catch ( IOException e ) {
             return -1;
          }
          return System.currentTimeMillis() - start;
       }

       void showResults() {
          Collections.sort( results );
          for( Result result : results )
             System.out.printf( "%-30.30s : %d\n", result.site, result.time );
          System.out.println("------------------");
       }

       public void start( String [] args )
       {
          Runnable showResultsAction = new Runnable() {
             public void run() {
                showResults();
                results.clear();
             } };
          barrier = new CyclicBarrier( args.length, showResultsAction );

          for ( final String site : args )
             new Thread() {
                public void run() {
                   while( true ) {
                      long time = timeConnect( site );
                      results.add( new Result( time, site ) );
                      try {
                         barrier.await();
                      } catch ( BrokenBarrierException e ) { return;
                      } catch ( InterruptedException e ) { return; }
                   }
                }
             }.start();
       }

       public static void main( String [] args ) throws IOException {
          new SiteTimer().start( args );
       }
    }

    Exchanger<ByteBuffer> xchange = new Exchanger<ByteBuffer>();

    // thread 1
    Buffer nextBuf = xchange.exchange( buffer1 ); // blocks

    // thread 2
    Buffer nextBuf = xchange.exchange( buffer2 );

    // buffers exchanged, both threads continue...