Zio-kafka, faster than java-kafka

TlDR: Concurrency and pre-fetching gives zio-kafka a higher consumer throughput than the default java Kafka client for most workloads.

Zio-kafka is an asynchronous Kafka client based on the ZIO async runtime. It allows you to consume from Kafka with a ZStream per partition. ZStreams are streams on steroids; they have all the stream operators known from e.g. RX-Java and Monix, and then some. ZStreams are supported by ZIO's super efficient runtime.

So the claim is that zio-kafka consumes faster than the regular Java client. But zio-kafka wraps the default java Kafka client. So how can it be faster? The trick is that with zio-kafka, processing (meaning your code) runs in parallel with the code that pulls records from the Kafka broker.

Obviously, there is some overhead in distributing the received records to the streams that need to process it. So the question then is, when is the extra overhead of zio-kafka less than the gain by parallel processing? Can we estimate this? In turns out we can! For this estimate we use the benchmarks that zio-kafka runs from GitHub. In particular, we look at these 2 benchmarks with runs from 2024-11-30:

• the throughput benchmark, uses zio-kafka and takes around 592 ms
• the kafkaClients benchmark, uses Kafka java client directly and takes around 544 ms

Both benchmarks:

• run using JMH on a 4 core github runner with 16GB RAM
• consume 50k records of ~512 bytes from 6 partitions
• process the records by counting the number of incoming records, batch by batch
• are using the same consumer settings, in particular max.poll.records is set to 1000
• the broker runs in the same JVM as the consumer, so there is almost no networking overhead
• do not commit offsets (but note that committing does not change the outcome of this article)

First we calculate the overhead of java-kafka. For this we assume that counting the number of records takes no time at all. This is reasonable, a count operation is just a few CPU instructions, nothing compared to fetching something over the network, even if it is on the same machine. Therefore, in the figure, the 'processing' rectangle collapses to a thick vertical line.

Polling and processing as blocks of execution on a thread/fiber.

We also assume that every poll returns the same amount of records, and takes the same amount of time. This is not how it works in practise, but when the records are already available we're probably not that far off. As the program consumes 50k records, and each poll returns a batch of 1k records, there are 50 polls. Therefore, the overhead of java-kafka is 544/50 ≈ 11 ms per poll.

Now we can calculate the overhead of zio-kakfa. Again we assume that processing takes no time at all, so that every millisecond that the zio-kafka benchmark takes longer can be attributed to zio-kafka's overhead. The zio-kakfa benchmark runs longer for 592-544 = 48 ms. Therefore zio-kafka's overhead is 48/50 ≈ 1ms per poll.

Now lets look at a more realistic scenario where processing does take time.

Polling and processing as blocks of execution on a thread/fiber.

As you can see a java-kafka program alternates between polling and processing, while the zio-kafka program processes the records in parallel distributed over 6 streams (1 stream per partition, each stream runs independently on its own ZIO fiber). In the figure we assume that the work is evenly distributed, but unless the load is really totally skewed, it won't matter that much in practice due to zio-kafka's per-partition pre-fetching. As you can see the zio-kafka program in this scenario is much faster. But is this a realistic example? When do zio-kafka programs become faster than java-kafka programs?

Observe that how long polling takes is not important for this question because polling time is the same for both programs. So we will only look at the time between polls. We use (p) for the processing time per batch for the java-kafka program. For the zio-kafka program time between polls is equal to zio-kafka's overhead (o) plus processing time per batch divided by the number of partitions (n). So we want to know for which p:

$\mathrm{o}+\frac{\mathrm{p}}{\mathrm{n}}\le \mathrm{p}$

This solves to:

$\mathrm{p}\ge \frac{\mathrm{o\; n}}{\mathrm{n\; -\; 1}}$

For the benchmark we fill in the zio-kakfa overhead (o = 1 ms) and the number of partitions (n = 6) and we get: p ≥ 1.2 ms. (Remember, this is processing time per batch of 1000 records.)

In the previous paragraph we assumed that processing is IO intensive. When the processing is compute intensive, ZIO can not actually run more fibers in parallel than the number of cores available. In our example we have 4 cores, using n = 4 gives p ≥ 1.3 ms which is still a very low tipping point.

Conclusion

For IO loads, or with enough cores available, even quite low processing times per batch, makes zio-kafka consume faster than the plain java Kafka library. The plain java consumer is only faster for trivial processing tasks like counting.

If you'd like to know what this looks like in practice, you can take a look at this real-world example application: Kafka Big-Query Express. This is a zio-kafka application recently open sourced by Adevinta. Here is the Kafka consumer code. (Disclaimer: I designed and partly implemented this application.)

Scheduling tasks and sharing state with streams

Recently we built a system that needs to perform 2 tasks. Taks 1 runs every 15 minutes, task 2 runs every 2 minutes. Task 1 kicks off some background jobs (an upload to BigQuery), task 2 checks upon the results of these background jobs and does some cleanup when they are done (delete the uploaded files). The two tasks need to share information back and forth about what jobs are running in the background.

Now think to yourself (ignore the title for now 😉), what would be the most elegant way to implement this? I suspect that most developers will come with a solution that involves some locking, synchronisation and global state. For example by sharing the information through a transactional database, or by using a semaphore to prevent the two tasks from running at the same time plus sharing information in a global variable. This is understandable, most programming environments do not provide better techniques for these kinds of problems at all!

However, if your environment supports streams and has some kind of scheduling, here are two tricks you can use: one for the scheduling of the tasks, the second for sharing information without a global variable.

Here is an example for the first written in Scala using the ZIO streams library. Read on for an explanation.

import zio._ import zio.stream._ def performTask1: Task[Unit] = ??? def performTask2: Task[Unit] = ??? // An enumeration (scala 2 style) for our task. sealed trait BusinessTask object Task1 extends BusinessTask object Task2 extends BusinessTask ZStream.mergeAllUnbounded()( ZStream.fromSchedule(Schedule.fixed(15.minutes)).as(Task1), ZStream.fromSchedule(Schedule.fixed(2.minutes)).as(Task2) ) .mapZIO { case Task1 => performTask1 case Task2 => performTask2 } .runDrain

We create 2 streams, each stream contains sequential numbers, emitted upon a schedule. As you can see, the schedule corresponds directly with the requirements. We do not really care for the sequential numbers, so with stream operator as we convert the stream's emitted values to a value from the BusinessTask enumeration.

Then we merge the two streams. We now have a stream that emits the two enumeration values at the time the corresponding task should run. This is already a big win! Even when the two schedules produce an item at the same time, the tasks will run sequentially. This is because by default streams are evaluated without parallelism.

We are not there yet though. The tasks need to share information. They could access a shared variable but then we still have tightly coupled components and no guarantees that the shared variable is used correctly.

Also, wouldn't it be great if performTask1 and performTask2 are functions that can be tested in isolation? With streams this is possible.

Here is the second part of the idea. Again, read on for an explanation.

case class State(...) val initialState = State(...) def performTask1(state: State): Task[State] = ??? def performTask2(state: State): Task[State] = ??? ZStream.mergeAllUnbounded()( ZStream.fromSchedule(Schedule.fixed(15.minutes)).as(Task1), ZStream.fromSchedule(Schedule.fixed(2.minutes)).as(Task2) ) .scanZIO(initialState) { (state, task) => task match { case Task1 => performTask1(state) case Task2 => performTask2(state) } } .runDrain

We have changed the signatures of the performTask* methods. Also, the mapZIO operator has been replaced with scanZIO. The stream operator scanZIO works much like foldLeft on collections. Like foldLeft, it accepts an initial state, and a function that combines the accumulated state plus the next stream element (of type BusinessTask) and converts those into the next state.

Stream operator scanZIO also emits the new states. This allows further common processing. For example we can persist the state to disk, or collect custom metrics about the state.

Conclusion

Using libraries with higher level constructs like streams, we can express straightforward requirements in a straightforward way. With a few lines of code we have solved the scheduling requirement, and showed an elegant way of sharing information between tasks without global variables.

Speed up ZIOs with memoization

TLDR: You can do ZIO memoization in just a few lines, however, use zio-cache for more complex use cases.

Recently I was working on fetching Avro schema's from a schema registry. Avro schema's are immutable and therefore perfectly cacheable. Also, the number of possible schema's is limited so cache evictions are not needed. We can simply cache every schema for ever in a plain hash-map. So, we are doing memoization.

Since this was the first time I did this in a ZIO based application, I looked around for existing solutions. What I wanted is something like this:

def fetchSchema(schemaId: String): Task[Schema] = { val fetchFromRegistry: Task[Schema] = ??? fetchFromRegistry.memoizeBy(key = schemaId) }

Frankly, I was a bit disappointed that ZIO does not already support this out of the box. However, as you'll see in this article, the proposed syntax only works for simple use cases. (Actually, there is ZIO.memoize but that is even simpler and only caches the result for a single ZIO instance, not for any instance that gives the same value.)

Let's continue anyway and implement it ourselves.

The idea is that method memoizeBy first looks in a map using the given key. If the value is not present, we get the result from the original Zio and store it in the map. If the value is present, it will be used and the original Zio is not executed.

A map, yes, we need also need to give the method a map! The map might be used and updated concurrently. I choose to wrap an immutable map in a Ref, but you could also use a ConcurrentMap.

Here we go:

import zio._ import scala.collection.immutable.Map implicit class ZioMemoizeBy[R, E, A](zio: ZIO[R, E, A]) { def memoizeBy[B](cacheRef: Ref[Map[B, A]], key: B): ZIO[R, E, A] = { for { cache <- cacheRef.get value <- cache.get(key) match { case Some(value) => ZIO.succeed(value) case None => zio.tap(value => cacheRef.update(_.updated(key, value))) } } yield value } }

That is it, just a few lines of code to put in some corner of your application.

Here is a full example with memoizeBy using Service Pattern 2.0:

import org.apache.avro.Schema import zio._ import scala.collection.immutable.Map trait SchemaFetcher { def fetchSchema(schemaId: String): Task[Schema] } object SchemaFetcherLive { val layer: ZLayer[Any, Throwable, SchemaFetcher] = ZLayer { for { // Create the Ref and the Map: schemaCacheRef <- Ref.make(Map.empty[String, Schema]) } yield SchemaFetcherLive(schemaCacheRef) } } case class SchemaFetcherLive( schemaCache: Ref[Map[String, Schema]] ) extends SchemaFetcher { def fetchSchema(schemaId: String): Task[Schema] = { val fetchFromRegistry: Task[Schema] = ... // Use memoizeBy to make fetchFromRegistry more efficient! fetchFromRegistry.memoizeBy(schemaCache, schemaId) } }

Discussion

Note how we're using the default immutable Map. Because it is immutable, all threads can read from the map at the same time without synchronization. We only need some synchronization using Ref, to atomically replace the map after a new element was added.

When multiple requests for the same key come in at roughly the same time, both are executed, and both lead to an update of the map. This is not as advanced as e.g. zio-cache, which detects multiple simultaneous requests for the same key. In the presented use case this is not a problem and very unlikely to happen often anyway.

Can we improve further? Yes, we can! If you look at method fetchSchema in the example, you see that a fetchFromRegistry ZIO is constructed, but we do not use it when the value is already present. And even worse, the value already being present is the common case! This is not very efficient. If efficiency is a problem, another API is needed. Zio-cache does not have this problem. In zio-cache the cache is aware of how to look up new values (it is a loading cache). So here is a trade off: efficiency with a more complex API, or highly readable code.

Using zio-cache

For completeness, here is (almost) the same example using zio-cache:

import org.apache.avro.Schema import zio._ import zio.cache.{Cache, Lookup} trait SchemaFetcher { def fetchSchema(schemaId: String): Task[Schema] } object ZioCacheSchemaFetcherLive { val layer: ZLayer[SomeService, Throwable, SchemaFetcher] = ZLayer { for { someService <- ZIO.service[SomeService] // the fetching logic can use someService: fetchFromRegistry: String => Task[Schema] = ??? // create the cache: cache <- Cache.make( capacity = 1000, timeToLive = Duration.Infinity, lookup = Lookup(fetchFromRegistry) ) } yield ZioCacheSchemaFetcherLive(cache) } } case class ZioCacheSchemaFetcherLive( cache: Cache[String, Throwable, Schema] ) extends SchemaFetcher { def fetchSchema(schemaId: String): Task[Schema] = { // use the loader cache: cache.get(schemaId) } }

We now need a reference to fetchFromRegistry while constructing the layer. This complicates the code a bit; we can no longer define fetchFromRegistry in the case class. In the example we pull a SomeService so that we can put the definition of fetchFromRegistry into the for comprehension and stick to Service Pattern 2.0. Perhaps we should completely move it to another service so that we can write lookup = Lookup(someService.fetchFromRegistry). That, I'll leave as an exercise to the reader.

Conclusion

For simple use cases like fetching Avro schema's, this article presents an appropriately light weight way to do memoization. If you need more features such as eviction and detection of concurrent invocations, I recommend zio-cache.

Update 2024-01-24

Here is a version of cachedBy that only fetches a value once, even when two fibers request it concurrently. The second fiber is semantically blocked until the first fiber has produced the value.

import zio._ object ZioCaching { implicit class ZioCachedBy[R, E, A](zio: ZIO[R, E, A]) { def cachedBy[B](cacheRef: Ref[Map[B, Promise[E, A]]], key: B): ZIO[R, E, A] = { for { newPromise <- Promise.make[E, A] actualPromise <- cacheRef.modify { cache => cache.get(key) match { case Some(existingPromise) => (existingPromise, cache) case None => (newPromise, cache + (key -> newPromise)) } } _ <- ZIO.when(actualPromise eq newPromise) { zio.intoPromise(newPromise) } value <- actualPromise.await } yield value } } }

Breaking the Circuit Breaker

The circuit breaker is this wonderful pattern to protect your application against resources that fail slowly. The idea is that you stop trying to use a resource when it has too many failures. Regular retries test the resource and will make the resource available again. The benefit is that your application can react quickly to a failed resource instead of hogging CPU, threads, network, etc. while you are waiting to find out the resource is unavailable.

So what's wrong?

Its the metaphor. In the classical description a circuit breaker has 3 states: the open state, the closed state and the half-open state. So what does it mean when the circuit breaker is open? When is a bridge open? When you can drive over it, or when you can sail through it? Only when you look at the first image you may see that a traditional open circuit breaker stops flow of electricity. To us that translates to no usage of the resource. In the 'closed' state electricity flows, which translates to having access to our resource. Now read that again and see if you can remember that!

Then we have a half-open state? Again, look at the first image. For such a switch half-open is still open. (A half-open bridge lets no traffic trough at all but that is another topic.) Why do we need the half-open state anyway? In the classical description we attempt to use the resource once while in this state. If it fails just once, we go back to the open state. This seems like a good idea, but let us think of modern networked applications. In such applications many requests are done simultaneously. So as soon as we switch to the half-open state for a retry, many, maybe hundreds of request will immediately try to use the resource, even if it is still down. This is exactly what we were trying to prevent!

Stop!

Although the circuit breaker is a great invention, I think we need a new metaphor, or at least some new terminology.

No more half-open

The first thing we can do is get rid of the half-open state. Instead, when its time to retry, we just let 1 client through to the resource. While that check is in progress we keep denying access to the resource for other clients; we stay in the same state. Only when the single check succeeds, we switch to the state in which we allow full access to the resource.

No more open

The second thing we need to do is to end the confusion on what it means to be 'open'. Instead I propose we call this state the broken state. No further explanation required. Good. In this state we do the regular retries.

Finally, to make things symmetric, I propose to rename the 'closed' state to flow state as all requests are granted.

Metaphor

Above I proposed new terminology but I failed to provide a new metaphor. Unfortunately metaphors are hard to find and too easy to get wrong. Perhaps a good metaphor should be related to the fact that we are limiting the number of errors we tolerate from a resource. If you have an idea, please let me know in a comment. I hope you liked my little rant. Any comments are always welcome.

—   ❧   —

Postscript: Sentries and the circuit breaker

The Sentries library contains a highly optimized circuit breaker implementation for Scala programs. The ideas in this article developed while writing Sentries. Feel free to have a look. As you can see there are only 2 states, the FlowState and the BrokenState. Note that the retryAt in BrokenState is a val; it can not be changed after initialization. When it is time to retry we replace the broken state with a new instance (in method attemptResetBrokenState).

Introducing Sentries

I recently needed a friendly to use circuit breaker in a Scala program. What I found was okay, but not nearly good enough for high volume, highly concurrent applications. So I set out to make it better.

After some iterations of changing I realized that the circuit breaker could be combined with other stuff like rate limiting, monitoring and such. Now, three months later, Sentries is ready for the world.

Here is an example usage: Please visit Sentries on Githib and let me know what you think.

Update 2012-09-04: Version 0.2 will be available in Maven central around 2012-09-04 18:00 GMT.

Update 2012-10-08: Version 0.3 has just landed in Maven central. Its only feature is the new 'clean()' method on the sentry registry which is useful during testing.

Update 2013-02-06: Version 0.5 is out. This is the first to support Scala 2.10. Breaking change is that all durations are now of type akka.util.Duration or scala.concurrent.duration.Duration (depending on the Scala version) instead of Long.

Breaking a Java HashSet

Can it be?

Set set1 = new HashSet(5); Set set2 = new HashSet(5); // add of bunch of strings to both sets assert set1.equals(set2) == false;

Yes it can!

We actually had this problem in an integration test. The cause was that the strings to one set were added concurrently. Interestingly, the sets seemed to be the same, when printed they contained the same strings, just in a different order (which is also interesting). However, the internals were apparently so damaged that an equals invocation return false.

We replaced the HashSet with a ConcurrentSkipSet, and all was fine again.

Conclusion: Okay, so this is a boring conclusion, but just be careful with using normal collections in concurrent situations.

Lock-less singleton pattern

Although the singleton pattern is now know as an anti-pattern, I think it still is a valid choice when you only need one instance in a particular context. Anyway, in Java there are several ways to implement a singleton.

For example this one uses a synchronized block:

private Singleton singleton = null; protected Singleton createSingleton() { synchronized (this) { // locking on 'this' for simplicity if (singleton == null) { singleton = new Singleton(); } return singleton; } }

To prevent locking all the time you can use the double-check idiom on a modern JVM. But I just thought about another implementation that switches locking for preventing code re-ordering, making it a lock-less implementation:

private AtomicReference singletonRef = new AtomicReference(null); protected Singleton createSingleton() { Singleton singleton = singletonRef.get(); if (singleton == null) { singleton = new Singleton(); if (!singletonRef.weakCompareAndSet(null, singleton)) { singleton = singletonRef.get(); } } return singleton; }

There is only one catch to this implementation: although createSingleton() will always return the same instance, it must be allowed to call the constructor of Singleton twice without side-effects.

Update 2011-06-18: changed compareAndSet to weakCompareAndSet as suggested by Adriano.

Improving Jamon’s performance

I was browsing through Jamon’s code to see why it is so much slower under high contention then Simon (see my previous article). In this article I will present these differences and show you a way to speed up Jamon right now.

Differences
Jamon has more features then Simon like listeners and data ranges. This obviously need some computing, if only to see if they are used. Another interesting difference is that Jamon uses double’s for aggregate statistics, whereas Simon uses long’s. I don’t really see the advantage of double’s (please surprise me) and it is probably a bit slower as well.

Synchronization
Jamon uses 5 synchronization points during a timing operation. There are 3 to get and create the monitor: on the MonitorFactory method, on the map of all monitor datas and on the map of all data range definitions. The last 2 synchronization points are for starting and stopping the timer (on the monitor data).

Simon (does not support data ranges) uses only 3 synchronization points: one for getting the monitor (on the SimonManager method) and 2 for starting and stopping the timer (on the timer itself).
Simon 2.0 no longer needs synchronization on starting a timer and therefore only needs 2 synchronization points.

Jamon defines its maps for monitors and data ranges as Collections.synchronizedMap(new HashMap(50)). However, it is not necessary to synchronize on these maps when the code that uses these maps is already synchronized on the MonitorFactory (in method getMonitor). Unfortunately when the synchronization wrapper is removed, all code that uses these maps will need to be analyzed to see if they properly synchronize.

Another solution would be to use a map that needs less synchronization: Java 5’s ConcurrentHashMap! Luckily Jamon provides a method to change the map implementation for exactly this reason. I included the following line in my test application (from my previous post):

MonitorFactory.setMap(new ConcurrentHashMap(200));

And these are the results. The measurements are in ms, the scale is logarithmic.

Quite astonishing: Jamon is no longer 5 up to 200 times slower, but just a bit slower for low contention, up to 15 times slower under heavy contention, and 2 times slower under extreme contention. Note that only the map for monitors was changed, the map for range definitions (always empty in my test) was not changed.

Conclusions
Even though Jamon is quite a bit slower under heavy contention, there is room for improvement. By calling a single method, you can make Jamon 20 times faster right now.

Evaluating Simon - Java monitoring

Recently a new monitoring kid appeared on the block: Java Simon (not to be confused with Dejal's Simon or some other simons).

Simon claims to be the successor of JAMon. As I just started a project to improve the documentation of Jamon, and integrate this better with Wicket projects, I thought this would be a good time to evaluate Simon and also to compare it to Jamon. Here are the results.

Features
Simon (only just started) can measure task durations (min, max, count, maximum concurrency and some timestamps) and has task counters. Other statistics can be added (supported now: standard deviation). The JDBC proxy driver can be used to monitor database usage. The monitors are grouped in a tree. You can disable each monitor individually, or per complete subtree.

Jamon adds the ability to measure in any unit (for example euro value of a transaction) and has quite a few monitors for Tomcat, Jetty, etc. Jamon does not organize monitors in a tree, so monitors can be disabled individually, or all at the same time.

So Jamon clearly wins on the feature front, but if you're just into the basic stuff Simon is fine.

Documentation
One of the weak points of Jamon is its documentation (hence my little project). Simon on the other hand starts with a good set of wiki pages. If these are maintained properly, Simon is the clear winner here.

Simon interface
Simon's interface is fairly simple. One gets a monitor through the singleton SimonManager, and call start and stop on it:

Stopwatch simon = SimonManager.getStopwatch("simon.test").start(); // Do something simon.stop();

(I will not discuss the Jamon interface further, but making a measurement is virtually the same.)

Getting data out of a Simon monitor (a simon) is a bit more difficult. You can just get the measurements like this:

Stopwatch simon = SimonManager.getStopwatch("simon.test"); long hitCount = simon.getCounter(); long max = simon.getMax(); long total = simon.getTotal(); simon.reset();

But that would be very thread unsafe. As each method is synchronized separately, the three measurements may represent different moments in time.

The idiomatic solution would be to use method sample:

Stopwatch simon = SimonManager.getStopwatch("simon.test"); Map<String, String> sample = simon.sample(true); long hitCount = Long.parseLong(sample.get("counter")); long max = Long.parseLong(sample.get("max")); long total = Long.parseLong(sample.get("total"));

This is more how it should be except for the boolean flag [F3, N7], and the return type of sample: a map! First of all, maps are very memory hungry [G21], secondly I have to assume that the values are longs [CE6] and parse them back from a String [GE37], and thirdly the keys are magic values [CE6, G25].

Here is another approach:

Stopwatch simon = SimonManager.getStopwatch("simon.test"); long hitCount; long max; long total; synchronized (simon) { hitCount = simon.getCounter(); max = simon.getMax(); total = simon.getTotal(); simon.reset(); }

Ouch. This works but uses client-side locking. That this is possible at all is a mistake in the Java language (as admitted by James Gosling, sorry, I lost the reference). Even so, Simon should not make it possible for clients to get the same lock that is used to protect the monitor data. This should be the only way to get actual data.

Recommendation: move all data retrieval methods (getMax(), getTotal(), etc.) from the monitor to an immutable data class, e.g. StopwatchMeasurements. Method sample should return instances of this class, or subclasses when a different analyzer was configured.

Update 2008-12-25: added following section on visiting all simons.
Update 2009-01-0: the iterator now actually works.

As I am using Jarep to browse through historic measurements, I want to expose the Simon data just like Jarep's servlet exposer for Jamon does. In other words: we need to visit each simon and extract the data as described above.

There are 2 approaches to visit all nodes. First is to get all simon names from SimonManager and then request them one by one. The other is to walk the tree of simons. I choose the latter. Unfortunately there is no tree iterator so I created one with commons-collections:

Simon rootSimon = SimonManager.getRootSimon(); ObjectGraphIterator simonIterator = new ObjectGraphIterator(rootSimon, new Transformer() { private Object previous; public Object transform(Object input) { if (input == previous) { previous = null; return input; } else { previous = input; Simon currentSimon = (Simon) input; return new IteratorChain( new SingletonIterator(currentSimon), currentSimon.getChildren().iterator()); } }});

Admittedly ugly, but it works. Still there is one glaring bug here. The children are stored in a regular TreeSet. When we request the children through getChildren we get a collection that wraps the original set. Unfortunately that means that when an simon is added while we iterate the complete simon tree, we get a ConcurrentModificationException! Not quite acceptable.

Recommendation: store children of a simon in a CopyOnWriteArrayList instead of a TreeSet. Another option is change the complete set of simons in the manager into a ConcurrentMap and provide an iterator over that.

Performance To measure the performance of Simon and Jamon I wrote a test application that mimics a random web applications. It has a variable number of threads to mimic the number of concurrent requests that are being handled. There is one monitor to measure the duration of the entire request, one monitor to measure a specific request, one monitor to measure the service layer, and 0 to 3 monitors to measure the database layer.

The test application was run from IntelliJ with Soy Latte 1.0.3 (a 1.6 JDK, part of the OpenJDK for Debian) on a 2.4GHz Intel Core 2 Duo.

The measurements below display the time spend in obtaining, starting and stopping a single monitor. Measurements were done in nanoseconds, but are displayed here in milliseconds (logarithmic scale).

Simon's performance is pretty good! The overhead in my test application doesn't come anywhere near the millisecond range even in highly concurrent situations. Jamon is not so bad either, even though its 5 to 200 times slower, depending on thread contention. With 5 monitors in a request, the overhead per request slowly grows to 0.7 ms for Simon, and quickly grows to 12ms for Jamon.

Note: the test application does nothing else but getting monitors and doing measurements. So the right side of the chart is unrealistic for most environments.

Improving the performance
Even though the performance is pretty good. I think it can be improved further. I investigated two potential performance improvements:

• Remove the ThreadLocal variable in StopWatchImpl.
• Replace synchronized HashMap with a ConcurrentHashMap in EnabledManager.

Remove the ThreadLocal variable in StopWatchImpl Stopwatch method start stores the start time in a thread local. When you call stop the start time is retrieved and used to calculate the duration.

ThreadLocals have seen an enormous performance improvement in recent JVMs, but you are still better off without.

The idea is as follows: method start will no longer return the monitor itself but an object that contains only the start time and a stop method. Lets call it a StopwatchMeasurement. Only in the stop method will it be necessary to synchronize and update the counters.
The advantages are:

• We no longer need to synchronize the start method.
• We no longer need the thread local variable to store the start time.
• We no longer need the start(key) and stop(key) variants.
• The stopwatch is no longer susceptible for memory leak when a malicious client keeps starting monitors in new threads, but never stops them (someone forgot a finally).
• The stopwatch is no longer susceptible for memory leak when a malicious client keeps starting monitors with start(key) with a new key, but never stops them (again, someone forgot a finally).

Disadvantage is that we need to use something else for the active counter. An AtomicLong would be very suitable as it uses hardware assisted atomic operations (and therefore has no synchronization overhead).
If it is desirable to keep the firstUsage and lastUsege timestamps (I don't use them), these can lift along with the start time and be stored during the stop method. Note: this may be unacceptable if you need information on long running operations.

To test it out I hacked StopwatchImpl to return a StopwatchMeasurement from the start method. If you look at the implementation you see that StopwatchMeasurement implements Stopwatch so that the idiomatic usage still works:

Stopwatch simon = SimonManager.getStopwatch("simon.test").start(); // Do something simon.stop();

Of course this is a kludge that should be improved upon for a real implementation. In addition, I made no attempt to keep the active counter and the first/lastUsage timestamps.

Here are the results (again in ms per measurement):

Not bad! Although the results are quite dramatic for the higher thread counts, it still shaves off 20% in situations with less contention.

Recommendation: remove the thread local from StopwatchImpl and change the API as described above.

Replace synchronized HashMap with a ConcurrentHashMap in EnabledManager
The second thing I noticed is that getting a monitor from the SimonManager (the SimonManager.getStopwatch("simon.test") part) happens in a synchronized block and uses a plain HashMap as store. Since Java 5 there is a ConcurrentHashMap implementation which supposedly performs better in almost any circumstances.

Lets see. I remove some synchronized keywords, changed the map type, and made the code to add new monitors thread safe. This resulted in a modified EnabledManager.

Here are the results. The graph displays a few attempts that differ in the concurrency level parameter to the ConcurrentHashMap constructor. For more accuracy this graph displays the improvement in nanoseconds.

To my big surprise there is no measurable difference with modest contention. During high contention there is a reasonable positive effect that is suddenly converted to a negative effect for very high contention. Only when the currency level is set very high, does the effect stay positive.

Recommendation: don't change the manager.

Conclusions Although Simon is only young, it is off with a good start. Its clearly faster then Jamon and also on the documentation front Simon wins. Jamon has much more features so it is still a valid choice. When the recommendations in this article are followed, Simon could be better still.

Discussion
Follow the discussion of this evaluation on the Java Simon mailing list.

References
Code smells from Clean Code by Robin C. Martin:
[F3] Flag arguments
[N7] Names should describe side effects
[G14] Feature Envy
[G21] Understand the algorithm
[G25] Replace magic numbers with named constants

Code smells by me:
[CE6] Incomplete contract
[GE37] Unnecessary conversions

Asynchronous cache updates

This is the second article on Java concurrency. The first is Java thread control.

Many applications use more or less static reference data such as postal codes, exchange rates and externally stored text. Often this type of data can be safely cached in memory for performance reasons. To make sure that the cached data does not become stale for too long, it can be reloaded every 10 minutes, once a day or on any other schedule you might like.

Rather then creating custom code for every type of reference data, one can extract the hard concurrency code to a utility class. This article presents the PeriodicExecutor, a utility class I wrote years ago, and proved to be so useful and reliable that is has been traveling with me from project to project.

Here are the requirements I had in mind while writing PeriodicExecutor:

1. Support any data retrieval and any cache mechanism.
2. Refresh the cache in the background, do not slow down request handling.
3. No idle threads, only start a thread when one is needed.
4. Be resilient against errors, retry when they occur.
5. Support flexible reloading schedules.
6. Support initial loading of the memory cache.

Lets discuss these requirements and see how they are implemented.

1. Support any data retrieval and any cache mechanism.
Each type of data source needs its own retrieval code. E.g. JDBC code for database stuff and things like HttpClient for remote stuff. The cache implementation can also be different from case to case. For example in one case we just need to store a String, in another case we first transform the data to a Map. Therefore the utility class will only solve the the multi-threaded scheduling problem.

PeriodicExecutor solves this by letting the client code provide its reload task as a Runnable. The task is also responsible to make the results available when reloading is finished.

2. Refresh the cache in the background, do not slow down request handling.
When its time for a data reload, it does not make sense to delay normal request processing until the reload is done. Remember, we are caching for performance reasons! As a consequence PeriodicExecutor executes the client task in a separate thread.

PeriodicExecutor guarantees that the task is never run in parallel, but the client task must make sure that there are no concurrency problems the moment the cache is updated.

3. No idle threads, only start a thread when one is needed.
There are 2 approaches to thread handling. We can either start a thread upon initialization and keep that thread running all the time, or we can start a new thread each (short) time a reload is required. The first approach has the disadvantage that more resources are used then is necessary (though since Java 5 we could share an executor service with all reloaders). Inconvenient is that we need explicit code to shut down the thread (or executor service). Advantage of this approach is that a thread can schedule reloads precisely.

I chose the second approach: each time the cache is accessed by some code, that code must call method PeriodicExecutor#requestStart() to see if a reload is dictated by the schedule. If that is the case, the reload is started in a newly started background thread. A small disadvantage to this approach is that after a long period of inactivity, the first caller will trigger a reload, but until the reload is finished all cache users will see stale data.

4. Be resilient against errors, retry when they occur.
The reference data may be retrieved from an unreliable data source. When the reload fails, it should be tried again later. Normal processing can continue as the new background thread isolates it from exceptions. In addition, the reload task should only update the memory cache until after the reload succeeded.

PeriodicExecutor detects reload errors by catching exceptions thrown by the reload task. When an error occurred the next call to requestStart() will trigger a second reload attempt. After 2 failed attempts the next reload task is delayed for a couple of minutes to prevent trashing.

5. Support flexible reloading schedules.
PeriodicExecutor normally reloads with a fixed delay. However, it is easy to override this by providing your own next reload time. A future improvement would be to extract this code according to the strategy pattern.

6. Support initial loading of the memory cache.
Initially the memory cache is empty. As we are talking about reference data, it does not make sense to continue for most programs. PeriodicExecutor therefore also support synchronous loading and exception rethrowing. During a synchronous load the caller is blocked until the reload task is done. Before one task completed, the default is to run synchronously and rethrow exceptions from the reload task. Once one task completed, the default is to run asynchronously and to swallow exceptions (they are logged of course).

Example
Enough theory, how do you use it? Here is a simple example:

public class CachingPostalCodeService { private final Object postalCacheLock = new Object(); private List<PostalCode> postalCache; private PeriodicExecutor postalCacheReloader; public CachingPostalCodeService() { postalCacheReloader = new PeriodicExecutor( TimeUnit.MINUTES.toMillis(10), new Runnable() { public void run() { refreshPostalCache(); } public String toString() { return "Postal code cache reloader"; } }); } public List<PostalCode> getPostalCodes() { postalCacheReloader.requestStart(); synchronized (postalCacheLock) { return postalCache; } } private void refreshPostalCache() { List<PostalCode> newPostalCodes = .... // Don't forget to make newPostalCodes unmodifiable. synchronized (postalCacheLock) { postalCache = newPostalCodes; } } }

Notice that only a limited amount of code is needed to get all the discussed features. In this example a list of postal codes is cached in memory. The only user of the cache, method getPostalCodes, calls requestStart every time. Initially, when no data is in the cache, requestStart will block until the data has been loaded. After 10 minutes of use, the cache will be reloaded in the background. Note how the example synchronizes on postalCacheLock to prevent concurrency problems during the cache update.

Variations
Sometimes you only need to reload once a day, preferably early in the morning. The following example shows how one can override scheduled reload to 5 o'clock in the morning.

public CachingPostalCodeService() { postalCacheReloader = new PeriodicExecutor( 0, new Runnable() { // ...as above... }) { @Override protected long nextExecuteTime(long currentExecute) { return getNextTime(5); } }; } /** * @param hour the requested hour * @return the next time it is the given hour * in the JVM default time zone */ private static long getNextTime(int hour) { Calendar cal = Calendar.getInstance(); cal.set(Calendar.HOUR_OF_DAY, hour); cal.set(Calendar.MINUTE, 0); cal.set(Calendar.SECOND, 0); cal.set(Calendar.MILLISECOND, 0); if (cal.getTimeInMillis() < System.currentTimeMillis()) { cal.add(Calendar.DAY_OF_MONTH, 1); } return cal.getTimeInMillis(); }

If you need more control on when to execute synchronously or not, take a look at method PeriodicExecutor#requestStart(boolean).

Download
Download PeriodicExecutor. PeriodicExecutor can also be used with Java 1.4 (and perhaps 1.3 as well).

Other options
Probably the most complete FOS library for task scheduling is Quartz.

Enjoy!

Java thread control

As promised in my book review of Effective java 2nd edition, here is an article that shows a neat trick to control threads.

The recommended way to stop a thread according to the book is to use a volatile boolean field as a flag to properly stop a running thread:

public class StopThread { private static volatile boolean stopRequested; public StopThread() { Thread t = new Thread(new Runnable() { public void run() { int i = 0; while (!stopRequested) i++; } }); t.start(); TimeUnit.SECONDS.sleep(1); stopRequested = true; } }

(See the book to understand why the volatile keyword is essential.) This is all nice, but this thread can only be started and stopped once. Here is a better implementation:

public class StopThread { private static volatile Thread currentThread; public StopThread() { Thread t = new Thread(new Runnable() { public void run() { int i = 0; while (currentThread == Thread.currentThread()) i++; } }); currentThread = t; t.start(); TimeUnit.SECONDS.sleep(1); currentThread = null; } }

By storing a reference to the running thread, the running thread itself can easily see whether the rest of the system also thinks it should be running. Restarting the thread becomes easy now. Just reference a new thread instance and the old thread will die.

The previous program just demonstrates the principle. A better starting point is the following program:

public class RestartableThread { private AtomicInteger threadNumber = new AtomicInteger(0); private volatile Thread currentThread; public RestartableThread() { startThread(); } private void startThread() { currentThread = new Thread( new RestartableRunnable(), "Restartable thread " + threadNumber.getAndIncrement()); currentThread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { public void uncaughtException(Thread t, Throwable e) { startThread(); logger.error(String.format("Thread '%s' has an uncaught exception and was restarted", t.getName()), e); } }); currentThread.start(); } public void shutdown() { currentThread = null; } private class RestartableRunnable implements Runnable { public void run() { while (true) { if (currentThread != Thread.currentThread()) { return; } try { // Do a bit of work. Make sure this takes a limited time. } catch (Exception ex) { logger.error("exception", ex); } } } } }

You can stop the thread by calling the shutdown() method. Failure to do so on program exit will prevent the JVM from stopping.
The runnable contains a try/catch to prevent the thread from dying. As a double fail safety mechanism, the uncaught exception handler is triggered for unhandled Throwables such as out of memory errors. The exception handler then simply starts a new thread so that execution continues. There are subtle advantages to this double approach. Exceptions are normally not so catastrophic, the thread could just continue. Out of memory errors however can be better dealt with by letting the thread terminate and make all its used memory available for the garbage collector.
If for any other reason you want to (re-)start the thread (for example because it is programmed to run only for a limited time), simply call startThread().

Make sure that the run condition is checked regularly. Consumer threads that wait on something like a queue should wake up regularly. For example, if you are waiting on new entries in a BlockingQueue this is better done with poll() then with get().

Conclusion
Despite the presence of the new and shiny executor services, it is still sometimes necessary to write your own thread. I hope that this little article will get you started on a robust yet flexible implementation.

Next thread article: a helper for asynchronous cache updates.