Class: Parallel::JobFactory
- Inherits:
-
Object
- Object
- Parallel::JobFactory
- Defined in:
- lib/parallel.rb
Instance Method Summary collapse
-
#initialize(source, mutex) ⇒ JobFactory
constructor
A new instance of JobFactory.
- #next ⇒ Object
-
#pack(item, index) ⇒ Object
generate item that is sent to workers just index is faster + less likely to blow up with unserializable errors.
- #size ⇒ Object
-
#unpack(data) ⇒ Object
unpack item that is sent to workers.
Constructor Details
#initialize(source, mutex) ⇒ JobFactory
Returns a new instance of JobFactory.
76 77 78 79 80 81 82 |
# File 'lib/parallel.rb', line 76 def initialize(source, mutex) @lambda = (source.respond_to?(:call) && source) || queue_wrapper(source) @source = source.to_a unless @lambda # turn Range and other Enumerable-s into an Array @mutex = mutex @index = -1 @stopped = false end |
Instance Method Details
#next ⇒ Object
84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 |
# File 'lib/parallel.rb', line 84 def next if producer? # - index and item stay in sync # - do not call lambda after it has returned Stop item, index = @mutex.synchronize do return if @stopped item = @lambda.call @stopped = (item == Parallel::Stop) return if @stopped [item, @index += 1] end else index = @mutex.synchronize { @index += 1 } return if index >= size item = @source[index] end [item, index] end |
#pack(item, index) ⇒ Object
generate item that is sent to workers just index is faster + less likely to blow up with unserializable errors
113 114 115 |
# File 'lib/parallel.rb', line 113 def pack(item, index) producer? ? [item, index] : index end |
#size ⇒ Object
103 104 105 106 107 108 109 |
# File 'lib/parallel.rb', line 103 def size if producer? Float::INFINITY else @source.size end end |
#unpack(data) ⇒ Object
unpack item that is sent to workers
118 119 120 |
# File 'lib/parallel.rb', line 118 def unpack(data) producer? ? data : [@source[data], data] end |