swift - Is CurrentValueSubject thread safe? - Stack Overflow

admin管理员组

文章数量:1406222

I was under the impression that Combine is supposed to be thread safe, even though the documentation is sparse. I noticed that when publishing to a CurrentValueSubject on a thread other than the one it was initialized in, there is a consistent crash on some internal dispatch_assert_queue_fail.

This is happening in a package where StrictConcurrency is enable. If I disable it, the crash go away. Interestingly when I put it back the crash that was consistent doesn't come back until I clear my DerivedData.

Is that expected? What is recommended if I want a thread safe solution to publish values to a Publisher while keeping strict concurrency on ?

Full code:

@preconcurrency import Combine

final class CombineTests: XCTestCase {
  
  @MainActor
  func testExample() throws {
    let exp = expectation(description: "received 1")
    let currentValue = CurrentValueSubject<Int, Never>(0)
    let cancellable = currentValue.eraseToAnyPublisher().sink { value in
      if value == 1 {
        exp.fulfill()
      }
    }
    
    DispatchQueue.global(qos: .background).async {
      currentValue.send(1)
    }
    
    wait(for: [exp], timeout: 1)
    _ = cancellable
  }
}

I was under the impression that Combine is supposed to be thread safe, even though the documentation is sparse. I noticed that when publishing to a CurrentValueSubject on a thread other than the one it was initialized in, there is a consistent crash on some internal dispatch_assert_queue_fail.

This is happening in a package where StrictConcurrency is enable. If I disable it, the crash go away. Interestingly when I put it back the crash that was consistent doesn't come back until I clear my DerivedData.

Is that expected? What is recommended if I want a thread safe solution to publish values to a Publisher while keeping strict concurrency on ?

Full code:

@preconcurrency import Combine

final class CombineTests: XCTestCase {
  
  @MainActor
  func testExample() throws {
    let exp = expectation(description: "received 1")
    let currentValue = CurrentValueSubject<Int, Never>(0)
    let cancellable = currentValue.eraseToAnyPublisher().sink { value in
      if value == 1 {
        exp.fulfill()
      }
    }
    
    DispatchQueue.global(qos: .background).async {
      currentValue.send(1)
    }
    
    wait(for: [exp], timeout: 1)
    _ = cancellable
  }
}

• swift
• thread-safety
• combine
• swift-concurrency

Share Improve this question Follow asked Mar 6 at 6:29 GuigGuig 10.5k88 gold badges7575 silver badges146146 bronze badges 2

• Recommendation is to fed about Combine and use Concurrency. – Cy-4AH Commented Mar 6 at 7:41
• Safe concurrency with locks has its place. There are a lot of things that are lost when suspending the context. – Guig Commented Mar 6 at 17:53

Add a comment  | 

1 Answer 1

Sorted by: Reset to default 3

According to this Swift Forums thread, it is safe to call send from any thread.

What went wrong here is similar to here. sink should have been declared to take a @Sendable closure, since values can be received from any thread. In reality, it isn't declared that way, so in Swift 6 there is a check inserted at the start of the closure to check that the code is still running on the same global actor (the main actor).

To fix this, you can either receive values on DispatchQueue.main:

currentValue.receive(on: DispatchQueue.main).sink { value in ... }

or you can manually mark the closure as @Sendable, so that the same-executor check is not added.

currentValue.sink { @Sendable value in ... }

本文标签： swiftIs CurrentValueSubject thread safeStack Overflow