Ruby：ファイルの一部を分割して読み込みスレッド数に応じて

Question

高度なI/Oバウンドのプロセスを高速化するために複数のスレッドを使用したいと考えています。私は、ループ内のCSVファイルから取得したIDを使って休憩サービスを呼び出すことができるようにしたいと考えています。私が知ることができなかったことは、使いたいスレッドの量に応じてファイルをエレガントにチャンクに分割する方法です。私は困惑午前、と私は解決策は、コードブロックCSV.foreach(id_file.csv") do |id|...である必要がある場所 ...

require 'benchmark' 
require 'csv' 

FILE_RECORD_COUNT = File.open("path-to-csv","r").readlines.size 

def setup(thread_count) 
    threads = [] 
    thread_count.times do 
    threads << Thread.new do 
     fetches_per_thread = FILE_RECORD_COUNT/thread_count 

     fetches_per_thread.times do 
     CSV.foreach("id_file.csv") do |id| 
      response = RestClient.get("https://api.examplerest/names?id={#id}",{accept: :json}) 
      # do some quick validation... 
     end 
     end 
    end 
    end 

    threads.each(&:join) 
end 

def run_benchmark 
    Benchmark.bm(20) do |bm| 
    [1, 2, 3, 5, 6, 10, 15, 30, 100].each do |thread_count| 
     bm.report("with #{thread_count} threads") do 
     setup(thread_count) 
     end 
    end 
    end 
end 

：

CSVファイルには、そのようなIDの単一の列が含まれています。私は、データを動的に分割して各スレッドに送り、残りの呼び出しをしたいと思います。私は手動でファイルを分割することができますが、私はそれを避けたいと思います。

私はこれをオンラインで見つかった例のベンチマークにしようとしました。スイートスポットがスレッド数のどこにあるのかを確認しました。

EDIT： BernardKの答えを使用して、私は私のコードスレッドを実行することができたし、次の結果を返しました：

     | user | system | total | real | 
with 1 threads   5.125000 2.594000 7.719000 (40.416162) 
with 2 threads   1.625000 2.015000 3.640000 (28.571521) 
with 3 threads   1.578000 1.625000 3.203000 (17.210526) 
with 4 threads   1.578000 1.235000 2.813000 ( 8.496068) 
with 5 threads   1.406000 1.250000 2.656000 ( 6.779216) 
with 10 threads  1.875000 1.328000 3.203000 ( 5.069487) 
with 15 threads  2.016000 1.640000 3.656000 ( 4.285426) 
with 30 threads  2.125000 1.625000 3.750000 ( 3.817084) 
with 100 threads  2.281000 1.375000 3.656000 ( 3.943304) 

これはテストランでしたが、本当にこのようなスレッドを高速化する方法を示していますRubyコード！

Answer 1

ファイルは、Enumerable#each_sliceを使用してチャンクで分割された配列に読み込まれます。

require 'benchmark' 
require 'csv' 

@file_name = 'xxx.txt' 
file = File.open(@file_name, 'w') 
1000.times do | i | 
    file.puts "#{i.to_s}" 
end 
file.close 

@lines = [] 
CSV.foreach(@file_name) { | line | @lines << line } 
FILE_RECORD_COUNT = @lines.size 
puts FILE_RECORD_COUNT 

def setup(thread_count) 
    puts "----- thread_count=#{thread_count}" 
    threads = [] 
    fetches_per_thread = FILE_RECORD_COUNT/thread_count 
    puts "----- fetches_per_thread=#{fetches_per_thread}" 
    raise 'invalid slice size' if fetches_per_thread < 1 

    @lines.each_slice(fetches_per_thread) do | slice | 
    threads << Thread.new do 
     puts "===== slice from #{slice.first} to #{slice.last}" 
     slice.each do | id | 
#  puts id 
#   response = RestClient.get("https://api.examplerest/names/{#id}",{accept: :json}) 
      # do some quick validation... 
     end # slice.each 
    end # Thread.new 
    end # @lines.each_slice 

    threads.each(&:join) 
end # def setup 

def run_benchmark 
    Benchmark.bm(20) do |bm| 
    [1, 2, 3, 5, 6, 10, 15, 30, 100].each do |thread_count| 
     bm.report("with #{thread_count} threads") do 
     setup(thread_count) 
     end 
    end 
    end 
end 

run_benchmark 

実行：

$ -------------------------------- 
-bash: --------------------------------: command not found 
$ ruby -w t.rb 
1000 
          user  system  total  real 
with 1 threads  ----- thread_count=1 
----- fetches_per_thread=1000 
===== slice from ["0"] to ["999"] 
    0.000000 0.000000 0.000000 ( 0.000288) 
with 2 threads  ----- thread_count=2 
----- fetches_per_thread=500 
===== slice from ["0"] to ["499"] 
===== slice from ["500"] to ["999"] 
    0.000000 0.000000 0.000000 ( 0.000318) 
with 3 threads  ----- thread_count=3 
----- fetches_per_thread=333 
===== slice from ["0"] to ["332"] 
===== slice from ["666"] to ["998"] 
===== slice from ["999"] to ["999"] 
===== slice from ["333"] to ["665"] 
    0.000000 0.000000 0.000000 ( 0.000549) 
with 5 threads  ----- thread_count=5 
----- fetches_per_thread=200 
===== slice from ["0"] to ["199"] 
===== slice from ["200"] to ["399"] 
===== slice from ["400"] to ["599"] 
===== slice from ["600"] to ["799"] 
===== slice from ["800"] to ["999"] 
    0.000000 0.000000 0.000000 ( 0.000536) 
with 6 threads  ----- thread_count=6 
----- fetches_per_thread=166 
===== slice from ["166"] to ["331"] 
===== slice from ["664"] to ["829"] 
===== slice from ["830"] to ["995"] 
===== slice from ["996"] to ["999"] 
===== slice from ["0"] to ["165"] 
===== slice from ["332"] to ["497"] 
===== slice from ["498"] to ["663"] 
    0.000000 0.000000 0.000000 ( 0.000735) 
with 10 threads  ----- thread_count=10 
----- fetches_per_thread=100 
===== slice from ["900"] to ["999"] 
... 
===== slice from ["190"] to ["199"] 
===== slice from ["200"] to ["209"] 
===== slice from ["210"] to ["219"] 
===== slice from ["220"] to ["229"] 
===== slice from ["230"] to ["239"] 
===== slice from ["240"] to ["249"] 
... 
===== slice from ["970"] to ["979"] 
===== slice from ["980"] to ["989"] 
===== slice from ["990"] to ["999"] 
===== slice from ["20"] to ["29"] 
===== slice from ["30"] to ["39"] 
    0.000000 0.000000 0.000000 ( 0.011656) 

その後、私は--------------------------------------を見つけて、実行の先頭にジャンプするターミナルでfindコマンドを使用します。