tsan: avoid false positives related to epoll

[dvyukov]

Summary

An application can use the mere fact of epoll_wait returning an fd
as synchronization with the write on the fd that triggered the notification.
This pattern come up in an internal networking server (b/229276331).

If an fd is added to epoll, setup a link from the fd to the epoll fd
and use it for synchronization as well.

compiler-rt/lib/tsan/rtl/tsan_fd.h

 void FdClose(ThreadState *thr, uptr pc, int fd, bool write = true);
 void FdFileCreate(ThreadState *thr, uptr pc, int fd);
 void FdDup(ThreadState *thr, uptr pc, int oldfd, int newfd, bool write);
 void FdPipeCreate(ThreadState *thr, uptr pc, int rfd, int wfd);
 void FdEventCreate(ThreadState *thr, uptr pc, int fd);
 void FdSignalCreate(ThreadState *thr, uptr pc, int fd);
 void FdInotifyCreate(ThreadState *thr, uptr pc, int fd);
 void FdPollCreate(ThreadState *thr, uptr pc, int fd);
+void FdPollAdd(ThreadState *thr, uptr pc, int epfd, int fd);
 void FdSocketCreate(ThreadState *thr, uptr pc, int fd);
 void FdSocketAccept(ThreadState *thr, uptr pc, int fd, int newfd);
 void FdSocketConnecting(ThreadState *thr, uptr pc, int fd);
 void FdSocketConnect(ThreadState *thr, uptr pc, int fd);
 bool FdLocation(uptr addr, int *fd, Tid *tid, StackID *stack);
 void FdOnFork(ThreadState *thr, uptr pc);
 
 uptr File2addr(const char *path);
 uptr Dir2addr(const char *path);
 
 }  // namespace __tsan
 
 #endif  // TSAN_INTERFACE_H

compiler-rt/lib/tsan/rtl/tsan_fd.cpp

 const int kTableSize = kTableSizeL1 * kTableSizeL2;
 
 struct FdSync {
   atomic_uint64_t rc;
 };
 
 struct FdDesc {
   FdSync *sync;
+  // This is used to establish write -> epoll_wait synchronization
+  // where epoll_wait receives notification about the write.
+  atomic_uintptr_t aux_sync;  // FdSync*
   Tid creation_tid;
   StackID creation_stack;
 };
 
 struct FdContext {
   atomic_uintptr_t tab[kTableSizeL1];
   // Addresses used for synchronization.
   FdSync globsync;
static void init(ThreadState *thr, uptr pc, int fd, FdSync *s,
     bool write = true) {
   FdDesc *d = fddesc(thr, pc, fd);
   // As a matter of fact, we don't intercept all close calls.
   // See e.g. libc __res_iclose().
   if (d->sync) {
     unref(thr, pc, d->sync);
     d->sync = 0;
   }
+  unref(thr, pc,
+        reinterpret_cast<FdSync *>(
+            atomic_load(&d->aux_sync, memory_order_relaxed)));
+  atomic_store(&d->aux_sync, 0, memory_order_relaxed);
   if (flags()->io_sync == 0) {
     unref(thr, pc, s);
   } else if (flags()->io_sync == 1) {
     d->sync = s;
   } else if (flags()->io_sync == 2) {
     unref(thr, pc, s);
     d->sync = &fdctx.globsync;
   }
void FdRelease(ThreadState *thr, uptr pc, int fd) {
   if (bogusfd(fd))
     return;
   FdDesc *d = fddesc(thr, pc, fd);
   FdSync *s = d->sync;
   DPrintf("#%d: FdRelease(%d) -> %p\n", thr->tid, fd, s);
   MemoryAccess(thr, pc, (uptr)d, 8, kAccessRead);
   if (s)
     Release(thr, pc, (uptr)s);
+  if (uptr aux_sync = atomic_load(&d->aux_sync, memory_order_acquire))
+    Release(thr, pc, aux_sync);
 }
 
 void FdAccess(ThreadState *thr, uptr pc, int fd) {
   DPrintf("#%d: FdAccess(%d)\n", thr->tid, fd);
   if (bogusfd(fd))
     return;
   FdDesc *d = fddesc(thr, pc, fd);
   MemoryAccess(thr, pc, (uptr)d, 8, kAccessRead);
if (!MustIgnoreInterceptor(thr)) {
       }
     }
     // We need to clear it, because if we do not intercept any call out there
     // that creates fd, we will hit false postives.
     MemoryResetRange(thr, pc, (uptr)d, 8);
   }
   unref(thr, pc, d->sync);
   d->sync = 0;
+  unref(thr, pc,
+        reinterpret_cast<FdSync *>(
+            atomic_load(&d->aux_sync, memory_order_relaxed)));
+  atomic_store(&d->aux_sync, 0, memory_order_relaxed);
   d->creation_tid = kInvalidTid;
   d->creation_stack = kInvalidStackID;
 }
 
 void FdFileCreate(ThreadState *thr, uptr pc, int fd) {
   DPrintf("#%d: FdFileCreate(%d)\n", thr->tid, fd);
   if (bogusfd(fd))
     return;

 
 void FdPollCreate(ThreadState *thr, uptr pc, int fd) {
   DPrintf("#%d: FdPollCreate(%d)\n", thr->tid, fd);
   if (bogusfd(fd))
     return;
   init(thr, pc, fd, allocsync(thr, pc));
 }
 
+void FdPollAdd(ThreadState *thr, uptr pc, int epfd, int fd) {
+  DPrintf("#%d: FdPollAdd(%d, %d)\n", thr->tid, epfd, fd);
+  if (bogusfd(epfd) || bogusfd(fd))
+    return;
+  FdDesc *d = fddesc(thr, pc, fd);
+  // Associate fd with epoll fd only once.
+  // While an fd can be associated with multiple epolls at the same time,
+  // or with different epolls during different phases of lifetime,
+  // synchronization semantics (and examples) of this are unclear.
+  // So we don't support this for now.
+  // If we change the association, it will also create lifetime management
+  // problem for FdRelease which accesses the aux_sync.
+  if (atomic_load(&d->aux_sync, memory_order_relaxed))
+    return;
+  FdDesc *epd = fddesc(thr, pc, epfd);
+  FdSync *s = epd->sync;
+  if (!s)
+    return;
+  uptr cmp = 0;
+  if (atomic_compare_exchange_strong(
+          &d->aux_sync, &cmp, reinterpret_cast<uptr>(s), memory_order_release))
+    ref(s);
+}
+
 void FdSocketCreate(ThreadState *thr, uptr pc, int fd) {
   DPrintf("#%d: FdSocketCreate(%d)\n", thr->tid, fd);
   if (bogusfd(fd))
     return;
   // It can be a UDP socket.
   init(thr, pc, fd, &fdctx.socksync);
 }
 

compiler-rt/lib/tsan/rtl/tsan_interceptors_posix.cpp

 }
 
 TSAN_INTERCEPTOR(int, epoll_ctl, int epfd, int op, int fd, void *ev) {
   SCOPED_TSAN_INTERCEPTOR(epoll_ctl, epfd, op, fd, ev);
   if (epfd >= 0)
     FdAccess(thr, pc, epfd);
   if (epfd >= 0 && fd >= 0)
     FdAccess(thr, pc, fd);
-  if (op == EPOLL_CTL_ADD && epfd >= 0)
+  if (op == EPOLL_CTL_ADD && epfd >= 0) {
+    FdPollAdd(thr, pc, epfd, fd);
     FdRelease(thr, pc, epfd);
+  }
   int res = REAL(epoll_ctl)(epfd, op, fd, ev);
   return res;
 }
 
 TSAN_INTERCEPTOR(int, epoll_wait, int epfd, void *ev, int cnt, int timeout) {
   SCOPED_TSAN_INTERCEPTOR(epoll_wait, epfd, ev, cnt, timeout);
   if (epfd >= 0)
     FdAccess(thr, pc, epfd);

compiler-rt/test/tsan/Linux/epoll_norace.cpp

+// RUN: %clangxx_tsan -O1 %s -o %t && %run %t 2>&1 | FileCheck %s
+
+// The test captures what some high-performance networking servers do.
+// One thread writes to an fd, and another just receives an epoll
+// notification about the write to synchronize with the first thread
+// w/o actually reading from the fd.
+
+#include "../test.h"
+#include <errno.h>
+#include <sys/epoll.h>
+#include <sys/eventfd.h>
+
+int main() {
+  int efd = epoll_create(1);
+  if (efd == -1)
+    exit(printf("epoll_create failed: %d\n", errno));
+  int fd = eventfd(0, 0);
+  if (fd == -1)
+    exit(printf("eventfd failed: %d\n", errno));
+  epoll_event event = {.events = EPOLLIN | EPOLLET};
+  if (epoll_ctl(efd, EPOLL_CTL_ADD, fd, &event))
+    exit(printf("epoll_ctl failed: %d\n", errno));
+  pthread_t th;
+  pthread_create(
+      &th, nullptr,
+      +[](void *arg) -> void * {
+        long long to_add = 1;
+        if (write((long)arg, &to_add, sizeof(to_add)) != sizeof(to_add))
+          exit(printf("write failed: %d\n", errno));
+        return nullptr;
+      },
+      (void *)(long)fd);
+  struct epoll_event events[1] = {};
+  if (epoll_wait(efd, events, 1, -1) != 1)
+    exit(printf("epoll_wait failed: %d\n", errno));
+  close(fd);
+  pthread_join(th, nullptr);
+  close(efd);
+  fprintf(stderr, "DONE\n");
+}
+
+// CHECK: DONE