implement pool cancellation #15
Assignees
Labels
feature
New feature or request
Comments
|
some legacy code: test-producer.go func CancelProducerAfter[I, R any](
_ context.Context,
delay time.Duration,
funcs ...any,
) {
fmt.Printf(" >>> 💤 CancelAfter - Sleeping before requesting cancellation (%v) ...\n", delay)
<-time.After(delay)
if len(funcs) > 0 {
c, ok := funcs[0].(context.CancelFunc)
if ok && c != nil {
fmt.Printf(" >>> CancelAfter - 🛑🛑🛑 cancellation submitted.\n")
c()
fmt.Printf(" >>> CancelAfter - ➖➖➖ CANCELLED\n")
}
} else {
fmt.Printf(" >>> CancelAfter - ✖️✖️✖️ cancellation attempt benign.\n")
}
}Output that shows an example run of the worker-pool with cancellation ... |
|
🎉🎉🎉 Found the resolution to the cancellation problem, it was in the producer. The producer could get blocked on sending the job, at which point it does not see the cancellation, until this fix was applied (pre-empt the job send, with a select on the context, newly passed into the item() function): func (p *Producer[I, R]) item(ctx context.Context) bool {
result := true
i := p.provider()
j := async.Job[I]{
ID: fmt.Sprintf("JOB-ID:%v", uuid.NewString()),
Input: i,
}
fmt.Printf(">>>> ✨ producer.item, 🟠 waiting to post item: '%+v'\n", i)
select {
case <-ctx.Done():
fmt.Println(">>>> 💠 producer.item - done received ⛔⛔⛔")
result = false
case p.JobsCh <- j:
}
p.Count++
if result {
fmt.Printf(">>>> ✨ producer.item, 🟢 posted item: '%+v'\n", i)
} else {
fmt.Printf(">>>> ✨ producer.item, 🔴 item NOT posted: '%+v'\n", i)
}
return result
}Output that shows an example run showing the cancellation working ...The producer run loop, checks the return value of the item() function and terminates its loop if the return value = false, indicating the job was not sent, due to the done event occurring. |
plastikfan
added a commit
that referenced
this issue
Aug 15, 2023
plastikfan
added a commit
that referenced
this issue
Aug 15, 2023
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Output that shows an example run showing the end of run being triggered by stopping the producer ...
We have a couple of options for cancellation.
1️⃣ follow the same route as stopping the producer. Might be easier to implement, but may not be in the spirt of golang concurrency, IE, we should be able to cancel the context and get the logic correct so that GRs are not leaked and there are no deadlocks or race conditions.
2️⃣ implement cancellation via context: This should be the way to fix this, but currently, this is not working due to deadlock.
Given a worker pool, its interface is a channel of jobs and a context in which to run. So we know that the pool will complete when the producer stops. So what about cancellation. If we cancel the producer, the producer can follow the same procedure for a stop. Or do we issue a cancellation to the producer and the pool? If we cancel the pool, how are we guaranteed that the producer is stooped?
What we should say is that the producer and the pool work under the same context. So if the context is cancelled, then the producer should stop and the pool should react to the cancellation.
STOP: When the producer indicates no more work, it closes the job channel. In response to this closure, the pool, closes its worker job channel, exits its run loop to enter the drain loop. When the worker sees that the workers job channel is closed, it exits it run loop, then sends a finished notification to the pool.
So here we have 2 notification for the pool, ie the source jobs closure and the multiple finished notifications from the workers. This interaction works fine.
The text was updated successfully, but these errors were encountered: