From cbca5d0add67a8ed07ee6a4601832d0e739c0220 Mon Sep 17 00:00:00 2001 From: Alexander Shabalin Date: Wed, 11 Mar 2020 15:54:04 +0300 Subject: [PATCH] Detect leaked workers (#3934) --- backend.native/tests/build.gradle | 21 +++ .../tests/runtime/memory/leak_memory.kt | 5 + .../leak_memory_with_worker_termination.kt | 10 ++ .../tests/runtime/workers/leak_worker.kt | 9 ++ .../org/jetbrains/kotlin/KotlinNativeTest.kt | 12 +- runtime/src/launcher/cpp/launcher.cpp | 7 +- runtime/src/main/cpp/Memory.cpp | 17 +- runtime/src/main/cpp/Porting.cpp | 9 ++ runtime/src/main/cpp/Porting.h | 1 + runtime/src/main/cpp/Runtime.cpp | 16 ++ runtime/src/main/cpp/Runtime.h | 2 + runtime/src/main/cpp/Worker.cpp | 145 ++++++++++++++---- runtime/src/main/cpp/Worker.h | 6 + 13 files changed, 217 insertions(+), 43 deletions(-) create mode 100644 backend.native/tests/runtime/memory/leak_memory.kt create mode 100644 backend.native/tests/runtime/workers/leak_memory_with_worker_termination.kt create mode 100644 backend.native/tests/runtime/workers/leak_worker.kt diff --git a/backend.native/tests/build.gradle b/backend.native/tests/build.gradle index e5f779a57af..c44954be029 100644 --- a/backend.native/tests/build.gradle +++ b/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 = "\n\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'] diff --git a/backend.native/tests/runtime/memory/leak_memory.kt b/backend.native/tests/runtime/memory/leak_memory.kt new file mode 100644 index 00000000000..487662fff28 --- /dev/null +++ b/backend.native/tests/runtime/memory/leak_memory.kt @@ -0,0 +1,5 @@ +import kotlinx.cinterop.* + +fun main() { + StableRef.create(Any()) +} diff --git a/backend.native/tests/runtime/workers/leak_memory_with_worker_termination.kt b/backend.native/tests/runtime/workers/leak_memory_with_worker_termination.kt new file mode 100644 index 00000000000..32a9d0a3a69 --- /dev/null +++ b/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 +} diff --git a/backend.native/tests/runtime/workers/leak_worker.kt b/backend.native/tests/runtime/workers/leak_worker.kt new file mode 100644 index 00000000000..1df5bfac135 --- /dev/null +++ b/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()) +} diff --git a/build-tools/src/main/kotlin/org/jetbrains/kotlin/KotlinNativeTest.kt b/build-tools/src/main/kotlin/org/jetbrains/kotlin/KotlinNativeTest.kt index aa90d3683ce..43c048bf70e 100644 --- a/build-tools/src/main/kotlin/org/jetbrains/kotlin/KotlinNativeTest.kt +++ b/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: diff --git a/runtime/src/launcher/cpp/launcher.cpp b/runtime/src/launcher/cpp/launcher.cpp index 560b557f3ee..54e98291c42 100644 --- a/runtime/src/launcher/cpp/launcher.cpp +++ b/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; } diff --git a/runtime/src/main/cpp/Memory.cpp b/runtime/src/main/cpp/Memory.cpp index c6c78d03dac..2ac2e7cc1ef 100644 --- a/runtime/src/main/cpp/Memory.cpp +++ b/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); diff --git a/runtime/src/main/cpp/Porting.cpp b/runtime/src/main/cpp/Porting.cpp index c6d1571a272..43321a03082 100644 --- a/runtime/src/main/cpp/Porting.cpp +++ b/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 diff --git a/runtime/src/main/cpp/Porting.h b/runtime/src/main/cpp/Porting.h index aed271fe9c0..32a1677883a 100644 --- a/runtime/src/main/cpp/Porting.h +++ b/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. diff --git a/runtime/src/main/cpp/Runtime.cpp b/runtime/src/main/cpp/Runtime.cpp index 801f16fe7ab..ecd5b4de9e8 100644 --- a/runtime/src/main/cpp/Runtime.cpp +++ b/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" diff --git a/runtime/src/main/cpp/Runtime.h b/runtime/src/main/cpp/Runtime.h index 22d5ae44fda..153b8d70a11 100644 --- a/runtime/src/main/cpp/Runtime.h +++ b/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 diff --git a/runtime/src/main/cpp/Worker.cpp b/runtime/src/main/cpp/Worker.cpp index 56787f1c10f..47f78e62142 100644 --- a/runtime/src/main/cpp/Worker.cpp +++ b/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 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 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(nextWorkerId(), errorReporting, customName); + worker = konanConstructInstance(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 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 futures_; KStdUnorderedMap workers_; + KStdUnorderedMap 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(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(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) diff --git a/runtime/src/main/cpp/Worker.h b/runtime/src/main/cpp/Worker.h index 885c9bc7a9a..39792c96ba2 100644 --- a/runtime/src/main/cpp/Worker.h +++ b/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);