Merge pull request swiftlang#36216 from apple/tsan-swift-concurrency2

Author: swift-ci

stdlib/public/Concurrency/Actor.cpp

@@ -145,6 +145,8 @@ void swift::swift_job_run(Job *job, ExecutorRef executor) {
 }
 
 void swift::runJobInExecutorContext(Job *job, ExecutorRef executor) {
+  _swift_tsan_acquire(job);
+
   if (auto task = dyn_cast<AsyncTask>(job)) {
     // Update the active task in the current thread.
     ActiveTask::set(task);
@@ -162,6 +164,8 @@ void swift::runJobInExecutorContext(Job *job, ExecutorRef executor) {
     // There's no extra bookkeeping to do for simple jobs.
     job->runSimpleInFullyEstablishedContext(executor);
   }
+
+  _swift_tsan_release(job);
 }
 
 AsyncTask *swift::swift_task_getCurrent() {
@@ -1435,6 +1439,8 @@ void swift::swift_task_switch(AsyncTask *task, ExecutorRef currentExecutor,
 void swift::swift_task_enqueue(Job *job, ExecutorRef executor) {
   assert(job && "no job provided");
 
+  _swift_tsan_release(job);
+
   if (executor.isGeneric())
     return swift_task_enqueueGlobal(job);
 

stdlib/public/Concurrency/CMakeLists.txt

@@ -65,6 +65,7 @@ add_swift_target_library(swift_Concurrency ${SWIFT_STDLIB_LIBRARY_BUILD_TYPES} I
   TaskGroup.swift
   TaskLocal.cpp
   TaskLocal.swift
+  ThreadSanitizer.cpp
   Mutex.cpp
   ${swift_concurrency_objc_sources}
 

stdlib/public/Concurrency/Task.cpp

@@ -64,10 +64,12 @@ FutureFragment::Status AsyncTask::waitFuture(AsyncTask *waitingTask) {
     switch (queueHead.getStatus()) {
     case Status::Error:
     case Status::Success:
+      _swift_tsan_acquire(static_cast<Job *>(this));
       // The task is done; we don't need to wait.
       return queueHead.getStatus();
 
     case Status::Executing:
+      _swift_tsan_release(static_cast<Job *>(waitingTask));
       // Task is now complete. We'll need to add ourselves to the queue.
       break;
     }
@@ -102,6 +104,8 @@ void AsyncTask::completeFuture(AsyncContext *context, ExecutorRef executor) {
     hadErrorResult = true;
   }
 
+  _swift_tsan_release(static_cast<Job *>(this));
+
   // Update the status to signal completion.
   auto newQueueHead = WaitQueueItem::get(
     hadErrorResult ? Status::Error : Status::Success,
@@ -135,6 +139,8 @@ void AsyncTask::completeFuture(AsyncContext *context, ExecutorRef executor) {
       waitingContext->fillWithSuccess(fragment);
     }
 
+    _swift_tsan_acquire(static_cast<Job *>(waitingTask));
+
     // Enqueue the waiter on the global executor.
     // TODO: allow waiters to fill in a suggested executor
     swift_task_enqueueGlobal(waitingTask);

stdlib/public/Concurrency/TaskPrivate.h

@@ -56,6 +56,11 @@ void donateThreadToGlobalExecutorUntil(bool (*condition)(void*),
                                        void *context);
 #endif
 
+/// release() establishes a happens-before relation with a preceding acquire()
+/// on the same address.
+void _swift_tsan_acquire(void *addr);
+void _swift_tsan_release(void *addr);
+
 // ==== ------------------------------------------------------------------------
 
 namespace {

stdlib/public/Concurrency/ThreadSanitizer.cpp

@@ -0,0 +1,63 @@
+//===--- ThreadSanitizer.cpp - Thread Sanitizer support -------------------===//
+//
+// This source file is part of the Swift.org open source project
+//
+// Copyright (c) 2014 - 2021 Apple Inc. and the Swift project authors
+// Licensed under Apache License v2.0 with Runtime Library Exception
+//
+// See https://swift.org/LICENSE.txt for license information
+// See https://swift.org/CONTRIBUTORS.txt for the list of Swift project authors
+//
+//===----------------------------------------------------------------------===//
+//
+// Thread Sanitizer support for the Swift Task runtime
+//
+//===----------------------------------------------------------------------===//
+
+#include "TaskPrivate.h"
+
+#if defined(_WIN32)
+#define NOMINMAX
+#include <windows.h>
+#else
+#include <dlfcn.h>
+#endif
+
+namespace {
+using TSanFunc = void(void *);
+TSanFunc *tsan_acquire, *tsan_release;
+
+TSanFunc *loadSymbol(const char *name) {
+#if defined(_WIN32)
+  return (TSanFunc *)GetProcAddress(GetModuleHandle(NULL), name);
+#else
+  return (TSanFunc *)dlsym(RTLD_DEFAULT, name);
+#endif
+}
+
+swift::swift_once_t initOnceToken;
+void initializeThreadSanitizer(void *unused) {
+  tsan_acquire = loadSymbol("__tsan_acquire");
+  tsan_release = loadSymbol("__tsan_release");
+}
+} // anonymous namespace
+
+void swift::_swift_tsan_acquire(void *addr) {
+  swift_once(&initOnceToken, initializeThreadSanitizer, nullptr);
+  if (tsan_acquire) {
+    tsan_acquire(addr);
+  }
+}
+
+void swift::_swift_tsan_release(void *addr) {
+  swift_once(&initOnceToken, initializeThreadSanitizer, nullptr);
+  if (tsan_release) {
+    tsan_release(addr);
+  }
+}
+
+SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(c)
+void swift_task_set_tsan_hooks(TSanFunc *acquire, TSanFunc *release) {
+  tsan_acquire = acquire;
+  tsan_release = release;
+}

test/Concurrency/Runtime/actor_counters.swift

@@ -63,6 +63,12 @@ func runTest(numCounters: Int, numWorkers: Int, numIterations: Int) async {
 
 @main struct Main {
   static func main() async {
-    await runTest(numCounters: 10, numWorkers: 100, numIterations: 1000)
+    // Useful for debugging: specify counter/worker/iteration counts
+    let args = CommandLine.arguments
+    let counters = args.count >= 2 ? Int(args[1])! : 10
+    let workers = args.count >= 3 ? Int(args[2])! : 100
+    let iterations = args.count >= 4 ? Int(args[3])! : 1000
+    print("counters: \(counters), workers: \(workers), iterations: \(iterations)")
+    await runTest(numCounters: counters, numWorkers: workers, numIterations: iterations)
   }
 }

test/Sanitizers/tsan/actor_counters.swift

@@ -0,0 +1,83 @@
+// RUN: %target-run-simple-swift(-Xfrontend -enable-experimental-concurrency %import-libdispatch -parse-as-library -sanitize=thread)
+
+// REQUIRES: executable_test
+// REQUIRES: concurrency
+// REQUIRES: libdispatch
+// REQUIRES: tsan_runtime
+// UNSUPPORTED: linux
+// UNSUPPORTED: windows
+
+#if canImport(Darwin)
+import Darwin
+#elseif canImport(Glibc)
+import Glibc
+#endif
+
+actor Counter {
+  private var value = 0
+  private let scratchBuffer: UnsafeMutableBufferPointer<Int>
+
+  init(maxCount: Int) {
+    scratchBuffer = .allocate(capacity: maxCount)
+    scratchBuffer.initialize(repeating: 0)
+  }
+
+  func next() -> Int {
+    let current = value
+
+    // Make sure we haven't produced this value before
+    assert(scratchBuffer[current] == 0)
+    scratchBuffer[current] = 1
+
+    value = value + 1
+    return current
+  }
+}
+
+
+func worker(identity: Int, counters: [Counter], numIterations: Int) async {
+  for i in 0..<numIterations {
+    let counterIndex = Int.random(in: 0 ..< counters.count)
+    let counter = counters[counterIndex]
+    let nextValue = await counter.next()
+    print("Worker \(identity) calling counter \(counterIndex) produced \(nextValue)")
+  }
+}
+
+func runTest(numCounters: Int, numWorkers: Int, numIterations: Int) async {
+  // Create counter actors.
+  var counters: [Counter] = []
+  for i in 0..<numCounters {
+    counters.append(Counter(maxCount: numWorkers * numIterations))
+  }
+
+  // Create a bunch of worker threads.
+  var workers: [Task.Handle<Void, Error>] = []
+  for i in 0..<numWorkers {
+    workers.append(
+      Task.runDetached { [counters] in
+        usleep(UInt32.random(in: 0..<100) * 1000)
+        await worker(identity: i, counters: counters, numIterations: numIterations)
+      }
+    )
+  }
+
+  // Wait until all of the workers have finished.
+  for worker in workers {
+    try! await worker.get()
+  }
+
+  print("DONE!")
+}
+
+@main struct Main {
+  static func main() async {
+    // Useful for debugging: specify counter/worker/iteration counts
+    let args = CommandLine.arguments
+    let counters = args.count >= 2 ? Int(args[1])! : 10
+    let workers = args.count >= 3 ? Int(args[2])! : 100
+    let iterations = args.count >= 4 ? Int(args[3])! : 1000
+    print("counters: \(counters), workers: \(workers), iterations: \(iterations)")
+    await runTest(numCounters: counters, numWorkers: workers, numIterations: iterations)
+  }
+}

test/Sanitizers/tsan/async_let_fibonacci.swift

@@ -0,0 +1,56 @@
+// RUN: %target-run-simple-swift(-Xfrontend -enable-experimental-concurrency %import-libdispatch -parse-as-library -sanitize=thread)
+
+// REQUIRES: executable_test
+// REQUIRES: concurrency
+// REQUIRES: libdispatch
+// REQUIRES: tsan_runtime
+
+#if canImport(Darwin)
+import Darwin
+#elseif canImport(Glibc)
+import Glibc
+#endif
+
+func fib(_ n: Int) -> Int {
+  var first = 0
+  var second = 1
+  for _ in 0..<n {
+    let temp = first
+    first = second
+    second = temp + first
+  }
+  return first
+}
+
+func asyncFib(_ n: Int) async -> Int {
+  if n == 0 || n == 1 {
+    return n
+  }
+
+  async let first = await asyncFib(n-2)
+  async let second = await asyncFib(n-1)
+
+  // Sleep a random amount of time waiting on the result producing a result.
+  usleep(UInt32.random(in: 0..<100) * 1000)
+
+  let result = await first + second
+
+  // Sleep a random amount of time before producing a result.
+  usleep(UInt32.random(in: 0..<100) * 1000)
+
+  return result
+}
+
+func runFibonacci(_ n: Int) async {
+  let result = await asyncFib(n)
+
+  print()
+  print("Async fib = \(result), sequential fib = \(fib(n))")
+  assert(result == fib(n))
+}
+
+@main struct Main {
+  static func main() async {
+    await runFibonacci(10)
+  }
+}

test/Sanitizers/tsan/basic_future.swift

@@ -0,0 +1,85 @@
+// RUN: %target-run-simple-swift(-Xfrontend -enable-experimental-concurrency  %import-libdispatch -parse-as-library -sanitize=thread)
+
+// REQUIRES: executable_test
+// REQUIRES: concurrency
+// REQUIRES: libdispatch
+// REQUIRES: tsan_runtime
+
+import Dispatch
+
+#if canImport(Darwin)
+import Darwin
+#elseif canImport(Glibc)
+import Glibc
+#endif
+
+enum HomeworkError: Error, Equatable {
+  case dogAteIt(String)
+}
+
+func formGreeting(name: String) async -> String {
+  return "Hello \(name) from async world"
+}
+
+func testSimple(
+  name: String, dogName: String, shouldThrow: Bool, doSuspend: Bool
+) async {
+  print("Testing name: \(name), dog: \(dogName), shouldThrow: \(shouldThrow) doSuspend: \(doSuspend)")
+
+  var completed = false
+
+  let taskHandle: Task.Handle<String, Error> = Task.runDetached {
+    let greeting = await formGreeting(name: name)
+
+    // If the intent is to test suspending, wait a bit so the second task
+    // can complete.
+    if doSuspend {
+      print("- Future sleeping")
+      sleep(1)
+    }
+
+    if (shouldThrow) {
+      print("- Future throwing")
+      throw HomeworkError.dogAteIt(dogName + " the dog")
+    }
+
+    print("- Future returning normally")
+    return greeting + "!"
+  }
+
+  // If the intent is not to test suspending, wait a bit so the first task
+  // can complete.
+  if !doSuspend {
+    print("+ Reader sleeping")
+    sleep(1)
+  }
+
+  do {
+    print("+ Reader waiting for the result")
+    let result = try await taskHandle.get()
+    completed = true
+    print("+ Normal return: \(result)")
+    assert(result == "Hello \(name) from async world!")
+  } catch HomeworkError.dogAteIt(let badDog) {
+    completed = true
+    print("+ Error return: HomeworkError.dogAteIt(\(badDog))")
+    assert(badDog == dogName + " the dog")
+  } catch {
+    fatalError("Caught a different exception?")
+  }
+
+  assert(completed)
+  print("Finished test")
+}
+
+
+@main struct Main {
+  static func main() async {
+    await testSimple(name: "Ted", dogName: "Hazel", shouldThrow: false, doSuspend: false)
+    await testSimple(name: "Ted", dogName: "Hazel", shouldThrow: true, doSuspend: false)
+    await testSimple(name: "Ted", dogName: "Hazel", shouldThrow: false, doSuspend: true)
+    await testSimple(name: "Ted", dogName: "Hazel", shouldThrow: true, doSuspend: true)
+
+    print("Done")
+  }
+}