java - 知道 Apache Spark 中 Java ForkJoinPool 中哪个线程进入哪个处理器？

Question

目标:当我 fork 一个线程时，知道它将落在哪个处理器上。那可能吗？不管基本方法是否有效，这个狭隘的问题有一个好的答案吗？谢谢。

(现在我需要为每个线程制作一个类的副本，在该线程中写入它并稍后将它们全部合并。使用同步方法是不可能的，因为我的Java专家老板认为这是一个坏主意，经过大量讨论后我同意。如果我知道每个线程将登陆哪个处理器，我只需要制作与处理器一样多的该类的副本。)

我们使用 Apache Spark 将作业分布在集群中，但在我们的应用程序中，运行一个大型执行程序，然后在集群中的每台机器上执行一些我们自己的多线程处理是有意义的。

我可以节省大量的深度复制如果我知道线程被发送到哪个处理器，这可能吗？我输入了我们的代码，但这可能更多是一个概念性问题:

当我深入到compute()的“执行任务”部分时，我能知道它正在哪个处理器上运行吗？

public class TholdExecutor extends RecursiveTask<TholdDropEvaluation> {

    final static Logger logger = LoggerFactory.getLogger(TholdExecutor.class);

    private List<TholdDropResult> partitionOfN = new ArrayList<>();
    private int coreCount;
    private int desiredPartitionSize; // will be updated by whatever is passed into the constructor per-chromosome
    private TholdDropEvaluation localDropEvaluation; // this DropEvaluation
    private TholdDropResult mSubI_DR;


    public TholdExecutor(List<TholdDropResult> subsetOfN, int cores, int partSize, TholdDropEvaluation passedDropEvaluation, TholdDropResult mDrCopy) {
        partitionOfN = subsetOfN;
        coreCount = cores;
        desiredPartitionSize = partSize;

        // the TholdDropEvaluation needs to be a copy for each thread? It can't be the same one passed to threads ... so ...
        TholdDropEvaluation localDropEvaluation = makeDECopy(passedDropEvaluation); // THIS NEEDS TO BE A DEEP COPY OF THE DROP EVAL!!! NOT THE ORIGINAL!!

        // we never modify the TholdDropResult that is passed in, we just need to read it all on the same JVM/worker, so
        mSubI_DR = mDrCopy; // this is purely a reference and can point to the passed in value (by reference, right?)

    }

    // this makes a deep copy of the TholdDropEvaluation for each thread, we copy the SharingRun's startIndex and endIndex only,
    // as LEG events will be calculated during the subsequent dropComparison. The constructor for TholdDropEvaluation must set
    // LEG events to zero.
    private void makeDECopy(TholdDropEvaluation passedDropEvaluation) {
        TholdDropEvaluation tholdDropEvaluation = new TholdDropEvaluation();

        // iterate through the SharingRuns in the SharingRunList from the TholdDropEval that was passed in
        for (SharingRun sr : passedDropEvaluation.getSharingRunList()) {
            SharingRun ourSharingRun = new SharingRun();
            ourSharingRun.startIndex = sr.startIndex;
            ourSharingRun.endIndex = sr.endIndex;

            tholdDropEvaluation.addSharingRun(ourSharingRun);
        }
        return tholdDropEvaluation
    }

    @Override
    protected TholdDropEvaluation compute() {

        int simsToDo = partitionOfN.size();
        UUID tag = UUID.randomUUID();

        long computeStartTime = System.nanoTime();

        if (simsToDo <= desiredPartitionSize) {
            logger.debug("IN MULTI-THREAD compute() --- UUID {}:Evaluating partitionOfN sublist length", tag, simsToDo);

            // job within size limit, do the task and return the completed TholdDropEvaluation
            // iterate through each TholdDropResult in the sub-partition and do the dropComparison to the refernce mSubI_DR,
            // writing to the copy of the DropEval in tholdDropEvaluation
            for (TholdDropResult currentResult : partitionOfN) {

                mSubI_DR.dropComparison(currentResult, localDropEvaluation);

            }
        } else {

            // job too large, subdivide and call this recursively
            int half = simsToDo / 2;
            logger.info("Splitting UUID = {}, half is {} and simsToDo is {}", tag, half, simsToDo );
            TholdExecutor nextExec = new TholdExecutor(partitionOfN.subList(0, half), coreCount, desiredPartitionSize, tholdDropEvaluation, mSubI_DR);
            TholdExecutor futureExec = new TholdExecutor(partitionOfN.subList(half, simsToDo), coreCount, desiredPartitionSize, tholdDropEvaluation, mSubI_DR);
            nextExec.fork();
            TholdDropEvaluation futureEval = futureExec.compute();
            TholdDropEvaluation nextEval = nextExec.join();
            tholdDropEvaluation.merge(futureEval);
            tholdDropEvaluation.merge(nextEval);
        }

        logger.info("{} Compute time is {} ns",tag, System.nanoTime() - computeStartTime);

        // NOTE: this was inside the else block in Rob's example, but don't we want it outside the block so it's returned
        // whether
        return tholdDropEvaluation;
    }
}

Answer 1

即使您可以弄清楚线程最初将在哪里运行，也没有理由假设它会在其余生中一直驻留在该处理器/核心上。对于任何大到足以值得产生线程的成本的任务来说，很可能它不会，因此您需要完全控制它的运行位置以提供这种级别的保证。

据我所知，Java 内部没有控制从线程到处理器核心的映射的标准机制。通常，这称为“线程关联”或“处理器关联”。例如，在 Windows 和 Linux 上，您可以使用以下命令进行控制:

• Windows:SetThreadAffinityMask
• Linux:sched_setaffinity或pthread_setaffinity_np

因此理论上您可以编写一些 C 和 JNI 代码，使您能够在您关心的 Java 主机上对其进行足够的抽象以使其正常工作。

对于您似乎面临的实际问题，这感觉像是错误的解决方案，因为您最终从操作系统调度程序中撤回了选项，这可能不允许它做出最明智的调度决策，从而导致总运行时间增加。除非您将不寻常的工作负载和建模/查询处理器信息/拓扑降低到 NUMA 和共享缓存级别，否则它应该比大多数工作负载更好地确定在哪里运行线程。你可以。除了您在调用 main() 后显式创建的线程之外，您的 JVM 通常还运行大量附加线程。此外，我不想对您今天(甚至明天)运行的 JVM 可能决定自行决定如何处理线程关联做出任何 promise 。

话虽如此，潜在的问题似乎是您希望每个线程拥有一个对象的实例。通常，这比预测线程将在何处运行，然后在任何时间点手动计算出 N 个处理器和 M 个线程之间的映射要容易得多。通常您会使用“线程本地存储”(TLS)来解决这个问题。

大多数语言都以某种形式提供这个概念。在 Java 中，这是通过 ThreadLocal 提供的类(class)。给出的链接文档中有一个示例:

 public class ThreadId {
     // Atomic integer containing the next thread ID to be assigned
     private static final AtomicInteger nextId = new AtomicInteger(0);

     // Thread local variable containing each thread's ID
     private static final ThreadLocal<Integer> threadId =
         new ThreadLocal<Integer>() {
             @Override protected Integer initialValue() {
                 return nextId.getAndIncrement();
         }
     };

     // Returns the current thread's unique ID, assigning it if necessary
     public static int get() {
         return threadId.get();
     }
 }

本质上，您关心两件事:

1. 当您调用 get() 时，它会返回属于当前线程的值(对象)
2. 如果您在当前没有任何内容的线程中调用 get，它将调用您实现的 initialValue()，这允许您构造或获取新对象。

因此，在您的场景中，您可能希望从只读全局版本深层复制某些本地状态的初始版本。

最后一点需要注意的是:如果你的目标是分而治之；在许多线程上做一些工作，然后将所有结果合并为一个答案，合并部分通常称为缩减。在这种情况下，您可能正在寻找 MapReduce这可能是使用归约的最著名的并行形式。