Как использовать рубиновые волокна, чтобы не блокировать IO
Я должен загрузить набор файлов в каталоге к S3. Так как больше чем 90% времени, требуемого загружать, потрачены, ожидая запроса HTTP для окончания, я хочу выполнить несколько из них сразу так или иначе.
Волокна могут помочь мне с этим вообще? Они описаны как способ решить этот вид проблемы, но я не могу думать ни о каком способе, которым я могу сделать любую работу, в то время как http называет блоки.
Каким-либо путем я могу решить эту проблему без потоков?
2 ответа
Я не разбираюсь в волокнах в 1.9, но обычные потоки из 1.8.6 могут решить эту проблему. Попробуйте использовать очередь 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.
Вы мог бы использовать для этого отдельные процессы вместо потоков:
#!/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