ThreadPoolExecutor плохо масштабируется между 6->62 потоками

У меня есть приложение, которому необходимо определить и выполнить около 3,9 миллиардов единиц работы. Он выполняет это небольшими партиями, от 1000 до 100 000 единиц работы (в основном на меньшем конце этого масштаба). Мне приходится группировать их, потому что каждая партия должна быть оценена, и эта оценка влияет на исходные данные для следующей партии.

Проблема в том, что я пытаюсь выполнить работу по оценке каждой единицы работы в многопоточном коде по соображениям пропускной способности/производительности. Вначале он хорошо масштабируется, но на более крупных серверах масштабируется отрицательно. Подробности ниже.

У меня есть ThreadPoolExecutor, который я создаю следующим образом:

    // Leave 2 CPUs unused for system processing, minimum of 1.
    numOfCores = Math.max(1, numOfCores - 2);
    this.backgroundExecutor = new ThreadPoolExecutor(
            /* coreSize */ numOfCores, 
            /* maximumPoolSize */ numOfCores, 
            /* keepAliveTime */ 10, TimeUnit.SECONDS,
            /* workQueue */ new LinkedBlockingQueue<Runnable>(), 
            threadFactory);

Основной поток приложения помещает работу в очередь небольшими порциями, возможно, от 1000 до 100 000 запросов, используя этот метод:

public void addRequest(R taskRequest) {
    this.taskList.add(() -> {
        // This is a callable - so it's code that will run in the background thread.
        final ThreadedBuilderThread<T> currentThread = (ThreadedBuilderThread<T>) Thread.currentThread();
        T newObject = taskRequest.build();
        if (taskRequest.isBestOnlyMode() && newObject.getValue() > Double.NEGATIVE_INFINITY) {
            T bestObject = currentThread.bestObjectReference.getPlain();
            if (bestObject == null || newObject.getValue() > bestObject.getValue()
                    || (newObject.getValue() == bestObject.getValue() && newObject.getId() < bestObject.getId())) {
                currentThread.bestObjectReference.setPlain(newObject);
            }
            // Don't return any object at this point, because we won't know which one is
            // best until all objects are built.
            return null;
        } else {
            // Not in "best only" mode, so return the object we just built.
            return newObject;
        }
    });
}

После добавления каждого пакета запросов основной поток ожидает результатов следующим образом:

public Collection<T> waitAndGetResults(boolean bestOnlyMode) {

    Collection<Future<T>> resultsFutureList;
    try {
        // NOTE: .invokeAll() waits for all the tasks to finish.
        resultsFutureList = this.backgroundExecutor.invokeAll(taskList); // , 10, TimeUnit.SECONDS);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }

    final Collection<T> allResultList = new ArrayList<>();

    for (final Future<T> f : resultsFutureList) {
        final T t;
        try {
            t = f.get(10, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            throw new RuntimeException("Background Thread was interrupted!", e);
        } catch (ExecutionException e) {
            throw new RuntimeException("Background Thread has encounted an uncaught exception!", e);
        } catch (TimeoutException e) {
            throw new RuntimeException("Background Thread has timed out!", e);
        }
        if (!bestOnlyMode) {
            allResultList.add(t);
        }
    }
    // Verify we got all of our work done (because we might have timed-out).
    if (!bestOnlyMode && allResultList.size() != taskList.size()) {
        throw new AssertionError(
                "Requested " + taskList.size() + " buildable objects, but only got " + allResultList.size() + ".");
    }

    // Clear the task list for the next group of tasks to be added later
    this.taskList.clear();

    if (!bestOnlyMode) {
        // We are not in best only mode, so return the full list of results
        for (ThreadedBuilderThread<T> th : threadList) {
            th.bestObjectReference.setRelease(null);
        }
        return allResultList;
    }

    // Otherwise, we are in best only mode
    double overallBestValue = Double.NEGATIVE_INFINITY;
    T overallBestObject = null;

    // Iterate through the builder threads and get the best object built for each
    // one.
    for (final ThreadedBuilderThread<T> th : threadList) {
        final T threadBestObject = th.bestObjectReference.get();
        if (threadBestObject != null && threadBestObject.getValue() > overallBestValue) {
            // This thread had the best object (so far)
            overallBestValue = threadBestObject.getValue();
            overallBestObject = threadBestObject;
        }
    }

    final Collection<T> bestResultList = new ArrayList<>();
    if (overallBestObject != null) {
        bestResultList.add(overallBestObject);
    }

    // Clear the thread map for the next batch of tasks
    for (ThreadedBuilderThread<T> th : threadList) {
        th.bestObjectReference.setRelease(null);
    }

    return bestResultList;
}

ПРОБЛЕМА

Я запускаю этот код на серверах, развернутых в облаке Amazon AWS. Когда я работаю на экземпляре с 8 виртуальными ЦП, я получаю лучшее время работы по настенным часам, чем когда я работаю на экземпляре с 4 виртуальными ЦП. Но когда я пытаюсь масштабировать до 64 виртуальных ЦП, время работы настенных часов составляет почти 200% от времени работы 8 виртуальных ЦП.

Я рассмотрел свою реализацию и удалил все источники конфликтов потоков, которые только мог придумать, и получил скромные улучшения, но общая проблема отрицательного масштабирования все еще сохраняется.

Есть ли у кого-нибудь мысли о том, как я могу это отладить или, что еще лучше, какой возможный источник разногласий может быть?