Asynchronous pruning for RubyThreadPoolExecutor #1082
Conversation
joshuay03
force-pushed
the
better-ruby-thread-pool-executor-pruning
branch
15 times, most recently
from
b6e5656 to
f90d46d
Compare
joshuay03
force-pushed
the
better-ruby-thread-pool-executor-pruning
branch
14 times, most recently
from
c5eca7d to
89b67f4
Compare
joshuay03
force-pushed
the
better-ruby-thread-pool-executor-pruning
branch
10 times, most recently
from
547428b to
a675844
Compare
lib/concurrent-ruby/concurrent/collection/ruby_timeout_queue.rb
Outdated
Show resolved
Hide resolved
joshuay03
force-pushed
the
better-ruby-thread-pool-executor-pruning
branch
7 times, most recently
from
fd02056 to
6cedac3
Compare
lib/concurrent-ruby/concurrent/collection/ruby_timeout_queue.rb
Outdated
Show resolved
Hide resolved
lib/concurrent-ruby/concurrent/collection/ruby_timeout_queue.rb
Outdated
Show resolved
Hide resolved
joshuay03
force-pushed
the
better-ruby-thread-pool-executor-pruning
branch
4 times, most recently
from
defd0f7 to
9739e9a
Compare
lib/concurrent-ruby/concurrent/collection/ruby_timeout_queue.rb
Outdated
Show resolved
Hide resolved
joshuay03
force-pushed
the
better-ruby-thread-pool-executor-pruning
branch
from
9739e9a to
6aa4ee5
Compare
There was a problem hiding this comment.
The Queue stuff looks ready to me, i.e. I reviewed and approve that part.
The pool changes it's hard for me to tell because I haven't had time to understand the implementation in details (and likely won't have time for that soon).
It would be great if another concurrent-ruby maintainer could help review that (and they should feel free to merge this PR after they approve)
| throw :stop | ||
| end | ||
|
|
||
| prunable = false |
There was a problem hiding this comment.
Is it impossible if prunable_capacity is <= 0 that a worker would need to be prunable again?
i.e. what if prunable_capacity is 0 because all workers are busy but later on the pool is no longer running? and then we should prune/finish all workers?
| timeout = prunable && my_pool.running? ? my_idletime : nil | ||
| case message = my_queue.pop(timeout: timeout) | ||
| when nil | ||
| if my_pool.prunable_capacity.positive? |
There was a problem hiding this comment.
| if my_pool.prunable_capacity.positive? | |
| if my_pool.prunable_capacity > 0 |
I think this is clearer
| @@ -95,8 +97,16 @@ def remaining_capacity | |||
| end | |||
|
|
|||
| # @!visibility private | |||
| def remove_busy_worker(worker) | |||
| synchronize { ns_remove_busy_worker worker } | |||
| def prunable_capacity | |||
There was a problem hiding this comment.
Could you add a comment documenting what does this return?
| def remove_busy_worker(worker) | ||
| synchronize { ns_remove_busy_worker worker } | ||
| def prunable_capacity | ||
| synchronize { ns_prunable_capacity } |
There was a problem hiding this comment.
Isn't there a risk the value is outdated by the time it's used since it will be used outside of this synchronize block?
|
|
||
| @next_gc_time = Concurrent.monotonic_time + @gc_interval | ||
| def ns_prunable_capacity |
There was a problem hiding this comment.
Could you explain the logic here? Is it how many workers can be removed at the current time?
Closes #1066
Closes #1075
Alternative to #1079
Implementation is based on the discussion in the linked issues.