You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
Thread pools are neither a new idea nor an implementation of the actor pattern. Nevertheless, thread pools are still an extremely relevant concurrency tool. Every time a thread is created then subsequently destroyed there is overhead. Creating a pool of reusable worker threads then repeatedly' dipping into the pool can have huge performace benefits for a long-running application like a service. Ruby's blocks provide an excellent mechanism for passing a generic work request to a thread, making Ruby an excellent candidate language for thread pools.
3
+
A Thread Pool is an abstraction that you can give a unit of work to, and the work will be executed by one of possibly several threads in the pool. One motivation for using thread pools is the overhead of creating and destroying threads. Creating a pool of reusable worker threads then repeatedly re-using threads from the pool can have huge performace benefits for a long-running application like a service.
4
4
5
-
The inspiration for thread pools in this library is Java's `java.util.concurrent` implementation of [thread pools](java.util.concurrent). The `java.util.concurrent` library is a well-designed, stable, scalable, and battle-tested concurrency library. It provides three different implementations of thread pools. One of those implementations is simply a special case of the first and doesn't offer much advantage in Ruby, so only the first two (`FixedThreadPool` and `CachedThreadPool`) are implemented here.
5
+
`concurrent-ruby` also offers some higher level abstractions than thread pools. For many problems, you will be better served by using one of these -- if you are thinking of using a thread pool, we especially recommend you look at and understand [Future](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/Future.html)s before deciding to use thread pools directly instead. Futures are implemented using thread pools, but offer a higher level abstraction.
6
6
7
-
Thread pools share common behavior defined by several mixin modules including `Executor`. The most important method is `post` (aliased with the left-shift operator `<<`). The `post` method sends a block to the pool for future processing.
7
+
But there are some problems for which directly using a thread pool is an appropriate solution. Or, you may wish to make your own thread pool to run Futures on, to be separate or have different characteristics than the global thread pool that Futures run on by default.
8
8
9
-
A running thread pool can be shutdown in an orderly or disruptive manner. Once a thread pool has been shutdown it cannot be started again. The `shutdown` method can be used to initiate an orderly shutdown of the thread pool. All new `post` calls will reject the given block and immediately return `false`. Threads in the pool will continue to process all in-progress work and will process all tasks still in the queue. The `kill` method can be used to immediately shutdown the pool. All new `post` calls will reject the given block and immediately return `false`. Ruby's `Thread.kill` will be called on all threads in the pool, aborting all in-progress work. Tasks in the queue will be discarded.
9
+
Thread pools are considered 'executors' -- an object you can give a unit of work to, to have it exeucted. In fact, thread pools are the main kind of executor you will see, others are mainly for testing or odd edge cases. In some documentation or source code you'll see reference to an 'executor' -- this is commonly a thread pool, or else something similar that executes units of work (usually supplied as ruby blocks).
10
10
11
-
A client thread can choose to block and wait for pool shutdown to complete. This is useful when shutting down an application and ensuring the app doesn't exit before pool processing is complete. The method `wait_for_termination` will block for a maximum of the given number of seconds then return `true` if shutdown completed successfully or `false`. When the timeout value is `nil` the call will block indefinitely. Calling `wait_for_termination` on a stopped thread pool will immediately return `true`.
11
+
## FixedThreadPool
12
12
13
-
Predicate methods are provided to describe the current state of the thread pool. Provided methods are `running?`, `shuttingdown?`, and `shutdown?`. The `shutdown` method will return true regardless of whether the pool was shutdown wil `shutdown` or `kill`.
13
+
A [FixedThreadPool](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/FixedThreadPool.html) contains a fixed number of threads. When you give a unit if work to it, an available thread will be used to execute.
14
14
15
-
### FixedThreadPool
15
+
~~~ruby
16
+
pool =Concurrent::FixedThreadPool.new(5) # 5 threads
17
+
pool.post do
18
+
# some parallel work
19
+
end
20
+
# As with all thread pools, execution resumes immediately here in the caller thread,
21
+
# while work is concurrently being done in the thread pool, at some possibly future point.
22
+
~~~
23
+
24
+
What happens if you post new work when all (eg) 5 threads are currently busy? It will be added to a queue, and executed when a thread becomes available. In a `FixedThreadPool`, if you post work to the pool much faster than the work can be completed, the queue may grow without bounds, as the work piles up in the holding queue, using up memory without bounds. To limit the queue and apply some form of 'back pressure' instead, you can use the more configurable `ThreadPoolExecutor` (See below).
25
+
26
+
If you'd like to base the number of threads in the pool on the number of processors available, your code can consult [Concurrent.processor_count](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/ProcessorCounter.html#processor_count-instance_method).
27
+
28
+
The `FixedThreadPool` is based on the semantics used in Java for [java.util.concurrent.Executors.newFixedThreadPool(int nThreads)](https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Executors.html#newFixedThreadPool(int))
16
29
17
-
From the docs:
30
+
## CachedThreadPool
18
31
19
-
> Creates a thread pool that reuses a fixed number of threads operating off a shared unbounded queue.
20
-
> At any point, at most `nThreads` threads will be active processing tasks. If additional tasks are submitted
21
-
> when all threads are active, they will wait in the queue until a thread is available. If any thread terminates
22
-
> due to a failure during execution prior to shutdown, a new one will take its place if needed to execute
23
-
> subsequent tasks. The threads in the pool will exist until it is explicitly `shutdown`.
32
+
A [CachedThreadPool](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/CachedThreadPool.html) will create as many threads as neccesary for work posted to it. If you post work to a `CachedThreadPool` when all it's existing threads are busy, it will create a new thread to execute that work, and then keep that thread cached for future work. Cached threads are reclaimed (destroyed) after they are idle for a while.
33
+
34
+
CachedThreadPools typically improve the performance of programs that execute many short-lived asynchronous tasks.
35
+
36
+
~~~ruby
37
+
pool =Concurrent::CachedThreadPool.new
38
+
pool.post do
39
+
# some parallel work
40
+
end
41
+
~~~
24
42
25
-
#### Examples
43
+
The behavior of `CachedThreadPool` is based on Java's [java.util.concurrent.Executors.newCachedThreadPool()](https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Executors.html#newCachedThreadPool())
26
44
27
-
```ruby
28
-
require'concurrent'
45
+
If you'd like to configure a maximum number of threads, you can use the more general configurable `ThreadPoolExecutor`.
A [ThreadPoolExecutor](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/ThreadPoolExecutor.html) is a general-purpose thread pool that can be configured to have various behaviors.
35
50
36
-
pool.post(1,2,3){|*args| sleep(10) }
37
-
pool <<proc{ sleep(10) }
38
-
pool.size #=> 5
51
+
The `CachedThreadPool` and `FixedThreadPool` are simply `ThreadPoolExecutor`s with certain configuration pre-determined. For instance, to create a `ThreadPoolExecutor` that works just like a `FixedThreadPool.new 5`, you could:
If you want to provide a maximum queue size, you may also consider the `overflow_policy` -- what will happen if work is posted to a pool when the queue of waiting work has reached the maximum size? Available policies:
46
62
47
-
pool.size #=> 0
48
-
pool.status #=> []
49
-
pool.shutdown? #=> true
50
-
```
63
+
* abort: Raise a `Concurrent::RejectedExecutionError` exception and discard the task. (default policy)
64
+
* discard: Silently discard the task and return nil as the task result.
65
+
* caller_runs: The work will be executed in the thread of the caller, instead of being given to another thread in the pool
51
66
52
-
### CachedThreadPool
67
+
~~~ruby
68
+
pool =Concurrent::ThreadPoolExecutor.new(
69
+
min_threads:5,
70
+
max_threads:5,
71
+
max_queue:100,
72
+
overflow_policy::caller_runs
73
+
)
74
+
~~~
53
75
54
-
From the docs:
76
+
Similarly, you can create something similar to a `CachedThreadPool`, but with a maximum number of threads. With an unbounded queue:
55
77
56
-
> Creates a thread pool that creates new threads as needed, but will reuse previously constructed threads when
57
-
> they are available. These pools will typically improve the performance of programs that execute many short-lived
58
-
> asynchronous tasks. Calls to [`post`] will reuse previously constructed threads if available. If no existing
59
-
> thread is available, a new thread will be created and added to the pool. Threads that have not been used for
60
-
> sixty seconds are terminated and removed from the cache. Thus, a pool that remains idle for long enough will
61
-
> not consume any resources. Note that pools with similar properties but different details (for example,
62
-
> timeout parameters) may be created using [`CachedThreadPool`] constructors.
78
+
~~~ruby
79
+
pool =Concurrent::ThreadPoolExecutor.new(
80
+
min_threads:3, # create 3 threads at startup
81
+
max_threads:10, # create at most 10 threads
82
+
max_queue:0, # unbounded queue of work waiting for an available thread
83
+
)
84
+
~~~
63
85
64
-
#### Examples
86
+
Or, with a variable number of threads like a CachedThreadPool, but with a bounded queue and an overflow_policy:
65
87
66
-
```ruby
67
-
require'concurrent'
88
+
~~~ruby
89
+
pool =Concurrent::ThreadPoolExecutor.new(
90
+
min_threads:3, # create 3 threads at startup
91
+
max_threads:10, # create at most 10 threads
92
+
max_queue:100, # at most 100 jobs waiting in the queue,
93
+
overflow_policy::abort
94
+
)
95
+
~~~
68
96
69
-
pool =Concurrent::CachedThreadPool.new
97
+
ThreadPoolExecutors with `min_threads` and `max_threads` set to different values will ordinarily reclaim idle threads. You can supply an `idletime` argument, number of seconds that a thread may be idle before being reclaimed. The default is 60 seconds.
70
98
71
-
pool.size #=> 0
72
-
pool.running? #=> true
73
-
pool.status #=> []
99
+
`concurrent-ruby` thread pools are based on designs from `java.util.concurrent` -- a well-designed, stable, scalable, and battle-tested concurrency library. The `ThreadPoolExecutor` is based on Java [java.util.concurrent.ThreadPoolExecutor](https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ThreadPoolExecutor.html), and is in fact implemented with a Java ThreadPoolExecutor when running under JRuby. For more information on the design and concepts, you may find the Java documentation helpful:
A running thread pool can be shutdown in an orderly or disruptive manner. Once a thread pool has been shutdown it cannot be started again.
86
108
87
-
pool.shutdown #=> :shuttingdown
88
-
pool.status #=> []
89
-
pool.wait_for_termination
109
+
The `shutdown` method can be used to initiate an orderly shutdown of the thread pool. All new post calls will reject the given block and immediately return false. Threads in the pool will continue to process all in-progress work and will process all tasks still in the queue.
90
110
91
-
pool.size #=> 0
92
-
pool.status #=> []
93
-
pool.shutdown? #=> true
94
-
```
111
+
The `kill` method can be used to immediately shutdown the pool. All new post calls will reject the given block and immediately return false. Ruby's Thread.kill will be called on all threads in the pool, aborting all in-progress work. Tasks in the queue will be discarded.
95
112
96
-
### Other Executors
113
+
The method `wait_for_termination` can be used to block and wait for pool shutdown to complete. This is useful when shutting down an application and ensuring the app doesn't exit before pool processing is complete. The method wait_for_termination will block for a maximum of the given number of seconds then return true if shutdown completed successfully or false. When the timeout value is nil the call will block indefinitely. Calling wait_for_termination on a stopped thread pool will immediately return true.
97
114
98
-
There are several other thread pools and executors in this library. See the API documentation for more information:
115
+
~~~ruby
116
+
# tell the pool to shutdown in an orderly fashion, allowing in progress work to complete
117
+
pool.shutdown
118
+
# now wait for all work to complete, wait as long as it takes
119
+
pool.wait_for_termination
120
+
~~~
121
+
122
+
You can check for current pool status:
123
+
124
+
~~~ruby
125
+
pool.running?
126
+
pool.shuttingdown? # in process of shutting down, can't take any more work
127
+
pool.shutdown? # it's done
128
+
~~~
129
+
130
+
The `shutdown?` method will return true for a stopped pool, regardless of whether the pool was stopped with `shutdown` or `kill`.
131
+
132
+
## Other Executors
133
+
134
+
There are several other thread pools and executors in the `concurrent-ruby` library. See the API documentation for more information:
For efficiency, Concurrent Ruby provides a few global thread pools. These executors are used by the higher-level abstractions for running asynchronous operations without creating new threads more often than necessary. These executors are lazy-loaded so they do not create overhead when not needed. The global executors may also be accessed directly if desired. For more information regarding the global thread pools and their configuration, refer to the [API documentation](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/Configuration.html).
147
+
## Global Thread Pools
114
148
115
-
#### Changing the Global Thread Pool
149
+
Concurrent Ruby provides several global thread pools. Higher-level abstractions use global thread pools, by default, for running asynchronous operations without creating new threads more often than necessary. These executors are lazy-loaded so they do not create overhead when not needed. The global executors may also be accessed directly if desired. For more information regarding the global thread pools and their configuration, refer to the [API documentation](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/Configuration.html).
116
150
117
-
It should rarely be necessary to reconfigure the global executors. If necessary, it is possible to change the gem configuration during application initialization. Gem configration must be done *before* the global executors are lazy-loaded. Once the global thread pools are initialized they may no longer be reconfigured. Doing so will raise an exception.
151
+
When using a higher-level abstraction, which ordinarily uses a global thread pool, you may wish to instead supply your own thread pool, for separation of work, or to control the thread pool behavior with configuration.
0 commit comments