[tsan] Add support for GCD target queues

[kubamracek]

Summary

GCD (libdispatch) has a concept of “target queues”: Each queue has either an implicit or explicit target queue, where the task is handed over to when it’s time to execute it. For example, a concurrent queue can have a serial target queue (effectively making the first queue serial), or multiple queues can have the same serial target queue (which means tasks in all the queues are mutually excluded). Thus we need to acquire-release semantics on the full “chain” of target queues.

This patch changes the way we Acquire() and Release() when executing tasks in queues. Now we’ll walk the chain of target queues and synchronize on each queue that is serial (or when dealing with a barrier block). This should avoid false positives when using dispatch_set_target_queue().

[dvyukov]

LGTM with nits

lib/tsan/rtl/tsan_libdispatch_mac.cc

112 ↗

(On Diff #75323)

Move this out of the loop, no need to execute acquire on the same addr for each queue.

126 ↗

(On Diff #75323)

Move this out of the loop, no need to execute release on the same addr for each queue.

compiler-rt/trunk/lib/tsan/rtl/tsan_libdispatch_mac.cc

 
 static bool IsQueueSerial(dispatch_queue_t q) {
   CHECK_EQ(dispatch_queue_offsets.dqo_width_size, 2);
   uptr width = *(uint16_t *)(((uptr)q) + dispatch_queue_offsets.dqo_width);
   CHECK_NE(width, 0);
   return width == 1;
 }
 
-static dispatch_queue_t GetTargetQueueFromSource(dispatch_source_t source) {
+static dispatch_queue_t GetTargetQueueFromQueue(dispatch_queue_t q) {
   CHECK_EQ(dispatch_queue_offsets.dqo_target_queue_size, 8);
-  dispatch_queue_t target_queue =
-      *(dispatch_queue_t *)(((uptr)source) +
-                            dispatch_queue_offsets.dqo_target_queue);
-  CHECK_NE(target_queue, 0);
-  return target_queue;
+  dispatch_queue_t tq = *(
+      dispatch_queue_t *)(((uptr)q) + dispatch_queue_offsets.dqo_target_queue);
+  return tq;
+}
+
+static dispatch_queue_t GetTargetQueueFromSource(dispatch_source_t source) {
+  dispatch_queue_t tq = GetTargetQueueFromQueue((dispatch_queue_t)source);
+  CHECK_NE(tq, 0);
+  return tq;
 }
 
 static tsan_block_context_t *AllocContext(ThreadState *thr, uptr pc,
                                           dispatch_queue_t queue,
                                           void *orig_context,
                                           dispatch_function_t orig_work) {
   tsan_block_context_t *new_context =
       (tsan_block_context_t *)user_alloc(thr, pc, sizeof(tsan_block_context_t));
   new_context->queue = queue;
   new_context->orig_context = orig_context;
   new_context->orig_work = orig_work;
   new_context->free_context_in_callback = true;
   new_context->submitted_synchronously = false;
   new_context->is_barrier_block = false;
   return new_context;
 }
 
-static void dispatch_callback_wrap(void *param) {
-  SCOPED_INTERCEPTOR_RAW(dispatch_callback_wrap);
-  tsan_block_context_t *context = (tsan_block_context_t *)param;
-  bool is_queue_serial = context->queue && IsQueueSerial(context->queue);
-  uptr sync_ptr = (uptr)context->queue ?: context->non_queue_sync_object;
+#define GET_QUEUE_SYNC_VARS(context, q)                      \
+  bool is_queue_serial = q && IsQueueSerial(q);              \
+  uptr sync_ptr = (uptr)q ?: context->non_queue_sync_object; \
+  uptr serial_sync = (uptr)sync_ptr;                         \
+  uptr concurrent_sync = ((uptr)sync_ptr) + sizeof(uptr);    \
+  bool serial_task = context->is_barrier_block || is_queue_serial
 
-  uptr serial_sync = (uptr)sync_ptr;
-  uptr concurrent_sync = ((uptr)sync_ptr) + sizeof(uptr);
+static void dispatch_sync_pre_execute(ThreadState *thr, uptr pc,
+                                      tsan_block_context_t *context) {
   uptr submit_sync = (uptr)context;
-  bool serial_task = context->is_barrier_block || is_queue_serial;
-
   Acquire(thr, pc, submit_sync);
+
+  dispatch_queue_t q = context->queue;
+  do {
+    GET_QUEUE_SYNC_VARS(context, q);
     Acquire(thr, pc, serial_sync);
     if (serial_task) Acquire(thr, pc, concurrent_sync);
 
+    if (q) q = GetTargetQueueFromQueue(q);
+  } while (q);
+}
+
+static void dispatch_sync_post_execute(ThreadState *thr, uptr pc,
+                                       tsan_block_context_t *context) {
+  uptr submit_sync = (uptr)context;
+  if (context->submitted_synchronously) Release(thr, pc, submit_sync);
+
+  dispatch_queue_t q = context->queue;
+  do {
+    GET_QUEUE_SYNC_VARS(context, q);
+    Release(thr, pc, serial_task ? serial_sync : concurrent_sync);
+
+    if (q) q = GetTargetQueueFromQueue(q);
+  } while (q);
+}
+
+static void dispatch_callback_wrap(void *param) {
+  SCOPED_INTERCEPTOR_RAW(dispatch_callback_wrap);
+  tsan_block_context_t *context = (tsan_block_context_t *)param;
+
+  dispatch_sync_pre_execute(thr, pc, context);
+
   SCOPED_TSAN_INTERCEPTOR_USER_CALLBACK_START();
   context->orig_work(context->orig_context);
   SCOPED_TSAN_INTERCEPTOR_USER_CALLBACK_END();
 
-  Release(thr, pc, serial_task ? serial_sync : concurrent_sync);
-  if (context->submitted_synchronously) Release(thr, pc, submit_sync);
+  dispatch_sync_post_execute(thr, pc, context);
 
   if (context->free_context_in_callback) user_free(thr, pc, context);
 }
 
 static void invoke_block(void *param) {
   dispatch_block_t block = (dispatch_block_t)param;
   block();
 }

compiler-rt/trunk/test/tsan/Darwin/gcd-target-queue-norace.mm

+// RUN: %clang_tsan %s -o %t -framework Foundation
+// RUN: %env_tsan_opts=ignore_interceptors_accesses=1 %run %t 2>&1 | FileCheck %s
+
+#import <Foundation/Foundation.h>
+
+long global;
+
+int main(int argc, const char *argv[]) {
+  dispatch_queue_t target_queue = dispatch_queue_create(NULL, DISPATCH_QUEUE_SERIAL);
+  dispatch_queue_t q1 = dispatch_queue_create(NULL, DISPATCH_QUEUE_CONCURRENT);
+  dispatch_queue_t q2 = dispatch_queue_create(NULL, DISPATCH_QUEUE_CONCURRENT);
+  dispatch_set_target_queue(q1, target_queue);
+  dispatch_set_target_queue(q2, target_queue);
+
+  for (int i = 0; i < 100000; i++) {
+    dispatch_async(q1, ^{
+      global++;
+
+      if (global == 200000) {
+        dispatch_sync(dispatch_get_main_queue(), ^{
+          CFRunLoopStop(CFRunLoopGetCurrent());
+        });
+      }
+    });
+    dispatch_async(q2, ^{
+      global++;
+
+      if (global == 200000) {
+        dispatch_sync(dispatch_get_main_queue(), ^{
+          CFRunLoopStop(CFRunLoopGetCurrent());
+        });
+      }
+    });
+  }
+
+  CFRunLoopRun();
+  NSLog(@"Done.");
+  return 0;
+}
+
+// CHECK-NOT: WARNING: ThreadSanitizer