ExecutorService with huge number of tasks

Question

I have a list of files and a list of analyzers that analyze those files. Number of files can be large (200,000) and number of analyzers (1000). So total number of operations can be really large (200,000,000). Now, I need to apply multithreading to speed things up. I followed this approach:

ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
for (File file : listOfFiles) {
  for (Analyzer analyzer : listOfAnalyzers){
    executor.execute(() -> {
      boolean exists = file.exists();
      if(exists){
        analyzer.analyze(file);
      }
    });
  }
}
executor.shutdown();
executor.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);

But the problem of this approach is that it's taking too much from memory and I guess there is better way to do it. I'm still beginner at java and multithreading.

Answer 1

Where are 200M tasks going to reside? Not in memory, I hope, unless you plan to implement your solution in a distributed fashion. In meantime, you need to instantiate an ExecutorService that does not accumulate a massive queue. Use with the "caller runs policy" (see here) when you create the service. If you try to put another task in the queue when it's already full, you'll end up executing it yourself, which is probably what you want.

OTOH, now that I look at your question more conscientiously, why not analyze a single file concurrently? Then the queue is never larger than the number of analyzers. That's what I'd do, frankly, since I'd like a readable log that has a message for each file as I load it, in the correct order.

I apologize for not being more helpful:

analysts.stream().map(analyst -> executor.submit(() -> analyst.analyze(file))).map(Future::get);

Basically, create bunch of futures for a single file, then wait for all of them before you move on.

Answer 2

One idea is to employ fork/join algorithm and group items (files) into batches in order to process them individually.

My suggestion is the following:

1. Firstly, filter out all files that do not exist - they occupy resources unnecessarily.

2. The following pseudo-code demonstrates the algorithm that might help you out:

   public static class CustomRecursiveTask extends RecursiveTask<Integer {
   
   private final Analyzer[] analyzers;
   
   private final int threshold;
   
   private final File[] files;
   
   private final int start;
   
   private final int end;
   
   public CustomRecursiveTask(Analyzer[] analyzers,
                              final int threshold,
                              File[] files,
                              int start,
                              int end) {
       this.analyzers = analyzers;
       this.threshold = threshold;
       this.files = files;
       this.start = start;
       this.end = end;
   }
   
   @Override
   protected Integer compute() {
       final int filesProcessed = end - start;
       if (filesProcessed < threshold) {
           return processSequentially();
       } else {
           final int middle = (start + end) / 2;
           final int analyzersCount = analyzers.length;
   
           final ForkJoinTask<Integer> left =
                   new CustomRecursiveTask(analyzers, threshold, files, start, middle);
           final ForkJoinTask<Integer> right =
                   new CustomRecursiveTask(analyzers, threshold, files, middle + 1, end);
           left.fork();
           right.fork();
   
           return left.join() + right.join();
       }
   }
   
   private Integer processSequentially() {
       for (int i = start; i < end; i++) {
           File file = files[i];   
           for(Analyzer analyzer : analyzers) { analyzer.analyze(file) };
       }
   
       return 1;
   }
   }
   

And the usage looks the following way:

 public static void main(String[] args) {
    final Analyzer[] analyzers = new Analyzer[]{};
    final File[] files = new File[] {};

    final int threshold = files.length / 5;

    ForkJoinPool.commonPool().execute(
            new CustomRecursiveTask(
                    analyzers,
                    threshold,
                    files,
                    0,
                    files.length
            )
    );
}

Notice that depending on constraints you can manipulate the task's constructor arguments so that the algorithm will adjust to the amount of files.

You could specify different thresholds let's say depending on the amount of files.

final int threshold;
if(files.length > 100_000) {
   threshold = files.length / 4;
} else {
   threshold = files.length / 8;
}

You could also specify the amount of worker threads in ForkJoinPool depending on the input amount.

Measure, adjust, modify, you will solve the problem eventually.

Hope that helps.

UPDATE:

If the result analysis is of no interest, you could replace the RecursiveTask with RecursiveAction. The pseudo-code adds auto-boxing overhead in between.