Detect leaked workers (#3934)

backend.native/tests/build.gradle

@@ -1029,6 +1029,20 @@ task enumIdentity(type: KonanLocalTest) {
     source = "runtime/workers/enum_identity.kt"
 }
 
+standaloneTest("leakWorker") {
+    source = "runtime/workers/leak_worker.kt"
+    flags = ['-g']
+    expectedExitStatusChecker = { it != 0 }
+    outputChecker = { s -> s.contains("Unfinished workers detected, 1 workers leaked!") }
+}
+
+standaloneTest("leakMemoryWithWorkerTermination") {
+    source = "runtime/workers/leak_memory_with_worker_termination.kt"
+    flags = ['-g']
+    expectedExitStatusChecker = { it != 0 }
+    outputChecker = { s -> s.contains("Memory leaks detected, 1 objects leaked!") }
+}
+
 task superFunCall(type: KonanLocalTest) {
     goldValue = "<fun:C><fun:C1>\n<fun:C><fun:C3>\n"
     source = "codegen/basics/superFunCall.kt"
@@ -2734,6 +2748,13 @@ standaloneTest("cycle_collector") {
     source = "runtime/memory/cycle_collector.kt"
 }
 
+standaloneTest("leakMemory") {
+    source = "runtime/memory/leak_memory.kt"
+    flags = ['-g']
+    expectedExitStatusChecker = { it != 0 }
+    outputChecker = { s -> s.contains("Memory leaks detected, 1 objects leaked!") }
+}
+
 standaloneTest("mpp1") {
     source = "codegen/mpp/mpp1.kt"
     flags = ['-tr', '-Xmulti-platform']

backend.native/tests/runtime/memory/leak_memory.kt

@@ -0,0 +1,5 @@
+import kotlinx.cinterop.*
+
+fun main() {
+    StableRef.create(Any())
+}

backend.native/tests/runtime/workers/leak_memory_with_worker_termination.kt

@@ -0,0 +1,10 @@
+import kotlin.native.concurrent.*
+import kotlinx.cinterop.*
+
+fun main() {
+    val worker = Worker.start()
+    // Make sure worker is initialized.
+    worker.execute(TransferMode.SAFE, {}, {}).result;
+    StableRef.create(Any())
+    worker.requestTermination().result
+}

backend.native/tests/runtime/workers/leak_worker.kt

@@ -0,0 +1,9 @@
+import kotlin.native.concurrent.*
+import kotlinx.cinterop.*
+
+fun main() {
+    val worker = Worker.start()
+    // Make sure worker is initialized.
+    worker.execute(TransferMode.SAFE, {}, {}).result;
+    StableRef.create(Any())
+}

build-tools/src/main/kotlin/org/jetbrains/kotlin/KotlinNativeTest.kt

@@ -197,7 +197,10 @@ open class KonanLocalTest : KonanTest() {
     override var testLogger = Logger.SILENT
 
     @Input @Optional
-    var expectedExitStatus = 0
+    var expectedExitStatus: Int? = null
+
+    @Input @Optional
+    var expectedExitStatusChecker: (Int) -> Boolean = { it == (expectedExitStatus ?: 0) }
 
     /**
      * Should this test fail or not.
@@ -266,9 +269,12 @@ open class KonanLocalTest : KonanTest() {
             exitCode + other.exitCode)
 
     private fun ProcessOutput.check() {
-        val exitCodeMismatch = exitCode != expectedExitStatus
+        val exitCodeMismatch = !expectedExitStatusChecker(exitCode)
         if (exitCodeMismatch) {
-            val message = "Expected exit status: $expectedExitStatus, actual: $exitCode"
+            val message = if (expectedExitStatus != null)
+                "Expected exit status: $expectedExitStatus, actual: $exitCode"
+            else
+                "Actual exit status doesn't match with exit status checker: $exitCode"
             check(expectedFail) { """
                     |Test failed. $message
                     |stdout:

runtime/src/launcher/cpp/launcher.cpp

@@ -19,6 +19,7 @@
 #include "Runtime.h"
 #include "KString.h"
 #include "Types.h"
+#include "Worker.h"
 
 #ifndef KONAN_ANDROID
 
@@ -55,7 +56,11 @@ extern "C" RUNTIME_USED int Init_and_run_start(int argc, const char** argv, int
 
   KInt exitStatus = Konan_run_start(argc, argv);
 
-  if (memoryDeInit) Kotlin_deinitRuntimeIfNeeded();
+  if (memoryDeInit) {
+    if (Kotlin_memoryLeakCheckerEnabled())
+      WaitNativeWorkersTermination();
+    Kotlin_deinitRuntimeIfNeeded();
+  }
 
   return exitStatus;
 }

runtime/src/main/cpp/Memory.cpp

@@ -114,7 +114,6 @@ FrameOverlay exportFrameOverlay;
 volatile int allocCount = 0;
 volatile int aliveMemoryStatesCount = 0;
 
-KBoolean g_checkLeaks = KonanNeedDebugInfo;
 KBoolean g_hasCyclicCollector = true;
 
 // TODO: can we pass this variable as an explicit argument?
@@ -1752,11 +1751,13 @@ void deinitMemory(MemoryState* memoryState) {
   atomicAdd(&pendingDeinit, 1);
 #if USE_GC
   bool lastMemoryState = atomicAdd(&aliveMemoryStatesCount, -1) == 0;
-  bool checkLeaks = g_checkLeaks && lastMemoryState;
+  bool checkLeaks = Kotlin_memoryLeakCheckerEnabled() && lastMemoryState;
   if (lastMemoryState) {
    garbageCollect(memoryState, true);
 #if USE_CYCLIC_GC
    // If there are other pending deinits (rare situation) - just skip the leak checker.
+   // This may happen when there're several threads with Kotlin runtimes created
+   // by foreign code, and that code stops those threads simultaneously.
    if (atomicGet(&pendingDeinit) > 1) {
      checkLeaks = false;
    }
@@ -1789,11 +1790,9 @@ void deinitMemory(MemoryState* memoryState) {
 #else
 #if USE_GC
   if (IsStrictMemoryModel && allocCount > 0 && checkLeaks) {
-    char buf[1024];
-    konan::snprintf(buf, sizeof(buf),
+    konan::consoleErrorf(
         "Memory leaks detected, %d objects leaked!\n"
         "Use `Platform.isMemoryLeakCheckerActive = false` to avoid this check.\n", allocCount);
-    konan::consoleErrorUtf8(buf, konan::strnlen(buf, sizeof(buf)));
     konan::abort();
   }
 #endif  // USE_GC
@@ -3010,14 +3009,6 @@ void Kotlin_Any_share(ObjHeader* obj) {
   shareAny(obj);
 }
 
-KBoolean Konan_Platform_getMemoryLeakChecker() {
-  return g_checkLeaks;
-}
-
-void Konan_Platform_setMemoryLeakChecker(KBoolean value) {
-  g_checkLeaks = value;
-}
-
 void AddTLSRecord(MemoryState* memory, void** key, int size) {
   auto* tlsMap = memory->tlsMap;
   auto it = tlsMap->find(key);

runtime/src/main/cpp/Porting.cpp

@@ -115,6 +115,15 @@ void consolePrintf(const char* format, ...) {
   consoleWriteUtf8(buffer, rv);
 }
 
+void consoleErrorf(const char* format, ...) {
+  char buffer[1024];
+  va_list args;
+  va_start(args, format);
+  int rv = vsnprintf_impl(buffer, sizeof(buffer) - 1, format, args);
+  va_end(args);
+  consoleErrorUtf8(buffer, rv);
+}
+
 // Thread execution.
 #if !KONAN_NO_THREADS
 

runtime/src/main/cpp/Porting.h

@@ -27,6 +27,7 @@ namespace konan {
 // Console operations.
 void consoleInit();
 void consolePrintf(const char* format, ...);
+void consoleErrorf(const char* format, ...);
 void consoleWriteUtf8(const void* utf8, uint32_t sizeBytes);
 void consoleErrorUtf8(const void* utf8, uint32_t sizeBytes);
 // Negative return value denotes that read wasn't successful.

runtime/src/main/cpp/Runtime.cpp

@@ -74,6 +74,8 @@ void InitOrDeinitGlobalVariables(int initialize, MemoryState* memory) {
   }
 }
 
+KBoolean g_checkLeaks = KonanNeedDebugInfo;
+
 constexpr RuntimeState* kInvalidRuntime = nullptr;
 
 THREAD_LOCAL_VARIABLE RuntimeState* runtimeState = kInvalidRuntime;
@@ -113,9 +115,11 @@ void deinitRuntime(RuntimeState* state) {
   InitOrDeinitGlobalVariables(DEINIT_THREAD_LOCAL_GLOBALS, state->memoryState);
   if (lastRuntime)
     InitOrDeinitGlobalVariables(DEINIT_GLOBALS, state->memoryState);
+  auto workerId = GetWorkerId(state->worker);
   WorkerDeinit(state->worker);
   DeinitMemory(state->memoryState);
   konanDestructInstance(state);
+  WorkerDestroyThreadDataIfNeeded(workerId);
 }
 
 void Kotlin_deinitRuntimeCallback(void* argument) {
@@ -268,4 +272,16 @@ void Kotlin_zeroOutTLSGlobals() {
     InitOrDeinitGlobalVariables(DEINIT_THREAD_LOCAL_GLOBALS, runtimeState->memoryState);
 }
 
+bool Kotlin_memoryLeakCheckerEnabled() {
+  return g_checkLeaks;
+}
+
+KBoolean Konan_Platform_getMemoryLeakChecker() {
+  return g_checkLeaks;
+}
+
+void Konan_Platform_setMemoryLeakChecker(KBoolean value) {
+  g_checkLeaks = value;
+}
+
 }  // extern "C"

runtime/src/main/cpp/Runtime.h

@@ -53,6 +53,8 @@ void AppendToInitializersTail(struct InitNode*);
 // Zero out all Kotlin thread local globals.
 void Kotlin_zeroOutTLSGlobals();
 
+bool Kotlin_memoryLeakCheckerEnabled();
+
 #ifdef __cplusplus
 }
 #endif

runtime/src/main/cpp/Worker.cpp

@@ -71,6 +71,11 @@ enum JobKind {
   JOB_EXECUTE_AFTER = 3
 };
 
+enum class WorkerKind {
+  kNative,  // Workers created using Worker.start public API.
+  kOther,   // Any other kind of workers.
+};
+
 struct Job {
   enum JobKind kind;
   union {
@@ -106,7 +111,10 @@ typedef KStdOrderedSet<Job, JobCompare> DelayedJobSet;
 
 class Worker {
  public:
-  Worker(KInt id, bool errorReporting, KRef customName) : id_(id), errorReporting_(errorReporting), terminated_(false) {
+  Worker(KInt id, bool errorReporting, KRef customName, WorkerKind kind)
+      : id_(id),
+        kind_(kind),
+        errorReporting_(errorReporting) {
     name_ = customName != nullptr ? CreateStablePointer(customName) : nullptr;
     pthread_mutex_init(&lock_, nullptr);
     pthread_cond_init(&cond_, nullptr);
@@ -114,6 +122,8 @@ class Worker {
 
   ~Worker();
 
+  void startEventLoop();
+
   void putJob(Job job, bool toFront);
   void putDelayedJob(Job job);
 
@@ -135,8 +145,13 @@ class Worker {
 
   KNativePtr name() const { return name_; }
 
+  WorkerKind kind() const { return kind_; }
+
+  pthread_t thread() const { return thread_; }
+
  private:
   KInt id_;
+  WorkerKind kind_;
   KStdDeque<Job> queue_;
   DelayedJobSet delayed_;
   // Stable pointer with worker's name.
@@ -146,7 +161,8 @@ class Worker {
   pthread_cond_t cond_;
   // If errors to be reported on console.
   bool errorReporting_;
-  bool terminated_;
+  bool terminated_ = false;
+  pthread_t thread_ = 0;
 };
 
 #else  // WITH_WORKERS
@@ -240,7 +256,6 @@ class Future {
   pthread_cond_t cond_;
 };
 
-
 class State {
  public:
   State() {
@@ -258,11 +273,11 @@ class State {
     pthread_cond_destroy(&cond_);
   }
 
-  Worker* addWorkerUnlocked(bool errorReporting, KRef customName) {
+  Worker* addWorkerUnlocked(bool errorReporting, KRef customName, WorkerKind kind) {
     Worker* worker = nullptr;
     {
       Locker locker(&lock_);
-      worker = konanConstructInstance<Worker>(nextWorkerId(), errorReporting, customName);
+      worker = konanConstructInstance<Worker>(nextWorkerId(), errorReporting, customName, kind);
       if (worker == nullptr) return nullptr;
       workers_[worker->id()] = worker;
     }
@@ -274,6 +289,10 @@ class State {
     Locker locker(&lock_);
     auto it = workers_.find(id);
     if (it == workers_.end()) return;
+    Worker* worker = it->second;
+    if (worker->kind() == WorkerKind::kNative) {
+      terminating_native_workers_[id] = worker->thread();
+    }
     workers_.erase(it);
   }
 
@@ -432,11 +451,58 @@ class State {
   KInt nextWorkerId() { return currentWorkerId_++; }
   KInt nextFutureId() { return currentFutureId_++; }
 
+  void destroyWorkerThreadDataUnlocked(KInt id) {
+    Locker locker(&lock_);
+    auto it = terminating_native_workers_.find(id);
+    if (it == terminating_native_workers_.end()) return;
+    // If this worker was not joined, detach it to free resources.
+    pthread_detach(it->second);
+    terminating_native_workers_.erase(it);
+  }
+
+  void waitNativeWorkersTerminationUnlocked() {
+    std::vector<pthread_t> threadsToWait;
+    {
+      Locker locker(&lock_);
+
+      checkNativeWorkersLeakLocked();
+
+      for (auto& kvp : terminating_native_workers_) {
+        RuntimeAssert(!pthread_equal(kvp.second, pthread_self()), "Native worker is joining with itself");
+        threadsToWait.push_back(kvp.second);
+      }
+      terminating_native_workers_.clear();
+    }
+
+    for (auto thread : threadsToWait) {
+      pthread_join(thread, nullptr);
+    }
+  }
+
+  void checkNativeWorkersLeakLocked() {
+    size_t remainingNativeWorkers = 0;
+    for (const auto& kvp : workers_) {
+      Worker* worker = kvp.second;
+      if (worker->kind() == WorkerKind::kNative) {
+        ++remainingNativeWorkers;
+      }
+    }
+
+    if (remainingNativeWorkers != 0) {
+      konan::consoleErrorf(
+        "Unfinished workers detected, %lu workers leaked!\n"
+        "Use `Platform.isMemoryLeakCheckerActive = false` to avoid this check.\n",
+        remainingNativeWorkers);
+      konan::abort();
+    }
+  }
+
  private:
   pthread_mutex_t lock_;
   pthread_cond_t cond_;
   KStdUnorderedMap<KInt, Future*> futures_;
   KStdUnorderedMap<KInt, Worker*> workers_;
+  KStdUnorderedMap<KInt, pthread_t> terminating_native_workers_;
   KInt currentWorkerId_;
   KInt currentFutureId_;
   KInt currentVersion_;
@@ -486,28 +552,10 @@ void Future::cancelUnlocked() {
 // Defined in RuntimeUtils.kt.
 extern "C" void ReportUnhandledException(KRef e);
 
-void* workerRoutine(void* argument) {
-  Worker* worker = reinterpret_cast<Worker*>(argument);
-
-  WorkerResume(worker);
-  Kotlin_initRuntimeIfNeeded();
-
-  do {
-    if (worker->processQueueElement(true) == JOB_TERMINATE) break;
-  } while (true);
-
-  // Runtime deinit callback could be called when TLS is already zeroed out, so clear memory
-  // here explicitly. to make sure leak detector properly works.
-  Kotlin_zeroOutTLSGlobals();
-
-  return nullptr;
-}
-
 KInt startWorker(KBoolean errorReporting, KRef customName) {
-  Worker* worker = theState()->addWorkerUnlocked(errorReporting != 0, customName);
+  Worker* worker = theState()->addWorkerUnlocked(errorReporting != 0, customName, WorkerKind::kNative);
   if (worker == nullptr) return -1;
-  pthread_t thread = 0;
-  pthread_create(&thread, nullptr, workerRoutine, worker);
+  worker->startEventLoop();
   return worker->id();
 }
 
@@ -642,10 +690,18 @@ KNativePtr detachObjectGraphInternal(KInt transferMode, KRef producer) {
 
 }  // namespace
 
+KInt GetWorkerId(Worker* worker) {
+#if WITH_WORKERS
+  return worker->id();
+#else
+  return 0;
+#endif  // WITH_WORKERS
+}
+
 Worker* WorkerInit(KBoolean errorReporting) {
 #if WITH_WORKERS
   if (::g_worker != nullptr) return ::g_worker;
-  Worker* worker = theState()->addWorkerUnlocked(errorReporting != 0, nullptr);
+  Worker* worker = theState()->addWorkerUnlocked(errorReporting != 0, nullptr, WorkerKind::kOther);
   ::g_worker = worker;
   return worker;
 #else
@@ -660,6 +716,18 @@ void WorkerDeinit(Worker* worker) {
 #endif  // WITH_WORKERS
 }
 
+void WorkerDestroyThreadDataIfNeeded(KInt id) {
+#if WITH_WORKERS
+  theState()->destroyWorkerThreadDataUnlocked(id);
+#endif
+}
+
+void WaitNativeWorkersTermination() {
+#if WITH_WORKERS
+  theState()->waitNativeWorkersTerminationUnlocked();
+#endif
+}
+
 Worker* WorkerSuspend() {
 #if WITH_WORKERS
   auto* result = ::g_worker;
@@ -714,6 +782,31 @@ Worker::~Worker() {
   pthread_cond_destroy(&cond_);
 }
 
+namespace {
+
+void* workerRoutine(void* argument) {
+  Worker* worker = reinterpret_cast<Worker*>(argument);
+
+  WorkerResume(worker);
+  Kotlin_initRuntimeIfNeeded();
+
+  do {
+    if (worker->processQueueElement(true) == JOB_TERMINATE) break;
+  } while (true);
+
+  // Runtime deinit callback could be called when TLS is already zeroed out, so clear memory
+  // here explicitly. to make sure leak detector properly works.
+  Kotlin_zeroOutTLSGlobals();
+
+  return nullptr;
+}
+
+}  // namespace
+
+void Worker::startEventLoop() {
+  pthread_create(&thread_, nullptr, workerRoutine, this);
+}
+
 void Worker::putJob(Job job, bool toFront) {
   Locker locker(&lock_);
   if (toFront)

runtime/src/main/cpp/Worker.h

@@ -6,8 +6,14 @@
 
 class Worker;
 
+KInt GetWorkerId(Worker* worker);
+
 Worker* WorkerInit(KBoolean errorReporting);
 void WorkerDeinit(Worker* worker);
+// Clean up all associated thread state, if this was a native worker.
+void WorkerDestroyThreadDataIfNeeded(KInt id);
+// Wait until all terminating native workers finish termination. Expected to be called once.
+void WaitNativeWorkersTermination();
 
 Worker* WorkerSuspend();
 void WorkerResume(Worker* worker);