Ruby concurrent/traitement asynchrone (avec un cas d'utilisation simple)

Question

Je regardais les capacités de traitement parallèle/asynchrone de ruby ​​et lisais de nombreux articles et articles de blog. Je regardais à travers eventmachine, fibres, Revactor, Reia, etc, etc. Malheureusement, je n'ai pas pu trouver une solution simple, efficace (et non-IO-blocage) pour cette utilisation très simple cas:

File.open('somelogfile.txt') do |file| 
    while line = file.gets  # (R) Read from IO 
    line = process_line(line) # (P) Process the line 
    write_to_db(line)   # (W) Write the output to some IO (DB or file) 
    end 
end 

est-ce que vous pouvez voir, mon petit script effectue trois opérations lues (R), processus (P) & écrire (W). Supposons - pour simplifier - que chaque opération prend exactement 1 unité de temps (par exemple 10ms), le code actuel serait donc quelque chose comme ça (5 lignes):

Time:  123456789(15 units in total) 
Operations: RPWRPWRPWRPWRPW 

Mais, je voudrais qu'il fasse quelque chose comme ceci:

Time:  1234567 (7 units in total) 
Operations: RRRRR 
      PPPPP 
       WWWWW 

Évidemment, je peux exécuter trois procédés (lecteur, le processeur & écrivain) et transmettre lignes de lecture de lecteur dans la file d'attente de processeur et ensuite passer les lignes traitées dans la file d'attente d'écriture (tous coordonnés par exemple RabbitMQ) . Mais, le cas d'utilisation est si simple, il ne se sent pas juste.

Des indices sur la façon dont cela pourrait être fait (sans passer de Ruby à Erlang, à Closure ou à Scala)?

Answer 1

Vérifiez la pêche (http://peach.rubyforge.org/). Faire un parallèle "chacun" ne pouvait pas être plus simple. Cependant, comme le dit la documentation, vous devrez exécuter sous JRuby pour utiliser le thread natif de la JVM.

Voir la réponse de Jorg Mittag à this SO question pour beaucoup de détails sur les capacités de multithreading des différents interprètes de Ruby.

Answer 2

Si vous avez besoin qu'il soit vraiment parallèle (à partir d'un seul processus), je crois que vous devrez utiliser JRuby pour obtenir des vrais threads natifs et pas de GIL.

Vous pouvez utiliser quelque chose comme DRB pour distribuer le traitement entre plusieurs processus/cœurs, mais pour votre utilisation, c'est un peu trop. , Vous pourriez plutôt essayer d'avoir plusieurs processus en utilisant des tuyaux communicating:

$ cat somelogfile.txt | ruby ./proc-process | ruby ./proc-store 

Dans ce scénario chaque pièce est son propre processus qui peut fonctionner en parallèle mais la communication en utilisant STDIN/STDOUT. C'est probablement l'approche la plus facile (et la plus rapide) de votre problème.

# proc-process 
while line = $stdin.gets do 
    # do cpu intensive stuff here 
    $stdout.puts "data to be stored in DB" 
    $stdout.flush # this is important 
end 

# proc-store 
while line = $stdin.gets do 
    write_to_db(line) 
end