[K/N] Extract RunLoopSource ^KT-63423

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

@@ -18,9 +18,9 @@
 #include "Utils.hpp"
 #include "Logging.hpp"
 #include "objc_support/AutoreleasePool.hpp"
+#include "objc_support/RunLoopSource.hpp"
 
 #if KONAN_OBJC_INTEROP
-#include <CoreFoundation/CoreFoundation.h>
 #include <CoreFoundation/CFRunLoop.h>
 #endif
 
@@ -119,19 +119,7 @@ private:
 #if KONAN_OBJC_INTEROP
     class ProcessingLoopWithCFImpl final : public ProcessingLoop {
     public:
-        explicit ProcessingLoopWithCFImpl(FinalizerProcessor& owner) :
+        explicit ProcessingLoopWithCFImpl(FinalizerProcessor& owner) : owner_(owner), runLoopSource_([this]() noexcept { handleNewFinalizers(); }) {}
-                owner_(owner),
-                sourceContext_{
-                        0, this, nullptr, nullptr, nullptr, nullptr, nullptr, nullptr, nullptr,
-                        [](void* info) {
-                            auto& self = *static_cast<ProcessingLoopWithCFImpl*>(info);
-                            self.handleNewFinalizers();
-                        }},
-                runLoopSource_(CFRunLoopSourceCreate(nullptr, 0, &sourceContext_)) {}
-
-        ~ProcessingLoopWithCFImpl() override {
-            CFRelease(runLoopSource_);
-        }
 
         void notify() override {
             // wait until runLoop_ ptr is published
@@ -139,7 +127,7 @@ private:
                 std::this_thread::yield();
             }
             // notify
-            CFRunLoopSourceSignal(runLoopSource_);
+            runLoopSource_.signal();
             CFRunLoopWakeUp(runLoop_);
         }
 
@@ -149,12 +137,10 @@ private:
 
         void body() override {
             objc_support::AutoreleasePool autoreleasePool;
-            auto mode = kCFRunLoopDefaultMode;
+            {
-            CFRunLoopAddSource(CFRunLoopGetCurrent(), runLoopSource_, mode);
+                auto subscription = runLoopSource_.attachToCurrentRunLoop();
-
+                CFRunLoopRun();
-            CFRunLoopRun();
+            }
-
-            CFRunLoopRemoveSource(CFRunLoopGetCurrent(), runLoopSource_, mode);
             runLoop_.store(nullptr, std::memory_order_release);
         }
 
@@ -176,9 +162,8 @@ private:
 
         FinalizerProcessor& owner_;
         int64_t finishedEpoch_ = 0;
-        CFRunLoopSourceContext sourceContext_;
+        objc_support::RunLoopSource runLoopSource_;
         std::atomic<CFRunLoopRef> runLoop_ = nullptr;
-        CFRunLoopSourceRef runLoopSource_;
     };
 #endif
 

kotlin-native/runtime/src/main/cpp/objc_support/RunLoopSource.hpp

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2010-2023 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the LICENSE file.
+ */
+
+#pragma once
+
+#if KONAN_HAS_FOUNDATION_FRAMEWORK
+
+#include <CoreFoundation/CoreFoundation.h>
+#include <CoreFoundation/CFRunLoop.h>
+
+#include "RawPtr.hpp"
+#include "Utils.hpp"
+#include "objc_support/ObjectPtr.hpp"
+
+namespace kotlin::objc_support {
+
+// Smart pointer around `CFRunLoopSource`.
+//
+// To attach to the current run loop use `attachToCurrentRunLoop`, it'll be detached when the returned `Subscription`
+// is destroyed.
+// `RunLoopSource` is initialized with `std::function<void()>`, which will be called every time the run loop processes
+// this source.
+// To notify attaced run loops that this source has more work, use `signal`.
+// The underlying raw pointer is available via `handle`.
+class RunLoopSource : private Pinned {
+public:
+    // A token that `RunLoopSource` is attached to a run loop.
+    //
+    // Must be destroyed on the same thread that called `attachToCurrentRunLoop`.
+    class [[nodiscard]] Subscription : private Pinned {
+        // TODO: Consider making it movable.
+    public:
+        ~Subscription() {
+            RuntimeAssert(*runLoop_ == CFRunLoopGetCurrent(), "Must be destroyed on the same thread as created");
+            CFRunLoopRemoveSource(*runLoop_, *owner_->source_, mode_);
+            owner_->unregisterSubscription(*this);
+        }
+
+    private:
+        friend class RunLoopSource;
+
+        Subscription(RunLoopSource& owner, CFRunLoopMode mode) noexcept :
+            owner_(&owner), runLoop_(object_ptr_retain, CFRunLoopGetCurrent()), mode_(mode) {
+            owner_->registerSubscription(*this);
+            CFRunLoopAddSource(*runLoop_, *owner_->source_, mode);
+        }
+
+        raw_ptr<RunLoopSource> owner_;
+        object_ptr<CFRunLoopRef> runLoop_;
+        CFRunLoopMode mode_;
+    };
+
+    // Create `RunLoopSource` with `callback` that will be invoked each time, when an attached run loop processes this source.
+    explicit RunLoopSource(std::function<void()> callback) noexcept :
+        callback_(std::move(callback)),
+        sourceContext_{0, &callback_, nullptr, nullptr, nullptr, nullptr, nullptr, nullptr, nullptr, &perform},
+        source_(CFRunLoopSourceCreate(nullptr, 0, &sourceContext_)) {}
+
+    ~RunLoopSource() {
+        auto* subscription = activeSubscription_.load(std::memory_order_relaxed);
+        RuntimeAssert(subscription == nullptr, "Expected no active subscription, but was %p", subscription);
+    }
+
+    // Raw pointer to `CFRunLoopSource`.
+    auto handle() noexcept { return *source_; }
+
+    // Signal the attached run loops, that this `RunLoopSource` has some work.
+    void signal() noexcept { CFRunLoopSourceSignal(*source_); }
+
+    // Attach this `RunLoopSource` to the current thread's run loop.
+    [[nodiscard]] std::unique_ptr<Subscription> attachToCurrentRunLoop(CFRunLoopMode mode = kCFRunLoopDefaultMode) noexcept {
+        return std::unique_ptr<Subscription>(new Subscription(*this, mode));
+    }
+
+private:
+    static void perform(void* callback) noexcept { static_cast<decltype(callback_)*>(callback)->operator()(); }
+
+    void registerSubscription(Subscription& subscription) noexcept {
+        Subscription* actual = nullptr;
+        activeSubscription_.compare_exchange_strong(actual, &subscription, std::memory_order_relaxed);
+        RuntimeAssert(
+                actual == nullptr, "Cannot have more than one active subscription. Trying to regiser %p but %p is already active",
+                &subscription, actual);
+    }
+
+    void unregisterSubscription(Subscription& subscription) noexcept {
+        Subscription* actual = &subscription;
+        activeSubscription_.compare_exchange_strong(actual, nullptr, std::memory_order_relaxed);
+        RuntimeAssert(actual == &subscription, "Expected %p to be active subscription. But was %p", &subscription, actual);
+    }
+
+    std::function<void()> callback_; // TODO: std::function_ref?
+    CFRunLoopSourceContext sourceContext_;
+    object_ptr<CFRunLoopSourceRef> source_;
+    std::atomic<Subscription*> activeSubscription_ = nullptr;
+};
+
+} // namespace kotlin::objc_support
+
+#endif

kotlin-native/runtime/src/main/cpp/objc_support/RunLoopSourceTest.cpp

@@ -0,0 +1,159 @@
+/*
+ * Copyright 2010-2023 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the LICENSE file.
+ */
+
+#if KONAN_HAS_FOUNDATION_FRAMEWORK
+
+#include "RunLoopSource.hpp"
+
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+#include "RunLoopTestSupport.hpp"
+#include "ScopedThread.hpp"
+
+using namespace kotlin;
+
+namespace {
+
+class Checkpoint : private Pinned {
+public:
+    Checkpoint() noexcept = default;
+
+    void checkpoint() noexcept { value_.fetch_add(1, std::memory_order_release); }
+
+    template <typename F>
+    int64_t performAndWaitChange(F&& f) noexcept {
+        auto initial = value_.load(std::memory_order_acquire);
+        std::invoke(std::forward<F>(f));
+        while (value_.load(std::memory_order_relaxed) <= initial) {
+            std::this_thread::yield();
+        }
+        auto final = value_.load(std::memory_order_acquire);
+        return final - initial;
+    }
+
+private:
+    std::atomic<int64_t> value_ = 0;
+};
+
+class MainThreadShutdownToken : private Pinned {
+public:
+    MainThreadShutdownToken() noexcept = default;
+
+    ~MainThreadShutdownToken() { CFRunLoopStop(CFRunLoopGetMain()); }
+};
+
+} // namespace
+
+TEST(RunLoopSourceTest, MainThread) {
+    ASSERT_EQ(CFRunLoopGetMain(), CFRunLoopGetCurrent());
+    Checkpoint checkpoint;
+    objc_support::RunLoopSource source([&]() noexcept { checkpoint.checkpoint(); });
+    auto subscription = source.attachToCurrentRunLoop();
+    auto thread = ScopedThread([&]() noexcept {
+        MainThreadShutdownToken token;
+        ASSERT_THAT(
+                checkpoint.performAndWaitChange([&]() noexcept {
+                    source.signal();
+                    CFRunLoopWakeUp(CFRunLoopGetMain());
+                }),
+                1);
+        ASSERT_THAT(
+                checkpoint.performAndWaitChange([&]() noexcept {
+                    source.signal();
+                    CFRunLoopWakeUp(CFRunLoopGetMain());
+                }),
+                1);
+    });
+    CFRunLoopRun();
+}
+
+TEST(RunLoopSourceTest, SignalRunsOnce) {
+    Checkpoint checkpoint;
+    objc_support::RunLoopSource source([&]() noexcept { checkpoint.checkpoint(); });
+    objc_support::test_support::RunLoopInScopedThread runLoop([&]() noexcept { return source.attachToCurrentRunLoop(); });
+    ASSERT_THAT(
+            checkpoint.performAndWaitChange([&]() noexcept {
+                source.signal();
+                runLoop.wakeUp();
+            }),
+            1);
+    ASSERT_THAT(
+            checkpoint.performAndWaitChange([&]() noexcept {
+                source.signal();
+                runLoop.wakeUp();
+            }),
+            1);
+}
+
+TEST(RunLoopSourceTest, ConnectToDifferentLoop) {
+    Checkpoint checkpoint;
+    objc_support::RunLoopSource source([&]() noexcept { checkpoint.checkpoint(); });
+    {
+        objc_support::test_support::RunLoopInScopedThread runLoop([&]() noexcept { return source.attachToCurrentRunLoop(); });
+        ASSERT_THAT(
+                checkpoint.performAndWaitChange([&]() noexcept {
+                    source.signal();
+                    runLoop.wakeUp();
+                }),
+                1);
+    }
+    {
+        objc_support::test_support::RunLoopInScopedThread runLoop([&]() noexcept { return source.attachToCurrentRunLoop(); });
+        ASSERT_THAT(
+                checkpoint.performAndWaitChange([&]() noexcept {
+                    source.signal();
+                    runLoop.wakeUp();
+                }),
+                1);
+    }
+}
+
+TEST(RunLoopSourceTest, Reconnect) {
+    Checkpoint checkpoint1;
+    Checkpoint checkpoint2;
+    objc_support::RunLoopSource source([&]() noexcept { checkpoint1.checkpoint(); });
+    objc_support::test_support::RunLoopInScopedThread runLoop([&]() noexcept { return 0; });
+
+    std::unique_ptr<objc_support::RunLoopSource::Subscription> subscription;
+
+    // Subscribe. Check that `source` is attached. Unsubscribe
+    checkpoint2.performAndWaitChange([&]() noexcept {
+        runLoop.schedule([&]() noexcept {
+            subscription = source.attachToCurrentRunLoop();
+            checkpoint2.checkpoint();
+        });
+    });
+    checkpoint1.performAndWaitChange([&]() noexcept {
+        source.signal();
+        runLoop.wakeUp();
+    });
+    checkpoint2.performAndWaitChange([&]() noexcept {
+        runLoop.schedule([&]() noexcept {
+            subscription = nullptr;
+            checkpoint2.checkpoint();
+        });
+    });
+
+    // Repeat the procedure.
+    checkpoint2.performAndWaitChange([&]() noexcept {
+        runLoop.schedule([&]() noexcept {
+            subscription = source.attachToCurrentRunLoop();
+            checkpoint2.checkpoint();
+        });
+    });
+    checkpoint1.performAndWaitChange([&]() noexcept {
+        source.signal();
+        runLoop.wakeUp();
+    });
+    checkpoint2.performAndWaitChange([&]() noexcept {
+        runLoop.schedule([&]() noexcept {
+            subscription = nullptr;
+            checkpoint2.checkpoint();
+        });
+    });
+}
+
+#endif

kotlin-native/runtime/src/main/cpp/objc_support/RunLoopTestSupport.hpp

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2010-2023 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the LICENSE file.
+ */
+
+#pragma once
+
+#if KONAN_HAS_FOUNDATION_FRAMEWORK
+
+#include <condition_variable>
+#include <mutex>
+#include <CoreFoundation/CFRunLoop.h>
+
+#include "ScopedThread.hpp"
+#include "Utils.hpp"
+#include "objc_support/ObjectPtr.hpp"
+
+namespace kotlin::objc_support::test_support {
+
+class RunLoopInScopedThread : private Pinned {
+public:
+    template <typename Init>
+    explicit RunLoopInScopedThread(Init init) noexcept :
+        thread_([&]() noexcept {
+            [[maybe_unused]] auto state = init();
+            {
+                std::unique_lock guard{stateMutex_};
+                runLoop_.reset(object_ptr_retain, CFRunLoopGetCurrent());
+                RuntimeAssert(*runLoop_ != nullptr, "Current run loop cannot be null");
+                RuntimeAssert(state_ == State::kInitial, "Expected state to be %d but was %d", State::kInitial, state_);
+                state_ = State::kRunning;
+            }
+            initializedCV_.notify_one();
+            while (true) {
+                CFRunLoopRun();
+                std::unique_lock guard{stateMutex_};
+                if (state_ != State::kRunning) {
+                    RuntimeAssert(state_ == State::kWillStop, "Expected state to be %d but was %d", State::kWillStop, state_);
+                    RuntimeAssert(*runLoop_ == nullptr, "Stored run loop must have been nulled");
+                    state_ = State::kStopped;
+                    break;
+                }
+            }
+        }) {
+        std::unique_lock guard{stateMutex_};
+        initializedCV_.wait(guard, [this]() noexcept { return state_ >= State::kRunning; });
+    }
+
+    ~RunLoopInScopedThread() {
+        object_ptr<CFRunLoopRef> runLoop;
+        {
+            std::unique_lock guard{stateMutex_};
+            runLoop = std::move(runLoop_);
+            RuntimeAssert(state_ == State::kRunning, "Expected state to be %d but was %d", State::kRunning, state_);
+            state_ = State::kWillStop;
+        }
+        CFRunLoopStop(*runLoop);
+    }
+
+    CFRunLoopRef handle() noexcept { return *runLoop_; }
+
+    void wakeUp() noexcept { CFRunLoopWakeUp(*runLoop_); }
+
+    void schedule(std::function<void()> f) noexcept {
+        CFRunLoopPerformBlock(*runLoop_, kCFRunLoopDefaultMode, ^{
+            f();
+        });
+        CFRunLoopWakeUp(*runLoop_);
+    }
+
+private:
+    enum class State {
+        kInitial,
+        kRunning,
+        kWillStop,
+        kStopped,
+    };
+
+    std::mutex stateMutex_;
+    std::condition_variable initializedCV_;
+    object_ptr<CFRunLoopRef> runLoop_;
+    State state_ = State::kInitial;
+    ScopedThread thread_;
+};
+
+} // namespace kotlin::objc_support::test_support
+
+#endif