Ensure clocks properly respect leeways and use raw time for calculations for continuous clocks (#58396)

* Ensure clocks properly respect leeways and use raw time for calculations for continuous clocks

* slurp up time calcualtions to inline forms

* ensure the tolerance codepaths get tested as well

* Use task local storage for task sources in leeway based calculations

* use comparison assertions for leeway based calculations

* Whitespace cleanup

* Use the CLOCK_MONOTONIC value for linux to be compatible with linux scheduling for dispatch

* remove incorrect paren in macro check for OpenBSD

* Address feedback and remove stray ;

Author: phausler

stdlib/public/Concurrency/Clock.cpp

@@ -37,7 +37,12 @@ void swift_get_time(
       clock_gettime(CLOCK_BOOTTIME, &continuous);
       *seconds = continuous.tv_sec;
       *nanoseconds = continuous.tv_nsec;
-#elif (defined(__APPLE__) || defined(__OpenBSD__)) && HAS_TIME
+#elif defined(__APPLE__) && HAS_TIME
+      struct timespec continuous;
+      clock_gettime(CLOCK_MONOTONIC_RAW, &continuous);
+      *seconds = continuous.tv_sec;
+      *nanoseconds = continuous.tv_nsec;
+#elif defined(__OpenBSD__) && HAS_TIME
       struct timespec continuous;
       clock_gettime(CLOCK_MONOTONIC, &continuous);
       *seconds = continuous.tv_sec;
@@ -63,7 +68,7 @@ void swift_get_time(
     case swift_clock_id_suspending: {
 #if defined(__linux__) && HAS_TIME
       struct timespec suspending;
-      clock_gettime(CLOCK_MONOTONIC_RAW, &suspending);
+      clock_gettime(CLOCK_MONOTONIC, &suspending);
       *seconds = suspending.tv_sec;
       *nanoseconds = suspending.tv_nsec;
 #elif defined(__APPLE__) && HAS_TIME
@@ -111,7 +116,12 @@ switch (clock_id) {
       clock_getres(CLOCK_BOOTTIME, &continuous);
       *seconds = continuous.tv_sec;
       *nanoseconds = continuous.tv_nsec;
-#elif (defined(__APPLE__) || defined(__OpenBSD__)) && HAS_TIME
+#elif defined(__APPLE__) && HAS_TIME
+      struct timespec continuous;
+      clock_getres(CLOCK_MONOTONIC_RAW, &continuous);
+      *seconds = continuous.tv_sec;
+      *nanoseconds = continuous.tv_nsec;
+#elif defined(__OpenBSD__) && HAS_TIME
       struct timespec continuous;
       clock_getres(CLOCK_MONOTONIC, &continuous);
       *seconds = continuous.tv_sec;

stdlib/public/Concurrency/DispatchGlobalExecutor.inc

@@ -24,11 +24,9 @@
 
 #if SWIFT_CONCURRENCY_ENABLE_DISPATCH
 #include <dispatch/dispatch.h>
-
 #if !defined(_WIN32)
 #include <dlfcn.h>
 #endif
-
 #endif
 
 // Ensure that Job's layout is compatible with what Dispatch expects.
@@ -226,6 +224,74 @@ static void swift_task_enqueueGlobalWithDelayImpl(JobDelay delay,
 }
 
 #define DISPATCH_UP_OR_MONOTONIC_TIME_MASK  (1ULL << 63)
+#define DISPATCH_WALLTIME_MASK  (1ULL << 62)
+#define DISPATCH_TIME_MAX_VALUE (DISPATCH_WALLTIME_MASK - 1)
+
+struct __swift_job_source {
+  dispatch_source_t source;
+  Job *job;
+};
+
+static void _swift_run_job_leeway(struct __swift_job_source *jobSource) {
+  dispatch_source_t source = jobSource->source;
+  dispatch_release(source);
+  Job *job = jobSource->job;
+  auto task = dyn_cast<AsyncTask>(job);
+  assert(task && "provided job must be a task");
+  _swift_task_dealloc_specific(task, jobSource);
+  __swift_run_job(job);
+}
+
+#if defined(__i386__) || defined(__x86_64__) || !defined(__APPLE__)
+#define TIME_UNIT_USES_NANOSECONDS 1
+#else
+#define TIME_UNIT_USES_NANOSECONDS 0
+#endif
+
+#if TIME_UNIT_USES_NANOSECONDS
+// x86 currently implements mach time in nanoseconds
+// this is NOT likely to change
+static inline uint64_t
+platform_time(uint64_t nsec) {
+  return nsec;
+}
+#else
+#define DISPATCH_USE_HOST_TIME 1
+#if defined(__APPLE__)
+#if defined(__arm__) || defined(__arm64__)
+// Apple arm platforms currently use a fixed mach timebase of 125/3 (24 MHz)
+static inline uint64_t
+platform_time(uint64_t nsec) {
+  if (!nsec) {
+    return nsec;
+  }
+  if (nsec >= (uint64_t)INT64_MAX) {
+    return INT64_MAX;
+  }
+  if (nsec >= UINT64_MAX / 3ull) {
+    return (nsec / 125ull) * 3ull;
+  } else {
+    return (nsec * 3ull) / 125ull;
+  }
+}
+#endif
+#endif
+#endif
+
+static inline dispatch_time_t
+clock_and_value_to_time(int clock, long long deadline) {
+  uint64_t value = platform_time((uint64_t)deadline);
+  if (value >= DISPATCH_TIME_MAX_VALUE) {
+    return DISPATCH_TIME_FOREVER;
+  }
+  switch (clock) {
+  case swift_clock_id_suspending:
+    return value;
+  case swift_clock_id_continuous:
+    return value | DISPATCH_UP_OR_MONOTONIC_TIME_MASK;
+  }
+  __builtin_unreachable();
+}
 
 SWIFT_CC(swift)
 static void swift_task_enqueueGlobalWithDeadlineImpl(long long sec,
@@ -234,9 +300,7 @@ static void swift_task_enqueueGlobalWithDeadlineImpl(long long sec,
                                                      long long tnsec,
                                                      int clock, Job *job) {
   assert(job && "no job provided");
-
-  dispatch_function_t dispatchFunction = &__swift_run_job;
-  void *dispatchContext = job;
+  auto task = cast<AsyncTask>(job);
 
   JobPriority priority = job->getPriority();
 
@@ -245,20 +309,33 @@ static void swift_task_enqueueGlobalWithDeadlineImpl(long long sec,
   job->SchedulerPrivate[Job::DispatchQueueIndex] =
       DISPATCH_QUEUE_GLOBAL_EXECUTOR;
 
-  long long nowSec;
-  long long nowNsec;
-  swift_get_time(&nowSec, &nowNsec, (swift_clock_id)clock);
+  uint64_t deadline = sec * NSEC_PER_SEC + nsec;
+  dispatch_time_t when = clock_and_value_to_time(clock, deadline);
+  
+  if (tnsec != -1) {
+    uint64_t leeway = tsec * NSEC_PER_SEC + tnsec;
 
-  uint64_t delta = (sec - nowSec) * NSEC_PER_SEC + nsec - nowNsec;
+    dispatch_source_t source = 
+      dispatch_source_create(DISPATCH_SOURCE_TYPE_TIMER, 0, 0, queue);
+    dispatch_source_set_timer(source, when, DISPATCH_TIME_FOREVER, leeway);
 
-  dispatch_time_t when = dispatch_time(DISPATCH_TIME_NOW, delta);
+    size_t sz = sizeof(struct __swift_job_source);
 
-  if (clock == swift_clock_id_continuous) {
-    when |= DISPATCH_UP_OR_MONOTONIC_TIME_MASK;
+    struct __swift_job_source *jobSource = 
+        (struct __swift_job_source *)_swift_task_alloc_specific(task, sz);
+    
+    jobSource->job = job;
+    jobSource->source = source;
+
+    dispatch_set_context(source, jobSource);
+    dispatch_source_set_event_handler_f(source, 
+      (dispatch_function_t)&_swift_run_job_leeway);
+
+    dispatch_activate(source);
+  } else {
+    dispatch_after_f(when, queue, (void *)job, 
+      (dispatch_function_t)&__swift_run_job);
   }
-  // TODO: this should pass the leeway/tolerance along when it is not -1 nanoseconds
-  // either a dispatch_source can be created or a better dispatch_after_f can be made for this
-  dispatch_after_f(when, queue, dispatchContext, dispatchFunction);
 }
 
 SWIFT_CC(swift)

test/Concurrency/Runtime/clock.swift

@@ -17,8 +17,26 @@ var tests = TestSuite("Time")
         try! await clock.sleep(until: .now + .milliseconds(100))
       }
       // give a reasonable range of expected elapsed time
-      expectTrue(elapsed > .milliseconds(90))
-      expectTrue(elapsed < .milliseconds(200))
+      expectGT(elapsed, .milliseconds(90))
+      expectLT(elapsed, .milliseconds(200))
+    }
+
+    tests.test("ContinuousClock sleep with tolerance") {
+      let clock = ContinuousClock()
+      let elapsed = await clock.measure {
+        try! await clock.sleep(until: .now + .milliseconds(100), tolerance: .milliseconds(100))
+      }
+      // give a reasonable range of expected elapsed time
+      expectGT(elapsed, .milliseconds(90))
+      expectLT(elapsed, .milliseconds(300))
+    }
+
+    tests.test("ContinuousClock sleep longer") {
+      let elapsed = await ContinuousClock().measure {
+        try! await Task.sleep(until: .now + .seconds(1), clock: .continuous)
+      }
+      expectGT(elapsed, .seconds(1) - .milliseconds(90))
+      expectLT(elapsed, .seconds(1) + .milliseconds(200))
     }
 
     tests.test("SuspendingClock sleep") {
@@ -27,8 +45,26 @@ var tests = TestSuite("Time")
         try! await clock.sleep(until: .now + .milliseconds(100))
       }
       // give a reasonable range of expected elapsed time
-      expectTrue(elapsed > .milliseconds(90))
-      expectTrue(elapsed < .milliseconds(200))
+      expectGT(elapsed, .milliseconds(90))
+      expectLT(elapsed, .milliseconds(200))
+    }
+
+    tests.test("SuspendingClock sleep with tolerance") {
+      let clock = SuspendingClock()
+      let elapsed = await clock.measure {
+        try! await clock.sleep(until: .now + .milliseconds(100), tolerance: .milliseconds(100))
+      }
+      // give a reasonable range of expected elapsed time
+      expectGT(elapsed, .milliseconds(90))
+      expectLT(elapsed, .milliseconds(300))
+    }
+
+    tests.test("SuspendingClock sleep longer") {
+      let elapsed = await SuspendingClock().measure {
+        try! await Task.sleep(until: .now + .seconds(1), clock: .suspending)
+      }
+      expectGT(elapsed, .seconds(1) - .milliseconds(90))
+      expectLT(elapsed, .seconds(1) + .milliseconds(200))
     }
 
     tests.test("duration addition") {