Buffering ad alte prestazioni per un flusso di randelli

Question

Ho codice che consuma un numero elevato (milioni attualmente, alla fine miliardi) di array relativamente brevi (5-100 elementi) di numeri casuali e fa una matematica non molto faticosa con loro. I numeri casuali essendo, beh, random, idealmente mi piacerebbe generarli su più core, poiché la generazione casuale di numeri è> 50% del mio runtime in profiling. Tuttavia, ho difficoltà a distribuire un numero elevato di piccole attività in un modo non più lento rispetto all'approccio a thread singolo.

Il mio codice attualmente sembra qualcosa di simile:

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*/ 
     } 
    } 
} 

approcci che ho preso che hanno non lavorato sono:

• 1+ discussioni che popolano un ArrayBlockingQueue, e il mio ciclo principale di consumare e popolando l'array (il boxing/unboxing era killer qui)
• Generare i vettori con un Callable (che produce un futuro) mentre si fanno le parti non dipendenti del calcolo matematico (sembra che il sovraccarico della indiretta sia superiore a qualsiasi guadagno di parallelismo che ho ottenuto)
• Utilizzo di 2 ArrayBlockingQueue, ciascuno compilato da un thread, uno per il corto e uno per i lunghi array (ancora circa il doppio rispetto al diretto single-threaded Astuccio).

Non sono alla ricerca di "soluzioni" per il mio particolare problema tanto quanto gestire il caso generale di generazione di grandi flussi di primitive indipendenti piccole in parallelo e consumo da un singolo thread.

Answer 1

Questo è più efficiente rispetto all'utilizzo di una coda a causa;

• il payload è un array di double[] che significa che il thread in background può generare più dati prima di doverlo passare.
• tutti gli oggetti sono riciclati.

.

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++]; 
    } 
} 

---

caso è thread sicuro e sincronizzato. Questo significa che ogni thread ha bisogno del proprio Random da eseguire contemporaneamente.

come gestire il caso generale di generazione di grandi flussi di primitive indipendenti piccole in parallelo e consumo da un singolo thread.

vorrei utilizzare un Exchanger<double[][]> per popolare i valori in background come passare in modo efficiente (senza overhead molto GC)

Answer 2

Il problema con le prestazioni sembra essere il fatto che i singoli lavori sono troppo piccoli, quindi la maggior parte del tempo è trascorso facendo la sincronizzazione e l'accodamento dei processi stessi. Una cosa da considerare è non per generare un grande flusso di piccoli lavori ma per consegnare a ciascun thread di lavoro una raccolta di lavori di medie dimensioni che annota con la risposta.

Ad esempio, invece di scorrere il ciclo con il primo thread facendo l'iterazione # 0, il thread successivo facendo l'iterazione # 1, ... avrei il primo thread da iterazioni da # 0 a # 999 o alcuni di questi. Dovrebbero funzionare in modo indipendente e annotare una classe Job con la risposta dei loro calcoli. Quindi alla fine possono restituire l'intera collezione dei lavori che sono stati completati come Future.

La classe Job potrebbe essere qualcosa di simile al seguente:

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); 
     } 
    } 
}