[sending] Change CheckedContinuation/AsyncThrowingStream.Continuation APIs to use sending parameters and results.

gottesmm · gottesmm · commit ce1dd4339521 · 2024-06-01T23:25:16.000-07:00
Specifically:

1. CheckedContinuation.resume now takes a sending parameter.
2. Async{Throwing,}Stream.yield takes a sending parameter.
3. withCheckedContinuation returns a transferring parameter.

Importantly due to the previous changes around mangling, this is a mangling
neutral change.

rdar://120420024
diff --git a/lib/AST/ASTMangler.cpp b/lib/AST/ASTMangler.cpp
@@ -2981,7 +2981,7 @@ void ASTMangler::appendFunctionType(AnyFunctionType *fn, GenericSignature sig,
   assert((DWARFMangling || fn->isCanonical()) &&
          "expecting canonical types when not mangling for the debugger");
 
-  appendFunctionSignature(fn, sig, forDecl, NoFunctionMangling);
+  appendFunctionSignature(fn, sig, forDecl, NoFunctionMangling, isRecursedInto);
 
   bool mangleClangType = fn->getASTContext().LangOpts.UseClangFunctionTypes &&
                          fn->hasNonDerivableClangType();
diff --git a/stdlib/public/Concurrency/AsyncStream.swift b/stdlib/public/Concurrency/AsyncStream.swift
@@ -192,7 +192,7 @@ public struct AsyncStream<Element> {
     /// This can be called more than once and returns to the caller immediately
     /// without blocking for any awaiting consumption from the iteration.
     @discardableResult
-    public func yield(_ value: __owned Element) -> YieldResult {
+    public func yield(_ value: sending Element) -> YieldResult {
       storage.yield(value)
     }
 
@@ -422,11 +422,11 @@ extension AsyncStream.Continuation {
   /// blocking for any awaiting consumption from the iteration.
   @discardableResult
   public func yield(
-    with result: Result<Element, Never>
+    with result: __shared sending Result<Element, Never>
   ) -> YieldResult {
     switch result {
-      case .success(let val):
-        return storage.yield(val)
+    case .success(let val):
+      return storage.yield(val)
     }
   }
 
@@ -501,7 +501,7 @@ public struct AsyncStream<Element> {
     @discardableResult
     @available(SwiftStdlib 5.1, *)
     @available(*, unavailable, message: "Unavailable in task-to-thread concurrency model")
-    public func yield(_ value: __owned Element) -> YieldResult {
+    public func yield(_ value: sending Element) -> YieldResult {
       fatalError("Unavailable in task-to-thread concurrency model")
     }
     @available(SwiftStdlib 5.1, *)
@@ -571,7 +571,7 @@ extension AsyncStream.Continuation {
   @available(SwiftStdlib 5.1, *)
   @available(*, unavailable, message: "Unavailable in task-to-thread concurrency model")
   public func yield(
-    with result: Result<Element, Never>
+    with result: __shared sending Result<Element, Never>
   ) -> YieldResult {
     fatalError("Unavailable in task-to-thread concurrency model")
   }
diff --git a/stdlib/public/Concurrency/AsyncThrowingStream.swift b/stdlib/public/Concurrency/AsyncThrowingStream.swift
@@ -214,7 +214,7 @@ public struct AsyncThrowingStream<Element, Failure: Error> {
     /// This can be called more than once and returns to the caller immediately
     /// without blocking for any awaiting consumption from the iteration.
     @discardableResult
-    public func yield(_ value: __owned Element) -> YieldResult {
+    public func yield(_ value: sending Element) -> YieldResult {
       storage.yield(value)
     }
 
@@ -463,7 +463,7 @@ extension AsyncThrowingStream.Continuation {
   /// blocking for any awaiting consumption from the iteration.
   @discardableResult
   public func yield(
-    with result: Result<Element, Failure>
+    with result: __shared sending Result<Element, Failure>
   ) -> YieldResult where Failure == Error {
     switch result {
     case .success(let val):
@@ -547,7 +547,7 @@ public struct AsyncThrowingStream<Element, Failure: Error> {
     @discardableResult
     @available(SwiftStdlib 5.1, *)
     @available(*, unavailable, message: "Unavailable in task-to-thread concurrency model")
-    public func yield(_ value: __owned Element) -> YieldResult {
+    public func yield(_ value: sending Element) -> YieldResult {
       fatalError("Unavailable in task-to-thread concurrency model")
     }
     @available(SwiftStdlib 5.1, *)
@@ -610,7 +610,7 @@ extension AsyncThrowingStream.Continuation {
   @available(SwiftStdlib 5.1, *)
   @available(*, unavailable, message: "Unavailable in task-to-thread concurrency model")
   public func yield(
-    with result: Result<Element, Failure>
+    with result: __shared sending Result<Element, Failure>
   ) -> YieldResult where Failure == Error {
     fatalError("Unavailable in task-to-thread concurrency model")
   }
diff --git a/stdlib/public/Concurrency/CheckedContinuation.swift b/stdlib/public/Concurrency/CheckedContinuation.swift
@@ -159,7 +159,7 @@ public struct CheckedContinuation<T, E: Error>: Sendable {
   /// After `resume` enqueues the task, control immediately returns to
   /// the caller. The task continues executing when its executor is
   /// able to reschedule it.
-  public func resume(returning value: __owned T) {
+  public func resume(returning value: sending T) {
     if let c: UnsafeContinuation<T, E> = canary.takeContinuation() {
       c.resume(returning: value)
     } else {
@@ -206,7 +206,7 @@ extension CheckedContinuation {
   /// the caller. The task continues executing when its executor is
   /// able to reschedule it.
   @_alwaysEmitIntoClient
-  public func resume<Er: Error>(with result: Result<T, Er>) where E == Error {
+  public func resume<Er: Error>(with result: __shared sending Result<T, Er>) where E == Error {
     switch result {
       case .success(let val):
         self.resume(returning: val)
@@ -230,7 +230,7 @@ extension CheckedContinuation {
   /// the caller. The task continues executing when its executor is
   /// able to reschedule it.
   @_alwaysEmitIntoClient
-  public func resume(with result: Result<T, E>) {
+  public func resume(with result: __shared sending Result<T, E>) {
     switch result {
       case .success(let val):
         self.resume(returning: val)
@@ -291,9 +291,11 @@ public func withCheckedContinuation<T>(
   isolation: isolated (any Actor)? = #isolation,
   function: String = #function,
   _ body: (CheckedContinuation<T, Never>) -> Void
-) async -> T {
-  return await withUnsafeContinuation {
-    body(CheckedContinuation(continuation: $0, function: function))
+) async -> sending T {
+  return await Builtin.withUnsafeContinuation {
+    let unsafeContinuation = UnsafeContinuation<T, Never>($0)
+    return body(CheckedContinuation(continuation: unsafeContinuation,
+                                    function: function))
   }
 }
 
@@ -350,9 +352,11 @@ public func withCheckedThrowingContinuation<T>(
   isolation: isolated (any Actor)? = #isolation,
   function: String = #function,
   _ body: (CheckedContinuation<T, Error>) -> Void
-) async throws -> T {
-  return try await withUnsafeThrowingContinuation {
-    body(CheckedContinuation(continuation: $0, function: function))
+) async throws -> sending T {
+  return try await Builtin.withUnsafeThrowingContinuation {
+    let unsafeContinuation = UnsafeContinuation<T, Error>($0)
+    return body(CheckedContinuation(continuation: unsafeContinuation,
+                                    function: function))
   }
 }
 
@@ -393,7 +397,7 @@ internal func _createCheckedThrowingContinuation<T>(
 @_alwaysEmitIntoClient
 internal func _resumeCheckedContinuation<T>(
   _ continuation: CheckedContinuation<T, Never>,
-  _ value: __owned T
+  _ value: sending T
 ) {
   continuation.resume(returning: value)
 }
@@ -402,7 +406,7 @@ internal func _resumeCheckedContinuation<T>(
 @_alwaysEmitIntoClient
 internal func _resumeCheckedThrowingContinuation<T>(
   _ continuation: CheckedContinuation<T, Error>,
-  _ value: __owned T
+  _ value: sending T
 ) {
   continuation.resume(returning: value)
 }
@@ -411,7 +415,7 @@ internal func _resumeCheckedThrowingContinuation<T>(
 @_alwaysEmitIntoClient
 internal func _resumeCheckedThrowingContinuationWithError<T>(
   _ continuation: CheckedContinuation<T, Error>,
-  _ error: __owned Error
+  _ error: consuming Error
 ) {
   continuation.resume(throwing: error)
 }
diff --git a/test/Concurrency/sending_continuation.swift b/test/Concurrency/sending_continuation.swift
@@ -0,0 +1,105 @@
+// RUN: %target-swift-frontend -emit-sil -swift-version 6 -disable-availability-checking -enable-experimental-feature TransferringArgsAndResults -verify %s -o /dev/null -parse-as-library
+
+// REQUIRES: asserts
+// REQUIRES: concurrency
+
+////////////////////////
+// MARK: Declarations //
+////////////////////////
+
+class NonSendableKlass {}
+
+func useValue<T>(_ t: T) {}
+func useValueAsync<T>(_ t: T) async {}
+
+/////////////////
+// MARK: Tests //
+/////////////////
+
+@MainActor
+func withCheckedContinuation_1() async -> NonSendableKlass {
+  await withCheckedContinuation { continuation in
+    continuation.resume(returning: NonSendableKlass())
+  }
+}
+
+@MainActor
+func withCheckedContinuation_2() async -> NonSendableKlass {
+  await withCheckedContinuation { continuation in
+    let x = NonSendableKlass()
+    continuation.resume(returning: x)
+    // expected-error @-1 {{sending 'x' risks causing data races}}
+    // expected-note @-2 {{'x' used after being passed as a 'sending' parameter}}
+    useValue(x) // expected-note {{access can happen concurrently}}
+  }
+}
+
+@MainActor
+func withCheckedContinuation_3() async {
+  // x is main actor isolated since withCheckedContinuation is #isolated.
+  let x = await withCheckedContinuation { continuation in
+    let x = NonSendableKlass()
+    continuation.resume(returning: x)
+    // expected-error @-1 {{sending 'x' risks causing data races}}
+    // expected-note @-2 {{'x' used after being passed as a 'sending' parameter}}
+    useValue(x) // expected-note {{access can happen concurrently}}
+  }
+  await useValueAsync(x)
+  // expected-error @-1 {{sending 'x' risks causing data races}}
+  // expected-note @-2 {{sending main actor-isolated 'x' to nonisolated global function 'useValueAsync' risks causing data races between nonisolated and main actor-isolated uses}}
+}
+
+@MainActor
+func withCheckedContinuation_4() async {
+  // x is main actor isolated since withCheckedContinuation is #isolated.
+  let y = NonSendableKlass()
+  let x = await withCheckedContinuation { continuation in
+    continuation.resume(returning: y)
+    // expected-error @-1 {{sending 'y' risks causing data races}}
+    // expected-note @-2 {{main actor-isolated 'y' is passed as a 'sending' parameter}}
+    useValue(y)
+  }
+  await useValueAsync(x)
+  // expected-error @-1 {{sending 'x' risks causing data races}}
+  // expected-note @-2 {{sending main actor-isolated 'x' to nonisolated global function 'useValueAsync' risks causing data races between nonisolated and main actor-isolated uses}}
+}
+
+@MainActor func testAsyncStream() {
+  let (_, continuation) = AsyncStream.makeStream(of: NonSendableKlass.self)
+
+  continuation.yield(NonSendableKlass())
+  let x = NonSendableKlass()
+  continuation.yield(x) // expected-error {{sending 'x' risks causing data races}}
+  // expected-note @-1 {{'x' used after being passed as a 'sending' parameter; Later uses could race}}
+  useValue(x) // expected-note {{access can happen concurrently}}
+}
+
+@MainActor func testAsyncStreamContinuation() {
+  let _ = AsyncStream(NonSendableKlass.self) { continuation in
+    continuation.yield(NonSendableKlass())
+    let x = NonSendableKlass()
+    continuation.yield(x) // expected-error {{sending 'x' risks causing data races}}
+    // expected-note @-1 {{'x' used after being passed as a 'sending' parameter; Later uses could race}}
+    useValue(x) // expected-note {{access can happen concurrently}}
+  }
+}
+
+@MainActor func testAsyncThrowingStream() {
+  let (_, continuation) = AsyncThrowingStream.makeStream(of: NonSendableKlass.self)
+
+  continuation.yield(NonSendableKlass())
+  let x = NonSendableKlass()
+  continuation.yield(x) // expected-error {{sending 'x' risks causing data races}}
+  // expected-note @-1 {{'x' used after being passed as a 'sending' parameter; Later uses could race}}
+  useValue(x) // expected-note {{access can happen concurrently}}
+}
+
+@MainActor func testAsyncThrowingStreamContinuation() {
+  let _ = AsyncThrowingStream(NonSendableKlass.self) { continuation in
+    continuation.yield(NonSendableKlass())
+    let x = NonSendableKlass()
+    continuation.yield(x) // expected-error {{sending 'x' risks causing data races}}
+    // expected-note @-1 {{'x' used after being passed as a 'sending' parameter; Later uses could race}}
+    useValue(x) // expected-note {{access can happen concurrently}}
+  }
+}
diff --git a/test/Concurrency/sending_mangling.swift b/test/Concurrency/sending_mangling.swift
@@ -36,6 +36,9 @@ func testNoRemoveFunctionResultImmediateTypedFunctionWithArg() -> ((sending NonS
 func testNoRemoveFunctionResultImmedateTypedFunctionWithResult() -> (() -> sending NonSendableKlass) { fatalError() }
 
 struct MethodTest {
+  // CHECK: sil hidden [ossa] @sending_mangling.MethodTest.init(__owned sending_mangling.NonSendableKlass) -> sending_mangling.MethodTest : $@convention(method) (@sil_sending @owned NonSendableKlass, @thin MethodTest.Type) -> MethodTest {
+  init(_ x: sending NonSendableKlass) {}
+
   // CHECK: sil hidden [ossa] @sending_mangling.MethodTest.testMethodRemoveFunctionArg(__owned sending_mangling.NonSendableKlass) -> () : $@convention(method) (@sil_sending @owned NonSendableKlass, MethodTest) -> () {
   func testMethodRemoveFunctionArg(_ x: sending NonSendableKlass) {}
   
