# Process multiple request from different threads in batches on one processing thread

So I have a system that uses multiple threads to process data. These data could be processed individually but it would be better to process them in batches.

Lets assume we have a class Data, a class OtherData and a class Processor which implements Function<List<Data>, List<OtherData>>.

To process objects of type Data from multiple threads I designed two classes System and Evaluator.

public class System {
private final Evaluator evalThread;
private final Object sync = new Object();
private Function<List<Data>, List<OtherData>> processor;
private Map<Object, Data> inputMap;
private Map<Object, CompletableFuture<OtherData>> futureMap;
private List<Object> idList;

public System() {
processor = new Processor();
inputMap = new HashMap<>();
futureMap = new HashMap<>();
idList = new LinkedList<>();

evalThread = new Evaluator(processor, inputMap, futureMap, idList, sync);
}

public CompletableFuture<OtherData> process(Data data) {
Object id = new Object();

final CompletableFuture<OtherData> completableFuture = new CompletableFuture<>();

synchronized (sync) {
inputMap.put(id, data);
futureMap.put(id, completableFuture);

if (idList.size() >= 32) {
sync.notifyAll();
}
}

return completableFuture;
}
}

public class Evaluator implements Runnable {
private final Function<List<Data>, List<OtherData>> processor;
private final Map<Object, Data> inputMap;
private final Map<Object, CompletableFuture<OtherData>> futureMap;
private final List<Object> idList;
private final Object sync;

private AtomicBoolean keepRunning = new AtomicBoolean(true);

public Evaluator(Function<List<Data>, List<OtherData>> processor, Map<Object, Data> inputMap, Map<Object,
CompletableFuture<OtherData>> futureMap, List<Object> idList, Object sync) {
this.processor = processor;
this.inputMap = inputMap;
this.futureMap = futureMap;
this.idList = idList;
this.sync = sync;
}

@Override
public void run() {
synchronized (sync) {
while(keepRunning.get()) {
if (idList.size() > 0) {
List<Data> input = new LinkedList<>();

for (int i = 0; i < idList.size(); i++) {
}

List<OtherData> output = processor.apply(input);

for (int i = 0; i < idList.size(); i++) {
futureMap.get(idList.get(i)).complete(output.get(i));
}

idList.clear();
inputMap.clear();
futureMap.clear();
}

try {
sync.wait(20);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}



My idea was that any one can call process with singular data but the data will (if there are enough) be processed together with other Data objects.

Any suggestions for improvements or are there systems in the Java-framework that would fit this task better? Do you might see problems according to deadlocks, etc.?

• Can you explain the motivation for batching these tasks? If I read this correctly, you're running batches of 32, no more or less regardless of the request timing, which seems weird to me. – ShapeOfMatter Sep 7 at 18:48
• That is not correct. If an call to process insertes the 32th element to be processed. The eveluation thread wil be awoken to process all scheduled data and then will call sync.wait(20). So if in the next 20ms under 32 elements got scheduled, the thread will nontheless be awoken and process the scheduled data. – Ackdari Sep 8 at 10:43
• I think I understand it now. Still, can you talk about why the tasks are better batched? – ShapeOfMatter Sep 8 at 13:29
• Perhaps use Runnable instead of Thread – PPann Sep 9 at 6:45
• @ShapeOfMatter the data gets processed by an neuralnet which can faster evaluate multiple data at once compare to processing each data item on it own – Ackdari Sep 9 at 8:47

Well, technically you should really post complete code and not have missing definitions. Especially since Data and OtherData could instead be generic type arguments perhaps, or interfaces, so that it's actually clear what's happening. I'm just gonna imagine they're basically both Objects. Also Processor is undefined. If it was really not important just pass it in as an argument to System, as it is, this is incomplete too.

In System having a separate idList is pointless, the inputMap already has the right size for implementing back pressure (I wanted to find a good definition, but right now at least Wikipedia doesn't have one under that term; basically if the input can grow unlimited, the system might not be able to catch up if processing takes too long and you might end up with a congested (read: out of memory) system).

In Evaluator, x.size() > 0 could be !x.isEmpty(), that might potentially be cheaper, but it also expresses intent a bit more clearly: The check is really whether the map "isn't empty", not about how many items are in the container exactly.

InterruptedException is for control flow, don't just print a stack trace.

Also now that I read it, why's there three containers, inputMap, futureMap and idList all related to the same task? It'd be much easier if there was just a sequence of tuples, Tuple<Input, Future>, and then work through them. Then replace the List with a BlockingQueue<Tuple<Input, Future>> and it's already supporting the waiting / back pressure too. An ArrayBlockingQueue could be used to limit the number of elements, while a LinkedBlockingQueue could have an unbounded size (that's really not advisable though).

Lastly instead of keepRunning a tombstone object could then be inserted to cancel the thread, then there's also no need for polling via wait. That is, insert a new Tuple<>() (same as the IDs at the moment via new Object()) and when dequeuing from the input, check whether this element was inserted, due to the object identity being unique that's then safe to do.

So without spelling it all out, the main thread would look like this perhaps:

public void run() {
while (true) {
Tuple<Input, Future> tuple = input.take();
if (tuple == tombstone) {
return;
}
tuple.getFuture().complete(processor.apply(tuple.getInput()));
}
}


With some implementation of Tuple of course; a custom class would work too.

For the other questions: I can't see it deadlocking right now, the check for 32 and the notifyAll are a bit odd though.

Other tools in the Java tool set? Yes, I'd suggest starting with reading through the java.util.concurrent namespace. In particular the ExecutorService perhaps.

Use a CyclicBarrier with only the neural-net thread waiting, and decrement it after adding an item to the queue (which should probably be a ConcurrentLinkedQueue). Then it will automatically trigger after every 32 entries, at which point it should pop exactly 32 entries from the head of the queue (an operation that won't block if it's the queue's only consumer).

You can probably eliminate the synchronized block in process if you also replace both HashMap instances with ConcurrentHashMap. This will improve performance when process is called from more than a small constant number of threads (probably on the order of 2 or 3).