diff --git a/CHANGELOG.md b/CHANGELOG.md index 5a02bc5..7f4ee6b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,7 @@ **Unreleased:** - Subjects: fix a deadlock when sending values or termination concurrently with consumer cancellation (https://github.com/sideeffect-io/AsyncExtensions/issues/52). +- Subjects: preserve delivery order for concurrent sends while keeping consumer resumption outside the state lock (https://github.com/sideeffect-io/AsyncExtensions/issues/61). - Multicast: preserve upstream element order and prevent termination from overtaking values when consumers advance concurrently. **v0.5.2 - Oxygen:** diff --git a/Sources/AsyncSubjects/AsyncCurrentValueSubject.swift b/Sources/AsyncSubjects/AsyncCurrentValueSubject.swift index 482bc33..c42353e 100644 --- a/Sources/AsyncSubjects/AsyncCurrentValueSubject.swift +++ b/Sources/AsyncSubjects/AsyncCurrentValueSubject.swift @@ -45,6 +45,7 @@ public final class AsyncCurrentValueSubject: AsyncSubject where Element } let state: ManagedCriticalState + let delivery = OrderedDelivery() public var value: Element { get { @@ -67,28 +68,32 @@ public final class AsyncCurrentValueSubject: AsyncSubject where Element /// Sends a value to all consumers /// - Parameter element: the value to send public func send(_ element: Element) { - let channels = self.state.withCriticalRegion { state in + let shouldDrain = self.state.withCriticalRegion { state in state.current = element - return Array(state.channels.values) - } - // Resuming a consumer must not hold the lock used by its cancellation handler. - for channel in channels { - channel.send(element) + let channels = Array(state.channels.values) + return self.delivery.enqueue { + for channel in channels { + channel.send(element) + } + } } + if shouldDrain { self.delivery.drain() } } /// Finishes the async sequences with a normal ending. /// - Parameter termination: The termination to finish the subject. public func send(_ termination: Termination) { - let channels = self.state.withCriticalRegion { state in + let shouldDrain = self.state.withCriticalRegion { state in state.terminalState = termination let channels = Array(state.channels.values) state.channels.removeAll() - return channels - } - for channel in channels { - channel.finish() + return self.delivery.enqueue { + for channel in channels { + channel.finish() + } + } } + if shouldDrain { self.delivery.drain() } } func handleNewConsumer() -> (iterator: AsyncBufferedChannel.Iterator, unregister: @Sendable () -> Void) { diff --git a/Sources/AsyncSubjects/AsyncPassthroughSubject.swift b/Sources/AsyncSubjects/AsyncPassthroughSubject.swift index 6f15f1c..39811cb 100644 --- a/Sources/AsyncSubjects/AsyncPassthroughSubject.swift +++ b/Sources/AsyncSubjects/AsyncPassthroughSubject.swift @@ -42,6 +42,7 @@ public final class AsyncPassthroughSubject: AsyncSubject { } let state: ManagedCriticalState + let delivery = OrderedDelivery() public init() { self.state = ManagedCriticalState( @@ -52,27 +53,31 @@ public final class AsyncPassthroughSubject: AsyncSubject { /// Sends a value to all consumers /// - Parameter element: the value to send public func send(_ element: Element) { - let channels = self.state.withCriticalRegion { state in - return Array(state.channels.values) - } - // Resuming a consumer must not hold the lock used by its cancellation handler. - for channel in channels { - channel.send(element) + let shouldDrain = self.state.withCriticalRegion { state in + let channels = Array(state.channels.values) + return self.delivery.enqueue { + for channel in channels { + channel.send(element) + } + } } + if shouldDrain { self.delivery.drain() } } /// Finishes the subject with a normal ending. /// - Parameter termination: The termination to finish the subject public func send(_ termination: Termination) { - let channels = self.state.withCriticalRegion { state in + let shouldDrain = self.state.withCriticalRegion { state in state.terminalState = termination let channels = Array(state.channels.values) state.channels.removeAll() - return channels - } - for channel in channels { - channel.finish() + return self.delivery.enqueue { + for channel in channels { + channel.finish() + } + } } + if shouldDrain { self.delivery.drain() } } func handleNewConsumer() -> (iterator: AsyncBufferedChannel.Iterator, unregister: @Sendable () -> Void) { diff --git a/Sources/AsyncSubjects/AsyncReplaySubject.swift b/Sources/AsyncSubjects/AsyncReplaySubject.swift index f3fc334..a1ec596 100644 --- a/Sources/AsyncSubjects/AsyncReplaySubject.swift +++ b/Sources/AsyncSubjects/AsyncReplaySubject.swift @@ -34,6 +34,7 @@ public final class AsyncReplaySubject: AsyncSubject where Element: Send } let state: ManagedCriticalState + let delivery = OrderedDelivery() /// Creates a replay subject with a buffer of replayable values /// - Parameter bufferSize: the number of values that will be replayed to new consumers @@ -46,33 +47,37 @@ public final class AsyncReplaySubject: AsyncSubject where Element: Send /// Sends a value to all consumers /// - Parameter element: the value to send public func send(_ element: Element) { - let channels = self.state.withCriticalRegion { state in + let shouldDrain = self.state.withCriticalRegion { state in if state.buffer.count >= state.bufferSize && !state.buffer.isEmpty { state.buffer.removeFirst() } state.buffer.append(element) - return Array(state.channels.values) - } - // Resuming a consumer must not hold the lock used by its cancellation handler. - for channel in channels { - channel.send(element) + let channels = Array(state.channels.values) + return self.delivery.enqueue { + for channel in channels { + channel.send(element) + } + } } + if shouldDrain { self.delivery.drain() } } /// Finishes the subject with a normal ending. /// - Parameter termination: The termination to finish the subject. public func send(_ termination: Termination) { - let channels = self.state.withCriticalRegion { state in + let shouldDrain = self.state.withCriticalRegion { state in state.terminalState = termination let channels = Array(state.channels.values) state.channels.removeAll() state.buffer.removeAll() state.bufferSize = 0 - return channels - } - for channel in channels { - channel.finish() + return self.delivery.enqueue { + for channel in channels { + channel.finish() + } + } } + if shouldDrain { self.delivery.drain() } } func handleNewConsumer() -> (iterator: AsyncBufferedChannel.Iterator, unregister: @Sendable () -> Void) { diff --git a/Sources/AsyncSubjects/AsyncThrowingCurrentValueSubject.swift b/Sources/AsyncSubjects/AsyncThrowingCurrentValueSubject.swift index 1b10921..884cb78 100644 --- a/Sources/AsyncSubjects/AsyncThrowingCurrentValueSubject.swift +++ b/Sources/AsyncSubjects/AsyncThrowingCurrentValueSubject.swift @@ -47,6 +47,7 @@ public final class AsyncThrowingCurrentValueSubject: As } let state: ManagedCriticalState + let delivery = OrderedDelivery() public var value: Element { get { @@ -67,33 +68,35 @@ public final class AsyncThrowingCurrentValueSubject: As /// Sends a value to all consumers /// - Parameter element: the value to send public func send(_ element: Element) { - let channels = self.state.withCriticalRegion { state in + let shouldDrain = self.state.withCriticalRegion { state in state.current = element - return Array(state.channels.values) - } - // Resuming a consumer must not hold the lock used by its cancellation handler. - for channel in channels { - channel.send(element) + let channels = Array(state.channels.values) + return self.delivery.enqueue { + for channel in channels { + channel.send(element) + } + } } + if shouldDrain { self.delivery.drain() } } /// Finishes the subject with either a normal ending or an error. /// - Parameter termination: The termination to finish the subject. public func send(_ termination: Termination) { - let channels = self.state.withCriticalRegion { state in + let shouldDrain = self.state.withCriticalRegion { state in state.terminalState = termination let channels = Array(state.channels.values) state.channels.removeAll() - return channels - } - for channel in channels { - switch termination { - case .finished: - channel.finish() - case .failure(let error): - channel.fail(error) + return self.delivery.enqueue { + switch termination { + case .finished: + for channel in channels { channel.finish() } + case .failure(let error): + for channel in channels { channel.fail(error) } + } } } + if shouldDrain { self.delivery.drain() } } func handleNewConsumer( diff --git a/Sources/AsyncSubjects/AsyncThrowingPassthroughSubject.swift b/Sources/AsyncSubjects/AsyncThrowingPassthroughSubject.swift index f456db3..d1020ff 100644 --- a/Sources/AsyncSubjects/AsyncThrowingPassthroughSubject.swift +++ b/Sources/AsyncSubjects/AsyncThrowingPassthroughSubject.swift @@ -43,6 +43,7 @@ public final class AsyncThrowingPassthroughSubject: Asy } let state: ManagedCriticalState + let delivery = OrderedDelivery() public init() { self.state = ManagedCriticalState( @@ -53,32 +54,36 @@ public final class AsyncThrowingPassthroughSubject: Asy /// Sends a value to all consumers /// - Parameter element: the value to send public func send(_ element: Element) { - let channels = self.state.withCriticalRegion { state in - return Array(state.channels.values) - } - // Resuming a consumer must not hold the lock used by its cancellation handler. - for channel in channels { - channel.send(element) + let shouldDrain = self.state.withCriticalRegion { state in + let channels = Array(state.channels.values) + return self.delivery.enqueue { + for channel in channels { + channel.send(element) + } + } } + if shouldDrain { self.delivery.drain() } } /// Finishes the subject with either a normal ending or an error. /// - Parameter termination: The termination to finish the subject public func send(_ termination: Termination) { - let channels = self.state.withCriticalRegion { state in + let shouldDrain = self.state.withCriticalRegion { state in state.terminalState = termination let channels = Array(state.channels.values) state.channels.removeAll() - return channels - } - for channel in channels { - switch termination { - case .finished: - channel.finish() - case .failure(let error): - channel.fail(error) + return self.delivery.enqueue { + for channel in channels { + switch termination { + case .finished: + channel.finish() + case .failure(let error): + channel.fail(error) + } + } } } + if shouldDrain { self.delivery.drain() } } func handleNewConsumer( diff --git a/Sources/AsyncSubjects/AsyncThrowingReplaySubject.swift b/Sources/AsyncSubjects/AsyncThrowingReplaySubject.swift index f3f21c8..1470b71 100644 --- a/Sources/AsyncSubjects/AsyncThrowingReplaySubject.swift +++ b/Sources/AsyncSubjects/AsyncThrowingReplaySubject.swift @@ -35,6 +35,7 @@ public final class AsyncThrowingReplaySubject: AsyncSub } let state: ManagedCriticalState + let delivery = OrderedDelivery() public init(bufferSize: UInt) { self.state = ManagedCriticalState( @@ -45,38 +46,40 @@ public final class AsyncThrowingReplaySubject: AsyncSub /// Sends a value to all consumers /// - Parameter element: the value to send public func send(_ element: Element) { - let channels = self.state.withCriticalRegion { state in + let shouldDrain = self.state.withCriticalRegion { state in if state.buffer.count >= state.bufferSize && !state.buffer.isEmpty { state.buffer.removeFirst() } state.buffer.append(element) - return Array(state.channels.values) - } - // Resuming a consumer must not hold the lock used by its cancellation handler. - for channel in channels { - channel.send(element) + let channels = Array(state.channels.values) + return self.delivery.enqueue { + for channel in channels { + channel.send(element) + } + } } + if shouldDrain { self.delivery.drain() } } /// Finishes the subject with either a normal ending or an error. /// - Parameter termination: The termination to finish the subject public func send(_ termination: Termination) { - let channels = self.state.withCriticalRegion { state in + let shouldDrain = self.state.withCriticalRegion { state in state.terminalState = termination let channels = Array(state.channels.values) state.channels.removeAll() state.buffer.removeAll() state.bufferSize = 0 - return channels - } - for channel in channels { - switch termination { - case .finished: - channel.finish() - case .failure(let error): - channel.fail(error) + return self.delivery.enqueue { + switch termination { + case .finished: + for channel in channels { channel.finish() } + case .failure(let error): + for channel in channels { channel.fail(error) } + } } } + if shouldDrain { self.delivery.drain() } } func handleNewConsumer( diff --git a/Sources/Supporting/OrderedDelivery.swift b/Sources/Supporting/OrderedDelivery.swift new file mode 100644 index 0000000..bdc92c5 --- /dev/null +++ b/Sources/Supporting/OrderedDelivery.swift @@ -0,0 +1,35 @@ +import DequeModule + +/// Delivers subject events in the same order that their state updates are recorded. +/// The queue lock is released before a delivery resumes any consumer. +final class OrderedDelivery: @unchecked Sendable { + private struct State { + var pending: Deque<@Sendable () -> Void> = [] + var isDelivering = false + } + + private let state = ManagedCriticalState(State()) + + /// Queues a delivery and returns true when the caller should start draining. + func enqueue(_ delivery: @escaping @Sendable () -> Void) -> Bool { + self.state.withCriticalRegion { state in + state.pending.append(delivery) + guard !state.isDelivering else { return false } + state.isDelivering = true + return true + } + } + + /// Drains queued deliveries without holding the queue lock while they run. + func drain() { + while let delivery = self.state.withCriticalRegion({ state -> (@Sendable () -> Void)? in + guard !state.pending.isEmpty else { + state.isDelivering = false + return nil + } + return state.pending.popFirst() + }) { + delivery() + } + } +} diff --git a/Tests/AsyncSubjets/AsyncSubjectConcurrentSendOrderingTests.swift b/Tests/AsyncSubjets/AsyncSubjectConcurrentSendOrderingTests.swift new file mode 100644 index 0000000..aec871b --- /dev/null +++ b/Tests/AsyncSubjets/AsyncSubjectConcurrentSendOrderingTests.swift @@ -0,0 +1,179 @@ +import Dispatch +@testable import AsyncExtensions +import Foundation +import XCTest + +final class AsyncSubjectConcurrentSendOrderingTests: XCTestCase { + private let rounds = 200 + private let sendsPerProducer = 100 + + func test_AsyncPassthroughSubject_subscribers_receive_concurrent_sends_in_the_same_order() async { + var roundsWithDifferentOrder = 0 + for _ in 0..() + let received = await collect(subscribers: 2, from: subject) { + sendConcurrently(producers: 2, sendsPerProducer: sendsPerProducer) { subject.send($0) } + subject.send(.finished) + } + if received[0] != received[1] { roundsWithDifferentOrder += 1 } + } + XCTAssertEqual(roundsWithDifferentOrder, 0) + } + + func test_AsyncCurrentValueSubject_subscriber_ends_on_the_subject_value_after_concurrent_sends() async { + var roundsWithStaleSubscriber = 0 + for _ in 0..(-1) + var value = 0 + let received = await collect(subscribers: 1, from: subject) { + sendConcurrently(producers: 2, sendsPerProducer: sendsPerProducer) { subject.send($0) } + value = subject.value + subject.send(.finished) + } + if received[0].last != value { roundsWithStaleSubscriber += 1 } + } + XCTAssertEqual(roundsWithStaleSubscriber, 0) + } + + func test_AsyncReplaySubject_subscriber_ends_on_the_replayed_value_after_concurrent_sends() async { + var roundsWithStaleSubscriber = 0 + for _ in 0..(bufferSize: 1) + var replayedValue: Int? + let received = await collect(subscribers: 1, from: subject) { + sendConcurrently(producers: 2, sendsPerProducer: sendsPerProducer) { subject.send($0) } + replayedValue = subject.state.criticalState.buffer.last + subject.send(.finished) + } + if received[0].last != replayedValue { roundsWithStaleSubscriber += 1 } + } + XCTAssertEqual(roundsWithStaleSubscriber, 0) + } + + func test_AsyncThrowingPassthroughSubject_subscribers_receive_concurrent_sends_in_the_same_order() async { + var roundsWithDifferentOrder = 0 + for _ in 0..() + let received = await collect(subscribers: 2, from: subject) { + sendConcurrently(producers: 2, sendsPerProducer: sendsPerProducer) { subject.send($0) } + subject.send(.finished) + } + if received[0] != received[1] { roundsWithDifferentOrder += 1 } + } + XCTAssertEqual(roundsWithDifferentOrder, 0) + } + + func test_AsyncThrowingCurrentValueSubject_subscriber_ends_on_the_subject_value_after_concurrent_sends() async { + var roundsWithStaleSubscriber = 0 + for _ in 0..(-1) + var value = 0 + let received = await collect(subscribers: 1, from: subject) { + sendConcurrently(producers: 2, sendsPerProducer: sendsPerProducer) { subject.send($0) } + value = subject.value + subject.send(.finished) + } + if received[0].last != value { roundsWithStaleSubscriber += 1 } + } + XCTAssertEqual(roundsWithStaleSubscriber, 0) + } + + func test_AsyncThrowingReplaySubject_subscribers_receive_concurrent_sends_in_the_same_order() async { + var roundsWithDifferentOrder = 0 + for _ in 0..(bufferSize: UInt(sendsPerProducer * 2)) + let received = await collect(subscribers: 2, from: subject) { + sendConcurrently(producers: 2, sendsPerProducer: sendsPerProducer) { subject.send($0) } + subject.send(.finished) + } + if received[0] != received[1] { roundsWithDifferentOrder += 1 } + } + XCTAssertEqual(roundsWithDifferentOrder, 0) + } + + func test_cancellation_handler_can_send_to_its_subject_during_a_concurrent_send() async { + for _ in 0..<1_000 { + let subject = AsyncPassthroughSubject() + let iterator = subject.makeAsyncIterator() + let exited = DispatchSemaphore(value: 0) + let consumer = Task.detached { + defer { exited.signal() } + await withTaskCancellationHandler { + var iterator = iterator + while let _ = await iterator.next() {} + } onCancel: { + subject.send(-1) + } + } + + let suspensionDeadline = DispatchTime.now() + .seconds(1) + while !isSuspended(subject) && DispatchTime.now() < suspensionDeadline { + try? await Task.sleep(nanoseconds: 100_000) + } + guard isSuspended(subject) else { + XCTFail("consumer did not suspend in next()") + consumer.cancel() + return + } + + DispatchQueue.concurrentPerform(iterations: 2) { index in + if index == 0 { + consumer.cancel() + } else { + subject.send(1) + } + } + + let exitWaiter = Task.detached { + waitForSignal(exited, timeout: .seconds(1)) + } + guard await exitWaiter.value else { + XCTFail("consumer did not exit after cancellation") + consumer.cancel() + return + } + } + } + + private func collect( + subscribers: Int, + from subject: Subject, + produce: () -> Void + ) async -> [[Subject.Element]] where Subject.Element: Sendable { + let iterators = (0.. [Subject.Element] in + var iterator = iterator + var values: [Subject.Element] = [] + while let value = try? await iterator.next() { + values.append(value) + } + return values + } + } + produce() + var received: [[Subject.Element]] = [] + for consumer in consumers { + received.append(await consumer.value) + } + return received + } +} + +private func isSuspended(_ subject: AsyncPassthroughSubject) -> Bool { + guard let channel = subject.state.criticalState.channels.values.first else { return false } + if case .awaiting = channel.state.criticalState { return true } + return false +} + +private func waitForSignal(_ semaphore: DispatchSemaphore, timeout: DispatchTimeInterval) -> Bool { + semaphore.wait(timeout: .now() + timeout) == .success +} + +private func sendConcurrently(producers: Int, sendsPerProducer: Int, send: (Int) -> Void) { + DispatchQueue.concurrentPerform(iterations: producers) { producer in + for index in 0..