Jakie są zalety korzystania z nowej struktury rozwidleń / złączeń w porównaniu z prostym podzieleniem dużego zadania na N podzadań na początku, wysłaniem ich do puli wątków w pamięci podręcznej (z wykonawców ) i czekaniem na zakończenie każdego zadania? Nie widzę, jak użycie abstrakcji rozwidlenia / złączenia upraszcza problem lub sprawia, że ​​rozwiązanie jest bardziej wydajne od tego, co mamy od lat.

Na przykład równoległy algorytm rozmycia w przykładzie samouczka można zaimplementować w następujący sposób:

public class Blur implements Runnable {
    private int[] mSource;
    private int mStart;
    private int mLength;
    private int[] mDestination;

    private int mBlurWidth = 15; // Processing window size, should be odd.

    public ForkBlur(int[] src, int start, int length, int[] dst) {
        mSource = src;
        mStart = start;
        mLength = length;
        mDestination = dst;
    }

    public void run() {
        computeDirectly();
    }

    protected void computeDirectly() {
        // As in the example, omitted for brevity
    }
}

Podziel na początku i wyślij zadania do puli wątków:

// source image pixels are in src
// destination image pixels are in dst
// threadPool is a (cached) thread pool

int maxSize = 100000; // analogous to F-J's "sThreshold"
List<Future> futures = new ArrayList<Future>();

// Send stuff to thread pool:
for (int i = 0; i < src.length; i+= maxSize) {
    int size = Math.min(maxSize, src.length - i);
    ForkBlur task = new ForkBlur(src, i, size, dst);
    Future f = threadPool.submit(task);
    futures.add(f);
}

// Wait for all sent tasks to complete:
for (Future future : futures) {
    future.get();
}

// Done!

Zadania trafiają do kolejki puli wątków, z której są wykonywane, gdy stają się dostępne wątki robocze. Dopóki dzielenie jest wystarczająco szczegółowe (aby uniknąć konieczności czekania w szczególności na ostatnie zadanie), a pula wątków ma wystarczającą liczbę (co najmniej N procesorów) wątków, wszystkie procesory pracują z pełną prędkością, aż do zakończenia obliczeń.

Czy coś mi brakuje? Jaka jest wartość dodana korzystania z frameworka fork / join?

Myślę, że podstawowym nieporozumieniem jest to, że przykłady Fork / Join NIE pokazują kradzieży pracy, a jedynie pewien rodzaj standardowego dziel i rządź.

Kradzież pracy wyglądałaby tak: Pracownik B zakończył swoją pracę. Jest miły, więc rozgląda się i widzi Pracownika A, który nadal bardzo ciężko pracuje. Podchodzi i pyta: „Hej, chłopcze, mogę ci pomóc”. Odpowiedzi. „Super, mam to zadanie na 1000 jednostek. Jak dotąd ukończyłem 345, pozostawiając 655. Czy mógłbyś popracować nad numerami 673 do 1000, zrobię 346 do 672”. B mówi „OK, zacznijmy, żebyśmy mogli wcześniej iść do pubu”.

Widzisz - pracownicy muszą komunikować się między sobą, nawet gdy zaczynali prawdziwą pracę. To jest brakująca część w przykładach.

Z drugiej strony przykłady pokazują tylko coś w rodzaju „korzystaj z podwykonawców”:

Pracownik A: „Cholera, mam 1000 jednostek pracy. Za dużo dla mnie. Zrobię 500 osobiście i podwykonuję 500 komuś innemu”. Trwa to tak długo, aż wielkie zadanie zostanie podzielone na małe pakiety po 10 jednostek każda. Zostaną one wykonane przez dostępnych pracowników. Ale jeśli jedno opakowanie jest rodzajem trującej pigułki i trwa znacznie dłużej niż inne opakowania - pech, faza podziału dobiegła końca.

Jedyna pozostała różnica między rozwidleniem / połączeniem a dzieleniem zadania z góry jest taka: podczas rozdzielania z góry kolejka pracy jest pełna od samego początku. Przykład: 1000 jednostek, próg wynosi 10, więc kolejka ma 100 wpisów. Te pakiety są dystrybuowane do członków puli wątków.

Rozwidlanie / łączenie jest bardziej złożone i stara się zmniejszyć liczbę pakietów w kolejce:

• Krok 1: Umieść jeden pakiet zawierający (1 ... 1000) w kolejce
• Krok 2: Jeden pracownik pobiera pakiet (1 ... 1000) i zastępuje go dwoma pakietami: (1 ... 500) i (501 ... 1000).
• Krok 3: Jeden pracownik zdejmuje pakiet (500 ... 1000) i wypycha (500 ... 750) i (751 ... 1000).
• Krok n: Stos zawiera następujące pakiety: (1..500), (500 ... 750), (750 ... 875) ... (991..1000)
• Krok n + 1: pakiet (991..1000) jest pobierany i wykonywany
• Krok n + 2: pakiet (981..990) jest pobierany i wykonywany
• Krok n + 3: Pakiet (961..980) jest pobierany i dzielony na (961 ... 970) i ​​(971..980). ....

Widzisz: w trybie Fork / Join kolejka jest mniejsza (w przykładzie 6), a fazy „split” i „work” są przeplatane.

Gdy wielu pracowników jednocześnie pcha i pcha, interakcje nie są oczywiście tak jasne.

Jeśli masz n zajętych wątków, wszystkie pracują w 100% niezależnie, będzie to lepsze niż n wątków w puli Fork-Join (FJ). Ale to nigdy nie działa.

Może nie być w stanie precyzyjnie podzielić problemu na n równych części. Nawet jeśli to zrobisz, planowanie wątków nie jest sprawiedliwe. Skończysz czekając na najwolniejszy wątek. Jeśli masz wiele zadań, każde z nich może działać z równoległością mniejszą niż n-kierunkową (ogólnie bardziej wydajną), ale przechodzić do n-kierunkowej po zakończeniu innych zadań.

Dlaczego więc po prostu nie podzielimy problemu na kawałki rozmiaru FJ i nie popracujemy nad tym puli wątków. Typowe użycie FJ tnie problem na drobne kawałki. Wykonywanie tych czynności w przypadkowej kolejności wymaga dużej koordynacji na poziomie sprzętu. Koszty ogólne byłyby zabójcze. W FJ zadania są umieszczane w kolejce, którą wątek odczytuje w kolejności Last In First Out (LIFO / stack), a kradzież pracy (ogólnie w pracy podstawowej) jest wykonywana First In First Out (FIFO / „kolejka”). W rezultacie przetwarzanie długich macierzy może być wykonywane w dużej mierze sekwencyjnie, nawet jeśli jest podzielone na małe kawałki. (Jest również tak, że rozbicie problemu na małe, równe fragmenty w jednym wielkim wybuchu może nie być trywialne. Powiedzmy, że zajmujemy się jakąś formą hierarchii bez balansowania).

Wniosek: FJ pozwala na bardziej efektywne wykorzystanie wątków sprzętowych w nierównych sytuacjach, co zawsze będzie miało miejsce, jeśli masz więcej niż jeden wątek.

Ostateczny cel puli wątków i rozwidlenia / łączenia jest podobny: oba chcą maksymalnie wykorzystać dostępną moc procesora, aby uzyskać maksymalną przepustowość. Maksymalna przepustowość oznacza, że ​​jak najwięcej zadań powinno być wykonanych w długim okresie. Co jest do tego potrzebne? (W poniższym przykładzie założymy, że nie brakuje zadań obliczeniowych: zawsze jest wystarczająco dużo do zrobienia dla 100% wykorzystania procesora. Dodatkowo używam „CPU” jako odpowiedników rdzeni lub wirtualnych rdzeni w przypadku hiperwątkowości).

1. Przynajmniej musi być uruchomionych tyle wątków, ile jest dostępnych procesorów, ponieważ uruchomienie mniejszej liczby wątków pozostawi rdzeń niewykorzystany.
2. Maksymalnie musi być uruchomionych tyle wątków, ile jest dostępnych procesorów, ponieważ uruchomienie większej liczby wątków spowoduje dodatkowe obciążenie dla harmonogramu, który przypisuje procesory do różnych wątków, co powoduje, że część czasu procesora przechodzi do harmonogramu, a nie do naszego zadania obliczeniowego.

W ten sposób ustaliliśmy, że dla maksymalnej przepustowości musimy mieć dokładnie taką samą liczbę wątków jak procesory. W rozmytym przykładzie Oracle można zarówno wziąć pulę wątków o stałym rozmiarze z liczbą wątków równą liczbie dostępnych procesorów, albo użyć puli wątków. To nie ma znaczenia, masz rację!

Kiedy więc będziesz mieć kłopoty z pulami wątków? Dzieje się tak, jeśli wątek blokuje się , ponieważ Twój wątek czeka na zakończenie innego zadania. Przyjmijmy następujący przykład:

class AbcAlgorithm implements Runnable {
    public void run() {
        Future<StepAResult> aFuture = threadPool.submit(new ATask());
        StepBResult bResult = stepB();
        StepAResult aResult = aFuture.get();
        stepC(aResult, bResult);
    }
}

Widzimy tutaj algorytm, który składa się z trzech kroków A, B i C.A i B można wykonać niezależnie od siebie, ale krok C wymaga wyniku kroku A AND B. To, co robi ten algorytm, to przesłanie zadania A do puli wątków i wykonaj zadanie b bezpośrednio. Następnie wątek będzie czekał na wykonanie zadania A i będzie kontynuował od kroku C. Jeśli A i B zostaną zakończone w tym samym czasie, wszystko jest w porządku. Ale co, jeśli A trwa dłużej niż B? Może tak być, ponieważ dyktuje to natura zadania A, ale może też tak być, ponieważ na początku nie ma wątku dla zadania A i zadanie A musi czekać. (Jeśli dostępny jest tylko jeden procesor, a zatem pula wątków ma tylko jeden wątek, spowoduje to nawet zakleszczenie, ale na razie nie ma to znaczenia). Chodzi o to, że wątek, który właśnie wykonał zadanie Bblokuje cały wątek . Ponieważ mamy taką samą liczbę wątków jak procesory, a jeden wątek jest zablokowany, oznacza to, że jeden procesor jest bezczynny .

Fork / Join rozwiązuje ten problem: w ramach fork / Join napisałbyś ten sam algorytm w następujący sposób:

class AbcAlgorithm implements Runnable {
    public void run() {
        ATask aTask = new ATask());
        aTask.fork();
        StepBResult bResult = stepB();
        StepAResult aResult = aTask.join();
        stepC(aResult, bResult);
    }
}

Wygląda tak samo, prawda? Jednak wskazówka jest taka, aTask.join że nie będzie blokować . Zamiast tego w grę wchodzi kradzież pracy : wątek rozejrzy się za innymi zadaniami, które zostały rozwidlone w przeszłości i będzie kontynuował te. Najpierw sprawdza, czy zadania, które sam rozwidliły, rozpoczęły przetwarzanie. Więc jeśli A nie został jeszcze uruchomiony przez inny wątek, zrobi A następny, w przeciwnym razie sprawdzi kolejkę innych wątków i wykradnie ich pracę. Po zakończeniu tego innego zadania innego wątku sprawdzi, czy A jest teraz zakończone. Jeśli jest to powyższy algorytm może wywołać stepC. W przeciwnym razie będzie szukał kolejnego zadania do kradzieży. W ten sposób pule rozwidleń / złączeń mogą osiągnąć 100% wykorzystanie procesora, nawet w obliczu działań blokujących .

Jest jednak pułapka: kradzież pracy jest możliwa tylko na joinwezwanie ForkJoinTasks. Nie można tego zrobić w przypadku zewnętrznych akcji blokujących, takich jak oczekiwanie na inny wątek lub oczekiwanie na akcję we / wy. A co z tym, czekanie na zakończenie operacji we / wy jest częstym zadaniem? W takim przypadku, gdybyśmy mogli dodać dodatkowy wątek do puli Rozwidlania / Łączenia, który zostanie zatrzymany ponownie, gdy tylko akcja blokowania zostanie zakończona, będzie drugą najlepszą rzeczą do zrobienia. I ForkJoinPoolfaktycznie może to zrobić, jeśli używamy ManagedBlockers.

Fibonacci

W JavaDoc for RecursiveTask znajduje się przykład obliczania liczb Fibonacciego za pomocą Fork / Join. Aby zapoznać się z klasycznym rozwiązaniem rekurencyjnym, zobacz:

public static int fib(int n) {
    if (n <= 1) {
        return n;
    }
    return fib(n - 1) + fib(n - 2);
}

Jak wyjaśniono w JavaDocs, jest to całkiem prosty sposób obliczania liczb Fibonacciego, ponieważ ten algorytm ma złożoność O (2 ^ n), podczas gdy prostsze sposoby są możliwe. Jednak ten algorytm jest bardzo prosty i łatwy do zrozumienia, więc trzymamy się go. Załóżmy, że chcemy to przyspieszyć za pomocą Fork / Join. Naiwna implementacja wyglądałaby tak:

class Fibonacci extends RecursiveTask<Long> {
    private final long n;

    Fibonacci(long n) {
        this.n = n;
    }

    public Long compute() {
        if (n <= 1) {
            return n;
        }
        Fibonacci f1 = new Fibonacci(n - 1);
        f1.fork();
        Fibonacci f2 = new Fibonacci(n - 2);
        return f2.compute() + f1.join();
   }
}

Kroki, na które podzielone jest to zadanie, są zbyt krótkie i dlatego będą działać okropnie, ale możesz zobaczyć, jak szkielet ogólnie działa bardzo dobrze: dwa szczyty można obliczyć niezależnie, ale wtedy potrzebujemy ich obu do zbudowania ostatecznego wynik. Więc połowa jest zrobiona w innym wątku. Baw się dobrze, robiąc to samo z pulami wątków bez zakleszczenia (możliwe, ale nie tak proste).

Tylko dla kompletności: jeśli faktycznie chcesz obliczyć liczby Fibonacciego za pomocą tego rekurencyjnego podejścia, oto zoptymalizowana wersja:

class FibonacciBigSubtasks extends RecursiveTask<Long> {
    private final long n;

    FibonacciBigSubtasks(long n) {
        this.n = n;
    }

    public Long compute() {
        return fib(n);
    }

    private long fib(long n) {
        if (n <= 1) {
            return 1;
        }
        if (n > 10 && getSurplusQueuedTaskCount() < 2) {
            final FibonacciBigSubtasks f1 = new FibonacciBigSubtasks(n - 1);
            final FibonacciBigSubtasks f2 = new FibonacciBigSubtasks(n - 2);
            f1.fork();
            return f2.compute() + f1.join();
        } else {
            return fib(n - 1) + fib(n - 2);
        }
    }
}

Dzięki temu podzadania są znacznie mniejsze, ponieważ są dzielone tylko wtedy, gdy n > 10 && getSurplusQueuedTaskCount() < 2jest prawdziwe, co oznacza, że ​​istnieje znacznie więcej niż 100 wywołań metod do ( n > 10) i nie czekają już zadania bardzo man ( getSurplusQueuedTaskCount() < 2).

Na moim komputerze (4 rdzenie (8, licząc Hyper-Threading), procesor Intel (R) Core (TM) i7-2720QM @ 2,20 GHz) fib(50)zajmuje to 64 sekundy przy klasycznym podejściu i zaledwie 18 sekund przy podejściu Fork / Join, które to dość zauważalny zysk, choć nie tak duży, jak teoretycznie możliwy.

Podsumowanie

• Tak, w twoim przykładzie Fork / Join nie ma przewagi nad klasycznymi pulami wątków.
• Rozwidlanie / łączenie może drastycznie poprawić wydajność podczas blokowania
• Rozwidlenie / łączenie pozwala uniknąć niektórych problemów z zakleszczeniem

Fork / Join różni się od puli wątków, ponieważ implementuje kradzież pracy. Z Fork / Join

Podobnie jak w przypadku każdej usługi ExecutorService, struktura rozwidlenia / złączenia dystrybuuje zadania do wątków roboczych w puli wątków. Struktura rozwidlenia / złączenia jest odrębna, ponieważ wykorzystuje algorytm kradzieży pracy. Wątki robocze, którym zabrakło rzeczy do zrobienia, mogą kraść zadania z innych wątków, które są nadal zajęte.

Załóżmy, że masz dwa wątki i 4 zadania a, b, c, d, które trwają odpowiednio 1, 1, 5 i 6 sekund. Początkowo a i b są przypisywane do wątku 1, a c id do wątku 2. W puli wątków zajęłoby to 11 sekund. Z rozwidleniem / złączeniem wątek 1 kończy pracę i może ukraść pracę z wątku 2, więc zadanie d zostanie wykonane przez wątek 1. Wątek 1 wykonuje a, bi d, wątek 2 po prostu c. Całkowity czas: 8 sekund, a nie 11.

EDYCJA: Jak wskazuje Joonas, zadania niekoniecznie są wstępnie przydzielane do wątku. Idea rozwidlenia / łączenia polega na tym, że wątek może podzielić zadanie na wiele części. Tak więc, aby powtórzyć powyższe:

Mamy dwa zadania (ab) i (cd), które trwają odpowiednio 2 i 11 sekund. Wątek 1 rozpoczyna wykonywanie ab i dzieli go na dwa podzadania a i b. Podobnie z wątkiem 2 dzieli się na dwa podzadania c & d. Gdy wątek 1 zakończy a i b, może ukraść d z wątku 2.

Wszyscy powyżej mają rację, korzyści płynące z kradzieży pracy, ale aby rozwinąć dlaczego tak jest.

Podstawową korzyścią jest wydajna koordynacja między wątkami roboczymi. Praca musi zostać podzielona i ponownie złożona, co wymaga koordynacji. Jak widać w odpowiedzi AH powyżej, każdy wątek ma własną listę zadań. Ważną właściwością tej listy jest to, że jest posortowana (duże zadania u góry i małe zadania u dołu). Każdy wątek wykonuje zadania na dole swojej listy i kradnie zadania z górnej części innych list wątków.

Wynikiem tego jest:

• Początek i koniec list zadań można synchronizować niezależnie, zmniejszając rywalizację o listę.
• Znaczące poddrzewa pracy są dzielone i składane ponownie przez ten sam wątek, więc dla tych poddrzew nie jest wymagana koordynacja między wątkami.
• Kiedy wątek kradnie pracę, zajmuje duży kawałek, który następnie dzieli na własną listę
• Stal robocza oznacza, że ​​gwinty są prawie w pełni wykorzystywane do końca procesu.

Większość innych schematów dzielenia i podbijania wykorzystujących pule wątków wymaga większej komunikacji i koordynacji między wątkami.

W tym przykładzie Fork / Join nie dodaje żadnej wartości, ponieważ rozwidlanie nie jest potrzebne, a obciążenie jest równomiernie podzielone na wątki robocze. Tylko rozwidlenie / połączenie dodaje narzut.

Oto fajny artykuł na ten temat. Zacytować:

Ogólnie możemy powiedzieć, że ThreadPoolExecutor ma być preferowany, gdy obciążenie jest równomiernie podzielone na wątki robocze. Aby móc to zagwarantować, musisz dokładnie wiedzieć, jak wyglądają dane wejściowe. Z kolei ForkJoinPool zapewnia dobrą wydajność niezależnie od danych wejściowych, a zatem jest znacznie bardziej niezawodnym rozwiązaniem.

Inną ważną różnicą wydaje się być to, że w przypadku FJ można wykonać wiele złożonych faz „łączenia”. Rozważ sortowanie przez scalanie z http://faculty.ycp.edu/~dhovemey/spring2011/cs365/lecture/lecture18.html , wymagałoby zbyt dużej orkiestracji, aby wstępnie podzielić tę pracę. np. musisz wykonać następujące czynności:

• posortuj pierwszy kwartał
• posortuj drugi kwartał
• scal pierwsze 2 kwartały
• posortuj trzeci kwartał
• posortuj czwarty kwartał
• scal ostatnie 2 kwartały
• połącz 2 połówki

Jak określić, że musisz dokonać sortowania przed połączeniami, które ich dotyczą itp.

Zastanawiałem się, jak najlepiej zrobić określoną rzecz dla każdej listy pozycji. Myślę, że po prostu wstępnie podzielę listę i użyję standardowej puli wątków. FJ wydaje się najbardziej przydatne, gdy praca nie może być wstępnie podzielona na wystarczająco niezależne zadania, ale można ją rekurencyjnie podzielić na zadania, które są niezależne od siebie (np. Sortowanie połówek jest niezależne, ale scalanie 2 posortowanych połówek w posortowaną całość nie jest).

F / J ma również wyraźną zaletę, gdy masz kosztowne operacje scalania. Ponieważ dzieli się na strukturę drzewiastą, tylko log2 (n) łączy się, w przeciwieństwie do n scaleń z liniowym podziałem wątków. (To teoretycznie zakłada, że ​​masz tyle procesorów, ile wątków, ale nadal jest to zaleta) W przypadku zadania domowego musieliśmy scalić kilka tysięcy tablic 2D (wszystkie te same wymiary), sumując wartości w każdym indeksie. Z rozwidleniem złączenia i procesorami P czas zbliża się do log2 (n), gdy P zbliża się do nieskończoności.

1 2 3 .. 7 3 1 .... 8 5 4
4 5 6 + 2 4 3 => 6 9 9
7 8 9 .. 1 1 0 .... 8 9 9

Byłbyś zdumiony wydajnością ForkJoin w aplikacjach takich jak crawler. oto najlepszy samouczek, z którego możesz się nauczyć.

Logika Fork / Join jest bardzo prosta: (1) rozdziel (podziel) każde duże zadanie na mniejsze; (2) przetwarzaj każde zadanie w osobnym wątku (dzieląc je na jeszcze mniejsze zadania, jeśli to konieczne); (3) dołącz wyniki.

Jeśli problem jest taki, że musimy poczekać na zakończenie innych wątków (jak w przypadku sortowania tablicy lub sumy tablic), należy użyć złączenia rozwidlonego, ponieważ Executor (Executors.newFixedThreadPool (2)) będzie się dławił z powodu ograniczonego Liczba wątków. Pula forkjoin utworzy w tym przypadku więcej wątków, aby zakryć zablokowany wątek, aby zachować tę samą równoległość

Źródło: http://www.oracle.com/technetwork/articles/java/fork-join-422606.html

Problem z wykonawcami do implementacji algorytmów dziel i zwyciężaj nie jest związany z tworzeniem podzadań, ponieważ wywoływana może przesłać nowe podzadanie do swojego modułu wykonawczego i czekać na jego wynik w sposób synchroniczny lub asynchroniczny. Problem jest związany z równoległością: gdy wywołanie czeka na wynik innego wywoływanego, przechodzi w stan oczekiwania, marnując w ten sposób okazję do obsłużenia innego wywoływanego w kolejce do wykonania.

Struktura rozwidlenia / złączenia dodana do pakietu java.util.concurrent w Javie SE 7 dzięki wysiłkom Douga Lei wypełnia tę lukę

Źródło: https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ForkJoinPool.html

Pula próbuje utrzymać wystarczającą liczbę aktywnych (lub dostępnych) wątków przez dynamiczne dodawanie, zawieszanie lub wznawianie wewnętrznych wątków roboczych, nawet jeśli niektóre zadania są wstrzymane i czekają na dołączenie do innych. Jednak żadne takie regulacje nie są gwarantowane w przypadku zablokowania we / wy lub innej niezarządzanej synchronizacji

public int getPoolSize () Zwraca liczbę wątków roboczych, które zostały uruchomione, ale jeszcze nie zostały zakończone. Wynik zwracany przez tę metodę może różnić się od metody getParallelism (), gdy wątki są tworzone w celu zachowania równoległości, gdy inne są wspólnie blokowane.

Chciałbym dodać krótką odpowiedź dla tych, którzy nie mają zbyt wiele czasu na przeczytanie długich odpowiedzi. Porównanie zaczerpnięto z książki Applied Akka Patterns:

Decyzja o tym, czy użyć executora rozwidlenia złączenia, czy też modułu wykonującego pulę wątków, jest w dużej mierze oparta na tym, czy operacje w tym programie rozsyłającym będą blokowane. Wykonawca rozwidlenia złączenia zapewnia maksymalną liczbę aktywnych wątków, podczas gdy wykonawca puli wątków zapewnia stałą liczbę wątków. Jeśli wątki są zablokowane, executor rozwidlenia złączenia utworzy więcej, podczas gdy wykonawca puli wątków nie. W przypadku operacji blokowych generalnie lepiej jest korzystać z modułu wykonawczego puli wątków, ponieważ zapobiega on eksplodowaniu liczby wątków. Bardziej „reaktywne” operacje są lepsze w executorze fork-join.