Class: SimpleThreadPool

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

Overview

Simple thread pool for executing blocks in parallel in a controlled manner. Threads are not re-used by the pool to prevent any thread local variables from leaking out.

Instance Method Summary collapse

Constructor Details

#initialize(max_threads) ⇒ SimpleThreadPool

Returns a new instance of SimpleThreadPool.

Parameters:

  • max_threads (Integer) —

    The maximum number of threads to spawn.

Raises:

  • (ArgumentError) 


12
13
14
15
16
17
18
19
20
 
# File 'lib/simple_thread_pool.rb', line 12

def initialize(max_threads)
  raise ArgumentError, "max_threads must be at least 1" unless max_threads >= 1
  @max_threads = max_threads
  @lock = Mutex.new
  @condition = ConditionVariable.new
  @threads = []
  @processing_ids = []
  @fatal_exception = nil
end

Instance Method Details

#execute(id = nil) { ... } ⇒ void

This method returns an undefined value.

Call this method to spawn a thread to run the block. If the thread pool is already full, this method will block until a thread is free. The block is responsible for handling any StandardError that could be raised.

If a block dies from an exception that is not a StandardError (i.e. an Exception that indicates the process itself is no longer healthy, such as NoMemoryError or SystemStackError), the pool is stopped. Blocks passed to this method after that point are not run; instead this method waits for the in-flight threads to finish and then raises the exception that stopped the pool, so that callers never silently lose work. The exception is cleared when it is raised, so it is delivered to exactly one caller and the pool can be used again afterwards.

The optional id argument can be used to provide an identifier for a block. If one is provided, processing will be blocked if the same id is already being processed. This ensures that each unique id is executed one at a time sequentially.

Parameters:

  • id (String, Symbol) (defaults to: nil) —

    An optional identifier for the block.

Yields:

  • The block to execute in a thread.



43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/simple_thread_pool.rb', line 43

def execute(id = nil, &block)
  loop do
    stopped = @lock.synchronize do
      until @fatal_exception || can_add_thread?(id)
        @condition.wait(@lock)
      end

      if @fatal_exception
        true
      else
        @processing_ids << id unless id.nil?
        thread_added = false
        begin
          add_thread(id, block)
          thread_added = true
        ensure
          unless thread_added
            @processing_ids.delete(id) unless id.nil?
            @condition.broadcast
          end
        end
        false
      end
    end

    break unless stopped

    # The pool has been stopped. Drain the in-flight threads and raise the
    # exception that stopped it. This has to happen outside of the synchronize
    # block above because #finish acquires the same lock. If another caller
    # claimed the exception first then #finish returns normally and the pool
    # is usable again, so retry rather than silently dropping the block.
    finish
  end

  nil
end

#finish ⇒ void

This method returns an undefined value.

Call this method to block until all current threads have finished executing.

Exceptions raised by the blocks are not propagated; they are the block's responsibility to handle. The exception is that if a block died from an exception that was not a StandardError, that exception is re-raised here after every in-flight thread has finished. In-flight threads are always allowed to run to completion rather than being killed, so that they can run their own ensure blocks. That exception is cleared as it is raised, so it is delivered to exactly one caller and the pool can be used again afterwards.



92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
 
# File 'lib/simple_thread_pool.rb', line 92

def finish
  active_threads = @lock.synchronize do
    # Exclude the calling thread so that this is safe to call from inside a
    # block running in the pool; Thread#join raises if given the current
    # thread. #execute calls this method, so that is reachable indirectly.
    @threads.select { |thread| thread.alive? && !thread.equal?(Thread.current) }
  end
  active_threads.each do |thread|
    thread.join
  rescue Exception => e # standard:disable Lint/RescueException
    # Joining a thread that died with an unhandled exception re-raises that
    # exception; those are the block's responsibility to handle, so they are
    # not propagated here. Worker threads record the exception that killed
    # them, so anything else caught here was raised on the calling thread
    # (e.g. an Interrupt from a signal) and is always re-raised immediately so
    # that the calling thread stays responsive to signals.
    raise unless e.equal?(thread.thread_variable_get(WORKER_EXCEPTION))
  end

  # Claim the exception under the lock and clear it before raising so that it
  # is delivered to exactly one caller. The pool is usable again afterwards.
  fatal_exception = @lock.synchronize do
    exception = @fatal_exception
    @fatal_exception = nil
    exception
  end
  raise fatal_exception if fatal_exception

  nil
end

#synchronize { ... } ⇒ Object

Synchronize data access across the thread pool. This method will block waiting on the same internal Mutex the thread pool uses to manage scheduling threads.

Yields:

  • The block to execute in a synchronized manner.

Returns:

  • (Object) —

    The return value of the block.



129
130
131
 
# File 'lib/simple_thread_pool.rb', line 129

def synchronize(&block)
  @lock.synchronize(&block)
end