Execute millions of Runnable's with small memory footprint

Question

I have N longs which are IDs. For every ID I need to execute a Runnable (ie. I don't care about a return value) and wait until all of them are finished. Each Runnable can take from a few seconds to a few minutes and it's safe to run about 100 threads in parallel.

In our current solution, we use Executors.newFixedThreadPool(), call submit() for each ID and then call get() on each returned Future.

The code works well and it's very simple in that I don't have to deal with threads, complicated waiting logic, etc. It has a downside: memory footprint.

All the still enqueued Runnable's consume memory (much more than 8 bytes than would be needed by a long: these are my Java classes with some internal state), and all the N Future instances consume memory too (these are Java classes with state as well, which I only use for waiting but I don't need the actual results). I looked at a heap dump and I would estimate that a little over 1 GiB of memory is taken up for N=10 million. 10 million longs in an array would only consume 76 MiB.

Is there a way to solve this problem with only holding the IDs in memory, preferably without resorting to low-level concurrent programming?

Answer 1

Yes: you could have a shared queue of longs. You submit n Runnables to the executor, where n is the number of threads in the executor, at the end of the run method, you get the next long from the queue, and resubmit a new Runnable.

Answer 2

Instead of creating million of Runnables, create specific thread pool which takes longs as tasks. Instead of waiting tasks to finish with Future.get(), use CountdownLatch.

That thread pool could be implemented like this:

int N = 1000000;// number of tasks;
int T = 100; // number of threads;
CountdownLatch latch = new CountdownLatch(N);
ArrayBlockingQueue<Long> queue = new ArrayBlockingQueue<>();

for (int k=0; k<N; k++) {
   queue.put(createNumber(k));
}
for (int k=0; k<T; k++) {
  new WorkingThread().start();
}
CountdownLatch.await();

class WorkingThread extends Thread {
  public void run() {
      while (latch.getCount() != 0) {
           processNumber(queue.take());
           latch.countDown();
      }
  }
}

Answer 3

What about using ExecutorCompletionService? Something like the following (which may contain bugs, I didn't test it):

import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorCompletionService;
import java.util.function.LongFunction;

public class Foo {

  private final ExecutorCompletionService<Void> completionService;
  private final LongFunction<Runnable> taskCreator;
  private final long maxRunning; // max tasks running or queued

  public Foo(Executor executor, LongFunction<Runnable> taskCreator, long maxRunning) {
    this.completionService = new ExecutorCompletionService<>(executor);
    this.taskCreator = taskCreator;
    this.maxRunning = maxRunning;
  }

  public synchronized void processIds(long[] ids) throws InterruptedException {
    int completed = 0;

    int running = 0;
    for (long id : ids) {
      if (running < maxRunning) {
        completionService.submit(taskCreator.apply(id), null);
        running++;
      } else {
        completionService.take();
        running--;
        completed++;
      }
    }

    while (completed < ids.length) {
      completionService.take();
      completed++;
    }

  }

}

---

Another version of the above could use Semaphore and CountDownLatch, rather than a CompletionService.

public static void processIds(long[] ids, Executor executor,
                              int max, LongFunction<Runnable> taskSup) throws InterruptedException {
  CountDownLatch latch = new CountDownLatch(ids.length);
  Semaphore semaphore = new Semaphore(max);

  for (long id : ids) {
    semaphore.acquire();

    Runnable task = taskSup.apply(id);
    executor.execute(() -> {
      try {
        task.run();
      } finally {
        semaphore.release();
        latch.countDown();
      }
    });

  }

  latch.await();
}

Answer 4

This is the kind of thing I'd usually do with a Producer/Consummer pattern, and a BlockingQueue coordonnating the two, or, using Akka actors if I have the at hand on the project.

But I figured I'd suggest something quiete a bit different, relying on Java's Stream behavior.

The intuition is that the lazy execution of streams will be used to throttle the creation of work units, futures, and their results.

public static void main(String[] args) {
    // So we have a list of ids, I stream it
    // (note : if we have an iterator, you could group it by a batch of, say 100,
    // and then flat map each batch)
    LongStream ids = LongStream.range(0, 10_000_000L);
    // This is were the actual tasks will be dispatched
    ExecutorService executor = Executors.newFixedThreadPool(4);

    // For each id to compute, create a runnable, which I call "WorkUnit"
    Optional<Exception> error = ids.mapToObj(WorkUnit::new)
             // create a parralel stream
             // this allows the stream engine to launch the next instructions concurrently
            .parallel()
            // We dispatch ("parallely") the work units to a thread and have them execute
            .map(workUnit -> CompletableFuture.runAsync(workUnit, executor))
            // And then we wait for the unit of work to complete
            .map(future -> {
                try {
                    future.get();
                } catch (Exception e) {
                    // we do care about exceptions
                    return e;
                } finally {
                    System.out.println("Done with a work unit ");
                }
                // we do not care for the result
                return null;
            })
            // Keep exceptions on the stream
            .filter(Objects::nonNull)
            // Stop as soon as one is found
            .findFirst();


    executor.shutdown();
    System.out.println(error.isPresent());
}

To be honest, I'm not quiete sure the behaviour is guaranteed by the specification, but from my experience it works. Each one of the parallel "chunck" grabs a few ids, the feed it to the pipeline (map to a work unit, dispatch to the thread pool, wait for a result, filter for exceptions), which means an equilibrium is reached fairly quickly balancing the number of active work units to that of the executor.

If the number of parallel "chuncks" is to be fine tuned, one should follow up here : Custom thread pool in Java 8 parallel stream