[K/N] Process objc run loop on the finalizer thread ^KT-58851

Merge-request: KT-MR-11024
Merged-by: Alexey Glushko <aleksei.glushko@jetbrains.com>

kotlin-native/runtime/src/gc/common/cpp/FinalizerProcessor.hpp

@@ -17,6 +17,13 @@
 #include "Runtime.h"
 #include "ScopedThread.hpp"
 #include "Utils.hpp"
+#include "Logging.hpp"
+
+#if KONAN_OBJC_INTEROP
+#include "ObjCMMAPI.h"
+#include <CoreFoundation/CoreFoundation.h>
+#include <CoreFoundation/CFRunLoop.h>
+#endif
 
 namespace kotlin::gc {
 
@@ -26,7 +33,7 @@ public:
     // epochDoneCallback could be called on any subset of them.
     // If no new tasks are set, epochDoneCallback will be eventually called on last epoch
     explicit FinalizerProcessor(std::function<void(int64_t)> epochDoneCallback) noexcept :
-        epochDoneCallback_(std::move(epochDoneCallback)) {}
+        epochDoneCallback_(std::move(epochDoneCallback)), processingLoop_(*this) {}
 
     ~FinalizerProcessor() { StopFinalizerThread(); }
 
@@ -40,7 +47,7 @@ public:
         StartFinalizerThreadIfNone();
         FinalizerQueueTraits::add(finalizerQueue_, std::move(tasks));
         finalizerQueueEpoch_ = epoch;
-        finalizerQueueCondVar_.notify_all();
+        processingLoop_.notify();
     }
 
     void StopFinalizerThread() noexcept {
@@ -48,7 +55,7 @@ public:
             std::unique_lock guard(finalizerQueueMutex_);
             if (!finalizerThread_.joinable()) return;
             shutdownFlag_ = true;
-            finalizerQueueCondVar_.notify_all();
+            processingLoop_.notify();
         }
         finalizerThread_.join();
         shutdownFlag_ = false;
@@ -71,26 +78,7 @@ public:
                 initialized_ = true;
             }
             initializedCondVar_.notify_all();
-            int64_t finalizersEpoch = 0;
-            while (true) {
-                std::unique_lock lock(finalizerQueueMutex_);
-                finalizerQueueCondVar_.wait(lock, [this, &finalizersEpoch] {
-                    return !FinalizerQueueTraits::isEmpty(finalizerQueue_) || finalizerQueueEpoch_ != finalizersEpoch || shutdownFlag_;
-                });
-                if (FinalizerQueueTraits::isEmpty(finalizerQueue_) && finalizerQueueEpoch_ == finalizersEpoch) {
-                    newTasksAllowed_ = false;
-                    RuntimeAssert(shutdownFlag_, "Nothing to do, but no shutdownFlag_ is set on wakeup");
-                    break;
-                }
-                auto queue = std::move(finalizerQueue_);
-                finalizersEpoch = finalizerQueueEpoch_;
-                lock.unlock();
-                if (!FinalizerQueueTraits::isEmpty(queue)) {
-                    ThreadStateGuard guard(ThreadState::kRunnable);
-                    FinalizerQueueTraits::process(std::move(queue));
-                }
-                epochDoneCallback_(finalizersEpoch);
-            }
+            processingLoop_.body();
             {
                 std::unique_lock guard(initializedMutex_);
                 initialized_ = false;
@@ -105,6 +93,119 @@ public:
     }
 
 private:
+    // should be called under the finalizerQueueMutex_
+    bool shouldShutdown(int64_t lastProcessedEpoch) noexcept {
+        bool shouldShutdown = FinalizerQueueTraits::isEmpty(finalizerQueue_) && finalizerQueueEpoch_ == lastProcessedEpoch;
+        if (shouldShutdown) {
+            RuntimeAssert(shutdownFlag_, "Nothing to do, but no shutdownFlag_ is set on wakeup");
+        }
+        return shouldShutdown;
+    }
+
+    void processSingle(FinalizerQueue&& queue, int64_t currentEpoch) noexcept {
+        if (!FinalizerQueueTraits::isEmpty(queue)) {
+#if KONAN_OBJC_INTEROP
+            konan::AutoreleasePool autoreleasePool;
+#endif
+            ThreadStateGuard guard(ThreadState::kRunnable);
+            FinalizerQueueTraits::process(std::move(queue));
+        }
+        epochDoneCallback_(currentEpoch);
+    }
+
+#if KONAN_OBJC_INTEROP
+    class ProcessingLoop {
+    public:
+        explicit ProcessingLoop(FinalizerProcessor& owner) :
+                owner_(owner),
+                sourceContext_{
+                        0, this, nullptr, nullptr, nullptr, nullptr, nullptr, nullptr, nullptr,
+                        [](void* info) {
+                            auto& self = *reinterpret_cast<ProcessingLoop*>(info);
+                            self.handleNewFinalizers();
+                        }},
+                runLoopSource_(CFRunLoopSourceCreate(nullptr, 0, &sourceContext_)) {}
+
+        ~ProcessingLoop() {
+            CFRelease(runLoopSource_);
+        }
+
+        void notify() {
+            // wait until runLoop_ ptr is published
+            while (runLoop_.load(std::memory_order_acquire) == nullptr) {
+                std::this_thread::yield();
+            }
+            // notify
+            CFRunLoopSourceSignal(runLoopSource_);
+            CFRunLoopWakeUp(runLoop_);
+        }
+
+        void body() {
+            konan::AutoreleasePool autoreleasePool;
+            auto mode = kCFRunLoopDefaultMode;
+            CFRunLoopAddSource(CFRunLoopGetCurrent(), runLoopSource_, mode);
+            runLoop_.store(CFRunLoopGetCurrent(), std::memory_order_release);
+
+            CFRunLoopRun();
+
+            runLoop_.store(nullptr, std::memory_order_release);
+
+            CFRunLoopRemoveSource(CFRunLoopGetCurrent(), runLoopSource_, mode);
+        }
+    private:
+        void handleNewFinalizers() {
+            std::unique_lock lock(owner_.finalizerQueueMutex_);
+            if (owner_.shouldShutdown(finishedEpoch_)) {
+                owner_.newTasksAllowed_ = false;
+                CFRunLoopStop(runLoop_.load(std::memory_order_acquire));
+                return;
+            }
+            auto queue = std::move(owner_.finalizerQueue_);
+            int64_t currentEpoch = owner_.finalizerQueueEpoch_;
+            lock.unlock();
+
+            owner_.processSingle(std::move(queue), currentEpoch);
+            finishedEpoch_ = currentEpoch;
+        }
+
+        FinalizerProcessor& owner_;
+        int64_t finishedEpoch_ = 0;
+        CFRunLoopSourceContext sourceContext_;
+        std::atomic<CFRunLoopRef> runLoop_ = nullptr;
+        CFRunLoopSourceRef runLoopSource_;
+    };
+#else
+    class ProcessingLoop {
+    public:
+        explicit ProcessingLoop(FinalizerProcessor& owner) : owner_(owner) {}
+
+        void notify() {
+            owner_.finalizerQueueCondVar_.notify_all();
+        }
+
+        void body() {
+            int64_t finishedEpoch = 0;
+            while (true) {
+                std::unique_lock lock(owner_.finalizerQueueMutex_);
+                owner_.finalizerQueueCondVar_.wait(lock, [this, &finishedEpoch] {
+                    return !FinalizerQueueTraits::isEmpty(owner_.finalizerQueue_) || owner_.finalizerQueueEpoch_ != finishedEpoch || owner_.shutdownFlag_;
+                });
+                if (owner_.shouldShutdown(finishedEpoch)) {
+                    owner_.newTasksAllowed_ = false;
+                    break;
+                }
+                auto queue = std::move(owner_.finalizerQueue_);
+                auto currentEpoch = owner_.finalizerQueueEpoch_;
+                lock.unlock();
+                owner_.processSingle(std::move(queue), currentEpoch);
+                finishedEpoch = currentEpoch;
+            }
+        }
+    private:
+        FinalizerProcessor& owner_;
+    };
+#endif
+
     ScopedThread finalizerThread_;
     FinalizerQueue finalizerQueue_;
     std::condition_variable finalizerQueueCondVar_;
@@ -114,11 +215,14 @@ private:
     bool shutdownFlag_ = false;
     bool newTasksAllowed_ = true;
 
+    ProcessingLoop processingLoop_;
+
     std::mutex initializedMutex_;
     std::condition_variable initializedCondVar_;
     bool initialized_ = false;
 
     std::mutex threadCreatingMutex_;
+
 };
 
 } // namespace kotlin::gc

kotlin-native/runtime/src/main/cpp/ObjCInterop.mm

@@ -339,7 +339,7 @@ konan::AutoreleasePool::AutoreleasePool()
   : handle(objc_autoreleasePoolPush()) {}
 
 konan::AutoreleasePool::~AutoreleasePool() {
-  kotlin::ThreadStateGuard guard(kotlin::ThreadState::kNative);
+  kotlin::ThreadStateGuard guard(kotlin::ThreadState::kNative, true);
   objc_autoreleasePoolPop(handle);
 }