Combine的receive(on:)未将cancel回调派发到指定串行队列引发数据竞争问题咨询

根据Apple官方说明，receive(on:options:) 可将回调调度到指定队列执行。我们在下方代码中使用自定义串行调度队列，目的是避免对localOptionalCancellable的访问产生竞态，但实际运行时receiveCancel回调并未被调度到该队列执行，请问造成该现象的根本原因是什么？

官方文档原文说明：

你可以使用receive(on:options:)操作符在指定调度器上接收发布结果与完成事件，例如在主运行循环上执行UI相关工作。
...
在订阅者中执行任务时，优先使用receive(on:options:)而非显式操作调度队列，例如可替代以下这类编码模式：

问题复现代码 #

import Foundation
import Combine

class Example {
    private var localOptionalCancellable: AnyCancellable?
    private let dispatchQueue = DispatchQueue(label: "LocalQueue-\(UUID())")
    
    func misbehavingFunction() {
        self.dispatchQueue.async {
            self.localOptionalCancellable = Just(())
                .setFailureType(to: Error.self)
                .receive(on: self.dispatchQueue)
                .handleEvents(
                    receiveCancel: {
                        // Simultaneous accesses to 0x600000364e10, but modification requires exclusive access.
                        // Can be fixed by wrapping in self.dispatchQueue.async {}
                        self.localOptionalCancellable = nil
                    }
                )
                .sink(
                    receiveCompletion: { _ in },
                    receiveValue: { _ in
                        self.localOptionalCancellable = nil
                    }
                )
        }
    }
}

Example().misbehavingFunction()

崩溃堆栈追踪 #

Simultaneous accesses to 0x600000364e10, but modification requires exclusive access.
Previous access (a modification) started at  (0x10eeaf12a).
Current access (a modification) started at:
0    libswiftCore.dylib                 0x00007fff2ff7be50 swift_beginAccess + 568
3    Combine                            0x00007fff4ba73a40 Publishers.HandleEvents.Inner.cancel() + 71
4    Combine                            0x00007fff4ba74230 protocol witness for Cancellable.cancel() in conformance Publishers.HandleEvents<A>.Inner<A1> + 16
5    Combine                            0x00007fff4b9f10c0 Subscribers.Sink.cancel() + 652
6    Combine                            0x00007fff4b9f1500 protocol witness for Cancellable.cancel() in conformance Subscribers.Sink<A, B> + 16
7    Combine                            0x00007fff4b9dd2d0 AnyCancellable.cancel() + 339
8    Combine                            0x00007fff4b9dd5f0 AnyCancellable.__deallocating_deinit + 9
9    libswiftCore.dylib                 0x00007fff2ff7da20 _swift_release_dealloc + 16
13   Combine                            0x00007fff4b9f0da0 Subscribers.Sink.receive(_:) + 54
14   Combine                            0x00007fff4b9f14c0 protocol witness for Subscriber.receive(_:) in conformance Subscribers.Sink<A, B> + 16
15   Combine                            0x00007fff4ba73ed0 Publishers.HandleEvents.Inner.receive(_:) + 129
16   Combine                            0x00007fff4ba74170 protocol witness for Subscriber.receive(_:) in conformance Publishers.HandleEvents<A>.Inner<A1> + 16
17   Combine                            0x00007fff4ba26440 closure #1 in Publishers.ReceiveOn.Inner.receive(_:) + 167
18   libswiftDispatch.dylib             0x000000010e97cad0 thunk for @escaping @callee_guaranteed () -> () + 14
19   libdispatch.dylib                  0x00007fff20105323 _dispatch_call_block_and_release + 12
20   libdispatch.dylib                  0x00007fff20106500 _dispatch_client_callout + 8
21   libdispatch.dylib                  0x00007fff2010c12e _dispatch_lane_serial_drain + 715
22   libdispatch.dylib                  0x00007fff2010cde1 _dispatch_lane_invoke + 403
23   libdispatch.dylib                  0x00007fff20117269 _dispatch_workloop_worker_thread + 782
24   libsystem_pthread.dylib            0x00007fff6116391b _pthread_wqthread + 290
25   libsystem_pthread.dylib            0x00007fff61162b68 start_wqthread + 15
Fatal access conflict detected.

目前已验证通过在receiveCancel回调内手动将操作派发到串行队列可规避该崩溃，希望明确该场景下receive(on:)未覆盖cancel事件调度的具体原因。

receive(on:)的调度能力在Combine设计中是有明确边界的，它仅拦截从上游流向下游的两类事件：上游发布的Output值、上游发送的完成/失败事件。所有沿pipeline反向（从下游指向上游）传递的操作，完全不在它的调度覆盖范围内。

触发链路拆解 #

• receiveCancel回调的触发属于典型的反向流：当AnyCancellable被置空、或sink收到事件后主动取消订阅时，cancel指令会从订阅端（下游）沿着pipeline逐节点向上传递，这个过程会直接调用对应节点的cancel()方法，完全绕开receive(on:)的调度逻辑，不会被重新派发到你指定的队列。
• 结合崩溃栈可以看到，Publishers.HandleEvents.Inner.cancel()的调用是直接从下游Sink的cancel逻辑传递过来的，没有经过任何队列包装，自然不会被receive(on:)调度到指定队列执行。

崩溃本质说明 #

代码中receiveValue回调里执行self.localOptionalCancellable = nil的操作，会立刻触发原有AnyCancellable的deinit，进而向上传递cancel指令，触发receiveCancel回调。此时localOptionalCancellable正处于被外层赋值/置空的Swift独占访问状态，再嵌套执行置空操作就触发了编译器的独占访问检查崩溃。

补充认知 #

除了cancel事件，receive(on:)也不会调度下游向上游传递的request(_:) 请求事件，以及handleEvents里的receiveRequest回调。所有反向传递的pipeline操作，都不会被receive(on:)的调度逻辑拦截。

内容的提问来源于stack exchange，提问作者Dracula