Чтение файла N строк одновременно в ruby

Question

У меня есть большой файл (сотни мегабайт), который состоит из имен файлов, по одному на строку.

Мне нужно перебрать список имен файлов и отключить процесс для каждого имени файла. Мне нужно одновременно не более 8 разветвленных процессов, и я не хочу читать весь список имен файлов в ОЗУ сразу.

Я даже не знаю, с чего начать, кто-нибудь может мне помочь?

Answer 1

File.foreach("large_file").each_slice(8) do |eight_lines|
  # eight_lines is an array containing 8 lines.
  # at this point you can iterate over these filenames
  # and spawn off your processes/threads
end

Answer 2

Похоже, Модуль процесса будет полезен для этой задачи. Вот что я быстро набросал в качестве отправной точки:

include Process

i = 0
for line in open('files.txt') do
    i += 1
    fork { `sleep #{rand} && echo "#{i} - #{line.chomp}" >> numbers.txt` }

    if i >= 8
        wait # join any single child process
        i -= 1
    end
end

waitall # join all remaining child processes

Выход:

hello
goodbye

test1
test2
a
b
c
d
e
f
g
$ ruby b.rb
$ cat numbers.txt 
1 - hello
3 - 
2 - goodbye
5 - test2
6 - a
4 - test1
7 - b
8 - c
8 - d
8 - e
8 - f
8 - g

Это работает так:

• for line in open(XXX) будет лениво перебирать строки указанного вами файла.
• fork будет порождать дочерний процесс, выполняющий данный блок, и в этом случае мы используем обратные метки, чтобы указать что-то, что должно быть выполнено оболочкой. Обратите внимание, что rand возвращает здесь значение 0-1, поэтому мы спим меньше секунды, и я звоню line.chomp, чтобы удалить завершающий символ новой строки, полученный от line.
• Если мы накопили 8 или более процессов, вызовите wait, чтобы остановить все, пока один из них не вернется.
• Наконец, за пределами цикла вызовите waitall, чтобы присоединиться ко всем оставшимся процессам перед выходом из сценария.

Answer 3

arr = IO.readlines ("filename")

Answer 4

Стандартная библиотека документации для Очередь имеет

require 'thread'

queue = Queue.new

producer = Thread.new do
  5.times do |i|
    sleep rand(i) # simulate expense
    queue << i
    puts "#{i} produced"
  end
end

consumer = Thread.new do
  5.times do |i|
    value = queue.pop
    sleep rand(i/2) # simulate expense
    puts "consumed #{value}"
  end
end

consumer.join

Хотя я нахожу это немного многословным.

Википедия описывает это как шаблон пула потоков

Answer 5

Вот решение Марка, заключенное в класс ProcessPool, может быть полезно иметь его (и, пожалуйста, исправьте меня, если я допустил ошибку):

class ProcessPool
  def initialize pool_size
    @pool_size = pool_size
    @free_slots = @pool_size
  end

  def fork &p
    if @free_slots == 0
      Process.wait
      @free_slots += 1
    end
    @free_slots -= 1
    puts "Free slots: #{@free_slots}"
    Process.fork &p
  end

  def waitall
    Process.waitall
  end
end

pool = ProcessPool.new 8
for line in open('files.txt') do
  pool.fork { Kernel.sleep rand(10); puts line.chomp }
end
pool.waitall
puts 'finished'