Class: FastlaneCore::QueueWorker

Inherits:
Object
  • Object
show all
Defined in:
fastlane_core/lib/fastlane_core/queue_worker.rb

Overview

This dispatches jobs to worker threads and make it work in parallel. It's suitable for I/O bounds works and not for CPU bounds works. Use this when you have all the items that you'll process in advance. Simply enqueue them to this and call QueueWorker#start.

Constant Summary collapse

NUMBER_OF_THREADS = 
FastlaneCore::Helper.test? ? 1 : [ENV["DELIVER_NUMBER_OF_THREADS"], ENV["FL_NUMBER_OF_THREADS"], 10].map(&:to_i).find(&:positive?).clamp(1, ENV.fetch("FL_MAX_NUMBER_OF_THREADS", 10).to_i)

Instance Method Summary collapse

Constructor Details

#initialize(concurrency = NUMBER_OF_THREADS, &block) ⇒ QueueWorker

Returns a new instance of QueueWorker.

Parameters:

  • concurrency (Numeric) (defaults to: NUMBER_OF_THREADS) —
    • A number of threads to be created
  • block (Proc) —
    • A task you want to execute with enqueued items


13
14
15
16
17
 
# File 'fastlane_core/lib/fastlane_core/queue_worker.rb', line 13

def initialize(concurrency = NUMBER_OF_THREADS, &block)
  @concurrency = concurrency
  @block = block
  @queue = Queue.new
end

Instance Method Details

#batch_enqueue(jobs) ⇒ Object

Parameters:

  • jobs (Array<Object>) —
    • An array of arbitrary object that keeps parameters

Raises:

  • (ArgumentError) 


25
26
27
28
 
# File 'fastlane_core/lib/fastlane_core/queue_worker.rb', line 25

def batch_enqueue(jobs)
  raise(ArgumentError, "Enqueue Array instead of #{jobs.class}") unless jobs.kind_of?(Array)
  jobs.each { |job| enqueue(job) }
end

#enqueue(job) ⇒ Object

Parameters:

  • job (Object) —
    • An arbitrary object that keeps parameters


20
21
22
 
# File 'fastlane_core/lib/fastlane_core/queue_worker.rb', line 20

def enqueue(job)
  @queue.push(job)
end

#start ⇒ Object

Call this after you enqueued all the jobs you want to process This method blocks current thread until all the enqueued jobs are processed



32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'fastlane_core/lib/fastlane_core/queue_worker.rb', line 32

def start
  @queue.close

  threads = []
  results = Queue.new
  @concurrency.times do
    threads << Thread.new do
      # Exceptions are re-joined (and raised) below, so this just causes duplicate raised exceptions
      Thread.current.report_on_exception = false

      job = @queue.pop
      while job
        results << @block.call(job)
        job = @queue.pop
      end
    end
  end

  threads.each(&:join)

  # Convert Queue to Array
  real_results = []
  real_results << results.pop until results.empty?
  real_results
end