Module: TWPipeline::Parallel

Defined in:
lib/twpipeline/parallel.rb,
sig/twpipeline.rbs

Constant Summary collapse

CHUNK_CAP =

Returns:

  • (::Integer) 
1 << 30
CHUNK_SHARE =

Returns:

  • (::Integer) 
32
LINE_CAP =

Returns:

  • (::Integer) 
500_000

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.chunk_bytes(resources = TWPipeline.resources) ⇒ Object 



30
31
32
33
 
# File 'lib/twpipeline/parallel.rb', line 30

def chunk_bytes(resources = TWPipeline.resources)
  override = TWPipeline::Resources.parse(ENV["TWP_CHUNK_BYTES"])
  override || [resources.memory_bytes / CHUNK_SHARE, CHUNK_CAP].min
end

.collect(pid, reader) ⇒ Object 



84
85
86
87
88
89
90
 
# File 'lib/twpipeline/parallel.rb', line 84

def collect(pid, reader)
  reader.binmode
  payload = reader.read
  reader.close
  Process.wait(pid)
  Marshal.load(payload)
end

.fold(items, workers: TWPipeline.resources.cores, &block) ⇒ Object 



17
18
19
20
21
22
23
24
 
# File 'lib/twpipeline/parallel.rb', line 17

def fold(items, workers: TWPipeline.resources.cores, &block)
  list = items.to_a
  return [block.call(list)] if workers < 2 || list.length < 2

  slices(list, workers)
    .map { |slice| spawn([slice], &block) }
    .flat_map { |pid, reader| collect(pid, reader) }
end

.map(items, workers: TWPipeline.resources.cores, &block) ⇒ Object 



7
8
9
10
11
12
13
14
15
 
# File 'lib/twpipeline/parallel.rb', line 7

def map(items, workers: TWPipeline.resources.cores, &block)
  list = items.to_a
  return list.map(&block) if workers < 2 || list.length < 2

  slices(list, workers)
    .map { |slice| spawn(slice, &block) }
    .map { |pid, reader| collect(pid, reader) }
    .flatten(1)
end

.pipe(input, output, workers: TWPipeline.resources.cores, budget: chunk_bytes, &block) ⇒ Object 



35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
 
# File 'lib/twpipeline/parallel.rb', line 35

def pipe(input, output, workers: TWPipeline.resources.cores, budget: chunk_bytes, &block)
  read = 0
  written = 0
  bytes = 0
  buffer = []

  flush = lambda do
    map(buffer, workers: workers, &block).each do |row|
      Array(row).each do |line|
        output.puts(line)
        written += 1
      end
    end

    buffer = []
    bytes = 0
  end

  lines = input.respond_to?(:each_line) ? input.each_line : input.each
  lines.each do |line|
    read += 1
    buffer << line
    bytes += line.bytesize
    flush.call if bytes >= budget || buffer.length >= LINE_CAP
  end

  flush.call if buffer.any?
  {read: read, written: written}
end

.slices(list, workers) ⇒ Object 



65
66
67
68
 
# File 'lib/twpipeline/parallel.rb', line 65

def slices(list, workers)
  size = (list.length / workers.to_f).ceil
  list.each_slice([size, 1].max).to_a
end

.spawn(slice, &block) ⇒ Object 



70
71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/twpipeline/parallel.rb', line 70

def spawn(slice, &block)
  reader, writer = IO.pipe
  pid = fork do
    reader.close
    writer.binmode
    writer.write(Marshal.dump(slice.map(&block)))
    writer.close
    exit!(0)
  end

  writer.close
  [pid, reader]
end

Instance Method Details

#self?.chunk_bytes ⇒ ::Integer

Parameters:

Returns:

  • (::Integer) 


42
 
# File 'sig/twpipeline.rbs', line 42

def self?.chunk_bytes: (?Resources) -> ::Integer

#self?.collect ⇒ Object

Parameters:

  • pid (::Integer)
  • reader (::IO)

Returns:

  • (Object) 


46
 
# File 'sig/twpipeline.rbs', line 46

def self?.collect: (::Integer pid, ::IO reader) -> untyped

#self?.fold ⇒ void

This method returns an undefined value.



41
 
# File 'sig/twpipeline.rbs', line 41

def self?.fold: [T, R] (::Enumerable[T], ?workers: ::Integer) { (::Array[T]) -> R } -> ::Array[R]

#self?.map ⇒ void

This method returns an undefined value.



40
 
# File 'sig/twpipeline.rbs', line 40

def self?.map: [T, R] (::Enumerable[T], ?workers: ::Integer) { (T) -> R } -> ::Array[R]

#self?.pipe {|arg0| ... } ⇒ ::Hash[::Symbol, ::Integer]

Parameters:

  • input (Object)
  • output (Object)
  • workers: (::Integer)
  • budget: (::Integer)

Yields:

Yield Parameters:

  • arg0 (::String)

Yield Returns:

  • (Object)

Returns:

  • (::Hash[::Symbol, ::Integer]) 


43
 
# File 'sig/twpipeline.rbs', line 43

def self?.pipe: (untyped input, untyped output, ?workers: ::Integer, ?budget: ::Integer) { (::String) -> untyped } -> ::Hash[::Symbol, ::Integer]

#self?.slices ⇒ void

This method returns an undefined value.



44
 
# File 'sig/twpipeline.rbs', line 44

def self?.slices: [T] (::Array[T], ::Integer workers) -> ::Array[::Array[T]]

#self?.spawn ⇒ void

This method returns an undefined value.



45
 
# File 'sig/twpipeline.rbs', line 45

def self?.spawn: [T, R] (::Array[T]) { (T) -> R } -> [::Integer, ::IO]