Ruby параллелизм/асинхронная обработка (с простым прецедентом)

Question

Я изучал возможности параллельной/асинхронной обработки Ruby и читал много статей и сообщений в блогах. Я просмотрел EventMachine, волокнам, Revactor, Реи, и т.д., и т.д. К сожалению, я не смог найти простое, эффективное (и не-IO-блокирующий) решения для этого очень простого случая использования:

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 

ли вы можете видеть, мой маленький скрипт выполняет три операции чтения (R), процесс (P) & записи (W). Давайте предположим, что - для простоты - что каждая операция занимает ровно 1 единицу времени (например, 10 мс), поэтому текущий код будет делать что-то подобное (5 строк):

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

Но я хотел бы, чтобы сделать что-то как это:

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

Очевидно, что я мог бы запустить три процесса (читатель, процессор & писателя) и передать читать строки из считывателя в очередь процессора, а затем передать обработанные строки в очередь писателя (все координируется с помощью, например, RabbitMQ) , Но прецедент настолько прост, что он просто не чувствует себя хорошо.

Любые подсказки о том, как это можно сделать (без переключения с Ruby на Erlang, Closure или Scala)?

Answer 1

Отметьте персик (http://peach.rubyforge.org/). Выполнение параллели «каждый» не может быть проще. Однако, как говорится в документации, вам нужно запустить под JRuby, чтобы использовать собственную поточную передачу JVM.

См. Ответ Йорга Миттага на номер this SO question для подробного описания возможностей многопоточности различных интерпретаторов Ruby.

Answer 2

Если вам нужно, чтобы он был действительно параллельным (из одного процесса), я считаю, что вам нужно будет использовать JRuby для получения истинных собственных потоков и без GIL.

Вы можете использовать что-то вроде DRb для распространения обработки по нескольким процессам/ядрам, но для вашего использования это немного. Вместо этого вы могли бы попытаться иметь несколько процессов взаимодействуют с помощью труб:

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

В этом случае каждая часть его собственный процесс, который может работать параллельно, но общается с помощью STDIN/STDOUT. Это, пожалуй, самый простой (и самый быстрый) подход к вашей проблеме.

# 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