[Concurrency] Move-only Job, in addition to UnownedJob

Author: ktoso

include/swift/ABI/Task.h

@@ -131,6 +131,10 @@ class alignas(2 * alignof(void*)) Job :
     return Flags.getPriority();
   }
 
+  uint32_t getJobId() const {
+    return Id;
+  }
+
   /// Given that we've fully established the job context in the current
   /// thread, actually start running this job.  To establish the context
   /// correctly, call swift_job_run or runJobInExecutorContext.

include/swift/Runtime/Concurrency.h

@@ -741,6 +741,11 @@ bool swift_task_isOnExecutor(
     const Metadata *selfType,
     const SerialExecutorWitnessTable *wtable);
 
+/// Return the 64bit TaskID (if the job is an AsyncTask),
+/// or the 32bits of the job Id otherwise.
+SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swift)
+uint64_t swift_task_getJobTaskId(Job *job);
+
 #if SWIFT_CONCURRENCY_ENABLE_DISPATCH
 
 /// Enqueue the given job on the main executor.

stdlib/cmake/modules/AddSwiftStdlib.cmake

@@ -1801,7 +1801,7 @@ function(add_swift_target_library name)
   if (SWIFTLIB_IS_STDLIB)
     list(APPEND SWIFTLIB_SWIFT_COMPILE_FLAGS "-warn-implicit-overrides")
     list(APPEND SWIFTLIB_SWIFT_COMPILE_FLAGS "-Xfrontend;-enable-ossa-modules")
-    list(APPEND SWIFTLIB_SWIFT_COMPILE_FLAGS "-Xfrontend;-enable-lexical-lifetimes=false")
+    list(APPEND SWIFTLIB_SWIFT_COMPILE_FLAGS "-Xfrontend;-enable-lexical-lifetimes=true") # FIXME: undo this
   endif()
 
   if(NOT SWIFT_BUILD_RUNTIME_WITH_HOST_COMPILER AND NOT BUILD_STANDALONE AND

stdlib/public/BackDeployConcurrency/Executor.swift

@@ -74,6 +74,7 @@ public struct UnownedSerialExecutor: Sendable {
 @_silgen_name("_swift_task_enqueueOnExecutor")
 internal func _enqueueOnExecutor<E>(job: UnownedJob, executor: E)
 where E: SerialExecutor {
+  // TODO: something to do here?
   executor.enqueue(job)
 }
 

stdlib/public/Concurrency/CMakeLists.txt

@@ -148,6 +148,7 @@ add_swift_target_library(swift_Concurrency ${SWIFT_STDLIB_LIBRARY_BUILD_TYPES} I
     ${SWIFT_STANDARD_LIBRARY_SWIFT_FLAGS}
     -parse-stdlib
     -Xfrontend -enable-experimental-concurrency
+    -Xfrontend -enable-experimental-move-only # FIXME: REMOVE THIS
     ${SWIFT_RUNTIME_CONCURRENCY_SWIFT_FLAGS}
   ${swift_concurrency_options}
   INSTALL_IN_COMPONENT ${swift_concurrency_install_component}

stdlib/public/Concurrency/Executor.swift

@@ -15,24 +15,44 @@ import Swift
 /// A service that can execute jobs.
 @available(SwiftStdlib 5.1, *)
 public protocol Executor: AnyObject, Sendable {
+  // This requirement is repeated here as a non-override so that we
+  // get a redundant witness-table entry for it.  This allows us to
+  // avoid drilling down to the base conformance just for the basic
+  // work-scheduling operation.
+  @available(*, deprecated, message: "Implement enqueueJob instead")
   func enqueue(_ job: UnownedJob)
+
+  @available(SwiftStdlib 5.9, *)
+  func enqueueJob(_ job: __owned Job) // FIXME: figure out how to introduce in compatible way
 }
 
 /// A service that executes jobs.
 @available(SwiftStdlib 5.1, *)
 public protocol SerialExecutor: Executor {
-  // This requirement is repeated here as a non-override so that we
-  // get a redundant witness-table entry for it.  This allows us to
-  // avoid drilling down to the base conformance just for the basic
-  // work-scheduling operation.
+
   @_nonoverride
+  @available(*, deprecated, message: "Implement enqueueJob instead")
   func enqueue(_ job: UnownedJob)
 
+  @_nonoverride
+  @available(SwiftStdlib 5.9, *)
+  func enqueueJob(_ job: __owned Job) // FIXME: figure out how to introduce in compatible way
+
   /// Convert this executor value to the optimized form of borrowed
   /// executor references.
   func asUnownedSerialExecutor() -> UnownedSerialExecutor
 }
 
+@available(SwiftStdlib 5.9, *)
+extension Executor {
+  public func enqueue(_ job: UnownedJob) { // FIXME: this is bad; how could we deploy this nicely
+    fatalError("Please implement \(Self.self).enqueueJob(_:)")
+  }
+  public func enqueueJob(_ job: __owned Job) {
+    self.enqueue(UnownedJob(job))
+  }
+}
+
 /// An unowned reference to a serial executor (a `SerialExecutor`
 /// value).
 ///
@@ -81,12 +101,26 @@ public struct UnownedSerialExecutor: Sendable {
 @_silgen_name("swift_task_isOnExecutor")
 public func _taskIsOnExecutor<Executor: SerialExecutor>(_ executor: Executor) -> Bool
 
+/// Primarily a debug utility.
+///
+/// If the passed in Job is a Task, returns the complete 64bit TaskId,
+/// otherwise returns only the job's 32bit Id.
+///
+/// - Returns: the Id stored in this Job or Task, for purposes of debug printing
+@available(SwiftStdlib 5.9, *)
+@_silgen_name("swift_task_getJobTaskId")
+internal func _getJobTaskId(_ job: UnownedJob) -> UInt64
+
 // Used by the concurrency runtime
 @available(SwiftStdlib 5.1, *)
 @_silgen_name("_swift_task_enqueueOnExecutor")
-internal func _enqueueOnExecutor<E>(job: UnownedJob, executor: E)
+internal func _enqueueOnExecutor<E>(job unownedJob: UnownedJob, executor: E)
 where E: SerialExecutor {
-  executor.enqueue(job)
+  if #available(SwiftStdlib 5.9, *) {
+    executor.enqueueJob(Job(context: unownedJob.context))
+  } else {
+    executor.enqueue(unownedJob)
+  }
 }
 
 @available(SwiftStdlib 5.1, *)

stdlib/public/Concurrency/GlobalExecutor.cpp

@@ -150,6 +150,17 @@ bool swift::swift_task_isOnExecutor(HeapObject *executor,
     return swift_task_isOnExecutorImpl(executor, selfType, wtable);
 }
 
+uint64_t swift::swift_task_getJobTaskId(Job *job) {
+  if (auto task = dyn_cast<AsyncTask>(job)) {
+    // TaskID is actually:
+    //   32bits of Job's Id
+    // + 32bits stored in the AsyncTask
+    return task->getTaskId();
+  } else {
+    return job->getJobId();
+  }
+}
+
 /*****************************************************************************/
 /****************************** MAIN EXECUTOR  *******************************/
 /*****************************************************************************/

stdlib/public/Concurrency/PartialAsyncTask.swift

@@ -26,7 +26,18 @@ internal func _swiftJobRun(_ job: UnownedJob,
 @available(SwiftStdlib 5.1, *)
 @frozen
 public struct UnownedJob: Sendable {
-  private var context: Builtin.Job
+  internal var context: Builtin.Job
+
+  @usableFromInline
+  internal init(context: Builtin.Job) {
+    self.context = context
+  }
+
+  @available(SwiftStdlib 5.9, *)
+  @usableFromInline
+  internal init(_ job: __owned Job) {
+    self.context = job.context
+  }
 
   @available(SwiftStdlib 5.9, *)
   public var priority: Priority {
@@ -41,7 +52,7 @@ public struct UnownedJob: Sendable {
   }
 }
 
-@available(SwiftStdlib 5.1, *)
+@available(SwiftStdlib 5.9, *)
 extension UnownedJob {
   public struct Priority {
     public typealias RawValue = UInt8
@@ -55,6 +66,58 @@ extension UnownedJob {
   }
 }
 
+/// A unit of scheduleable work.
+///
+/// Unless you're implementing a scheduler,
+/// you don't generally interact with jobs directly.
+@available(SwiftStdlib 5.9, *)
+@frozen
+@_moveOnly
+public struct Job: Sendable {
+  internal var context: Builtin.Job
+
+  @usableFromInline
+  internal init(context: __owned Builtin.Job) {
+    self.context = context
+  }
+
+  public var priority: UnownedJob.Priority {
+    // let raw = _swift_concurrency_jobPriority(UnownedJob(context: self.context)) // TODO: do we also surface this or the base priority?
+    let raw = _taskCurrentPriority(context as! Builtin.NativeObject)
+    return UnownedJob.Priority(rawValue: raw)
+  }
+
+  // TODO: move only types cannot conform to protocols, so we can't conform to CustomStringConvertible;
+  //       we can still offer a description to be called explicitly though.
+  public var description: String {
+    "\(Self.self)(\(_getJobTaskId(UnownedJob(context: self.context)))"
+  }
+}
+
+@available(SwiftStdlib 5.9, *)
+extension UnownedSerialExecutor {
+  /// Run the job synchronously.
+  ///
+  /// This operation consumes the job.
+  @_alwaysEmitIntoClient
+  @inlinable
+  public func runJobSynchronously(_ job: __owned Job) {
+    UnownedJob(job)._runSynchronously(on: self)
+  }
+}
+
+@available(SwiftStdlib 5.9, *)
+extension SerialExecutor {
+  /// Run the job synchronously.
+  ///
+  /// This operation consumes the job.
+  @_alwaysEmitIntoClient
+  @inlinable
+  public func runJobSynchronously(_ job: __owned Job) {
+    UnownedJob(job)._runSynchronously(on: self.asUnownedSerialExecutor())
+  }
+}
+
 /// A mechanism to interface
 /// between synchronous and asynchronous code,
 /// without correctness checking.

test/Concurrency/Runtime/custom_executors_moveOnly_job.swift

@@ -0,0 +1,62 @@
+// RUN: %target-run-simple-swift( -Xfrontend -enable-experimental-move-only -Xfrontend -disable-availability-checking %import-libdispatch -parse-as-library) | %FileCheck %s
+
+// REQUIRES: concurrency
+// REQUIRES: executable_test
+// UNSUPPORTED: freestanding
+
+// UNSUPPORTED: back_deployment_runtime
+// REQUIRES: concurrency_runtime
+
+final class InlineExecutor: SerialExecutor, CustomStringConvertible {
+
+  public func enqueueJob(_ job: __owned Job) {
+    print("\(self): enqueue (job: \(job.description))")
+    runJobSynchronously(job)
+    print("\(self): after run")
+  }
+
+  public func asUnownedSerialExecutor() -> UnownedSerialExecutor {
+    return UnownedSerialExecutor(ordinary: self)
+  }
+
+  var description: Swift.String {
+    "InlineExecutor()"
+  }
+}
+
+let inlineExecutor = InlineExecutor()
+
+actor Custom {
+  var count = 0
+
+  nonisolated var unownedExecutor: UnownedSerialExecutor {
+    print("custom unownedExecutor")
+    return inlineExecutor.asUnownedSerialExecutor()
+  }
+
+  func report() async {
+    print("custom.count == \(count)")
+    count += 1
+  }
+}
+
+@available(SwiftStdlib 5.1, *)
+@main struct Main {
+  static func main() async {
+    print("begin")
+    let actor = Custom()
+    await actor.report()
+    await actor.report()
+    await actor.report()
+    print("end")
+  }
+}
+
+// CHECK:      begin
+// CHECK-NEXT: custom unownedExecutor
+// CHECK-NEXT: custom.count == 0
+// CHECK-NEXT: custom unownedExecutor
+// CHECK-NEXT: custom.count == 1
+// CHECK-NEXT: custom unownedExecutor
+// CHECK-NEXT: custom.count == 2
+// CHECK-NEXT: end

test/Concurrency/Runtime/custom_executors_priority.swift

@@ -35,7 +35,6 @@ actor Custom {
   }
 }
 
-@available(SwiftStdlib 5.1, *)
 @main struct Main {
   static func main() async {
     print("begin")
@@ -46,6 +45,11 @@ actor Custom {
     await Task() {
       await actor.report()
     }.value
+    await Task(priority: .low) {
+      print("P: \(Task.currentPriority)")
+      print("B: \(Task.currentBasePriority)")
+      await actor.report()
+    }.value
     print("end")
   }
 }

test/Concurrency/Runtime/custom_executors_protocol.swift

@@ -1,4 +1,4 @@
-// RUN: %target-run-simple-swift( -Xfrontend -disable-availability-checking %import-libdispatch -parse-as-library) | %FileCheck %s
+// RUN: %target-run-simple-swift( -Xfrontend -enable-experimental-move-only -Xfrontend -disable-availability-checking %import-libdispatch -parse-as-library) | %FileCheck %s
 
 // REQUIRES: concurrency
 // REQUIRES: executable_test
@@ -42,6 +42,12 @@ final class InlineExecutor: SpecifiedExecutor, Swift.CustomStringConvertible {
     print("\(self): after run")
   }
 
+  public func enqueueJob(_ job: __owned Job) {
+    print("\(self): enqueue")
+    runJobSynchronously(job)
+    print("\(self): after run")
+  }
+
   public func asUnownedSerialExecutor() -> UnownedSerialExecutor {
     return UnownedSerialExecutor(ordinary: self)
   }