class ConcurrentEnumerable::Parallel

Public Class Methods

new(list, opts = {}) click to toggle source
Calls superclass method
# File lib/concurrent_enumerable/parallel.rb, line 7
def initialize(list, opts = {})
  super list
  @opts = opts
  @executor = Concurrent::OptionsParser.get_executor_from(opts) || \
              Concurrent.configuration.global_task_pool
end

Public Instance Methods

parallel() click to toggle source
# File lib/concurrent_enumerable/parallel.rb, line 14
def parallel
  self
end
serial() click to toggle source
# File lib/concurrent_enumerable/parallel.rb, line 18
def serial
  __getobj__
end

Protected Instance Methods

run_in_threads(method, *args) { |item| ... } click to toggle source
# File lib/concurrent_enumerable/parallel.rb, line 24
def run_in_threads(method, *args, &block)
  if block
    latch = Concurrent::CountDownLatch.new(size)

    results = Array.new(size)

    index = Concurrent::AtomicFixnum.new(-1)

    __getobj__.each do |item|
      i = index.increment

      @executor.post do
        results[i] = yield(item)
        latch.count_down
      end
    end

    latch.wait
    results.send(method, *args) { |value| value }
  else
    send method, *args
  end
end
run_in_threads_return_original(method, *args, &block) click to toggle source
# File lib/concurrent_enumerable/parallel.rb, line 48
def run_in_threads_return_original(method, *args, &block)
  run_in_threads(method, *args, &block)

  self
end
run_in_threads_return_parallel(method, *args, &block) click to toggle source
# File lib/concurrent_enumerable/parallel.rb, line 54
def run_in_threads_return_parallel(method, *args, &block)
  Parallel.new run_in_threads(method, *args, &block), @opts
end