Chaotic Java

One of the most classic patterns in software is the producer-consumer pattern. There is a module producing data, and a module reading it for further processing. Moreover, in order to achieve better performance, usually there are many consumer modules running on many different threads while the producer (or several producers) run on its own thread. This allows to distribute processing work between threads, and in the multi-core, multi-processor environment of today, between physical processors as well. Concurrent frameworks (such as java.util.concurrent) provide out-of-the-box solutions for these kind of problems.

Same pattern, multiple machines

This pattern also works well when wanting to distribute tasks between different physical machines as well. The producer machine somehow creates information, and offloads the task of processing that information to other machines. Since we want to distribute the work between different machines without coupling issues, the data is written to a physical disk and using some sort of a distribution system (e.g. JMS queues), a notification containing the full path to the data is sent. An available consumer receiving the notification can then read the data from the physical disk and process it. The following diagram illustrates this simple idea:

It seems to me there are two reasons to write the data to disk: first, to allow the decoupling as mentioned before – by using the physical drive, the consumer is responsible to read the data when it receives the notification. Second, since (in my experience) most queue infrastructures are not very good at varying message sizes, so sending arbitrary sized data through them could become a major performance issue.

The classic way to read the data would be to fetch the file on the disk, open a FileInputStream, and start processing the data. However, when there is a lot of data to transfer, IO to and from the physical drive starts to become an issue as well. There are a few solutions to this:

1. Read all the data into a byte array, and process in-memory. When reading byte-per-byte from the FileInputStream, Java doesn’t know how much of the file you’d want to process. Therefore, it constantly fetches small bulks of bytes. If the file is small enough to be loaded to memory, a full-read could be made by using:

   
   byte[] bytes = new byte[file.length()];
   stream.read(bytes);
   

2. Read and process data in multiple threads. When receiving the notification, start a new thread and start reading and processing in that thread. To avoid the creation of too many threads, you can use a thread pool with a fixed size, a synchronous queue and a blocking rejection policy:

   
   ExecutorService executor =
     new ThreadPoolExecutor(nThreads, nThreads,
       0, TimeUnit.SECONDS,
       new SynchronousQueue<Runnable>(),
       new ThreadPoolExecutor.CallerRunsPolicy());
   

   This will make sure that only nThreads are running, and that if the number of running threads reaches it, no more notifications are read.

3. Combination of 1 and 2. Run multiple threads that first read the data into memory and then process it.
4. Use tar to group files together. This solves a core problem with multiple small files: opening and closing file streams is expensive as well as just reading the data. Therefore, if a lot of separate data files are involved in the process, it might be worth grouping them in bulks at the producer’s side, and then ungrouping them in-memory at the consumer’s side. This will work like option 1, except that it will have the added benefit of skipping the resources required for opening and closing multiple files. I use Tim Endres’ library for TarInputStream and TarOutputStream, but soon it will be part of Apache’s Commons Compression package.

I was recently rethinking about the need for the tar mechanism, so I tested it out by creating a small application which uses all aforementioned methods using different parameters. All four methods were tried on 10, 100 and 500 files, all of them sized at 1MB. The files’ data was created using Random.nextBytes(byte[]) method. For the multithreading tests I used 5, 10 and 100 threads for each option (either reading directly from file or first to a byte array). I ran the applications 5 times, and I provide the results here:

Conclusion

I then wanted to create an easy-to-read graph to show the differences between the approaches. I used the single threaded, reading directly from file approach as the baseline and divided the appropriate results other tests received with it. This gave the speed ratio between these approaches and the baseline approach. I removed the 10 items measurements as the numbers were too small to mean anything, and I present the outcome below:

I’ve coloured the baseline red, and the tar option with a darker blue. It’s clear to see that while creating multiple threads made some positive impact to performance, it doesn’t come close to what using tar did. What do you think? I’m happy to hear any ideas for boosting the performance on IO intensive tasks!