Overcoming Java 8 streams

Java 8 streams provide functional and LINQ-like features in a fluent API. But streams are not without drawbacks:

• Referenced methods and lambdas cannot throw checked exceptions
• Controlling the threads used, especially for parallel streams, is awkward
• Streams are not designed for extension

Overcoming these drawbacks requires a "look-a-like" API. For example, implementing java.util.stream.Stream does not help: none of the existing methods throw checked exceptions, and none of the existing stream factory helpers would return your implementation with new methods.

So I wrote my own, copying the existing stream API, updating the methods to throw checked exceptions:

hm.binkley.util.stream.CheckedStream ('develop' branch for now)

From the javadoc:

CheckedStream is a throwing Stream look-a-like with control over thread pool. It cannot be a Stream as it takes throwing versions of suppliers, functions and consumers. Otherwise it is a faithful reproduction.

Write this:

   long beanCount() throws SomeException, OtherException {
       checked(Stream.of(1, 2, 3)).
           map(this::someThrowingFunction).
           peek(That::oldBean).
           count();
   }

not this:

   long beanCount() throws SomeException, OtherException {
       try {
           Stream.of(1, 2, 3).
               map(i -> {
                   try {
                       someThrowingFunction(i);
                   } catch (final SomeException e) {
                       throw new RuntimeException(e);
                   }
               }).
               peek(i -> {
                   try {
                       That.oldBean(i);
                   } catch (final OtherException e) {
                       throw new RuntimeException(e);
                   }
               }).
               count();
       } catch (final RuntimeException e) {
           final Throwable x = e.getCause();
           if (x instanceof SomeException)
               throw (SomeException) x;
           if (x instanceof OtherException)
               throw (OtherException) x;
           throw e;
       }
   }

"Intentional" exceptions (checked exceptions plus CancellationException) have "scrubbed" stacktraces: frames from framework/glue packages are removed before the intentional exception is rethrown to calling code. Scrubbed stacktraces are much easier to understand, the framework and glue code having been removed.

To see the unscrubbed stacktrace, set the system property "hm.binkley.util.stream.CheckedStream.debug" to "true".

Controlling the thread pool used by Stream is a challenge. Deep in the implementation, it checks if being run in a ForkJoinTask, and uses that thread if so, otherwise using the common pool. So with CheckedStream write this:

       checked(stream, new ForkJoinPool()).
           map(currentThread()).
           forEach(System.out::println);

not this:

       try {
           new ForkJoinPool().submit(() -> stream.
                   map(currentThread()).
                   forEach(System.out::println)).
               get();
       } catch (final ExecutionException e) {
           final Throwable x = e.getCause();
           if (x instanceof Error)
               throw (Error) x;
           if (x instanceof RuntimeException)
               // Much tricker when stream functions throw runtime
               throw (RuntimeException) x;
           throw new Error(e); // We have no checked exceptions in this example
       }

Care is taken to respect lazy and terminal operations in using thread pools. Changing thread pool or thread mode mid-stream is supported, and are "immediate" operations: they terminate the existing stream, and start a new one with the changes:

stream.sequential().
    filter(this::someFilter).
    parallel(threads). // Existing lazy operations terminated
    map(this:someMapper).
    forEach(System.out::println);

Immediate operations ensure stream methods are run in the correct threading context.

I hope you'll agree: CheckedStream is nicer to use, especially with existing code using checked exceptions.

Suggestions, bug fixes, improvements welcome!

State of the art Java lock-free queues

Nitsan Wakart writes on optimizing SPSC lock-free queues (single producer, single consumer) with lots of code samples, performance data and refernces. Even the comments are interesting. An excellent read to understand state of the art in Java.

A random blub:

There are a few subtleties explored here:

• I'm using Unsafe for array access, this is nothing new and is a cut and paste job from the AtomicReferenceArray JDK class. This means I've opted out of the array bound checks we get when we use arrays in Java, but in this case it's fine since the ring buffer wrapping logic already assures correctness there. This is necessary in order to gain access to getVolatile/putOrdered.
• I switched Pressanas original field padding method with mine, mostly for consistency but also it's a more stable method of padding fields (read more on memory layout here).
• I doubled the padding to protect against pre-fetch caused false sharing (padding is omitted above, but have a look at the code).
• I replaced the POW final field with a constant (ELEMENT_SHIFT). This proved surprisingly significant in this case. Final fields are not optimized as aggressively as constants, partly due to the exploited backdoor in Java allowing the modification of final fields (here's Cliff Click's rant on the topic). ELEMENT_SHIFT is the shift for the sparse data and the shift for elements in the array (inferred from Unsafe.arrayIndexScale(Object[].class)) combined.

Is it live, or is it structured deferral?

A delightful ACM paper by Paul E. McKenney, Structured Deferral: Synchronization via Procrastination. The "motivating example":

In this example, Schrödinger would like to construct an in-memory database to keep track of the animals in his zoo. Births would of course result in insertions into this database, while deaths would result in deletions. The database is also queried by those interested in the health and welfare of Schrödinger's animals.

Schrödinger has numerous short-lived animals such as mice, resulting in high update rates. In addition, there is a surprising level of interest in the health of Schrödinger's cat,19 so much so that Schrödinger sometimes wonders whether his mice are responsible for most of these queries. Regardless of their source, the database must handle the large volume of cat-related queries without suffering from excessive levels of contention. Both accesses and updates are typically quite short, involving accessing or mutating an in-memory data structure, and therefore synchronization overhead cannot be ignored.

Schrödinger also understands, however, that it is impossible to determine exactly when a given animal is born or dies. For example, suppose that his cat's passing is to be detected by heartbeat. Seconds or even minutes will be required to determine that the poor cat's heart has in fact stopped. The shorter the measurement interval, the less certain the measurement, so that a pair of veterinarians examining the cat might disagree on exactly when death occurred. For example, one might declare death 30 seconds after the last heartbeat, while another might insist on waiting a full minute, in which case the veterinarians disagree on the state of the cat during the second half of the minute after the last heartbeat.

Fortunately, Heisenberg8 has taught Schrödinger how to cope with such uncertainty. Furthermore, the delay in detecting the cat's passing permits use of synchronization via procrastination. After all, given that the two veterinarians' pronouncements of death were separated by a full 30 seconds, a few additional milliseconds of software procrastination is perfectly acceptable.

Among the pleasures in this paper: Linux's RCU, hazard pointers, and "read-only cat-heavy performance of Schrödinger's hash table".

Enjoy!

JUnit testing that a call blocks

Certain that I missed an existing solution, I post this in hopes of helping anyone who also misses it. But if you know a better way, please comment.

Usage

class SomeTest {
    @Test(timeout = 1000L)
    public void shouldBlock() {
        assertBlocks(new BlockingCall() {
            @Override
            public void call()
                    throws InterruptedException {
                // My blocking code here
            }
        });
    }
}

Implementation

import javax.annotation.Nonnull;
import java.util.Timer;
import java.util.TimerTask;

import static java.lang.String.format;
import static java.lang.Thread.currentThread;
import static org.junit.Assert.fail;

/**
 * {@code BlockingCall} supports blocking call assertion for JUnit tests.
 *
 * @author <a href="mailto:binkley@alumni.rice.edu">B. K. Oxley (binkley)</a>
 */
public interface BlockingCall {
    /**
     * Calls blocking code.  The blocking code must throw {@code
     * InterruptedException} when its current thread is interrupted.
     *
     * @throws InterruptedException if interrupted.
     */
    void call()
            throws InterruptedException;

    /** Wrapper class for block assertion. */
    public static final class Assert {
        /**
         * Asserts the given <var>code</var> blocks at least 100ms.  Interrupts
         * the blocking code after 100ms and checks {@code InterruptedException}
         * was thrown.  When not blocking, appends <var>block</var> to the
         * failure message.
         *
         * @param block the blocking call, never missing
         */
        public static void assertBlocks(@Nonnull final BlockingCall block) {
            final Timer timer = new Timer(true);
            try {
                final Thread current = currentThread();
                timer.schedule(new TimerTask() {
                    @Override
                    public void run() {
                        current.interrupt();
                    }
                }, 100L);
                block.call();
                fail(format("Did not block: %s", block));
            } catch (final InterruptedException ignored) {
            } finally {
                timer.cancel();
            }
        }
    }
}

UPDATE: Idioms like this would be far more attractive with Java 8 lambdas:

public void shouldBlock() {
    assertBlocks(() -> { /* blocking code */ });
}

Baker's Treadmill and Disruptor

Flying Frog Consultancy in Cambridge, England posts a nice comparison between Disruptor and low-latency GC algorithms:

The core of [Disruptor's] idea is to accomplish all of this message passing using a single shared data structure, the disruptor, rather than using several separate concurrent queues.

Baker's Treadmill is a low-latency garbage collection algorithm. Allocated blocks in the heap are linked together to form a cyclic doubly-linked list that is divided into four segments using four iterators that chase each other around the ring as heap blocks are allocated, traced and freed.

Cleverly observed with a good diagram. As they say, read the whole thing.

Martin Thompson on lock cost

Anything Martin Thompson says about performance is usually worth a good study. No exception is Locks & Condition Variables - Latency Impact.

His takeaway points:

1. The one-way latency to signal a change is pretty much the same as what is considered current state of the art for network hops between nodes via a switch.
2. The impact is clear when letting the OS choose what CPUs the threads get scheduled on rather than pinning them manually.

Peregrine: Help for programming concurrency

Good news from Columbia University. Peregrine is automated software to aid in programming concurrency. More in the ACM: Major Breakthrough Improves Software Reliability and Security:

"Our main finding in developing Peregrine is that we can make threads deterministic in an efficient and stable way: Peregrine can compute a plan for allowing when and where a thread can 'change lanes' and can then place barriers between the lanes, allowing threads to change lanes only at fixed locations, following a fixed order," says Columbia professor Junfeng Yang. "Once Peregrine computes a good plan without collisions for one group of threads, it can reuse the plan on subsequent groups to avoid the cost of computing a new plan for each new group."

Non-portable

Elliott Hughes quips:

Oh, yeah... The "_np" suffix means "non-portable". The pthread_getattr_np(3) function is available on glibc and bionic, but not (afaik) on Mac OS. But let's face it; any app that genuinely needs to query stack addresses and sizes is probably going to contain an #ifdef or two anyway...

I am a bit late but more marvel from Trisha of LMAX, Dissecting the Disruptor: Demystifying Memory Barriers. How can one not love this writing?

Compilers and CPUs can re-order instructions, provided the end result is the same, to try and optimise performance. Inserting a memory barrier tells the CPU and the compiler that what happened before that command needs to stay before that command, and what happens after needs to stay after. All similarities to a trip to Vegas are entirely in your own mind.

And with diagrams.

Farwell LinkedBlockingQueue

More astonishing figures for the LMAX disruptor. More results like this and I feel a new JDK concurrency framework coming soon.

UPDATE:Trisha posts more explanation of the interesting technical tricks in Disruptor: magic cache line padding.

New Shimmer for Java Concurrency

At work we use this Java idiom to coordinate start up:

public class LatchExecutorService
        implements ExecutorService {
    private final ExecutorService threadPool
            = newSingleThreadExecutor();
    private final CountDownLatch latch;

    public LatchExecutorService(final int count) {
        latch = new CountDownLatch(count);
        threadPool.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    latch.await();
                    threadPool.shutdown();
                } catch (final InterruptedException e) {
                    threadPool.shutdownNow();
                }
            }
        });
    }

    public void countDown() {
        latch.countDown();
    }

    public int getCount() {
        return latch.getCount();
    }

    public <T> Future<T> submit(final Callable<T> task) {
        try {
            return threadPool.submit(task);
        } catch (final RejectedExecutionException e) {
            task.run();
        }
    }

    public <T> Future<T> submit(final Runnable task,
            final T result) {
        try {
            return threadPool.submit(task, result);
        } catch (final RejectedExecutionException e) {
            task.run();
        }
    }

    public Future<?> submit(final Runnable task) {
        try {
            return threadPool.submit(task);
        } catch (final RejectedExecutionException e) {
            task.run();
        }
    }

    // Delegate remaining executor service methods
    // Cut for brevity
}

This class is like a count down latch but with the extra "runnable" feature of a cyclic barrier. When the latch fully counts down, it executes any registered runnables in order of submission. And it is also an executor service.

Some explanation of the unusual submit semantics. The intent of the class is use at start up, to delay actions that depend on some common initialization but guarantee their execution without coordination of parties.

Say there are four worker threads at start up performing essential work. The rest of the program works in a reduced state until these four complete. Rather than carry four independent signals, I collapsed all into LatchExecutorService.

As tasks are submitted, LatchExecutorService delays their execution until its latch fully counts down. With a blocking task at the head of the single-threaded inner executor service, all other tasks must wait on the latch. When the latch triggers, the blocking task clears and all waiting tasks execute on the inner executor service thread.

Once the latch triggers, I do not need the inner executor service any longer except to clear out waiting tasks so I call shutdown() and free up that thread.

I want any new tasks submitted to run immediately. Rather than have them fail or need to coordinate between slow submitters and fast count downers, I run submitted tasks in their caller's thread on rejection. This guarantees new tasks eventually execute without any coordination, either on the original inner executor service or directly at submission.

You can also view LatchExecutorService as an inside-out cyclic barrier. Rather than one runnable registered at construction to execute when all parties signal, callers can provide as many runnables as they like at any time.

It's a count down latch; it's a cyclic barrier; it's an executor service. It's New Shimmer.

Upgrading read lock to write lock in Java

Setup

I ran into a great question on stackoverflow while researching a concurrency question: how to you upgrade a read lock to a write lock?

The answer is easy: you can't. Trying to do so gives you deadlock.

The discussion covers a DAG data structure, but I wanted to try a simpler case.

Simplest case

Say you have an idempotent "resource" (intentionally vague):

public interface Resource {
    void something() throws IOException;
}

The resource can spoil, and you want try another instance automatically. The obvious thing:

public class RetryResourceFacade
        implements Resource {
    private Resource resource;

    public RetryResourceFacade() {
        refreshResource();
    }

    public void something() {
        while (true) {
            try {
                resource.something();
                return;
            } catch (final IOException e) {
                refreshResource();
            }
        }
    }

    private void refreshResource() {
        while (true) {
            try {
                resource = newResource();
                return;
            } catch (final IOException e) {
                // try another one
            }
        }
    }

    public static Resource newResource()
            throws IOException {
        return null; // unlikely in real life
    }
}

Essentially when a resource spoils, you try another one until one works. But this is horrible for concurrent code: how to fix it?

Thread-safe

public class RetryResourceFacade
        implements Resource {
    private final Object lock = new Object();

    private volatile Resource resource;

    public RetryResourceFacade() {
        refreshResource();
    }

    public void something() {
        synchronized(lock) {
            while (true)
                try {
                    resource.something();
                    return;
                } catch (final IOException e) {
                    refreshResource();
                }
            }
        }
    }

    private void refreshResource() {
        while (true) {
            try {
                resource = newResource();
                return;
            } catch (final IOException e) {
                // try another one
            }
        }
    }

    public static Resource newResource()
            throws IOException {
        return null; // unlikely in real life
    }
}

Well, better, but still—do I really want to single-thread calls to something()? I can do yet better still:

More throughput

public class RetryResourceFacade
        implements Resource {
    private final Lock readLock;
    private final Lock writeLock;

    private volatile Resource resource;

    public RetryResourceFacade() {
        final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
        readLock = lock.readLock();
        writeLock = lock.writeLock();

        refreshResource();
    }

    public void something() {
        while (true) {
            readLock.lock();
            try {
                resource.something();
                readLock.unlock();
                return;
            } catch (final IOException e) {
                readLock.unlock();
                refreshResource();
            }
        }
    }

    private void refreshResource() {
        writeLock.lock();
        while (true) {
            try {
                resource = newResource();
                writeLock.unlock();
                return;
            } catch (final IOException e) {
                // try another one
            }
        }
    }

    public static Resource newResource()
            throws IOException {
        return null; // unlikely in real life
    }
}

By introducing read/write locks, I can let many callers use something() but single-thread refreshing the resource (I do not want several callers refreshing resource at once).

Best of all

But there is one final problem. What happens when two callers are both in something() and the same resource spoils? They both race to refreshResource(), one of them wins and gets a valid resource, but the other then throws that good one away and gets another new resource. Wasteful and embarrassing. The final solution:

public class RetryResourceFacade
        implements Resource {
    private final Lock readLock;
    private final Lock writeLock;

    private volatile Resource resource;
    private volatile boolean valid;

    public RetryResourceFacade() {
        final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
        readLock = lock.readLock();
        writeLock = lock.writeLock();

        refreshResource();
    }

    public void something() {
        while (true) {
            readLock.lock();
            try {
                resource.something();
                readLock.unlock();
                return;
            } catch (final IOException e) {
                valid = false;
                readLock.unlock();
                refreshResource();
            }
        }
    }

    private void refreshResource() {
        writeLock.lock();
        if (valid) {
            // Another caller already fixed resource
            writeLock.unlock();
            return;
        }

        while (true) {
            try {
                resource = newResource();
                valid = true;
                writeLock.unlock();
                return;
            } catch (final IOException e) {
                // try another one
            }
        }
    }

    public static Resource newResource()
            throws IOException {
        return null; // unlikely in real life
    }
}

Only refresh the resource if it is invalid. This is a little bit like handling spurious wakeup in that several threads compete for the resource, but only the first one gets to actually do something with it.

Key points

• Back to the setup: avoid deadlock. You cannot obtain the write lock while holding the read lock, so release the read lock first, then obtain the write lock.
• Use a status flag to ensure the write lock section is not executed needlessly.
• You must mark resource and valid as volatile to ensure "publish" semantics (Jeremy Mason has a great post on this), otherwise anything could happen (really).

UPDATE: Thanks to Darren Accardo for key ideas in this post; his threading is stronger than mine!

Actor thinking from Kresten Krab Thorup

Kresten Krab Thorup posts a long think on objects and processes, full of great references, which I enjoy. Maybe you will, too.

Keeping a current timestamp in a Java concurrent map

Tricker than I like: In Java, how do you keep a map with the values being the "most recent timestamp" for the key? How do you do this for a concurrent map with multiple threads updating?

boolean isLatest(final K key, final long newTimestamp,
        final ConcurrentMap<K, Long> map) {
    Long oldTimestamp = map.putIfAbsent(key, newTimestamp);

    // If we are first, there is no previous to compare
    // with
    if (null == oldTimestamp)
        return true;

    // Make sure we are still the most recent
    while (newTimestamp > oldTimestamp) {
        if (map.replace(key, oldTimestamp, newTimestamp))
            // No other thread has updated, we are most
            // recent
            return true;

        // Another thread got here first, recheck
        oldTimestamp = map.get(key);
    }

    // Another thread put in a newer update, we are not
    // most recent
    return false;
}

As with all multi-threaded coding, bugs are easy to make and hard to spot. If I have made one, please let me know!

UPDATE: Fixed a typo. Thanks, David!

Dataflow concurrency for Java

I ran across this interesting Scala class by Vaclav Pech which makes the data concurrent rather than the code (if I understood the author correctly). The ~ (extraction) and << (insertion) operators looked nifty.

Looking to do the same in Java, I see four key requirements:

1. Insertion can happen only once.
2. Extraction is idempotent (infinitely repeatable, non-modifying).
3. Extraction blocks until insertion has completed.
4. Insertion and extraction are both atomic.

With these in mind, I write this:

public class DataFlowReference<T> {
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition condition = lock.newCondition();
    private volatile T item;
    private volatile boolean set = false;
    private volatile boolean called = false;

    public T get()
            throws InterruptedException {
        lock.lockInterruptibly();
        try {
            while (true) {
                if (set)
                    return item;

                if (!called) {
                    called = true;
                    onFirstGet();
                    continue;
                }

                condition.await();
            }
        } finally {
            lock.unlock();
        }
    }

    public void setOnce(final T value) {
        lock.lock();
        try {
            if (set)
                return;

            set = true;
            item = value;

            condition.signalAll();
        } finally {
            lock.unlock();
        }
    }

    protected void onFirstGet() {
    }

    // Object

    @Override
    public boolean equals(final Object o) {
        if (this == o)
            return true;
        if (o == null || getClass() != o.getClass())
            return false;

        final DataFlowReference that = (DataFlowReference) o;

        return set && item.equals(that.item);
    }

    @Override
    public int hashCode() {
        return set ? item.hashCode() : super.hashCode();
    }

    @Override
    public String toString() {
        return "(" + set + ':' + item + ')';
    }
}

Use of DataFlowReference follows the examples from Vaclav's page: declare dataflow references, create threads whose runables use setOnce() and get(), invoke them all together.

onFirstGet() supports "just in time" supply of item and would call setOnce(T) with a fresh value.

UPDATE: The original version of this class had a horrible, obvious race condition. Caveat plicator.

Jumping the work queue in Executor

My server has an ExecutorService for job scheduling. It came up that we need to have some jobs "jump the queue", that is, go to the front of the line waiting for the next free thread. What to do?

A web search came up empty. After hemming and hawing, and polling other teams, I went with this solution (below). Comments and improvements welcome.

public class PriorityExecutor
        extends ThreadPoolExecutor {
    public PriorityExecutor() {
        super(0, Integer.MAX_VALUE, 60L, SECONDS,
                new PriorityBlockingQueue<Runnable>(11,
                        new PriorityTaskComparator()));
    }

    public PriorityExecutor(final ThreadFactory threadFactory) {
        super(0, Integer.MAX_VALUE, 60L, SECONDS,
                new PriorityBlockingQueue<Runnable>(11,
                        new PriorityTaskComparator()), threadFactory);
    }

    public PriorityExecutor(final RejectedExecutionHandler handler) {
        super(0, Integer.MAX_VALUE, 60L, SECONDS,
                new PriorityBlockingQueue<Runnable>(11,
                        new PriorityTaskComparator()), handler);
    }

    public PriorityExecutor(final ThreadFactory threadFactory,
            final RejectedExecutionHandler handler) {
        super(0, Integer.MAX_VALUE, 60L, SECONDS,
                new PriorityBlockingQueue<Runnable>(11,
                        new PriorityTaskComparator()), threadFactory, handler);
    }

    @Override
    protected <T> RunnableFuture<T> newTaskFor(final Callable<T> callable) {
        if (callable instanceof Important)
            return new PriorityTask<T>(((Important) callable).getPriority(),
                    callable);
        else
            return new PriorityTask<T>(0, callable);
    }

    @Override
    protected <T> RunnableFuture<T> newTaskFor(final Runnable runnable,
            final T value) {
        if (runnable instanceof Important)
            return new PriorityTask<T>(((Important) runnable).getPriority(),
                    runnable, value);
        else
            return new PriorityTask<T>(0, runnable, value);
    }

    public interface Important {
        int getPriority();
    }

    private static final class PriorityTask<T>
            extends FutureTask<T>
            implements Comparable<PriorityTask<T>> {
        private final int priority;

        public PriorityTask(final int priority, final Callable<T> tCallable) {
            super(tCallable);

            this.priority = priority;
        }

        public PriorityTask(final int priority, final Runnable runnable,
                final T result) {
            super(runnable, result);

            this.priority = priority;
        }

        @Override
        public int compareTo(final PriorityTask<T> o) {
            final long diff = o.priority - priority;
            return 0 == diff ? 0 : 0 > diff ? -1 : 1;
        }
    }

    private static class PriorityTaskComparator
            implements Comparator<Runnable> {
        @Override
        public int compare(final Runnable left, final Runnable right) {
            return ((PriorityTask) left).compareTo((PriorityTask) right);
        }
    }
}

Some points:

• In JDK5 you will need to add three of:

      @Override
      public Future<?> submit(final Runnable task) {
          if (task == null)
              throw new NullPointerException();
          final RunnableFuture<Object> ftask = newTaskFor(task, null);
          execute(ftask);
          return ftask;
      }
  

  These extension points were added in JDK6.
• It might be useful to include a ThreadFactory which gives important threads higher scheduling priority. It should reuse a factory from the constructor.
• Remember to take advantage of NORM_PRIORITY and his cohorts.

Callers create their own Runnable and Callable classes which implement Important to communicate their priority. Then simply submit instances of these as usual. Jobs with the highest priority jump to the front of the work queue in the executor waiting for the next free thread.