Closed jonmast closed 2 years ago
Yes looks like it's run while holding the lock.
Task 4 running on 'main'
Task 1 running on 'worker-1'
Task 4 complete
Task 5 running on 'main'
Task 1 complete
Task 2 running on 'worker-1'
Task 5 complete
Task 7 running on 'main'
Task 2 complete
Task 3 running on 'worker-1'
Task 7 complete
Task 9 running on 'main'
Task 3 complete
Task 6 running on 'worker-1'
Task 9 complete
Task 6 complete
Task 8 running on 'worker-1'
Task 8 complete
Task 10 running on 'worker-1'
Task 10 complete
That's what we want, isn't it?
Here's a deterministic test.
require 'concurrent-ruby'
log = Queue.new
executor = Concurrent::FixedThreadPool.new(1, max_queue: 2, fallback_policy: :caller_runs)
worker_unblocker = Concurrent::CountDownLatch.new(1)
executor_unblocker = Concurrent::CountDownLatch.new(1)
# Block the worker thread
executor << proc { worker_unblocker.wait }
# Fill the queue
executor << proc { log.push :queued }
executor << proc { log.push :queued }
t = Thread.new {
# Block in a caller_runs job
executor << proc { executor_unblocker.wait; log.push :unblocked }
}
# Wait until the caller_runs job is blocked
Thread.pass until t.status == 'sleep'
# Now unblock the worker thread
worker_unblocker.count_down
executor_unblocker.count_down
executor.shutdown
executor.wait_for_termination
# Ideally, we will see the queued jobs run before the caller_runs job unblocks
p [log.pop, log.pop, log.pop]
# [:unblocked, :queued, :queued] bad
# [:queued, :queued, :unblocked] good
Phew that was hard.
I have a PR that achieves that, if I get feedback that is is the desired behaviour.
Sorry I'm so slow to follow up but this is perfect, thanks @chrisseaton
I'm attempting to use a threadpool to implement bounded concurrency with the the
caller_runs
fallback_policy to add back pressure. This doesn't work as I'd expect, after the queue is full and the fallback policy is triggered subsequent jobs all get processed on the caller thread rather than being delegated to the workers.Reproduction script:
Output:
I suspect this is because the threadpool remains locked while the caller task is running, preventing worker threads from continuing. Adding a
sleep
between posting each task to the pool unblocks it and allows worker threads to continue processing.I don't know if this is considered a bug or not, but I found it very surprising and can't see an obvious reason it needs to work this way, I'd expect the worker threads to be able to pull items from the queue regardless of how long the caller task blocks.