Как использовать рубиновые волокна, чтобы избежать блокировки ввода-вывода

Question

Мне нужно загрузить кучу файлов в каталоге на S3. Поскольку более 90% времени, необходимого для загрузки, тратится на ожидание завершения http-запроса, я хочу каким-то образом выполнить несколько из них одновременно.

Могут ли волокна помочь мне в этом? Они описаны как способ решения такого рода проблем, но я не могу придумать, каким образом я могу выполнять какую-либо работу, пока блокируется вызов http.

Как я могу решить эту проблему без потоков?

Answer 1

Я не в курсе волокон в 1.9, но обычные нити из 1.8.6 могут решить эту проблему. Попробуйте использовать Queue http://ruby -doc.org / stdlib / libdoc / thread / rdoc / classes / Queue.html

Глядя на пример в документации, ваш потребитель принимает участие в загрузке. Он «потребляет» URL-адрес и файл и загружает данные. Производитель - это часть вашей программы, которая продолжает работать и находит новые файлы для загрузки.

Если вы хотите загрузить несколько файлов одновременно, просто запустите новую тему для каждого файла:

t = Thread.new do
  upload_file(param1, param2)
end
@all_threads << t

Затем, позже, в вашем коде «производителя» (который, помните, не обязательно должен быть в своем собственном потоке, это может быть основная программа):

@all_threads.each do |t|
  t.join if t.alive?
end

Очередь может быть либо @member_variable, либо $ global.

Answer 2

Аарон Паттерсон (@tenderlove) использует пример, почти такой же, как ваш, чтобы описать, почему вы можете и должны использовать потоки для достижения параллелизма в вашей ситуации.

Большинство библиотек ввода-вывода теперь достаточно умны, чтобы освободить GVL (глобальную блокировку виртуальной машины, или большинство людей знают ее как GIL или глобальную блокировку интерпретатора) при выполнении ввода-вывода. Для этого в С есть простой вызов функции. Вам не нужно беспокоиться о коде C, но для вас это означает, что большинство библиотек ввода-вывода, достойных их внимания, собираются выпустить GVL и позволить другим потокам выполняться, пока поток, выполняющий ввод-вывод, ожидает возвращения данных. .

Если то, что я только что сказал, сбивает с толку, вам не нужно слишком беспокоиться об этом. Главное, что вам нужно знать, это то, что если вы используете приличную библиотеку для выполнения ваших HTTP-запросов (или любой другой операции ввода-вывода в этом отношении ... базы данных, межпроцессного взаимодействия и т. Д.), Интерпретатор Ruby (MRI) достаточно умен, чтобы иметь возможность снять блокировку на интерпретаторе и позволить другим потокам выполняться, пока один поток ожидает ввода-вывода. Если следующий поток имеет свой собственный ввод-вывод для захвата, интерпретатор Ruby сделает то же самое (если предположить, что библиотека ввода-вывода построена для использования этой возможности Ruby, что, как я полагаю, в настоящее время используется чаще всего).

Итак, чтобы подвести итог тому, что я говорю, используйте темы! Вы должны увидеть выигрыш в производительности. Если нет, проверьте, использует ли ваша библиотека http функцию rb_thread_blocking_region () в C, и, если нет, выясните, почему нет. Возможно, есть веская причина, может быть, вам стоит подумать об использовании лучшей библиотеки.

Ссылка на видео Аарона Паттерсона находится здесь: http://www.youtube.com/watch?v=kufXhNkm5WU

Стоит посмотреть, хотя бы ради смеха, так как Аарон Паттерсон - один из самых смешных людей в Интернете.

Answer 3

Чтобы ответить на ваши актуальные вопросы:

Могут ли волокна помочь мне в этом?

Нет, они не могут. Йорг W Mittag объясняет, почему лучше .

Нет, вы не можете выполнять параллелизм с волокнами. Волокна просто не являются концепцией параллелизма, они представляют собой конструкцию потока управления, как исключения. В этом весь смысл Fibers: они никогда не работают параллельно, они сотрудничают, и они детерминированы. Волокна сопрограммы. (На самом деле, я никогда не понимал, почему их не просто называют сопрограммами.)

Единственная конструкция параллелизма в Ruby - это Thread.

Когда он говорит, что единственный параллельный конструкт в Ruby - это Thread, помните, что в Ruby есть много разных имплиментов, и что они различаются по своим реализациям потоков. Йорг еще раз дает отличный ответ на эти различия; и правильно делает вывод, что только что-то вроде JRuby (который использует потоки JVM, сопоставленные с собственными потоками) или разветвление вашего процесса - это то, как вы можете достичь истинного параллелизма.

Как я могу решить эту проблему без потоков?

Кроме разветвления вашего процесса, я бы также посоветовал вам взглянуть на EventMachine и что-то вроде em-http-request . Это управляемый событиями, неблокирующий, HTTP-клиент на основе реакторной схемы , который является асинхронным и не требует дополнительных потоков.

Answer 4

Вы можете использовать для этого отдельные процессы вместо потоков:

#!/usr/bin/env ruby

$stderr.sync = true

# Number of children to use for uploading
MAX_CHILDREN = 5

# Hash of PIDs for children that are working along with which file
# they're working on.
@child_pids = {}

# Keep track of uploads that failed
@failed_files = []

# Get the list of files to upload as arguments to the program
@files = ARGV


### Wait for a child to finish, adding the file to the list of those
### that failed if the child indicates there was a problem.
def wait_for_child
    $stderr.puts "    waiting for a child to finish..."
    pid, status = Process.waitpid2( 0 )
    file = @child_pids.delete( pid )
    @failed_files << file unless status.success?
end


### Here's where you'd put the particulars of what gets uploaded and
### how. I'm just sleeping for the file size in bytes * milliseconds
### to simulate the upload, then returning either +true+ or +false+
### based on a random factor.
def upload( file )
    bytes = File.size( file )
    sleep( bytes * 0.00001 )
    return rand( 100 ) > 5
end


### Start a child uploading the specified +file+.
def start_child( file )
    if pid = Process.fork
        $stderr.puts "%s: uploaded started by child %d" % [ file, pid ]
        @child_pids[ pid ] = file
    else
        if upload( file )
            $stderr.puts "%s: done." % [ file ]
            exit 0 # success
        else
            $stderr.puts "%s: failed." % [ file ]
            exit 255
        end
    end
end


until @files.empty?

    # If there are already the maximum number of children running, wait 
    # for one to finish
    wait_for_child() if @child_pids.length >= MAX_CHILDREN

    # Start a new child working on the next file
    start_child( @files.shift )

end


# Now we're just waiting on the final few uploads to finish
wait_for_child() until @child_pids.empty?

if @failed_files.empty?
    exit 0
else
    $stderr.puts "Some files failed to upload:",
        @failed_files.collect {|file| "  #{file}" }
    exit 255
end