[Concurrency] Updates after second SE pitch.

We no longer attempt to convert timestamps from the passed-in `Clock`
in order to allow any clock to work with any executor.  Instead,
executors that do not recognise a clock should call the `enqueue`
function on that `Clock`, which lets the `Clock` itself decide how
to proceed.

Additionally, rename `SchedulableExecutor` to `SchedulingExecutor`.

Author: al45tair

lib/SILGen/SILGen.cpp

@@ -523,7 +523,36 @@ FuncDecl *SILGenModule::getExit() {
 Type SILGenModule::getConfiguredExecutorFactory() {
   auto &ctx = getASTContext();
 
-  // Look in the main module for a typealias
+  // First look in the @main struct, if any
+  NominalTypeDecl *mainType = ctx.MainModule->getMainTypeDecl();
+  if (mainType) {
+    SmallVector<ValueDecl *, 1> decls;
+    auto identifier = ctx.getIdentifier("DefaultExecutorFactory");
+    mainType->lookupQualified(mainType,
+                              DeclNameRef(identifier),
+                              SourceLoc(),
+                              NL_RemoveNonVisible | NL_RemoveOverridden
+                              | NL_OnlyTypes | NL_ProtocolMembers,
+                              decls);
+    for (auto decl : decls) {
+      TypeDecl *typeDecl = dyn_cast<TypeDecl>(decl);
+      if (typeDecl) {
+        if (auto *nominalDecl = dyn_cast<NominalTypeDecl>(typeDecl)) {
+          return nominalDecl->getDeclaredType();
+        }
+
+        if (isa<AssociatedTypeDecl>(typeDecl)) {
+          // We ignore associatedtype declarations; those with a default will
+          // turn into a `typealias` instead.
+          continue;
+        }
+
+        return typeDecl->getDeclaredInterfaceType();
+      }
+    }
+  }
+
+  // Failing that, look at the top level
   Type factory = ctx.getNamedSwiftType(ctx.MainModule, "DefaultExecutorFactory");
 
   // If we don't find it, fall back to _Concurrency.PlatformExecutorFactory

stdlib/public/Concurrency/CFExecutor.swift

@@ -43,6 +43,7 @@ enum CoreFoundation {
 
 // .. Main Executor ............................................................
 
+/// A CFRunLoop-based main executor (Apple platforms only)
 @available(StdlibDeploymentTarget 6.2, *)
 final class CFMainExecutor: DispatchMainExecutor, @unchecked Sendable {
 
@@ -58,6 +59,7 @@ final class CFMainExecutor: DispatchMainExecutor, @unchecked Sendable {
 
 // .. Task Executor ............................................................
 
+/// A `TaskExecutor` to match `CFMainExecutor` (Apple platforms only)
 @available(StdlibDeploymentTarget 6.2, *)
 final class CFTaskExecutor: DispatchGlobalTaskExecutor,
                             @unchecked Sendable {

stdlib/public/Concurrency/Clock.swift

@@ -43,6 +43,28 @@ public protocol Clock<Duration>: Sendable {
 #endif
 }
 
+#if !SWIFT_STDLIB_TASK_TO_THREAD_MODEL_CONCURRENCY
+extension Clock {
+  // The default implementation works by creating a trampoline and calling
+  // the run() method.
+  @available(StdlibDeploymentTarget 6.2, *)
+  func enqueue(_ job: consuming ExecutorJob,
+               on executor: some Executor,
+               at instant: Instant, tolerance: Duration?) {
+    let trampoline = job.createTrampoline(to: executor)
+    run(trampoline, at: instant, tolerance: tolerance)
+  }
+
+  // Clocks that do not implement run will fatalError() if you try to use
+  // them with an executor that does not understand them.
+  @available(StdlibDeploymentTarget 6.2, *)
+  func run(_ job: consuming ExecutorJob,
+           at instant: Instant, tolerance: Duration?) {
+    fatalError("\(Self.self) does not implement run(_:at:tolerance:).")
+  }
+}
+#endif // !SWIFT_STDLIB_TASK_TO_THREAD_MODEL_CONCURRENCY
+
 @available(StdlibDeploymentTarget 5.7, *)
 extension Clock {
   /// Measure the elapsed time to execute a closure.
@@ -117,6 +139,7 @@ extension Clock {
 enum _ClockID: Int32 {
   case continuous = 1
   case suspending = 2
+  case walltime = 3
 }
 
 @available(StdlibDeploymentTarget 5.7, *)

stdlib/public/Concurrency/ContinuousClock.swift

@@ -201,3 +201,36 @@ extension ContinuousClock.Instant: InstantProtocol {
     rhs.duration(to: lhs)
   }
 }
+
+#if !$Embedded && !SWIFT_STDLIB_TASK_TO_THREAD_MODEL_CONCURRENCY
+@available(StdlibDeploymentTarget 6.2, *)
+extension ContinuousClock {
+
+  public func run(_ job: consuming ExecutorJob,
+                  at instant: Instant,
+                  tolerance: Duration?) {
+    guard let executor = Task<Never,Never>.currentSchedulingExecutor else {
+      fatalError("no scheduling executor is available")
+    }
+
+    executor.enqueue(job, at: instant,
+                     tolerance: tolerance,
+                     clock: self)
+  }
+
+  public func enqueue(_ job: consuming ExecutorJob,
+                      on executor: some Executor,
+                      at instant: Instant,
+                      tolerance: Duration?) {
+    if let schedulingExecutor = executor.asSchedulingExecutor {
+      schedulingExecutor.enqueue(job, at: instant,
+                                 tolerance: tolerance,
+                                 clock: self)
+    } else {
+      let trampoline = job.createTrampoline(to: executor)
+      run(trampoline, at: instant, tolerance: tolerance)
+    }
+  }
+
+}
+#endif

stdlib/public/Concurrency/CooperativeExecutor.swift

@@ -97,16 +97,77 @@ extension ExecutorJob {
   }
 }
 
+#if !$Embedded && !SWIFT_STDLIB_TASK_TO_THREAD_MODEL_CONCURRENCY
+/// A wait queue is a specialised priority queue used to run a timer.
+@available(StdlibDeploymentTarget 6.2, *)
+struct WaitQueue {
+  var queue: PriorityQueue<UnownedJob>
+  var clock: _ClockID
+
+  init(clock: _ClockID) {
+    queue = PriorityQueue(compare: {
+                            ExecutorJob($0).cooperativeExecutorTimestamp
+                              < ExecutorJob($1).cooperativeExecutorTimestamp
+                          })
+    self.clock = clock
+  }
+
+  var currentTime: CooperativeExecutor.Timestamp {
+    var now: CooperativeExecutor.Timestamp = .zero
+    unsafe _getTime(seconds: &now.seconds,
+                    nanoseconds: &now.nanoseconds,
+                    clock: clock.rawValue)
+    return now
+  }
+
+  mutating func enqueue(_ job: consuming ExecutorJob,
+                        after delay: CooperativeExecutor.Duration) {
+    let deadline = currentTime + delay
+    job.setupCooperativeExecutorTimestamp()
+    job.cooperativeExecutorTimestamp = deadline
+    queue.push(UnownedJob(job))
+  }
+
+  mutating func forEachReadyJob(body: (consuming ExecutorJob) -> ()) {
+    let now = currentTime
+    while let job = queue.pop(
+            when: {
+              ExecutorJob($0).cooperativeExecutorTimestamp < now
+            }) {
+      var theJob = ExecutorJob(job)
+      theJob.clearCooperativeExecutorTimestamp()
+      body(theJob)
+    }
+  }
+
+  var timeToNextJob: CooperativeExecutor.Duration? {
+    if let job = queue.top {
+      let deadline = ExecutorJob(job).cooperativeExecutorTimestamp
+      let now = currentTime
+      if deadline > now {
+        return deadline - now
+      } else {
+        return CooperativeExecutor.Duration(seconds: 0, nanoseconds: 0)
+      }
+    }
+    return nil
+  }
+}
+#endif
+
 /// A co-operative executor that can be used as the main executor or as a
 /// task executor.
 @available(StdlibDeploymentTarget 6.2, *)
-class CooperativeExecutor: Executor, @unchecked Sendable {
+final class CooperativeExecutor: Executor, @unchecked Sendable {
   var runQueue: PriorityQueue<UnownedJob>
-  var waitQueue: PriorityQueue<UnownedJob>
+  #if !$Embedded && !SWIFT_STDLIB_TASK_TO_THREAD_MODEL_CONCURRENCY
+  var suspendingWaitQueue = WaitQueue(clock: .suspending)
+  var continuousWaitQueue = WaitQueue(clock: .continuous)
+  #endif
   var shouldStop: Bool = false
 
   /// Internal representation of a duration for CooperativeExecutor
-  struct Duration {
+  struct Duration: Comparable {
     var seconds: Int64
     var nanoseconds: Int64
 
@@ -120,6 +181,16 @@ class CooperativeExecutor: Executor, @unchecked Sendable {
       self.seconds = seconds
       self.nanoseconds = attoseconds / 1_000_000_000
     }
+
+    static func == (lhs: Duration, rhs: Duration) -> Bool {
+      return lhs.seconds == rhs.seconds && lhs.nanoseconds == rhs.nanoseconds
+    }
+    static func < (lhs: Duration, rhs: Duration) -> Bool {
+      return lhs.seconds < rhs.seconds || (
+        lhs.seconds == rhs.seconds
+          && lhs.nanoseconds < rhs.nanoseconds
+      )
+    }
   }
 
   /// Internal representation of a timestamp for CooperativeExecutor
@@ -163,48 +234,53 @@ class CooperativeExecutor: Executor, @unchecked Sendable {
 
   public init() {
     runQueue = PriorityQueue(compare: { $0.priority > $1.priority })
-    waitQueue =
-      PriorityQueue(compare: {
-                      ExecutorJob($0).cooperativeExecutorTimestamp
-                        < ExecutorJob($1).cooperativeExecutorTimestamp
-                    })
   }
 
   public func enqueue(_ job: consuming ExecutorJob) {
     runQueue.push(UnownedJob(job))
   }
 
   public var isMainExecutor: Bool { true }
-
-  public var asSchedulable: any SchedulableExecutor { self }
 }
 
+#if !$Embedded && !SWIFT_STDLIB_TASK_TO_THREAD_MODEL_CONCURRENCY
 @available(StdlibDeploymentTarget 6.2, *)
-extension CooperativeExecutor: SchedulableExecutor {
-  var currentTime: Timestamp {
+extension CooperativeExecutor: SchedulingExecutor {
+
+  public var asScheduling: (any SchedulingExecutor)? {
+    return self
+  }
+
+  func currentTime(clock: _ClockID) -> Timestamp {
     var now: Timestamp = .zero
     unsafe _getTime(seconds: &now.seconds,
                     nanoseconds: &now.nanoseconds,
-                    clock: _ClockID.suspending.rawValue)
+                    clock: clock.rawValue)
     return now
   }
 
   public func enqueue<C: Clock>(_ job: consuming ExecutorJob,
                                 after delay: C.Duration,
                                 tolerance: C.Duration? = nil,
                                 clock: C) {
-    guard let swiftDuration = delay as? Swift.Duration else {
-      fatalError("Unsupported clock")
+    // If it's a clock we know, get the duration to wait
+    if let _ = clock as? ContinuousClock {
+      let continuousDuration = delay as! ContinuousClock.Duration
+      let duration = Duration(from: continuousDuration)
+      continuousWaitQueue.enqueue(job, after: duration)
+    } else if let _ = clock as? SuspendingClock {
+      let suspendingDuration = delay as! SuspendingClock.Duration
+      let duration = Duration(from: suspendingDuration)
+      suspendingWaitQueue.enqueue(job, after: duration)
+    } else {
+      clock.enqueue(job, on: self, at: clock.now.advanced(by: delay),
+                    tolerance: tolerance)
+      return
     }
-
-    let duration = Duration(from: swiftDuration)
-    let deadline = self.currentTime + duration
-
-    job.setupCooperativeExecutorTimestamp()
-    job.cooperativeExecutorTimestamp = deadline
-    waitQueue.push(UnownedJob(job))
   }
+
 }
+#endif
 
 @available(StdlibDeploymentTarget 6.2, *)
 extension CooperativeExecutor: RunLoopExecutor {
@@ -215,36 +291,47 @@ extension CooperativeExecutor: RunLoopExecutor {
   public func runUntil(_ condition: () -> Bool) throws {
     shouldStop = false
     while !shouldStop && !condition() {
-      // Process the timer queue
-      let now = currentTime
-      while let job = waitQueue.pop(when: {
-                                      ExecutorJob($0).cooperativeExecutorTimestamp <= now
-                                    }) {
-        var theJob = ExecutorJob(job)
-        theJob.clearCooperativeExecutorTimestamp()
-        runQueue.push(job)
+      #if !$Embedded && !SWIFT_STDLIB_TASK_TO_THREAD_MODEL_CONCURRENCY
+      // Process the timer queues
+      suspendingWaitQueue.forEachReadyJob {
+        runQueue.push(UnownedJob($0))
+      }
+      continuousWaitQueue.forEachReadyJob {
+        runQueue.push(UnownedJob($0))
       }
+      #endif
 
       // Now run any queued jobs
-      while let job = runQueue.pop() {
+      var runQ = runQueue.take()
+      while let job = runQ.pop() {
         unsafe ExecutorJob(job).runSynchronously(
           on: self.asUnownedSerialExecutor()
         )
       }
 
+      #if !$Embedded && !SWIFT_STDLIB_TASK_TO_THREAD_MODEL_CONCURRENCY
       // Finally, wait until the next deadline
-      if let job = waitQueue.top {
-        let deadline = ExecutorJob(job).cooperativeExecutorTimestamp
-        let now = self.currentTime
-        if deadline > now {
-          let toWait = deadline - now
-          _sleep(seconds: toWait.seconds,
-                 nanoseconds: toWait.nanoseconds)
+      var toWait: Duration? = suspendingWaitQueue.timeToNextJob
+
+      if let continuousToWait = continuousWaitQueue.timeToNextJob {
+        if toWait == nil ||  continuousToWait < toWait! {
+          toWait = continuousToWait
         }
-      } else {
+      }
+
+      if let toWait {
+        _sleep(seconds: toWait.seconds,
+               nanoseconds: toWait.nanoseconds)
+      } else if runQueue.isEmpty {
+        // Stop if no more jobs are available
+        break
+      }
+      #else // $Embedded || SWIFT_STDLIB_TASK_TO_THREAD_MODEL_CONCURRENCY
+      if runQueue.isEmpty {
         // Stop if no more jobs are available
         break
       }
+      #endif
     }
   }