Smart Batching in Java

1. Overview

In this tutorial, we’ll understand the smart batching pattern. We’ll first look at micro batching and its pros and cons, and then we’ll see how smart batching can alleviate its problems. We’ll also look at some examples of both patterns using simple Java data structures.

2. Micro Batching

We could consider micro batching as the base for the smart batching pattern. Although inferior, it’s the base on which we’ll build smart batching.

2.1. What Is Micro Batching?

Micro batching is an optimization technique for systems with workloads that consist of bursts of small tasks. Although they have a small computational overhead, they come with some kind of operation that supports a low amount of requests per second, for example, a write to an I/O device.

When we employ the micro batching pattern, we avoid processing incoming tasks individually. Instead, we aggregate them in a batch, and once it’s large enough, we process them together.

With this grouping technique, we can optimize resource utilization, especially when it comes to I/O devices. This approach helps us mitigate the latency introduced by handling bursts of small tasks one by one.

2.2. How Does It Work?

The simplest way to implement micro batching is to cache incoming tasks in a collection such as a Queue. Once the collection is over a certain size, dictated by the target system’s properties, we gather all tasks up to that limit and process them together.

Let’s create a minimal MicroBatcher class:

class MicroBatcher {
    Queue<String> tasksQueue = new ConcurrentLinkedQueue<>();
    Thread batchThread;
    int executionThreshold;
    int timeoutThreshold;

    MicroBatcher(int executionThreshold, int timeoutThreshold, Consumer<List<String>> executionLogic) {
        batchThread = new Thread(batchHandling(executionLogic));
        batchThread.setDaemon(true);
        batchThread.start();
        this.executionThreshold = executionThreshold;
        this.timeoutThreshold = timeoutThreshold;
    }

    void submit(String task) {
        tasksQueue.add(task);
    }

    Runnable batchHandling(Consumer<List<String>> executionLogic) {
        return () -> {
            while (!batchThread.isInterrupted()) {
                long startTime = System.currentTimeMillis();
                while (tasksQueue.size() < executionThreshold && (System.currentTimeMillis() - startTime) < timeoutThreshold) {
                    Thread.sleep(100);
                }
                List<String> tasks = new ArrayList<>(executionThreshold);
                while (tasksQueue.size() > 0 && tasks.size() < executionThreshold) {
                    tasks.add(tasksQueue.poll());
                }
                executionLogic.accept(tasks);
            }
        };
    }
}

Our batcher class has two important fields, tasksQueue, and batchThread.

As our Queue implementation, we select the ConcurrentLinkedQueue since it offers concurrent access and can grow as much as needed. This is where all submitted tasks reside. In our case, we represent them as simple String objects we provide as arguments to executionLogic that we define externally.

Additionally, our MicroBatcher has a dedicated Thread for batch handling. We must note that task submission and processing must be done in different threads. This decoupling is the most important part of latency minimization. This is because we let only one thread issue the slow requests while the rest can submit tasks as fast as needed since they aren’t blocked by the operation.

Finally, we define executionThreshold and timeoutThreshold. The first determines the number of tasks that must be buffered before we execute them. Its value depends on the target operation. For example, if we’re writing to a network device, the threshold should be equal to the max packet size. The second is the maximum amount of time we’ll wait for tasks to be buffered before we process them, even if executionThreshold hasn’t been reached.

2.3. Pros and Cons

We get many benefits by using the micro batcher pattern. First, it gives us increased throughput since tasks are submitted regardless of the state of execution, which means that our system is more responsive.

Additionally, by tuning the micro batcher, we can achieve proper utilization of the underlying resource (e.g., disk storage) and saturate it to the optimal level.

Finally, it conforms well to real-world traffic, which rarely is uniform and usually comes in bursts.

One of the most important cons of this implementation, however, is the fact that when the system isn’t under load, for example, at night, even a single request is forced to wait for the timeoutThreshold before being processed. This results in under-utilization of resources and, most importantly, a bad user experience.

3. Smart Batching

Enter smart batching, a modified version of micro batching. The difference is that we omit the timeoutThreshold and instead of waiting for the queue to fill up with tasks, we immediately execute any number of tasks up to executionThreshold.

With this simple change, we avoid the low traffic latency issue mentioned above while still keeping all the benefits of micro batching. The reason is that usually, the time it takes to process a batch of tasks is enough for the queue to fill up with the next batch. We thus have optimal resource usage and avoid blocking the execution of single tasks, in case that’s all that’s pending.

Let’s convert our MicroBatcher into SmartBatcher:

class SmartBatcher {
    BlockingQueue<String> tasksQueue = new LinkedBlockingQueue<>();
    Thread batchThread;
    int executionThreshold;
    boolean working = false;
    SmartBatcher(int executionThreshold, Consumer<List<String>> executionLogic) {
        batchThread = new Thread(batchHandling(executionLogic));
        batchThread.setDaemon(true);
        batchThread.start();
        this.executionThreshold = executionThreshold;
    }

    Runnable batchHandling(Consumer<List<String>> executionLogic) {
        return () -> {
            while (!batchThread.isInterrupted()) {
                List<String> tasks = new ArrayList<>(executionThreshold);
                while(tasksQueue.drainTo(tasks, executionThreshold) == 0) {
                    Thread.sleep(100);
                }
                working = true;
                executionLogic.accept(tasks);
                working = false;
            }
        };
    }
}

We changed three things in our new implementation. First, we removed timeoutThreshold. Second, we changed our Queue implementation to BlockingQueue. These support the drainTo() method, which works perfectly for our needs. Finally, we took advantage of this method to simplify our batchHandling() logic.

4. No-Batching vs. Batching Comparison

Let’s create an application class with a simple scenario to test the straightforward method against the batched approach:

class BatchingApp {
    public static void main(String[] args) throws Exception {
        final Path testPath = Paths.get("./test.txt");
        testPath.toFile().createNewFile();
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(100);
        Set<Future> futures = new HashSet<>();
        for (int i = 0; i < 50000; i++) {
            futures.add(executorService.submit(() -> {
                Files.write(testPath, Collections.singleton(Thread.currentThread().getName()), StandardOpenOption.APPEND);
            }));
        }
        long start = System.currentTimeMillis();
        for (Future future : futures) {
            future.get();
        }
        System.out.println("Time: " + (System.currentTimeMillis() - start));
        executorService.shutdown();
    }
}

We’ve selected a simple file write for the I/O operation. We create a test.txt file and write 50000 lines to it using 100 threads. Although the time displayed in the console will depend on the target hardware, here’s an example:

Time (ms): 4968

Even trying with different thread counts, the time still is around 4500 ms. It seems we’re hitting our hardware’s limit.

Let’s now switch to SmartBatcher:

class BatchingApp {
    public static void main(String[] args) throws Exception {
        final Path testPath = Paths.get("./testio.txt");
        testPath.toFile().createNewFile();
        SmartBatcher batcher = new SmartBatcher(10, strings -> {
            List<String> content = new ArrayList<>(strings);
            content.add("-----Batch Operation-----");
            Files.write(testPath, content, StandardOpenOption.APPEND);
        });

        for (int i = 0; i < 50000; i++) {
            batcher.submit(Thread.currentThread().getName() + "-1");
        }
        long start = System.currentTimeMillis();
        while (!batcher.finished());
        System.out.println("Time: " + (System.currentTimeMillis() - start));
    }
}

We added a finished() method to SmartBatcher to check when all tasks are finished:

boolean finished() {
    return tasksQueue.isEmpty() && !working;
}

Here’s the new time shown:

Time (ms): 1053

Even with executionThreshold of 10, we achieve a five-fold improvement. Increasing the threshold to 100 reduces the time to ~150 ms, almost 50 times faster than the simplistic approach.

As we can see, employing a simple technique that takes advantage of the underlying hardware characteristics can drastically boost our application performance. We should always keep in mind what our system is doing and the traffic it’s working with.

5. Conclusion

In this article, we had an overview of task batching techniques, specifically micro batching and smart batching. We saw potential use cases, the pros and cons of micro batching, and how smart batching can mitigate its shortcomings. Finally, we looked at a comparison between simple task execution and batched execution.