Describe the bug
It looks like subject can deadlock if task, that is observing its values is cancelled at the same time as send is called on it
Thread 1:
AsyncBufferedChannel.send acquires os_unfair_lockAsyncBufferedChannel.send attempts to resume continuation returning element- Resuming continuation attempts to acquire internal lock for task status
Thread 2:
- Task that awaits for values from subject is cancelled. This acquires internal lock for task state
unregister() is called and attempts to acquire os_unfair_lock
Thread 1 has os_unfair_lock, waits for internal lock
Thread 2 has internal lock, waits for os_unfair_lock
Situation can be reproduced with test like this, if run repeatedly for 100 times:
@Test
func send_deadlocks_when_cancelled_concurrently() throws {
let subject = AsyncPassthroughSubject<Int>()
let t1 = Task.detached {
for await _ in subject { }
}
let barrier = DispatchSemaphore(value: 0)
let done = DispatchSemaphore(value: 0)
DispatchQueue.global().async {
barrier.wait()
t1.cancel()
done.signal()
}
DispatchQueue.global().async {
barrier.wait()
subject.send(1)
done.signal()
}
barrier.signal()
barrier.signal()
done.wait()
done.wait()
}
It can also be reproduced using strict concurrency only, but setting up actual parallel execution is more complicated.
Additional context
Describe the bug
It looks like subject can deadlock if task, that is observing its values is cancelled at the same time as
sendis called on itThread 1:
AsyncBufferedChannel.sendacquiresos_unfair_lockAsyncBufferedChannel.sendattempts to resume continuation returning elementThread 2:
unregister()is called and attempts to acquireos_unfair_lockThread 1 has
os_unfair_lock, waits for internal lockThread 2 has internal lock, waits for
os_unfair_lockSituation can be reproduced with test like this, if run repeatedly for 100 times:
It can also be reproduced using strict concurrency only, but setting up actual parallel execution is more complicated.
Additional context