[Concurrency] Use dispatch cooperative queues when available.

These queues are better suited to the concurrency runtime.

Author: mikeash

include/swift/Runtime/EnvironmentVariables.h

@@ -36,6 +36,10 @@ extern swift::once_t initializeToken;
   }
 #include "../../../stdlib/public/runtime/EnvironmentVariables.def"
 
+// Wrapper around SWIFT_DEBUG_CONCURRENCY_ENABLE_COOPERATIVE_QUEUES that the
+// Concurrency library can call.
+SWIFT_RUNTIME_STDLIB_SPI bool concurrencyEnableCooperativeQueues();
+
 // Wrapper around SWIFT_ENABLE_ASYNC_JOB_DISPATCH_INTEGRATION that the
 // Concurrency library can call.
 SWIFT_RUNTIME_STDLIB_SPI bool concurrencyEnableJobDispatchIntegration();

stdlib/public/Concurrency/DispatchGlobalExecutor.inc

@@ -111,6 +111,8 @@ static constexpr size_t globalQueueCacheCount =
     static_cast<size_t>(JobPriority::UserInteractive) + 1;
 static std::atomic<dispatch_queue_t> globalQueueCache[globalQueueCacheCount];
 
+static constexpr size_t dispatchQueueCooperativeFlag = 4;
+
 #if defined(SWIFT_CONCURRENCY_BACK_DEPLOYMENT) || !defined(__APPLE__)
 extern "C" void dispatch_queue_set_width(dispatch_queue_t dq, long width);
 #endif
@@ -153,8 +155,14 @@ static dispatch_queue_t getGlobalQueue(JobPriority priority) {
   // If we don't have a queue cached for this priority, cache it now. This may
   // race with other threads doing this at the same time for this priority, but
   // that's OK, they'll all end up writing the same value.
-  queue = dispatch_get_global_queue((dispatch_qos_class_t)priority,
-                                    /*flags*/ 0);
+  if (runtime::environment::concurrencyEnableCooperativeQueues())
+    queue = dispatch_get_global_queue((dispatch_qos_class_t)priority,
+                                      dispatchQueueCooperativeFlag);
+  // If dispatch doesn't support dispatchQueueCooperativeFlag, it will return
+  // NULL. Fall back to a standard global queue.
+  if (!queue)
+    queue = dispatch_get_global_queue((dispatch_qos_class_t)priority,
+                                      /*flags*/ 0);
 
   // Unconditionally store it back in the cache. If we raced with another
   // thread, we'll just overwrite the entry with the same value.
@@ -164,6 +172,18 @@ static dispatch_queue_t getGlobalQueue(JobPriority priority) {
   return queue;
 }
 
+// Get a queue suitable for dispatch_after. Use the cooperative queues on OS
+// versions where they work with dispatch_after, and use a standard global
+// queue where cooperative queues don't work.
+static dispatch_queue_t getTimerQueue(JobPriority priority) {
+  // On newer OSes, we can use the cooperative queues.
+  if (__builtin_available(macOS 12.3, iOS 15.4, tvOS 15.4, watchOS 8.5, *))
+    return getGlobalQueue(priority);
+
+  // On older OSes, use a standard global queue.
+  return dispatch_get_global_queue((dispatch_qos_class_t)priority, /*flags*/ 0);
+}
+
 SWIFT_CC(swift)
 static void swift_task_enqueueGlobalImpl(Job *job) {
   assert(job && "no job provided");
@@ -216,7 +236,7 @@ static void swift_task_enqueueGlobalWithDelayImpl(JobDelay delay,
 
   JobPriority priority = job->getPriority();
 
-  auto queue = getGlobalQueue(priority);
+  auto queue = getTimerQueue(priority);
 
   job->SchedulerPrivate[Job::DispatchQueueIndex] =
       DISPATCH_QUEUE_GLOBAL_EXECUTOR;
@@ -306,7 +326,7 @@ static void swift_task_enqueueGlobalWithDeadlineImpl(long long sec,
 
   JobPriority priority = job->getPriority();
 
-  auto queue = getGlobalQueue(priority);
+  auto queue = getTimerQueue(priority);
 
   job->SchedulerPrivate[Job::DispatchQueueIndex] =
       DISPATCH_QUEUE_GLOBAL_EXECUTOR;

stdlib/public/runtime/EnvironmentVariables.cpp

@@ -242,6 +242,12 @@ bool swift_COWChecksEnabled() {
   return runtime::environment::SWIFT_DEBUG_ENABLE_COW_CHECKS();
 }
 
+SWIFT_RUNTIME_STDLIB_SPI bool concurrencyEnableCooperativeQueues() {
+  return runtime::environment::
+      SWIFT_DEBUG_CONCURRENCY_ENABLE_COOPERATIVE_QUEUES();
+}
+
+
 SWIFT_RUNTIME_STDLIB_SPI bool concurrencyEnableJobDispatchIntegration() {
   return runtime::environment::
       SWIFT_ENABLE_ASYNC_JOB_DISPATCH_INTEGRATION();

stdlib/public/runtime/EnvironmentVariables.def

@@ -65,6 +65,9 @@ VARIABLE(SWIFT_DEBUG_ENABLE_SHARED_CACHE_PROTOCOL_CONFORMANCES, bool, true,
 
 #endif
 
+VARIABLE(SWIFT_DEBUG_CONCURRENCY_ENABLE_COOPERATIVE_QUEUES, bool, true,
+         "Enable dispatch cooperative queues in the global executor.")
+
 #ifndef NDEBUG
 
 VARIABLE(SWIFT_DEBUG_RUNTIME_EXCLUSIVITY_LOGGING, bool, false,