java - 单线程性能更好

Question

我正在树搜索类(class)作业。我了解树搜索部分，但由于我有一些额外的时间，我想通过添加更多线程来加快速度。

最后的任务是接受一组约束、类和时隙，并输出包含所有这些类且满足所有约束的时间表。输入空作业或部分作业，输出完整的类作业。

我们的搜索设计得像一棵树，输入是根节点。函数div(n)如下:对于一个节点n，找到一个未使用的类C，对于每个未使用的slot S，在S中产生一个带有C的子节点. 为了使搜索更高效，我们使用了对节点质量进行排序的搜索控件，以便首先选择最好的候选者，而不会在不好的候选者上浪费时间

节点实现 Comparable，使用搜索控件实现 compareTo()。我使用优先级队列来存储等待处理的节点，因此“最佳”节点总是排在下一个。工作人员删除一个节点，应用 div() 并将子节点添加到优先级队列中。

我的第一个方法是使用共享优先级队列，特别是 PriorityBlockingQueue。性能很糟糕，因为队列几乎总是阻塞。

我试图通过添加后台工作程序和 ConcurrentLinkedQueue 缓冲区来修复它。工作人员将添加到缓冲区，工作人员将周期性地将元素从缓冲区移动到优先级队列。这也不起作用。

我发现的最佳性能是为每个工作人员提供自己的优先级队列。我猜这已经是最好的了，因为现在线程与其他人的行为没有联系。使用此配置，在 4C/8T 机器上，我得到了 ~2.5 的加速。我认为这里的瓶颈是为所有这些节点分配内存，但我在这里可能是错的。

来自搜索者:

private PriorityQueue<Schedule> workQueue;
private static volatile boolean shutdownSignal = false;
private Schedule best;

public Searcher(List<Schedule> instances) {
    workQueue = new PriorityQueue<>(instances);
}

public static void stop() {
    shutdownSignal = true;
}

/**
 * Run the search control starting with the first node in the workQueue
 */
@Override
public void run() {

    while (!shutdownSignal) {
        try {
            Schedule next = workQueue.remove();
            List<Schedule> children = next.div(checkBest);
            workQueue.addAll(children);
        } catch (Exception e) {
            //TODO: handle exception
        }
    }
    //For testing
    System.out.println("Shutting down: " + workQueue.size());
}

//passing a function as a parameter
Consumer<Schedule> checkBest = new Consumer<Schedule>() {
    public void accept(Schedule sched) {
        if (best == null || sched.betterThan(best)) {
            best = sched;
            Model.checkBest.accept(sched);
        }
    }
};

从时间表:

public List<Schedule> div(Consumer<Schedule> completion) {
    List<Schedule> n = new ArrayList<>();

    int selected = 0;
    List<Slot> available = Model.getSlots();
    List<Slot> allocated = getAssigned();

    while (allocated.get(selected) != null) {
        selected++;
    } // find first available slot to fill.
      // Iterate through all available slots
    for (Slot t : available) {
        //Prepare a fresh copy
        List<Slot> newAssignment = new ArrayList<>(allocated.size());
        Collections.copy(newAssignment, allocated);

        //assign the course to the timeslot
        newAssignment.set(selected, t);
        Schedule next = new Schedule(this, newAssignment);
        n.add(next);
    }

    /**
     * Filter out nodes which violate the hard constraints and which are solved, 
     * and check if they are the best in a calling thread
     */

    List<Schedule> unsolvedNodes = new ArrayList<>();

    for (Schedule schedule: n) {
        if (schedule.constr() && !schedule.solved()){
            unsolvedNodes.add(schedule);
            completion.accept(schedule);
        }
    }
    return unsolvedNodes;
}

Answer 1

我会说 fork-join 框架是适合您任务的工具。您需要从 ResursiveTask 扩展您的任务或 ResursiveAction并提交给ForkJoinPool .这是一个伪代码示例。此外，您的 shutdown 标志必须是 volatile。

public class Task extends RecursiveAction {
    private final Node<Integer> node;

    public Task(Node<Integer> node) {
      this.node = node;
    }

    @Override
    protected void compute() {
      // check result and stop recursion if needed

      List<Task> subTasks = new ArrayList<>();

      List<Node<Integer>> nodes = div(this.node);
      for (Node<Integer> node : nodes) {
        Task task = new Task(node);
        task.fork();
        subTasks.add(task);
      }

      for(Task task : subTasks) {
        task.join();
      }
    }

    public static void main(String[] args) {
        Node root = getRootNode();
        new ForkJoinPool().invoke(new Task(root));
    }