Class: ThreadLimiter
- Inherits:
-
Object
- Object
- ThreadLimiter
- Defined in:
- lib/threadlimiter/threadlimiter.rb,
lib/threadlimiter/version.rb
Overview
Fork threads like Thread.fork, but limit the number of concurrently running threads.
ThreadLimiter isn’t a thread pool. Each fork really starts a new thread.
Constant Summary collapse
- VERSION =
"0.2.0"
Class Method Summary collapse
-
.handle_clusters(enumeration, number_of_clusters, method_name, &block) ⇒ Object
:nodoc:.
-
.open(*args) ⇒ Object
Create and use a new ThreadLimiter and wait for all threads to finish.
Instance Method Summary collapse
-
#fork(*args, &block) ⇒ Object
Fork a thread.
-
#initialize(limit, options = {}) ⇒ ThreadLimiter
constructor
Initialize the ThreadLimiter.
-
#wait ⇒ Object
Wait for all threads to finish.
Constructor Details
#initialize(limit, options = {}) ⇒ ThreadLimiter
Initialize the ThreadLimiter. The optional parameter limit is the maximum number of concurrently running threads. Set limit to -1 or 0 to fork threads without limiting the number of concurrently running threads. Set options to true to start the new thread before waiting for resources.
24 25 26 27 28 29 30 31 |
# File 'lib/threadlimiter/threadlimiter.rb', line 24 def initialize(limit, ={}) @limit = limit # The maximum number of concurrently running threads. @running = 0 # The number of currently running threads. @noblock = [:noblock] @mutex = Mutex.new @cv = ConditionVariable.new end |
Class Method Details
.handle_clusters(enumeration, number_of_clusters, method_name, &block) ⇒ Object
:nodoc:
59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 |
# File 'lib/threadlimiter/threadlimiter.rb', line 59 def self.handle_clusters(enumeration, number_of_clusters, method_name, &block) # :nodoc: clusters = [] # One cluster per fork. last_pos = -1 res = [] enumeration.each do |object| last_pos += 1 (clusters[last_pos%number_of_clusters] ||= []) << object end clusters.__send__(method_name, -1) do |cluster| cluster.collect do |object| if block.arity > 1 and object.kind_of?(Enumerable) yield(*object.to_a) else yield(object) end end end.collect do |cluster| cluster + (cluster.length == clusters[0].length ? [] : [nil]) # Add padding nil, in order to be able to transpose end.transpose.each do |array| res.concat(array) end res[0..last_pos] # Remove padding nil. end |
.open(*args) ⇒ Object
Create and use a new ThreadLimiter and wait for all threads to finish.
9 10 11 12 13 14 15 16 17 |
# File 'lib/threadlimiter/threadlimiter.rb', line 9 def self.open(*args) thread_limiter = new(*args) begin yield(thread_limiter) ensure thread_limiter.wait end end |
Instance Method Details
#fork(*args, &block) ⇒ Object
Fork a thread. The given block is run within the thread. It behaves like Thread.fork(). In fact, it invokes Thread.fork() and returns its result. The list of arguments is passed to Thread.fork().
39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 |
# File 'lib/threadlimiter/threadlimiter.rb', line 39 def fork(*args, &block) if @limit <= 0 Thread.fork do yield(*args) end else cv_wait unless @noblock Thread.fork do cv_wait if @noblock begin yield(*args) ensure cv_signal end end end end |
#wait ⇒ Object
Wait for all threads to finish.
89 90 91 92 93 94 95 96 97 |
# File 'lib/threadlimiter/threadlimiter.rb', line 89 def wait @mutex.synchronize do while @running > 0 @cv.wait(@mutex) end end self end |