Jak zapobiegać głodowaniu aktorów w obecności innych długoletnich aktorów?

Question

To używa Scala 2.8 Aktorów. Mam długą pracę, którą można zrównoleglić. Składa się z około 650 000 jednostek pracy. Dzielę go do 2600 różnych oddzielnych podzadań, a dla każdego z nich utworzyć nowy Aktor:

actor { 
    val range = (0L to total by limit) 
    val latch = new CountDownLatch(range.length) 
    range.foreach { offset => 
    actor { 
     doExpensiveStuff(offset,limit) 
     latch.countDown 
    } 
    } 
    latch.await 
} 

to działa dość dobrze, ale ogólnie trwa 2 + H, aby zakończyć. Problem polega na tym, że w międzyczasie inni aktorzy, którzy tworzą zwyczajne zadania, wydają się być zagubieni przez pierwszych 2600 aktorów, którzy cierpliwie czekają, aż nadejdzie ich czas na wątek, ale czekali dłużej niż nowi aktorzy. zbliżać się.

Jak mogę uniknąć tego głodu?

myśli wstępne:

• Zamiast 2600 aktorów, użyj jednego aktora, który kolejno pługi przez stertę pracy. Nie przepadam za tym, ponieważ chciałbym, aby ta praca skończyła się wcześniej, dzieląc ją.
• Zamiast 2600 aktorów użyj dwóch aktorów, z których każdy przetwarza inną połowę całego zestawu zadań. To może działać lepiej, ale co, jeśli moja maszyna ma 8 rdzeni? Prawdopodobnie chciałbym wykorzystać więcej.

UPDATE

Niektórzy ludzie kwestionowali wykorzystanie Aktorzy w ogóle, zwłaszcza, że ​​zdolność przechodzenia wiadomość nie był używany w ciągu pracowników. Przyjąłem, że aktor był bardzo lekką abstrakcją wokół ThreadPool na poziomie lub w pobliżu tego samego poziomu wydajności, po prostu ręcznie kodując wykonanie oparte na ThreadPool. Więc napisałem trochę odniesienia:

import testing._ 
import java.util.concurrent._ 
import actors.Futures._ 

val count = 100000 
val poolSize = 4 
val numRuns = 100 

val ActorTest = new Benchmark { 
    def run = { 
    (1 to count).map(i => future { 
     i * i 
    }).foreach(_()) 
    } 
} 

val ThreadPoolTest = new Benchmark { 
    def run = { 
    val queue = new LinkedBlockingQueue[Runnable] 
    val pool = new ThreadPoolExecutor(
      poolSize, poolSize, 1, TimeUnit.SECONDS, queue) 
    val latch = new CountDownLatch(count) 
    (1 to count).map(i => pool.execute(new Runnable { 
     override def run = { 
     i * i 
     latch.countDown 
     } 
    })) 
    latch.await 
    } 
} 

List(ActorTest,ThreadPoolTest).map { b => 
    b.runBenchmark(numRuns).sum.toDouble/numRuns 
} 

// List[Double] = List(545.45, 44.35) 

użyłem Future abstrakcję w ActorTest uniknąć przekazując wiadomość z powrotem do innego uczestnika, aby zasygnalizować praca została wykonana. Zaskoczyło mnie, że mój kod aktora był ponad 10 razy wolniejszy. Zauważ, że utworzyłem również swój ThreadPoolExecutor z początkowym rozmiarem puli, z którym tworzona jest domyślna pula aktorów.

Patrząc wstecz, wydaje mi się, że prawdopodobnie nadużyłem abstrakcji Aktora. Zamierzam przyjrzeć się użyciu osobnych ThreadPools dla tych odrębnych, kosztownych, długotrwałych zadań.

Answer 1

Nieważne ile masz aktorów, jeśli nie jesteś konfigurowania harmonogramu wyraźnie, wszystkie z nich są wspierane ze pojedynczego widelca/join Scheduler (działa przeciw puli wątków o pojemności 4, jeśli się nie mylę). To stąd bierze się głód.

1. Należy próbować różnych szeregowania dla swojej puli aktorów, aby znaleźć taki, który pokazuje najlepszą wydajność (spróbuj ResizableThreadPoolScheduler, jeśli chcesz, aby zmaksymalizować równoległość używając jak najwięcej tematów jak to możliwe)
2. trzeba mieć oddzielny harmonogram dla ogromnej puli aktorów (inni aktorzy w twoim systemie tego nie używają)
3. Zgodnie z sugestią @DaGGeRRz możesz wypróbować framework Akka, który oferuje konfigurowalne narzędzia dyspozytorskie (np. kradzież ruchu, równoważenie obciążenia, dispacheser przenosi zdarzenia ze skrzynek pocztowych ruchliwych aktorów do bezczynnych aktorów)

Z uwag do domyślnych Actor realizacji:

System run-time może być skonfigurowany do korzystania większy rozmiar puli wątków (na przykład poprzez ustawienie właściwości actors.corePoolSize JVM). scheduler Sposób cechy Actor można przestawić zwracają ResizableThreadPoolScheduler, która zmienia wielkość w basenie z gwintu uniknąć głodu spowodowanego aktorów powołać się dowolne metody blokowania. Właściwość JVM można ustawić na false, w którym to przypadku domyślnie jest używana ResizableThreadPoolScheduler do wykonania aktorów.

Ponadto: interesujący thread on schedulers w scala-lang.

Answer 2

Nie użyłem aktorów z tą składnią, ale domyślnie myślę, że wszyscy aktorzy w scala używają puli wątków.

Zobacz How to designate a thread pool for actors

Answer 3

Z twojego przykładu wynika, że ​​nie musisz w ogóle używać aktorów, ponieważ nie przekazujesz wiadomości do swoich jednostek pracy, nie odpowiadasz, a nawet nie zapętlasz.

Dlaczego po prostu nie utworzyć obciążenia Future s, a następnie czekać na nich zakończeniu? W ten sposób, pod spodem wideł Dołącz Basen jest całkowicie swobodnie decydować na odpowiednim poziomie równoległości (tj # wątków) dla systemu:

import actors.Futures._ 
def mkFuture(i : Int) = future { 
    doExpensiveStuff(i, limit) 
} 
val fs = (1 to range by limit).map(mkFuture) 
awaitAll(timeout, fs) //wait on the work all finishing 

Zauważ, że idziesz tylko korzyścią od równoległości przez obróbkę więcej zadań współbieżnie niż twój system ma rdzenie, jeśli kosztowna praca nie jest związana z CPU (może to jest granica IO).