[tsan] Add support for GCD IO channels on Darwin

[kubamracek]

Summary

This patch adds interceptors for dispatch_io_*, dispatch_read and dispatch_write functions. This avoids false positives when using GCD IO. Adding several test cases.

[dvyukov]

lib/tsan/rtl/tsan_libdispatch_mac.cc

461

Please outline in a comment here what operations we want to synchronize with what.
It's difficult to restore the intention from there acquires and released on queues and channels scattered across all these functions.

469

dispatch_callback_wrap will Acquire on context address, which is at best wasteful.
What do we want to synchronize with what? Why do we acquire/release on the queue address? Do we want to synchronize all reads and writes? If we want to synchronize submission with callback execution, then existing dispatch_callback_wrap machinery will do what we need in more precise way (using context address).

497

Same here.

528

Documentation says:

The barrier operation applies to the channel’s file descriptor and not to a specific channel. In other words, if multiple channels are associated with the same file descriptor, a barrier operation scheduled on any of the channels acts as a barrier across all of the channels. All previously scheduled operations on any of those channels must complete before the barrier block is executed.

But here you synchronize on channel address. So I guess we will have some false positives.

Do you release on channel in read/write to synchronize with barriers? If so, it seems to be exactly the kind of reader/writer synchronization we added for barriers recently. Can we reuse that machinery? It captures precisely who needs to synchronize with whom.

543

When is the cleanup handler executed? The documentation is quite vague on this. Is it executed when the channel is closed? If so, should we use a more specific address to not interfere with all other operations (e.g. chan+8)?

587

Where is the pairing acquire?

[kubamracek]

lib/tsan/rtl/tsan_libdispatch_mac.cc

469

Yes, I want to synchronize submission with callback execution, plus the usual “within-one-queue” rules. I wanted to avoid malloc’ing the context, but I guess it’s okay here.

528

The io_barrier is a bit more tricky, since it’s not submitted to the same queue as the read&write callbacks, so user’s shouldn’t be using this to synchronize with their code. So I basically only care about synchronizing submission and execution.

543

The cleanup handler is *submitted* (executed asynchronously) when the channel is closed explicitly (dispatch_io_close) or implicitly by releasing all references to the channel (dispatch_release or removing the reference in ARC). The second case isn’t handled by this patch, because I didn’t want to get into the complexity of intercepting dispatch_release.

587

The Acquire is in the cleanup handler in dispatch_io_create_*.

[kubamracek]

lib/tsan/rtl/tsan_libdispatch_mac.cc

528

Ah, so the issue is that we don’t have a dispatch_queue_t parameter here, so we can’t simply use the regular dispatch_callback_wrap...

[dvyukov]

lib/tsan/rtl/tsan_libdispatch_mac.cc

528

dispatch_callback_wrap uses dispatch_queue_t only as synchronization address and to understand if the queue is serial or not.

What about adding is_serial to tsan_block_context_t, and then having 2 AllocContext's: one takes dispatch_queue_t and uses IsQueueSerial(q); second -- just takes uptr addr and bool is_serial from caller?

[kubamracek]

Updating the patch with review comments. Adding more testcases.

[dvyukov]

lib/tsan/rtl/tsan_libdispatch_mac.cc

456

Here is something I am missing.
Is new_context captured by reference or by value? Internet says that __block makes it captured by reference. But then doesn't new_h refer to a dangling stack location? new_context is stack allocated in this function, but new_h will be executed some time later.
And if it is captured by value, then synchronizing on address &new_context is pointless -- closure will have a different address.
What am I missing?

596

Also here: is it captured by value or by reference?
Or such variables somehow magically promoted to heap?

[kubamracek]

lib/tsan/rtl/tsan_libdispatch_mac.cc

456

__block turns the variable into a reference-counted object that is captured by reference. It’s actually allocated as part of the block. When the block is promoted to a heap object (via Block_copy), the variable is promoted as well. Note that &new_context is different before the Block_copy and after it (before it points to the stack, after it points to the heap).

Does this make sense? You can take a look at this via clang -rewrite-objc file.m -o - which will rewrite all Obj-C and all block stuff into pure C/C++ code. But the full mechanism is pretty complex and the generated code is hard to read.

[dvyukov]

lib/tsan/rtl/tsan_libdispatch_mac.cc

456

Okay

458

I wonder if it is possible to make dispatch_callback_wrap forward any number of arguments with some vaargs template magic. Then we could write just:
REAL(dispatch_read)(fd, length, q, dispatch_callback_wrap<dispatch_data_t, int>);
Up to you. No need to do in this change.

[kubamracek]

Thanks for the review!

lib/tsan/rtl/tsan_libdispatch_mac.cc

TSAN_INTERCEPTOR(void, dispatch_apply_f, size_t iterations,
                  void (*work)(void *, size_t)) {
   SCOPED_TSAN_INTERCEPTOR(dispatch_apply_f, iterations, queue, context, work);
   void (^new_block)(size_t) = ^(size_t iteration) {
     work(context, iteration);
   };
   WRAP(dispatch_apply)(iterations, queue, new_block);
 }
 
+typedef void (^fd_handler_t)(dispatch_data_t data, int error);
+typedef void (^cleanup_handler_t)(int error);
+
+#define DISPATCH_CALLBACK_WITH_BLOCK(q, b)                              \
+  dispatch_block_t new_block = ^(void) {                                \
+    b;                                                                  \
+  };                                                                    \
+  tsan_block_context_t ctx = { q, new_block, &invoke_and_release_block, \
+      false, false, false};                                             \
+  dispatch_callback_wrap(&ctx)
+
+TSAN_INTERCEPTOR(void, dispatch_read, dispatch_fd_t fd, size_t length,
+                 dispatch_queue_t q, fd_handler_t h) {
+  SCOPED_TSAN_INTERCEPTOR(dispatch_read, fd, length, q, h);
+  fd_handler_t new_h = ^(dispatch_data_t data, int error) {
+    {
+      SCOPED_INTERCEPTOR_RAW(dispatch_read_callback);
+      Acquire(thr, pc, (uptr)q);
+    }
+    DISPATCH_CALLBACK_WITH_BLOCK(q, h(data, error));
+  };
+  Release(thr, pc, (uptr)q);
+  REAL(dispatch_read)(fd, length, q, new_h);
+}
+
+TSAN_INTERCEPTOR(void, dispatch_write, dispatch_fd_t fd, dispatch_data_t data,
+                 dispatch_queue_t q, fd_handler_t h) {
+  SCOPED_TSAN_INTERCEPTOR(dispatch_write, fd, data, q, h);
+  fd_handler_t new_h = ^(dispatch_data_t data, int error) {
+    {
+      SCOPED_INTERCEPTOR_RAW(dispatch_write_callback);
+      Acquire(thr, pc, (uptr)q);
+    }
+    DISPATCH_CALLBACK_WITH_BLOCK(q, h(data, error));
+  };
+  Release(thr, pc, (uptr)q);
+  REAL(dispatch_write)(fd, data, q, new_h);
+}
+
+TSAN_INTERCEPTOR(void, dispatch_io_read, dispatch_io_t channel, off_t offset,
+                 size_t length, dispatch_queue_t q, dispatch_io_handler_t h) {
+  SCOPED_TSAN_INTERCEPTOR(dispatch_io_read, channel, offset, length, q, h);
+  dispatch_io_handler_t new_h = ^(bool done, dispatch_data_t data, int error) {
+    {
+      SCOPED_INTERCEPTOR_RAW(dispatch_io_read_callback);
+      Acquire(thr, pc, (uptr)channel);
+    }
+    DISPATCH_CALLBACK_WITH_BLOCK(q, h(done, data, error));
+  };
+  Release(thr, pc, (uptr)channel);
+  REAL(dispatch_io_read)(channel, offset, length, q, new_h);
+}
+
+TSAN_INTERCEPTOR(void, dispatch_io_write, dispatch_io_t channel, off_t offset,
+                 dispatch_data_t data, dispatch_queue_t q,
+                 dispatch_io_handler_t h) {
+  SCOPED_TSAN_INTERCEPTOR(dispatch_io_write, channel, offset, data, q, h);
+  dispatch_io_handler_t new_h = ^(bool done, dispatch_data_t data, int error) {
+    {
+      SCOPED_INTERCEPTOR_RAW(dispatch_io_write_callback);
+      Acquire(thr, pc, (uptr)channel);
+    }
+    DISPATCH_CALLBACK_WITH_BLOCK(q, h(done, data, error));
+  };
+  Release(thr, pc, (uptr)channel);
+  REAL(dispatch_io_write)(channel, offset, data, q, new_h);
+}
+
+TSAN_INTERCEPTOR(void, dispatch_io_barrier, dispatch_io_t channel,
+                 dispatch_block_t barrier) {
+  SCOPED_TSAN_INTERCEPTOR(dispatch_io_barrier, channel, barrier);
+  dispatch_block_t new_block = ^(void) {
+    {
+      SCOPED_INTERCEPTOR_RAW(dispatch_io_barrier_callback);
+      Acquire(thr, pc, (uptr)channel);
+    }
+    barrier();
+  };
+  Release(thr, pc, (uptr)channel);
+  REAL(dispatch_io_barrier)(channel, new_block);
+}
+
+TSAN_INTERCEPTOR(dispatch_io_t, dispatch_io_create, dispatch_io_type_t type,
+                 dispatch_fd_t fd, dispatch_queue_t q, cleanup_handler_t h) {
+  SCOPED_TSAN_INTERCEPTOR(dispatch_io_create, type, fd, q, h);
+  __block dispatch_io_t new_channel = nullptr;
+  cleanup_handler_t new_h = ^(int error) {
+    {
+      SCOPED_INTERCEPTOR_RAW(dispatch_io_create_callback);
+      Acquire(thr, pc, (uptr)new_channel);
+    }
+    DISPATCH_CALLBACK_WITH_BLOCK(q, h(error));
+  };
+  Release(thr, pc, (uptr)q);
+  new_channel = REAL(dispatch_io_create)(type, fd, q, new_h);
+  return new_channel;
+}
+
+TSAN_INTERCEPTOR(dispatch_io_t, dispatch_io_create_with_path,
+                 dispatch_io_type_t type, const char *path, int oflag,
+                 mode_t mode, dispatch_queue_t q, cleanup_handler_t h) {
+  SCOPED_TSAN_INTERCEPTOR(dispatch_io_create_with_path, type, path, oflag, mode,
+                          q, h);
+  __block dispatch_io_t new_channel = nullptr;
+  cleanup_handler_t new_h = ^(int error) {
+    {
+      SCOPED_INTERCEPTOR_RAW(dispatch_io_create_with_path_callback);
+      Acquire(thr, pc, (uptr)new_channel);
+    }
+    DISPATCH_CALLBACK_WITH_BLOCK(q, h(error));
+  };
+  Release(thr, pc, (uptr)q);
+  new_channel =
+      REAL(dispatch_io_create_with_path)(type, path, oflag, mode, q, new_h);
+  return new_channel;
+}
+
+TSAN_INTERCEPTOR(dispatch_io_t, dispatch_io_create_with_io,
+                 dispatch_io_type_t type, dispatch_io_t io, dispatch_queue_t q,
+                 cleanup_handler_t h) {
+  SCOPED_TSAN_INTERCEPTOR(dispatch_io_create_with_io, type, io, q, h);
+  __block dispatch_io_t new_channel = nullptr;
+  cleanup_handler_t new_h = ^(int error) {
+    {
+      SCOPED_INTERCEPTOR_RAW(dispatch_io_create_with_io_callback);
+      Acquire(thr, pc, (uptr)new_channel);
+    }
+    DISPATCH_CALLBACK_WITH_BLOCK(q, h(error));
+  };
+  Release(thr, pc, (uptr)q);
+  new_channel = REAL(dispatch_io_create_with_io)(type, io, q, new_h);
+  return new_channel;
+}
+
+TSAN_INTERCEPTOR(void, dispatch_io_close, dispatch_io_t channel,
+                 dispatch_io_close_flags_t flags) {
+  SCOPED_TSAN_INTERCEPTOR(dispatch_io_close, channel, flags);
+  Release(thr, pc, (uptr)channel);
+  return REAL(dispatch_io_close)(channel, flags);
+}
+
 }  // namespace __tsan
 
 #endif  // SANITIZER_MAC

test/tsan/Darwin/gcd-fd.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 my_global = 0;
+
+int main(int argc, const char *argv[]) {
+  fprintf(stderr, "Hello world.\n");
+
+  dispatch_queue_t queue = dispatch_queue_create("my.queue", DISPATCH_QUEUE_SERIAL);
+  dispatch_semaphore_t sem = dispatch_semaphore_create(0);
+
+  NSString *path = [NSTemporaryDirectory() stringByAppendingPathComponent:[NSString stringWithFormat:@"temp-gcd-io.%d", getpid()]];
+
+  dispatch_io_t channel = dispatch_io_create_with_path(DISPATCH_IO_STREAM, path.fileSystemRepresentation, O_CREAT | O_WRONLY,
+      0666, queue, ^(int error) { });
+  dispatch_io_set_high_water(channel, 1);
+
+  NSData *ns_data = [NSMutableData dataWithLength:1000];
+  dispatch_data_t data = dispatch_data_create(ns_data.bytes, ns_data.length, NULL, DISPATCH_DATA_DESTRUCTOR_DEFAULT);
+
+  my_global++;
+  dispatch_io_write(channel, 0, data, queue, ^(bool done, dispatch_data_t remainingData, int error) {
+    my_global++;
+    dispatch_async(queue, ^{
+      my_global++;
+      if (done) {
+        dispatch_semaphore_signal(sem);
+      }
+    });
+  });
+
+  dispatch_semaphore_wait(sem, DISPATCH_TIME_FOREVER);
+  my_global++;
+  dispatch_io_close(channel, 0);
+  channel = dispatch_io_create_with_path(DISPATCH_IO_STREAM, path.fileSystemRepresentation, O_RDONLY,
+      0, queue, ^(int error) { });
+  dispatch_io_set_high_water(channel, 1);
+
+  my_global++;
+  dispatch_io_read(channel, 0, SIZE_MAX, queue, ^(bool done, dispatch_data_t remainingData, int error) {
+    my_global++;
+    dispatch_async(queue, ^{
+      my_global++;
+      if (done) {
+        dispatch_semaphore_signal(sem);
+      }
+    });
+  });
+
+  dispatch_semaphore_wait(sem, DISPATCH_TIME_FOREVER);
+  my_global++;
+  fprintf(stderr, "Done.\n");
+  return 0;
+}
+
+// CHECK: Hello world.
+// CHECK-NOT: WARNING: ThreadSanitizer
+// CHECK: Done.

test/tsan/Darwin/gcd-io-cleanup.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 my_global = 0;
+
+int main(int argc, const char *argv[]) {
+  fprintf(stderr, "Hello world.\n");
+  
+  dispatch_queue_t queue = dispatch_queue_create("my.queue", DISPATCH_QUEUE_CONCURRENT);
+  dispatch_semaphore_t sem = dispatch_semaphore_create(0);
+  NSString *ns_path = [NSTemporaryDirectory() stringByAppendingPathComponent:[NSString stringWithFormat:@"temp-gcd-io.%d", getpid()]];
+  const char *path = ns_path.fileSystemRepresentation;
+  dispatch_io_t channel;
+  
+  dispatch_fd_t fd = open(path, O_CREAT | O_WRONLY, 0666);
+  my_global++;
+  channel = dispatch_io_create(DISPATCH_IO_STREAM, fd, queue, ^(int error) {
+    my_global++;
+    dispatch_semaphore_signal(sem);
+  });
+  if (! channel) abort();
+  my_global++;
+  dispatch_io_close(channel, 0);
+  dispatch_semaphore_wait(sem, DISPATCH_TIME_FOREVER);
+  
+  my_global++;
+  channel = dispatch_io_create_with_path(DISPATCH_IO_STREAM, path, O_CREAT | O_WRONLY, 0666, queue, ^(int error) {
+    my_global++;
+    dispatch_semaphore_signal(sem);
+  });
+  if (! channel) abort();
+  my_global++;
+  dispatch_io_close(channel, 0);
+  dispatch_semaphore_wait(sem, DISPATCH_TIME_FOREVER);
+  
+  my_global++;
+  dispatch_io_t other_channel = dispatch_io_create_with_path(DISPATCH_IO_STREAM, path, O_CREAT | O_WRONLY, 0666, queue, ^(int error) { });
+  channel = dispatch_io_create_with_io(DISPATCH_IO_STREAM, other_channel, queue, ^(int error) {
+    my_global++;
+    dispatch_semaphore_signal(sem);
+  });
+  if (! channel) abort();
+  my_global++;
+  dispatch_io_close(channel, 0);
+  dispatch_io_close(other_channel, 0);
+  dispatch_semaphore_wait(sem, DISPATCH_TIME_FOREVER);
+  
+  fprintf(stderr, "Done.\n");
+  return 0;
+}
+
+// CHECK: Hello world.
+// CHECK-NOT: WARNING: ThreadSanitizer
+// CHECK: Done.

test/tsan/Darwin/gcd-io-race.mm

+// RUN: %clang_tsan %s -o %t -framework Foundation
+// RUN: %env_tsan_opts=ignore_interceptors_accesses=1 %deflake %run %t 2>&1 | FileCheck %s
+
+#import <Foundation/Foundation.h>
+
+#import "../test.h"
+
+dispatch_queue_t queue;
+dispatch_data_t data;
+dispatch_semaphore_t sem;
+const char *path;
+
+long my_global = 0;
+
+int main(int argc, const char *argv[]) {
+  fprintf(stderr, "Hello world.\n");
+  print_address("addr=", 1, &my_global);
+  barrier_init(&barrier, 2);
+  
+  queue = dispatch_queue_create("my.queue", DISPATCH_QUEUE_CONCURRENT);
+  sem = dispatch_semaphore_create(0);
+  NSString *ns_path = [NSTemporaryDirectory() stringByAppendingPathComponent:[NSString stringWithFormat:@"temp-gcd-io.%d", getpid()]];
+  path = ns_path.fileSystemRepresentation;
+  NSData *ns_data = [NSMutableData dataWithLength:1000];
+  data = dispatch_data_create(ns_data.bytes, ns_data.length, NULL, DISPATCH_DATA_DESTRUCTOR_DEFAULT);
+  
+  dispatch_io_t channel = dispatch_io_create_with_path(DISPATCH_IO_STREAM, path, O_CREAT | O_WRONLY, 0666, queue, ^(int error) { });
+  if (! channel) abort();
+  dispatch_io_set_high_water(channel, 1);
+  
+  dispatch_io_write(channel, 0, data, queue, ^(bool done, dispatch_data_t remainingData, int error) {
+    my_global = 42;
+    barrier_wait(&barrier);
+  });
+
+  dispatch_io_write(channel, 0, data, queue, ^(bool done, dispatch_data_t remainingData, int error) {
+    barrier_wait(&barrier);
+    my_global = 42;
+
+    dispatch_semaphore_signal(sem);
+  });
+  
+  dispatch_semaphore_wait(sem, DISPATCH_TIME_FOREVER);
+  dispatch_io_close(channel, 0);
+  
+  fprintf(stderr, "Done.\n");
+  return 0;
+}
+
+// CHECK: Hello world.
+// CHECK: addr=[[ADDR:0x[0-9,a-f]+]]
+// CHECK: WARNING: ThreadSanitizer: data race
+// CHECK: Location is global 'my_global' {{(of size 8 )?}}at [[ADDR]] (gcd-io-race.mm.tmp+0x{{[0-9,a-f]+}})
+// CHECK: Done.

test/tsan/Darwin/gcd-io.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>
+
+dispatch_queue_t queue;
+dispatch_data_t data;
+dispatch_semaphore_t sem;
+const char *path;
+
+long my_global = 0;
+
+void test_dispatch_io_write() {
+  dispatch_io_t channel = dispatch_io_create_with_path(DISPATCH_IO_STREAM, path, O_CREAT | O_WRONLY, 0666, queue, ^(int error) { });
+  if (! channel) abort();
+  dispatch_io_set_high_water(channel, 1);
+  
+  my_global++;
+  dispatch_io_write(channel, 0, data, queue, ^(bool done, dispatch_data_t remainingData, int error) {
+    if (error) abort();
+    my_global++;
+    dispatch_async(queue, ^{
+      my_global++;
+      if (done) {
+        dispatch_semaphore_signal(sem);
+      }
+    });
+  });
+  
+  dispatch_semaphore_wait(sem, DISPATCH_TIME_FOREVER);
+  my_global++;
+  dispatch_io_close(channel, 0);
+}
+
+void test_dispatch_write() {
+  dispatch_fd_t fd = open(path, O_CREAT | O_WRONLY, 0666);
+  if (fd == -1) abort();
+  
+  my_global++;
+  dispatch_write(fd, data, queue, ^(dispatch_data_t data, int error) {
+    if (error) abort();
+    my_global++;
+    dispatch_async(queue, ^{
+      my_global++;
+      
+      dispatch_semaphore_signal(sem);
+    });
+  });
+  
+  dispatch_semaphore_wait(sem, DISPATCH_TIME_FOREVER);
+  my_global++;
+  close(fd);
+}
+
+void test_dispatch_io_read() {
+  dispatch_io_t channel = dispatch_io_create_with_path(DISPATCH_IO_STREAM, path, O_RDONLY,
+                       0, queue, ^(int error) { });
+  dispatch_io_set_high_water(channel, 1);
+  
+  my_global++;
+  dispatch_io_read(channel, 0, SIZE_MAX, queue, ^(bool done, dispatch_data_t remainingData, int error) {
+    if (error) abort();
+    my_global++;
+    dispatch_async(queue, ^{
+      my_global++;
+      if (done) {
+        dispatch_semaphore_signal(sem);
+      }
+    });
+  });
+  
+  dispatch_semaphore_wait(sem, DISPATCH_TIME_FOREVER);
+  my_global++;
+  dispatch_io_close(channel, 0);
+}
+
+void test_dispatch_read() {
+  dispatch_fd_t fd = open(path, O_RDONLY, 0);
+  if (fd == -1) abort();
+  
+  my_global++;
+  dispatch_read(fd, SIZE_MAX, queue, ^(dispatch_data_t data, int error) {
+    if (error) abort();
+    my_global++;
+    dispatch_async(queue, ^{
+      my_global++;
+      dispatch_semaphore_signal(sem);
+    });
+  });
+  
+  dispatch_semaphore_wait(sem, DISPATCH_TIME_FOREVER);
+  my_global++;
+  close(fd);
+}
+
+int main(int argc, const char *argv[]) {
+  fprintf(stderr, "Hello world.\n");
+  
+  queue = dispatch_queue_create("my.queue", DISPATCH_QUEUE_SERIAL);
+  sem = dispatch_semaphore_create(0);
+  NSString *ns_path = [NSTemporaryDirectory() stringByAppendingPathComponent:[NSString stringWithFormat:@"temp-gcd-io.%d", getpid()]];
+  path = ns_path.fileSystemRepresentation;
+  NSData *ns_data = [NSMutableData dataWithLength:1000];
+  data = dispatch_data_create(ns_data.bytes, ns_data.length, NULL, DISPATCH_DATA_DESTRUCTOR_DEFAULT);
+  
+  test_dispatch_io_write();
+  test_dispatch_write();
+  test_dispatch_io_read();
+  test_dispatch_read();
+  
+  fprintf(stderr, "Done.\n");
+  return 0;
+}
+
+// CHECK: Hello world.
+// CHECK-NOT: WARNING: ThreadSanitizer
+// CHECK: Done.