Wysokowydajne buforowanie strumienia potoków

Question

Mam kod, który pochłania dużą liczbę (obecnie miliony, a nawet miliardy) stosunkowo krótkich (5-100 elementów) tablic liczb losowych i wykonuje niektóre niezbyt intensywne zadania matematyczne im. Losowe liczby są, no cóż, przypadkowe, najlepiej chciałbym je wygenerować na wielu rdzeniach, ponieważ generowanie liczb losowych stanowi> 50% mojego środowiska wykonawczego w profilowaniu. Mam jednak trudności z rozprowadzaniem dużej liczby małych zadań w sposób, który nie jest wolniejszy od podejścia z pojedynczym gwintem.

Moje kodu aktualnie wygląda tak:

for(int i=0;i<1000000;i++){ 
    for(RealVector d:data){ 
     while(!converged){ 
      double[] shortVec = new double[5]; 
      for(int i=0;i<5;i++) shortVec[i]=rng.nextGaussian(); 
      double[] longerVec = new double[50]; 
      for(int i=0;i<50;i++) longerVec[i]=rng.nextGaussian(); 
      /*Do some relatively fast math*/ 
     } 
    } 
} 

Podejścia Wziąłem że mają nie pracował to:

• 1+ wątki Zapełnienie ArrayBlockingQueue, a moim głównym pętla spożywania i zapełnianie tablicy (boxing/unboxing był tu zabójcą)
• Generowanie wektorów z możliwością wywołania (przynoszącego przyszłość) podczas wykonywania niezależnych części matematyki (Wygląda na to, że nadwyżka nadwyżki przeważała nad każdym uzyskanym paralelizmem)
• Używanie 2 ArrayBlockingQueue, z których każdy jest wypełniony wątkiem, jeden dla krótkiego i jeden dla długich tablic (wciąż mniej więcej dwa razy wolniejszy niż bezpośredni jednowątkowy walizka).

Nie szukam "rozwiązań" dla mojego konkretnego problemu, tak jak radzić sobie z ogólnym przypadkiem generowania dużych strumieni małych, niezależnych prymitywów równolegle i konsumowania ich z jednego wątku.

Answer 1

Jest bardziej wydajny niż przy użyciu kolejki ponieważ;

• ładunek jest tablicą o numerze double[], co oznacza, że ​​wątek tła może wygenerować więcej danych, zanim będzie musiał go przekazać.
• wszystkie obiekty są poddawane recyklingowi.

.

public class RandomGenerator { 
    private final ExecutorService generator = Executors.newSingleThreadExecutor(new ThreadFactory() { 
     @Override 
     public Thread newThread(Runnable r) { 
      Thread t = new Thread(r, "generator"); 
      t.setDaemon(true); 
      return t; 
     } 
    }); 
    private final Exchanger<double[][]> exchanger = new Exchanger<>(); 
    private double[][] buffer; 
    private int nextRow = Integer.MAX_VALUE; 

    public RandomGenerator(final int rows, final int columns) { 
     buffer = new double[rows][columns]; 
     generator.submit(new Callable<Void>() { 
      @Override 
      public Void call() throws Exception { 
       Random random = new Random(); 
       double[][] buffer2 = new double[rows][columns]; 
       while (!Thread.interrupted()) { 
        for (int r = 0; r < rows; r++) 
         for (int c = 0; c < columns; c++) 
          buffer2[r][c] = random.nextGaussian(); 
        buffer2 = exchanger.exchange(buffer2); 
       } 
       return null; 
      } 
     }); 
    } 

    public double[] nextArray() throws InterruptedException { 
     if (nextRow >= buffer.length) { 
      buffer = exchanger.exchange(buffer); 
      nextRow = 0; 
     } 
     return buffer[nextRow++]; 
    } 
} 

---

Losowy wątek jest bezpieczne i zsynchronizowane. Oznacza to, że każdy wątek potrzebuje własnego Losowego do równoczesnego wykonywania.

sposób radzenia sobie z ogólnym przypadkiem generowania dużych strumieni małych, niezależnych elementów pierwotnych równolegle i pochłaniania ich z jednego wątku.

chciałbym użyć Exchanger<double[][]> do wypełnienia wartości w tle jak przekazać je skutecznie (bez większego narzutu GC)

Answer 2

Problem z wydajnością wydaje się, że poszczególne zadania są zbyt małe, więc większość czasu spędza się na synchronizacji i kolejkowaniu samych zadań. Jedną z rzeczy do rozważenia jest , a nie generowanie dużego strumienia małych zadań, ale dostarczanie do każdego działającego wątku średniej wielkości zbioru zadań, które będzie adnotować z odpowiedzią.

Na przykład, zamiast powtarzania przez pętlę z pierwszym wątkiem wykonującym iterację # 0, kolejny wątek wykonujący iterację # 1, ... Miałem pierwszy wątek do iteracji od 0 do 999 lub kilka takich. Powinny one pracować niezależnie i opisywać klasę Job z odpowiedzią na ich obliczenia. Następnie na końcu mogą zwrócić cały zbiór zadań, które zostały zakończone jako Future.

Twoja klasa Job może być coś jak poniżej:

public class Job { 
    Collection<RealVector> dataCollection; 
    Collection<SomeAnswer> answerCollection = new ArrayList<SomeAnswer>(); 
    public void run() { 
     for (RealVector d : dataCollection) { 
      // do the magic work on the vector 
      while(!converged){ 
       ... 
      } 
      // put the associated "answer" in another collection 
      answerCollection.add(someAnswer); 
     } 
    } 
}