Contents

module Workhorse
  # Abstraction layer of a simple thread pool implementation used by the worker.
  class Pool
    attr_reader :mutex

    def initialize(size)
      @size = size
      @executor = Concurrent::ThreadPoolExecutor.new(
        min_threads: 0,
        max_threads: @size,
        max_queue: 0,
        fallback_policy: :abort,
        auto_terminate: false
      )
      @mutex = Mutex.new
      @active_threads = Concurrent::AtomicFixnum.new(0)
    end

    # Posts a new work unit to the pool.
    def post
      mutex.synchronize do
        if idle.zero?
          fail 'All threads are busy.'
        end

        active_threads = @active_threads

        active_threads.increment

        @executor.post do
          begin
            yield
          ensure
            active_threads.decrement
          end
        end
      end
    end

    # Returns the number of idle threads.
    def idle
      @size - @active_threads.value
    end

    def wait
      @executor.wait_for_termination
    end

    # Shuts down the pool and waits for termination.
    def shutdown
      @executor.shutdown
      wait
    end
  end
end

Version data entries

13 entries across 13 versions & 1 rubygems

Version	Path
workhorse-0.3.9	lib/workhorse/pool.rb
workhorse-0.3.8	lib/workhorse/pool.rb
workhorse-0.3.7	lib/workhorse/pool.rb
workhorse-0.3.6	lib/workhorse/pool.rb
workhorse-0.3.5	lib/workhorse/pool.rb
workhorse-0.3.4	lib/workhorse/pool.rb
workhorse-0.3.3	lib/workhorse/pool.rb
workhorse-0.3.2	lib/workhorse/pool.rb
workhorse-0.3.1	lib/workhorse/pool.rb
workhorse-0.3.0	lib/workhorse/pool.rb
workhorse-0.2.0	lib/workhorse/pool.rb
workhorse-0.1.0	lib/workhorse/pool.rb
workhorse-0.0.3	lib/workhorse/pool.rb