From b2d957b99012cb8c7517cb53cd49eb71fc398812 Mon Sep 17 00:00:00 2001 From: Alexander Shabalin Date: Mon, 20 Nov 2023 17:22:22 +0100 Subject: [PATCH] [K/N] Extract RunLoopSource ^KT-63423 --- .../src/gc/common/cpp/FinalizerProcessor.hpp | 31 +--- .../main/cpp/objc_support/RunLoopSource.hpp | 102 +++++++++++ .../cpp/objc_support/RunLoopSourceTest.cpp | 159 ++++++++++++++++++ .../cpp/objc_support/RunLoopTestSupport.hpp | 88 ++++++++++ 4 files changed, 357 insertions(+), 23 deletions(-) create mode 100644 kotlin-native/runtime/src/main/cpp/objc_support/RunLoopSource.hpp create mode 100644 kotlin-native/runtime/src/main/cpp/objc_support/RunLoopSourceTest.cpp create mode 100644 kotlin-native/runtime/src/main/cpp/objc_support/RunLoopTestSupport.hpp diff --git a/kotlin-native/runtime/src/gc/common/cpp/FinalizerProcessor.hpp b/kotlin-native/runtime/src/gc/common/cpp/FinalizerProcessor.hpp index 473a05c1460..dc6c90dad2a 100644 --- a/kotlin-native/runtime/src/gc/common/cpp/FinalizerProcessor.hpp +++ b/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 #include #endif @@ -119,19 +119,7 @@ private: #if KONAN_OBJC_INTEROP class ProcessingLoopWithCFImpl final : public ProcessingLoop { public: - explicit ProcessingLoopWithCFImpl(FinalizerProcessor& owner) : - owner_(owner), - sourceContext_{ - 0, this, nullptr, nullptr, nullptr, nullptr, nullptr, nullptr, nullptr, - [](void* info) { - auto& self = *static_cast(info); - self.handleNewFinalizers(); - }}, - runLoopSource_(CFRunLoopSourceCreate(nullptr, 0, &sourceContext_)) {} - - ~ProcessingLoopWithCFImpl() override { - CFRelease(runLoopSource_); - } + explicit ProcessingLoopWithCFImpl(FinalizerProcessor& owner) : owner_(owner), runLoopSource_([this]() noexcept { handleNewFinalizers(); }) {} 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); - - CFRunLoopRun(); - - CFRunLoopRemoveSource(CFRunLoopGetCurrent(), runLoopSource_, mode); + { + auto subscription = runLoopSource_.attachToCurrentRunLoop(); + CFRunLoopRun(); + } 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 runLoop_ = nullptr; - CFRunLoopSourceRef runLoopSource_; }; #endif diff --git a/kotlin-native/runtime/src/main/cpp/objc_support/RunLoopSource.hpp b/kotlin-native/runtime/src/main/cpp/objc_support/RunLoopSource.hpp new file mode 100644 index 00000000000..501bc095fc2 --- /dev/null +++ b/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 +#include + +#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`, 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 owner_; + object_ptr 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 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 attachToCurrentRunLoop(CFRunLoopMode mode = kCFRunLoopDefaultMode) noexcept { + return std::unique_ptr(new Subscription(*this, mode)); + } + +private: + static void perform(void* callback) noexcept { static_cast(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 callback_; // TODO: std::function_ref? + CFRunLoopSourceContext sourceContext_; + object_ptr source_; + std::atomic activeSubscription_ = nullptr; +}; + +} // namespace kotlin::objc_support + +#endif \ No newline at end of file diff --git a/kotlin-native/runtime/src/main/cpp/objc_support/RunLoopSourceTest.cpp b/kotlin-native/runtime/src/main/cpp/objc_support/RunLoopSourceTest.cpp new file mode 100644 index 00000000000..e9c5608ed8c --- /dev/null +++ b/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 + int64_t performAndWaitChange(F&& f) noexcept { + auto initial = value_.load(std::memory_order_acquire); + std::invoke(std::forward(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 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 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 \ No newline at end of file diff --git a/kotlin-native/runtime/src/main/cpp/objc_support/RunLoopTestSupport.hpp b/kotlin-native/runtime/src/main/cpp/objc_support/RunLoopTestSupport.hpp new file mode 100644 index 00000000000..9f139511a14 --- /dev/null +++ b/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 +#include +#include + +#include "ScopedThread.hpp" +#include "Utils.hpp" +#include "objc_support/ObjectPtr.hpp" + +namespace kotlin::objc_support::test_support { + +class RunLoopInScopedThread : private Pinned { +public: + template + 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 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 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 runLoop_; + State state_ = State::kInitial; + ScopedThread thread_; +}; + +} // namespace kotlin::objc_support::test_support + +#endif \ No newline at end of file