Java 7 Multithread Fork Join Scheduler

Question

Я написал многопоточную программу с использованием Java fork/join framework в jre 1.7. Эта программа направлена ​​на поиск определенных точек, которые удовлетворяют заданному условию во всех узлах Quadtree (каждый листовой узел в квадранте может быть заполнен неограниченным числом точек, например, может быть нулем или 1000). Я протестировал ускорение многопоточной программы по сравнению с последовательной программой на 16-ядерном процессоре, а ускорение - только 1,3-1,5. Ниже псевдокод:

public class QuadtreeFindMultiThread extends RecursiveTask<IntArrayList> { 
private Quadtree T; 
private ObjectArrayList<Node> leaf_nodes; 
private ObjectArrayList<Entry> candidatePoints; 
private static int POINT_THRESHOLD = 50; 
private static int NODE_THRESHOLD = 1; 

public QuadtreeFindMultiThread(Quadtree T) { 
    this.T = T 
    this.leaf_nodes = T.get_nonempty_leaf_nodes(); 
    this.candidatePoints = new IntArrayList(); 
} 

private QuadtreeFindMultiThread(Quadtree T, IntArrayList leaf_nodes) { 
    this.T = T; 
    this.leaf_nodes = leaf_nodes; // reference copy 
    this.candidatePoints = new IntArrayList(); 
} 

private IntArrayList QuadtreeFind() { 
    //... 
      //... 
      return candidatePoints; 
} 

private int getPointNum(){ 
    int count = 0; 
    for(Node node:this.leaf_nodes){ 
     count += node.getAllPoints().size(); 
    } 
    return count; 
} 
@Override 
public IntArrayList compute() { 

    if (this.getPointNum() <= POINT_THRESHOLD || this.leaf_nodes.size() <= NODE_THRESHOLD) {// trivial problem, solve by single thread 
     this.candidatePoints = QuadtreeFind(); 

    } else {// START: divide and conquer 
    // Divide Step: partition this.leaf_nodes by direction: NW, NE, SW, SE 
     Partition leaf_nodes to four quadrants: leaf_nodes_NW, 
        leaf_nodes_NE, 
        leaf_nodes_SW, 
        leaf_nodes_SE 



    // Conquer Step 
    QuadtreeFindMultiThread thread_NW = new QuadtreeFindMultiThread(
       this.T, leaf_nodes_NW); 
    QuadtreeFindMultiThread thread_NE = new QuadtreeFindMultiThread(
       this.T, leaf_nodes_NE); 
      QuadtreeFindMultiThread thread_SW = new QuadtreeFindMultiThread(
       this.T, leaf_nodes_SW); 
      QuadtreeJoinMultiThread thread_SE = new QuadtreeFindMultiThread(
       this.T, leaf_nodes_SE); 
     // fork three new sub threads 
     thread_NE.fork(); 
     thread_SW.fork(); 
     thread_SE.fork(); 
     this.candidatePoints.addAll(thread_NW.compute()); // main thread 
     this.candidatePoints.addAll(thread_NE.join()); 
     this.candidatePoints.addAll(thread_SW.join()); 
     this.candidatePoints.addAll(thread_SE.join()); 

    }// END: divide and conquer 
    return this.candidatePoints; 
} 


} 

Я новичок в Java многопоточного программирования, почему эта программа получила так плохо форсировки по процессорной машине 16-ядро? Я также тестировал эту многопоточную программу на своем ноутбуке с двумя ядрами и двумя виртуальными ядрами, ускорение также приближается к 1.3-1.5. Производительность многопоточной программы на моем ноутбуке даже лучше, чем у 16-процессорного процессора.

Похоже, что политика расписания по умолчанию fork/join framefork - это LIFO, как я могу изменить FIFO?

Кстати, я обнаружил, что обработка некоторых листовых узлов, у которых много точек, занимает много времени обработки. Могу ли я изменить планировщик fork/join таким образом, чтобы он сначала обрабатывал узлы с большим количеством точек? Таким образом, он должен получить лучшую производительность. Спасибо!

Answer 1

Документации по этой структуре недостаточно. Это, конечно, легко понять неправильно. Эта структура предназначена для рекурсивного разложения сбалансированных деревьев (DAG.). Она не очень хорошо переносит неправильное использование, поскольку она была первоначально разработана как эксперимент в исследовательской работе.

Рамка хочет разделить влево, вправо. Left.fork(), right.compute(), left.join(). Таким образом, он спускается по листьям сбалансированного дерева. Завораживающие задачи возвращаются в свое дебе, надежно украденное другими нитками. Когда все идет так, как планировалось, каждый поток задает достаточно задач для других потоков и остается занятым.

Что вы делаете, это вернуть три задачи обратно в deque, а затем обработать один квад. Это не очень хорошо прокладывает работу. В итоге вы можете использовать несколько потоков, для которых много задач, а не много потоков, имеющих несколько ожидающих задач. Эта структура неспособна правильно распределить нагрузку.

Существует также проблема соединения(). Join() требует контекстного переключателя, чтобы освободить поток для другой работы. Эта структура не может выполнять контекстный переключатель, поэтому он создает «поток продолжения» для каждого соединения(), а затем выдает wait() для потока соединения.Со многими объединениями может возникнуть много проблем с созданием/уничтожением. Версия Java8 устраняет «продолжение потока», но часто останавливается (особенно так, как вы это делаете).

Попробуйте перепроектировать DAG и посмотреть, что произойдет. С 16 потоками он должен работать хорошо.

Answer 2

Хитрость для любой параллельной задачи, чтобы сбалансировать две различные задачи:

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

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

Оптимальный порог зависит от многих факторов - в основном от размера проблемы, но на них могут также сильно влиять различные архитектуры компьютеров, памяти и т. Д. Лучший способ понять это - сделать порог входным параметром и запустить тесты в разной степени.