From 45edfa067e574407ed538b79bf5c0865b866c541 Mon Sep 17 00:00:00 2001 From: Josiah VanderZee Date: Tue, 16 Jun 2026 13:27:08 -0500 Subject: [PATCH 01/14] Add unit tests for lock macros --- src/iocore/eventsystem/CMakeLists.txt | 3 + .../eventsystem/unit_tests/test_Lock.cc | 357 ++++++++++++++++++ 2 files changed, 360 insertions(+) create mode 100644 src/iocore/eventsystem/unit_tests/test_Lock.cc diff --git a/src/iocore/eventsystem/CMakeLists.txt b/src/iocore/eventsystem/CMakeLists.txt index a6f4d6fc436..e5c5bde6624 100644 --- a/src/iocore/eventsystem/CMakeLists.txt +++ b/src/iocore/eventsystem/CMakeLists.txt @@ -72,6 +72,9 @@ if(BUILD_TESTING) target_link_libraries(test_Action ts::inkevent configmanager Catch2::Catch2WithMain) add_catch2_test(NAME test_Action COMMAND test_Action) + add_executable(test_Lock unit_tests/test_Lock.cc) + target_link_libraries(test_Lock ts::inkevent configmanager Catch2::Catch2WithMain) + add_catch2_test(NAME test_Lock COMMAND test_Lock) endif() clang_tidy_check(inkevent) diff --git a/src/iocore/eventsystem/unit_tests/test_Lock.cc b/src/iocore/eventsystem/unit_tests/test_Lock.cc new file mode 100644 index 00000000000..ce835e6c635 --- /dev/null +++ b/src/iocore/eventsystem/unit_tests/test_Lock.cc @@ -0,0 +1,357 @@ +/** @file + + Catch2 unit tests for the Lock.h mutex acquisition macros. + + @section license License + + Licensed to the Apache Software Foundation (ASF) under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + */ + +#include "inkevent_test_fixtures.h" + +using inkevent_test::AtomicFlag; +using inkevent_test::EventProcessorListener; + +CATCH_REGISTER_LISTENER(EventProcessorListener) + +namespace +{ + +class HoldOnEThread : public Continuation +{ +public: + HoldOnEThread(ProxyMutex *self_mutex, Ptr &target) : Continuation(self_mutex), target_mutex(target) + { + SET_HANDLER(&HoldOnEThread::on_event); + } + + Ptr target_mutex; + AtomicFlag held; + AtomicFlag release; + AtomicFlag done; + +private: + int + on_event(int /* event ATS_UNUSED */, void * /* data ATS_UNUSED */) + { + SCOPED_MUTEX_LOCK(guard, target_mutex, this_ethread()); + held.set(); + release.wait_until_set(); + done.set(); + return 0; + } +}; + +} // namespace + +TEST_CASE("MUTEX_TRY_LOCK on an unheld ProxyMutex constructs a guard whose is_locked() reports the successful acquisition", + "[inkevent][lock]") +{ + Ptr m{new_ProxyMutex()}; + EThread *t = this_ethread(); + + MUTEX_TRY_LOCK(guard, m, t); + + REQUIRE(guard.is_locked()); + REQUIRE(m->thread_holding == t); + REQUIRE(m->nthread_holding == 1); +} + +TEST_CASE("MUTEX_TRY_LOCK against a contended ProxyMutex constructs a guard whose is_locked() reports the failed acquisition", + "[inkevent][lock][multithread]") +{ + Ptr contended{new_ProxyMutex()}; + Ptr cont_self{new_ProxyMutex()}; + HoldOnEThread holder{cont_self.get(), contended}; + + REQUIRE(eventProcessor.schedule_imm(&holder, ET_CALL) != nullptr); + REQUIRE(holder.held.wait_until_set()); + + EThread *t = this_ethread(); + MUTEX_TRY_LOCK(guard, contended, t); + + REQUIRE_FALSE(guard.is_locked()); + + holder.release.set(); + REQUIRE(holder.done.wait_until_set()); +} + +TEST_CASE("MUTEX_TRY_LOCK by the holding thread is reentrant and returns a guard reporting is_locked() == true", "[inkevent][lock]") +{ + Ptr m{new_ProxyMutex()}; + EThread *t = this_ethread(); + + MUTEX_TRY_LOCK(outer, m, t); + REQUIRE(outer.is_locked()); + REQUIRE(m->nthread_holding == 1); + + { + MUTEX_TRY_LOCK(inner, m, t); + REQUIRE(inner.is_locked()); + REQUIRE(m->nthread_holding == 2); + } + + REQUIRE(m->nthread_holding == 1); + REQUIRE(m->thread_holding == t); +} + +TEST_CASE("A MUTEX_TRY_LOCK guard releases the lock at scope exit, leaving the ProxyMutex unheld", "[inkevent][lock]") +{ + Ptr m{new_ProxyMutex()}; + EThread *t = this_ethread(); + + { + MUTEX_TRY_LOCK(guard, m, t); + REQUIRE(guard.is_locked()); + } + + REQUIRE(m->nthread_holding == 0); + REQUIRE(m->thread_holding == nullptr); +} + +TEST_CASE("MUTEX_RELEASE on a MUTEX_TRY_LOCK guard releases the lock early and flips is_locked() to false", "[inkevent][lock]") +{ + Ptr m{new_ProxyMutex()}; + EThread *t = this_ethread(); + + MUTEX_TRY_LOCK(guard, m, t); + REQUIRE(guard.is_locked()); + + MUTEX_RELEASE(guard); + + REQUIRE_FALSE(guard.is_locked()); + REQUIRE(m->nthread_holding == 0); + REQUIRE(m->thread_holding == nullptr); +} + +TEST_CASE("MUTEX_RELEASE invoked twice on the same MUTEX_TRY_LOCK guard is a no-op on the second call", "[inkevent][lock]") +{ + Ptr m{new_ProxyMutex()}; + EThread *t = this_ethread(); + + MUTEX_TRY_LOCK(guard, m, t); + MUTEX_RELEASE(guard); + REQUIRE_FALSE(guard.is_locked()); + + MUTEX_RELEASE(guard); + + REQUIRE_FALSE(guard.is_locked()); + REQUIRE(m->nthread_holding == 0); +} + +TEST_CASE("SCOPED_MUTEX_LOCK on an unheld ProxyMutex acquires the lock during construction", "[inkevent][lock]") +{ + Ptr m{new_ProxyMutex()}; + EThread *t = this_ethread(); + + SCOPED_MUTEX_LOCK(guard, m, t); + + REQUIRE(m->thread_holding == t); + REQUIRE(m->nthread_holding == 1); +} + +TEST_CASE("A SCOPED_MUTEX_LOCK guard releases the lock when its enclosing scope ends", "[inkevent][lock]") +{ + Ptr m{new_ProxyMutex()}; + EThread *t = this_ethread(); + + { + SCOPED_MUTEX_LOCK(guard, m, t); + REQUIRE(m->nthread_holding == 1); + } + + REQUIRE(m->nthread_holding == 0); + REQUIRE(m->thread_holding == nullptr); +} + +TEST_CASE("SCOPED_MUTEX_LOCK is reentrant when the calling EThread already holds the lock", "[inkevent][lock]") +{ + Ptr m{new_ProxyMutex()}; + EThread *t = this_ethread(); + + SCOPED_MUTEX_LOCK(outer, m, t); + REQUIRE(m->nthread_holding == 1); + + { + SCOPED_MUTEX_LOCK(inner, m, t); + REQUIRE(m->nthread_holding == 2); + REQUIRE(m->thread_holding == t); + } + + REQUIRE(m->nthread_holding == 1); + REQUIRE(m->thread_holding == t); +} + +TEST_CASE("MUTEX_RELEASE on a SCOPED_MUTEX_LOCK guard releases the lock early and renders the destructor a no-op", + "[inkevent][lock]") +{ + Ptr m{new_ProxyMutex()}; + EThread *t = this_ethread(); + + { + SCOPED_MUTEX_LOCK(guard, m, t); + MUTEX_RELEASE(guard); + + REQUIRE(m->nthread_holding == 0); + REQUIRE(m->thread_holding == nullptr); + } + + REQUIRE(m->nthread_holding == 0); + REQUIRE(m->thread_holding == nullptr); +} + +TEST_CASE("MUTEX_TAKE_LOCK acquires an unheld ProxyMutex and records the calling thread as its holder", "[inkevent][lock]") +{ + Ptr m{new_ProxyMutex()}; + EThread *t = this_ethread(); + + MUTEX_TAKE_LOCK(m, t); + + REQUIRE(m->thread_holding == t); + REQUIRE(m->nthread_holding == 1); + + MUTEX_UNTAKE_LOCK(m, t); +} + +TEST_CASE("MUTEX_TAKE_LOCK by the holding thread is reentrant and increments the reentry count without blocking", + "[inkevent][lock]") +{ + Ptr m{new_ProxyMutex()}; + EThread *t = this_ethread(); + + MUTEX_TAKE_LOCK(m, t); + MUTEX_TAKE_LOCK(m, t); + + REQUIRE(m->nthread_holding == 2); + REQUIRE(m->thread_holding == t); + + MUTEX_UNTAKE_LOCK(m, t); + MUTEX_UNTAKE_LOCK(m, t); + + REQUIRE(m->nthread_holding == 0); + REQUIRE(m->thread_holding == nullptr); +} + +TEST_CASE("After the holding EThread fully releases a contended ProxyMutex, MUTEX_TRY_LOCK on the main thread succeeds", + "[inkevent][lock][multithread]") +{ + Ptr contended{new_ProxyMutex()}; + Ptr cont_self{new_ProxyMutex()}; + HoldOnEThread holder{cont_self.get(), contended}; + + REQUIRE(eventProcessor.schedule_imm(&holder, ET_CALL) != nullptr); + REQUIRE(holder.held.wait_until_set()); + + holder.release.set(); + REQUIRE(holder.done.wait_until_set()); + + EThread *t = this_ethread(); + MUTEX_TRY_LOCK(guard, contended, t); + + REQUIRE(guard.is_locked()); + REQUIRE(contended->thread_holding == t); +} + +TEST_CASE("WEAK_SCOPED_MUTEX_LOCK on a non-null ProxyMutex acquires the lock during construction", "[inkevent][lock]") +{ + Ptr m{new_ProxyMutex()}; + EThread *t = this_ethread(); + + WEAK_SCOPED_MUTEX_LOCK(guard, m, t); + + REQUIRE(m->thread_holding == t); + REQUIRE(m->nthread_holding == 1); +} + +TEST_CASE("A WEAK_SCOPED_MUTEX_LOCK guard releases its lock when its enclosing scope ends", "[inkevent][lock]") +{ + Ptr m{new_ProxyMutex()}; + EThread *t = this_ethread(); + + { + WEAK_SCOPED_MUTEX_LOCK(guard, m, t); + REQUIRE(m->nthread_holding == 1); + } + + REQUIRE(m->nthread_holding == 0); + REQUIRE(m->thread_holding == nullptr); +} + +TEST_CASE("WEAK_SCOPED_MUTEX_LOCK on a null Ptr is a no-op that takes and releases no lock", "[inkevent][lock]") +{ + Ptr empty; + Ptr witness{new_ProxyMutex()}; + EThread *t = this_ethread(); + + REQUIRE(empty.get() == nullptr); + + { + WEAK_SCOPED_MUTEX_LOCK(guard, empty, t); + REQUIRE(witness->thread_holding == nullptr); + REQUIRE(witness->nthread_holding == 0); + } + + REQUIRE(witness->thread_holding == nullptr); + REQUIRE(witness->nthread_holding == 0); +} + +TEST_CASE("WEAK_MUTEX_TRY_LOCK on an unheld ProxyMutex constructs a guard whose is_locked() reports the successful acquisition", + "[inkevent][lock]") +{ + Ptr m{new_ProxyMutex()}; + EThread *t = this_ethread(); + + WEAK_MUTEX_TRY_LOCK(guard, m, t); + + REQUIRE(guard.is_locked()); + REQUIRE(m->thread_holding == t); + REQUIRE(m->nthread_holding == 1); +} + +TEST_CASE("WEAK_MUTEX_TRY_LOCK against a contended ProxyMutex constructs a guard whose is_locked() reports the failed acquisition", + "[inkevent][lock][multithread]") +{ + Ptr contended{new_ProxyMutex()}; + Ptr cont_self{new_ProxyMutex()}; + HoldOnEThread holder{cont_self.get(), contended}; + + REQUIRE(eventProcessor.schedule_imm(&holder, ET_CALL) != nullptr); + REQUIRE(holder.held.wait_until_set()); + + EThread *t = this_ethread(); + WEAK_MUTEX_TRY_LOCK(guard, contended, t); + + REQUIRE_FALSE(guard.is_locked()); + + holder.release.set(); + REQUIRE(holder.done.wait_until_set()); +} + +TEST_CASE("WEAK_MUTEX_TRY_LOCK on a null Ptr reports is_locked() == true while taking no lock", "[inkevent][lock]") +{ + Ptr empty; + Ptr witness{new_ProxyMutex()}; + EThread *t = this_ethread(); + + REQUIRE(empty.get() == nullptr); + + WEAK_MUTEX_TRY_LOCK(guard, empty, t); + + REQUIRE(guard.is_locked()); + REQUIRE(witness->thread_holding == nullptr); + REQUIRE(witness->nthread_holding == 0); +} From 5fac80afa22605c819d710c6a0074315235a0e76 Mon Sep 17 00:00:00 2001 From: Josiah VanderZee Date: Thu, 30 Jul 2026 08:38:38 -0500 Subject: [PATCH 02/14] Narrow improper tests to be useful --- .../eventsystem/unit_tests/test_Lock.cc | 36 ++++++++----------- 1 file changed, 14 insertions(+), 22 deletions(-) diff --git a/src/iocore/eventsystem/unit_tests/test_Lock.cc b/src/iocore/eventsystem/unit_tests/test_Lock.cc index ce835e6c635..0efd8e0c7a3 100644 --- a/src/iocore/eventsystem/unit_tests/test_Lock.cc +++ b/src/iocore/eventsystem/unit_tests/test_Lock.cc @@ -291,24 +291,6 @@ TEST_CASE("A WEAK_SCOPED_MUTEX_LOCK guard releases its lock when its enclosing s REQUIRE(m->thread_holding == nullptr); } -TEST_CASE("WEAK_SCOPED_MUTEX_LOCK on a null Ptr is a no-op that takes and releases no lock", "[inkevent][lock]") -{ - Ptr empty; - Ptr witness{new_ProxyMutex()}; - EThread *t = this_ethread(); - - REQUIRE(empty.get() == nullptr); - - { - WEAK_SCOPED_MUTEX_LOCK(guard, empty, t); - REQUIRE(witness->thread_holding == nullptr); - REQUIRE(witness->nthread_holding == 0); - } - - REQUIRE(witness->thread_holding == nullptr); - REQUIRE(witness->nthread_holding == 0); -} - TEST_CASE("WEAK_MUTEX_TRY_LOCK on an unheld ProxyMutex constructs a guard whose is_locked() reports the successful acquisition", "[inkevent][lock]") { @@ -341,10 +323,9 @@ TEST_CASE("WEAK_MUTEX_TRY_LOCK against a contended ProxyMutex constructs a guard REQUIRE(holder.done.wait_until_set()); } -TEST_CASE("WEAK_MUTEX_TRY_LOCK on a null Ptr reports is_locked() == true while taking no lock", "[inkevent][lock]") +TEST_CASE("WEAK_MUTEX_TRY_LOCK on a null Ptr reports is_locked() == true", "[inkevent][lock]") { Ptr empty; - Ptr witness{new_ProxyMutex()}; EThread *t = this_ethread(); REQUIRE(empty.get() == nullptr); @@ -352,6 +333,17 @@ TEST_CASE("WEAK_MUTEX_TRY_LOCK on a null Ptr reports is_locked() == WEAK_MUTEX_TRY_LOCK(guard, empty, t); REQUIRE(guard.is_locked()); - REQUIRE(witness->thread_holding == nullptr); - REQUIRE(witness->nthread_holding == 0); +} + +TEST_CASE("MUTEX_RELEASE on a null WEAK_MUTEX_TRY_LOCK guard clears is_locked() without unlocking a mutex", "[inkevent][lock]") +{ + Ptr empty; + EThread *t = this_ethread(); + + WEAK_MUTEX_TRY_LOCK(guard, empty, t); + REQUIRE(guard.is_locked()); + + MUTEX_RELEASE(guard); + + REQUIRE_FALSE(guard.is_locked()); } From 7f4755c4003caa8fcd913737b4c6043e36ae22ee Mon Sep 17 00:00:00 2001 From: Josiah VanderZee Date: Thu, 30 Jul 2026 09:56:46 -0500 Subject: [PATCH 03/14] Release lock before setting done flag --- src/iocore/eventsystem/unit_tests/test_Lock.cc | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/iocore/eventsystem/unit_tests/test_Lock.cc b/src/iocore/eventsystem/unit_tests/test_Lock.cc index 0efd8e0c7a3..41dac78948b 100644 --- a/src/iocore/eventsystem/unit_tests/test_Lock.cc +++ b/src/iocore/eventsystem/unit_tests/test_Lock.cc @@ -48,9 +48,11 @@ class HoldOnEThread : public Continuation int on_event(int /* event ATS_UNUSED */, void * /* data ATS_UNUSED */) { - SCOPED_MUTEX_LOCK(guard, target_mutex, this_ethread()); - held.set(); - release.wait_until_set(); + { + SCOPED_MUTEX_LOCK(guard, target_mutex, this_ethread()); + held.set(); + release.wait_until_set(); + } done.set(); return 0; } From 8cb5858fa30696568a25c9a7d3c9141185626195 Mon Sep 17 00:00:00 2001 From: Josiah VanderZee Date: Fri, 31 Jul 2026 07:56:18 -0500 Subject: [PATCH 04/14] Unfreeze waiting threads on exception --- src/iocore/eventsystem/unit_tests/test_Lock.cc | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/iocore/eventsystem/unit_tests/test_Lock.cc b/src/iocore/eventsystem/unit_tests/test_Lock.cc index 41dac78948b..690d29c1f52 100644 --- a/src/iocore/eventsystem/unit_tests/test_Lock.cc +++ b/src/iocore/eventsystem/unit_tests/test_Lock.cc @@ -39,6 +39,10 @@ class HoldOnEThread : public Continuation SET_HANDLER(&HoldOnEThread::on_event); } + // In case of an exception in a thread that would have set release, we set + // it here in order to unfreeze any threads that may be waiting on done. + ~HoldOnEThread() { release.set(); } + Ptr target_mutex; AtomicFlag held; AtomicFlag release; From 5991711a8ede0bc30b56ea8195bb7f6ce8f0b129 Mon Sep 17 00:00:00 2001 From: Josiah VanderZee Date: Fri, 31 Jul 2026 08:08:14 -0500 Subject: [PATCH 05/14] Treat wait timeout as test failure --- src/iocore/eventsystem/unit_tests/test_Lock.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/iocore/eventsystem/unit_tests/test_Lock.cc b/src/iocore/eventsystem/unit_tests/test_Lock.cc index 690d29c1f52..a116978caa4 100644 --- a/src/iocore/eventsystem/unit_tests/test_Lock.cc +++ b/src/iocore/eventsystem/unit_tests/test_Lock.cc @@ -55,7 +55,7 @@ class HoldOnEThread : public Continuation { SCOPED_MUTEX_LOCK(guard, target_mutex, this_ethread()); held.set(); - release.wait_until_set(); + REQUIRE(release.wait_until_set()); } done.set(); return 0; From 9230724e34c9c2cfe6a787f5f6bd34df71834263 Mon Sep 17 00:00:00 2001 From: Josiah VanderZee Date: Fri, 31 Jul 2026 08:16:27 -0500 Subject: [PATCH 06/14] Use non-fatal assertion for timeout --- src/iocore/eventsystem/unit_tests/test_Lock.cc | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/iocore/eventsystem/unit_tests/test_Lock.cc b/src/iocore/eventsystem/unit_tests/test_Lock.cc index a116978caa4..9f4c09db3e6 100644 --- a/src/iocore/eventsystem/unit_tests/test_Lock.cc +++ b/src/iocore/eventsystem/unit_tests/test_Lock.cc @@ -55,7 +55,8 @@ class HoldOnEThread : public Continuation { SCOPED_MUTEX_LOCK(guard, target_mutex, this_ethread()); held.set(); - REQUIRE(release.wait_until_set()); + // Non-fatal so that done gets set. + CHECK(release.wait_until_set()); } done.set(); return 0; From b0aef94fc6b8b6191f1f5cd471156274c9acb50b Mon Sep 17 00:00:00 2001 From: Josiah VanderZee Date: Fri, 31 Jul 2026 09:18:46 -0500 Subject: [PATCH 07/14] Enable thread safe assertions --- src/iocore/eventsystem/unit_tests/test_Lock.cc | 1 + 1 file changed, 1 insertion(+) diff --git a/src/iocore/eventsystem/unit_tests/test_Lock.cc b/src/iocore/eventsystem/unit_tests/test_Lock.cc index 9f4c09db3e6..500ea612926 100644 --- a/src/iocore/eventsystem/unit_tests/test_Lock.cc +++ b/src/iocore/eventsystem/unit_tests/test_Lock.cc @@ -21,6 +21,7 @@ limitations under the License. */ +#define CATCH_CONFIG_THREAD_SAFE_ASSERTIONS #include "inkevent_test_fixtures.h" using inkevent_test::AtomicFlag; From 9fa7ae1c417a7504d024910f1dc21d18ba4d7c74 Mon Sep 17 00:00:00 2001 From: Josiah VanderZee Date: Fri, 31 Jul 2026 09:51:31 -0500 Subject: [PATCH 08/14] Wait for cb to complete before cleanup --- src/iocore/eventsystem/unit_tests/test_Lock.cc | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/iocore/eventsystem/unit_tests/test_Lock.cc b/src/iocore/eventsystem/unit_tests/test_Lock.cc index 500ea612926..1f2c72fee3d 100644 --- a/src/iocore/eventsystem/unit_tests/test_Lock.cc +++ b/src/iocore/eventsystem/unit_tests/test_Lock.cc @@ -42,7 +42,11 @@ class HoldOnEThread : public Continuation // In case of an exception in a thread that would have set release, we set // it here in order to unfreeze any threads that may be waiting on done. - ~HoldOnEThread() { release.set(); } + ~HoldOnEThread() + { + release.set(); + done.wait_until_set(); + } Ptr target_mutex; AtomicFlag held; From 231c74d96a1a970dbb0a1df18bee222b2c3bb8ab Mon Sep 17 00:00:00 2001 From: Josiah VanderZee Date: Fri, 31 Jul 2026 10:19:45 -0500 Subject: [PATCH 09/14] Do not throw while holding lock --- src/iocore/eventsystem/unit_tests/test_Lock.cc | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/iocore/eventsystem/unit_tests/test_Lock.cc b/src/iocore/eventsystem/unit_tests/test_Lock.cc index 1f2c72fee3d..3954b60b044 100644 --- a/src/iocore/eventsystem/unit_tests/test_Lock.cc +++ b/src/iocore/eventsystem/unit_tests/test_Lock.cc @@ -141,7 +141,7 @@ TEST_CASE("MUTEX_RELEASE on a MUTEX_TRY_LOCK guard releases the lock early and f EThread *t = this_ethread(); MUTEX_TRY_LOCK(guard, m, t); - REQUIRE(guard.is_locked()); + CHECK(guard.is_locked()); MUTEX_RELEASE(guard); @@ -157,7 +157,7 @@ TEST_CASE("MUTEX_RELEASE invoked twice on the same MUTEX_TRY_LOCK guard is a no- MUTEX_TRY_LOCK(guard, m, t); MUTEX_RELEASE(guard); - REQUIRE_FALSE(guard.is_locked()); + CHECK_FALSE(guard.is_locked()); MUTEX_RELEASE(guard); @@ -353,7 +353,7 @@ TEST_CASE("MUTEX_RELEASE on a null WEAK_MUTEX_TRY_LOCK guard clears is_locked() EThread *t = this_ethread(); WEAK_MUTEX_TRY_LOCK(guard, empty, t); - REQUIRE(guard.is_locked()); + CHECK(guard.is_locked()); MUTEX_RELEASE(guard); From bf625b2857f662a06b2fd0f832e2f279069aab7d Mon Sep 17 00:00:00 2001 From: Josiah VanderZee Date: Tue, 4 Aug 2026 07:49:18 -0500 Subject: [PATCH 10/14] Harden `HoldOnEThread` utility Claude did not do a good job of handling edge cases cleanly. I have manually rewritten parts of `HoldOnEThread` to shut down cleanly in case of an exception, to avoid distracting side effects if a test fails. --- .../eventsystem/unit_tests/test_Lock.cc | 68 ++++++++++++------- 1 file changed, 44 insertions(+), 24 deletions(-) diff --git a/src/iocore/eventsystem/unit_tests/test_Lock.cc b/src/iocore/eventsystem/unit_tests/test_Lock.cc index 3954b60b044..4ca4a7a6189 100644 --- a/src/iocore/eventsystem/unit_tests/test_Lock.cc +++ b/src/iocore/eventsystem/unit_tests/test_Lock.cc @@ -38,34 +38,65 @@ class HoldOnEThread : public Continuation HoldOnEThread(ProxyMutex *self_mutex, Ptr &target) : Continuation(self_mutex), target_mutex(target) { SET_HANDLER(&HoldOnEThread::on_event); + this->callback_action = eventProcessor.schedule_imm(this, ET_CALL); } // In case of an exception in a thread that would have set release, we set // it here in order to unfreeze any threads that may be waiting on done. ~HoldOnEThread() { - release.set(); - done.wait_until_set(); + this->release.set(); + this->cancel_callback(); + this->done.wait_until_set(); } + bool + wait_for_callback_start() + { + return this->held.wait_until_set(); + } + + bool + wait_for_callback_finish() + { + this->release.set(); + return this->done.wait_until_set(); + } + +private: + Action *callback_action{}; Ptr target_mutex; AtomicFlag held; AtomicFlag release; AtomicFlag done; -private: int on_event(int /* event ATS_UNUSED */, void * /* data ATS_UNUSED */) { - { - SCOPED_MUTEX_LOCK(guard, target_mutex, this_ethread()); - held.set(); - // Non-fatal so that done gets set. - CHECK(release.wait_until_set()); + SCOPED_MUTEX_LOCK(guard, this->target_mutex, this_ethread()); + this->held.set(); + if (this->release.wait_until_set()) { + MUTEX_RELEASE(guard); + this->done.set(); } - done.set(); return 0; } + + void + cancel_callback() + { + SCOPED_MUTEX_LOCK(guard, this->mutex, this_ethread()); + if (this->is_expecting_callback()) { + this->callback_action->cancel(this); + this->done.set(); + } + } + + bool + is_expecting_callback() + { + return !this->held.is_set() && !this->callback_action->cancelled; + } }; } // namespace @@ -90,16 +121,12 @@ TEST_CASE("MUTEX_TRY_LOCK against a contended ProxyMutex constructs a guard whos Ptr cont_self{new_ProxyMutex()}; HoldOnEThread holder{cont_self.get(), contended}; - REQUIRE(eventProcessor.schedule_imm(&holder, ET_CALL) != nullptr); - REQUIRE(holder.held.wait_until_set()); + REQUIRE(holder.wait_for_callback_start()); EThread *t = this_ethread(); MUTEX_TRY_LOCK(guard, contended, t); REQUIRE_FALSE(guard.is_locked()); - - holder.release.set(); - REQUIRE(holder.done.wait_until_set()); } TEST_CASE("MUTEX_TRY_LOCK by the holding thread is reentrant and returns a guard reporting is_locked() == true", "[inkevent][lock]") @@ -265,11 +292,8 @@ TEST_CASE("After the holding EThread fully releases a contended ProxyMutex, MUTE Ptr cont_self{new_ProxyMutex()}; HoldOnEThread holder{cont_self.get(), contended}; - REQUIRE(eventProcessor.schedule_imm(&holder, ET_CALL) != nullptr); - REQUIRE(holder.held.wait_until_set()); - - holder.release.set(); - REQUIRE(holder.done.wait_until_set()); + REQUIRE(holder.wait_for_callback_start()); + REQUIRE(holder.wait_for_callback_finish()); EThread *t = this_ethread(); MUTEX_TRY_LOCK(guard, contended, t); @@ -323,16 +347,12 @@ TEST_CASE("WEAK_MUTEX_TRY_LOCK against a contended ProxyMutex constructs a guard Ptr cont_self{new_ProxyMutex()}; HoldOnEThread holder{cont_self.get(), contended}; - REQUIRE(eventProcessor.schedule_imm(&holder, ET_CALL) != nullptr); - REQUIRE(holder.held.wait_until_set()); + REQUIRE(holder.wait_for_callback_start()); EThread *t = this_ethread(); WEAK_MUTEX_TRY_LOCK(guard, contended, t); REQUIRE_FALSE(guard.is_locked()); - - holder.release.set(); - REQUIRE(holder.done.wait_until_set()); } TEST_CASE("WEAK_MUTEX_TRY_LOCK on a null Ptr reports is_locked() == true", "[inkevent][lock]") From bb9a9622e9a53da0ddc001cec692502edd3c6de7 Mon Sep 17 00:00:00 2001 From: Josiah VanderZee Date: Tue, 4 Aug 2026 10:09:56 -0500 Subject: [PATCH 11/14] Add assertions and comments for Copilot --- include/iocore/eventsystem/Action.h | 312 ++++++++++++------ .../eventsystem/unit_tests/test_Lock.cc | 7 + 2 files changed, 216 insertions(+), 103 deletions(-) diff --git a/include/iocore/eventsystem/Action.h b/include/iocore/eventsystem/Action.h index 04f2788cb04..7659f55e277 100644 --- a/include/iocore/eventsystem/Action.h +++ b/include/iocore/eventsystem/Action.h @@ -28,113 +28,109 @@ #include "iocore/eventsystem/Continuation.h" /** - Represents an operation initiated on a Processor. - - The Action class is an abstract representation of an operation - being executed by some Processor. A reference to an Action object - allows you to cancel an ongoing asynchronous operation before it - completes. This means that the Continuation specified for the - operation will not be called back. - - Actions or classes derived from Action are the typical return - type of methods exposed by Processors in the Event System and - throughout the IO Core libraries. - - The canceller of an action must be the state machine that will - be called back by the task and that state machine's lock must be - held while calling cancel. - - Processor implementers: - - You must ensure that no events are sent to the state machine after - the operation has been cancelled appropriately. - - Returning an Action: - - Processor functions that are asynchronous must return actions to - allow the calling state machine to cancel the task before completion. - Because some processor functions are reentrant, they can call - back the state machine before the returning from the call that - creates the actions. To handle this case, special values are - returned in place of an action to indicate to the state machine - that the action is already completed. - - - @b ACTION_RESULT_DONE The processor has completed the task - and called the state machine back inline. - - @b ACTION_RESULT_INLINE Not currently used. - - @b ACTION_RESULT_IO_ERROR Not currently used. - - To make matters more complicated, it's possible if the result is - ACTION_RESULT_DONE that state machine deallocated itself on the - reentrant callback. Thus, state machine implementers MUST either - use a scheme to never deallocate their machines on reentrant - callbacks OR immediately check the returned action when creating - an asynchronous task and if it is ACTION_RESULT_DONE neither read - nor write any state variables. With either method, it's imperative - that the returned action always be checked for special values and - the value handled accordingly. - - Allocation policy: - - Actions are allocated by the Processor performing the actions. - It is the processor's responsibility to handle deallocation once - the action is complete or cancelled. A state machine MUST NOT - access an action once the operation that returned the Action has - completed or it has cancelled the Action. - - Action pointer sanity checks must also check whether the lowest - bit of the pointer is 1. If it is 1, then the value must not be - treated as a pointer, and should be used as one of the values - defined below (e.g. ACTION_RESULT_DONE). - + Handle to an in-flight asynchronous operation. + + An Action is returned by a Processor when it accepts an asynchronous + request from a Continuation. Holding the Action lets the Continuation + cancel the operation before it completes; once cancelled, the Continuation + will not be called back for that operation. + + Processors that derive from Action attach additional state to the handle. + Processors that complete a request synchronously (re-entrantly) MAY return + a sentinel @c Action* (see @c MAKE_ACTION_RESULT) instead of a real + pointer; callers MUST check the low bit of the returned pointer to + distinguish a real Action from a sentinel before dereferencing. + + @par Ownership + Allocated by the Processor that returned the Action; deallocated by that + same Processor when the operation completes or is cancelled. Callers + MUST NOT delete an Action* and MUST NOT access an Action after the + operation it represents has completed or after they have called cancel(). + + @par Thread Safety + Not instance-thread-safe. The Continuation that initiated the Action is + the only legitimate canceller, and it MUST hold its own @c ProxyMutex + (the same mutex stored in @c Action::mutex) while calling cancel(). The + Processor MUST guarantee that no callbacks are delivered to a cancelled + Action. */ class Action { public: /** - Continuation that initiated this action. - - The reference to the initiating continuation is only used to - verify that the action is being cancelled by the correct - continuation. This field should not be accessed or modified - directly by the state machine. - + The Continuation that initiated this Action. + + @par Thread Safety + The owning Processor binds this field (via @c operator= or by + direct assignment) before the Action is made observable to a + canceller, and the value is stable from that point until the + Processor releases the Action. Rebinding (including to nullptr) + is the Processor's responsibility and MUST be serialized against + any concurrent cancel(). */ Continuation *continuation = nullptr; /** - Reference to the Continuation's lock. - - Keeps a reference to the Continuation's lock to preserve the - access to the cancelled field valid even when the state machine - has been deallocated. This field should not be accessed or - modified directly by the state machine. - + A retained reference to the initiating Continuation's @c ProxyMutex. + + Held independently of @c continuation so that @c cancelled remains + accessible under a valid lock even after the initiating Continuation + has been deallocated. + + @par Thread Safety + The owning Processor binds this field (via @c operator= or by + direct assignment) before the Action is made observable to a + canceller, and the value is stable from that point until the + Processor releases the Action. Rebinding (including to nullptr) + is the Processor's responsibility and MUST be serialized against + any concurrent cancel(). The retained reference keeps the + @c ProxyMutex alive for as long as it is held. */ Ptr mutex; /** - Internal flag used to indicate whether the action has been - cancelled. - - This flag is set after a call to cancel or cancel_action and - it should not be accessed or modified directly by the state - machine. - + Set to true after cancel() or cancel_action() is invoked. Initially + false. + + The owning Processor MAY clear this flag back to false when recycling + the Action for a new operation, before the recycled Action is + published to a canceller. + + @par Thread Safety + Plain @c bool. Readers and writers MUST hold @c this->mutex, except + that the owning Processor MAY clear it without the lock while the + Action is not yet published to any canceller. The Processor MUST + inspect this flag under @c this->mutex immediately before invoking + @c continuation, and MUST NOT invoke @c continuation if the flag is + set. */ bool cancelled = false; /** - Cancels the asynchronous operation represented by this action. + Cancels the asynchronous operation represented by this Action. + + After a successful return, no callback for this operation will be + delivered to @c continuation. Derived Processors may override this + method to release additional resources before flagging the Action as + cancelled. + + @param c The Continuation associated with this Action, or nullptr. + If non-null, MUST equal @c this->continuation. - This method is called by state machines willing to cancel an - ongoing asynchronous operation. Classes derived from Action may - perform additional steps before flagging this action as cancelled. - There are certain rules that must be followed in order to cancel - an action (see the Remarks section). + @pre @c this->cancelled is false. + @pre Caller MUST hold @c this->mutex. + @pre Caller is the Continuation referenced by @c this->continuation, + i.e. the same Continuation that initiated the Action. + @post @c this->cancelled is true. The Processor will not invoke + @c continuation for this operation. - @param c Continuation associated with this Action. + @par Errors + Cannot fail. Precondition violations are checked by @c ink_assert + in debug builds and produce undefined behavior in release builds. + @par Thread Safety + Caller-synchronized. The caller MUST hold @c this->mutex. Concurrent + cancellation from multiple threads is a precondition violation. */ virtual void cancel(Continuation *c = nullptr) @@ -145,15 +141,31 @@ class Action } /** - Cancels the asynchronous operation represented by this action. - - This method is called by state machines willing to cancel an - ongoing asynchronous operation. There are certain rules that - must be followed in order to cancel an action (see the Remarks - section). - - @param c Continuation associated with this Action. - + Flags the Action as cancelled without invoking any derived-class + cancellation logic. + + Performs only the base cancellation: marks the Action cancelled so + that the Processor will not invoke @c continuation for this + operation. Any cleanup that a derived Processor performs from its + overridden @c cancel() is skipped. + + @param c The Continuation associated with this Action, or nullptr. + If non-null, MUST equal @c this->continuation. + + @pre @c this->cancelled is false. + @pre Caller MUST hold @c this->mutex. + @pre Caller is the Continuation referenced by @c this->continuation, + i.e. the same Continuation that initiated the Action. + @post @c this->cancelled is true. The Processor will not invoke + @c continuation for this operation. + + @par Errors + Cannot fail. Precondition violations are checked by @c ink_assert + in debug builds and produce undefined behavior in release builds. + + @par Thread Safety + Caller-synchronized. The caller MUST hold @c this->mutex. Concurrent + cancellation from multiple threads is a precondition violation. */ void cancel_action(Continuation *c = nullptr) @@ -163,6 +175,28 @@ class Action cancelled = true; } + /** + Binds this Action to a Continuation and retains a reference to that + Continuation's mutex. + + @param acont The Continuation that will cancel and be called back on + this Action. May be nullptr to detach. + @return @p acont, for assignment chaining. + + @pre None. + @post @c this->continuation == @p acont. If @p acont is non-null, + @c this->mutex refers to the same @c ProxyMutex as + @c acont->mutex; otherwise @c this->mutex is null. The + @c cancelled flag is unchanged. + + @par Errors + Cannot fail. + + @par Thread Safety + Intended to be invoked by the Processor that owns the Action, before + the Action is published to other threads. Not safe against concurrent + cancel() once the Action has been published. + */ Continuation * operator=(Continuation *acont) { @@ -176,20 +210,92 @@ class Action } /** - Constructor of the Action object. Processor implementers are - responsible for associating this action with the proper - Continuation. + Constructs an Action with no associated Continuation and no retained + mutex. + + @pre None. + @post @c continuation is nullptr; @c mutex is null; + @c cancelled is false. + + @par Errors + Cannot fail. + @par Thread Safety + Safe to call from any thread. */ Action() {} + + /** + Releases the retained reference to the Continuation's mutex. + + @pre The Action's owning Processor has completed its lifecycle for + this Action (the operation has finished or has been cancelled). + @post The retained @c ProxyMutex reference is dropped; if this was + the last reference, the @c ProxyMutex is destroyed. + + @par Errors + Cannot fail. + + @par Thread Safety + The owning Processor invokes destruction; it MUST guarantee no other + thread accesses the Action concurrently. + */ virtual ~Action() {} }; +/** + Sentinel return value: the Processor completed the request inline and + has already invoked the Continuation. The caller MUST treat the returned + pointer as a tag, not a real @c Action*. Equivalent to + @c MAKE_ACTION_RESULT(1). + + When a caller observes this sentinel, the Continuation may have been + deallocated during the inline callback; the caller MUST NOT touch any + state that the Continuation owned after seeing this value. +*/ #define ACTION_RESULT_DONE MAKE_ACTION_RESULT(1) -#define ACTION_IO_ERROR MAKE_ACTION_RESULT(2) -// Use these classes by -// #define ACTION_RESULT_HOST_DB_OFFLINE -// MAKE_ACTION_RESULT(ACTION_RESULT_HOST_DB_BASE + 0) +/** + Sentinel return value: the Processor failed the request inline with an + I/O error and has already invoked the Continuation with a + Processor-specific error event. The caller MUST treat the returned + pointer as a tag, not a real @c Action*. Equivalent to + @c MAKE_ACTION_RESULT(2). + + When a caller observes this sentinel, the Continuation may have been + deallocated during the inline callback; the caller MUST NOT touch any + state that the Continuation owned after seeing this value. +*/ +#define ACTION_IO_ERROR MAKE_ACTION_RESULT(2) + +// Processors that need additional sentinels define them with +// MAKE_ACTION_RESULT, e.g. +// #define MY_PROCESSOR_BASE 3 +// #define ACTION_RESULT_MY_FAILURE MAKE_ACTION_RESULT(MY_PROCESSOR_BASE + 0) + +/** + Constructs a sentinel @c Action* from a small integer. + + The encoding shifts @p _x left by one bit and sets the low bit, so + every sentinel has bit 0 set and is therefore distinguishable from any + validly-aligned @c Action*. A receiver of an @c Action* MUST inspect + @c ((uintptr_t)p & 1) before dereferencing: when set, the value is a + sentinel that MUST be compared against @c ACTION_RESULT_* constants + rather than treated as an @c Action. + Sentinel values MUST NOT collide; the convention is for each Processor + that defines its own sentinels to reserve a numeric base constant and + define sentinels relative to that base. + + @param _x A non-negative integer expression. After the @c "<<1" shift + and @c "+1", the result MUST fit in @c uintptr_t. Behavior + for values that overflow the shift is undefined. + + @par Errors + Cannot fail. + + @par Thread Safety + Safe to evaluate from any thread; the expansion is a pure expression + with no shared state. +*/ #define MAKE_ACTION_RESULT(_x) (Action *)(((uintptr_t)((_x << 1) + 1))) diff --git a/src/iocore/eventsystem/unit_tests/test_Lock.cc b/src/iocore/eventsystem/unit_tests/test_Lock.cc index 4ca4a7a6189..f4197580e9d 100644 --- a/src/iocore/eventsystem/unit_tests/test_Lock.cc +++ b/src/iocore/eventsystem/unit_tests/test_Lock.cc @@ -24,6 +24,8 @@ #define CATCH_CONFIG_THREAD_SAFE_ASSERTIONS #include "inkevent_test_fixtures.h" +#include + using inkevent_test::AtomicFlag; using inkevent_test::EventProcessorListener; @@ -39,6 +41,7 @@ class HoldOnEThread : public Continuation { SET_HANDLER(&HoldOnEThread::on_event); this->callback_action = eventProcessor.schedule_imm(this, ET_CALL); + ink_assert(this->callback_action != nullptr); } // In case of an exception in a thread that would have set release, we set @@ -60,6 +63,8 @@ class HoldOnEThread : public Continuation wait_for_callback_finish() { this->release.set(); + // The callback can finish without setting done due to wait timeouts. We + // return false in that case. return this->done.wait_until_set(); } @@ -88,6 +93,7 @@ class HoldOnEThread : public Continuation SCOPED_MUTEX_LOCK(guard, this->mutex, this_ethread()); if (this->is_expecting_callback()) { this->callback_action->cancel(this); + ink_assert(this->callback_action->cancelled); this->done.set(); } } @@ -95,6 +101,7 @@ class HoldOnEThread : public Continuation bool is_expecting_callback() { + ink_assert(this->mutex->thread_holding == this_ethread()); return !this->held.is_set() && !this->callback_action->cancelled; } }; From bf255c38805ee24a2321d98fc2ca64071ec67b55 Mon Sep 17 00:00:00 2001 From: Josiah VanderZee Date: Tue, 4 Aug 2026 10:32:44 -0500 Subject: [PATCH 12/14] Revert documentation --- include/iocore/eventsystem/Action.h | 312 +++++++++------------------- 1 file changed, 103 insertions(+), 209 deletions(-) diff --git a/include/iocore/eventsystem/Action.h b/include/iocore/eventsystem/Action.h index 7659f55e277..04f2788cb04 100644 --- a/include/iocore/eventsystem/Action.h +++ b/include/iocore/eventsystem/Action.h @@ -28,109 +28,113 @@ #include "iocore/eventsystem/Continuation.h" /** - Handle to an in-flight asynchronous operation. - - An Action is returned by a Processor when it accepts an asynchronous - request from a Continuation. Holding the Action lets the Continuation - cancel the operation before it completes; once cancelled, the Continuation - will not be called back for that operation. - - Processors that derive from Action attach additional state to the handle. - Processors that complete a request synchronously (re-entrantly) MAY return - a sentinel @c Action* (see @c MAKE_ACTION_RESULT) instead of a real - pointer; callers MUST check the low bit of the returned pointer to - distinguish a real Action from a sentinel before dereferencing. - - @par Ownership - Allocated by the Processor that returned the Action; deallocated by that - same Processor when the operation completes or is cancelled. Callers - MUST NOT delete an Action* and MUST NOT access an Action after the - operation it represents has completed or after they have called cancel(). - - @par Thread Safety - Not instance-thread-safe. The Continuation that initiated the Action is - the only legitimate canceller, and it MUST hold its own @c ProxyMutex - (the same mutex stored in @c Action::mutex) while calling cancel(). The - Processor MUST guarantee that no callbacks are delivered to a cancelled - Action. + Represents an operation initiated on a Processor. + + The Action class is an abstract representation of an operation + being executed by some Processor. A reference to an Action object + allows you to cancel an ongoing asynchronous operation before it + completes. This means that the Continuation specified for the + operation will not be called back. + + Actions or classes derived from Action are the typical return + type of methods exposed by Processors in the Event System and + throughout the IO Core libraries. + + The canceller of an action must be the state machine that will + be called back by the task and that state machine's lock must be + held while calling cancel. + + Processor implementers: + + You must ensure that no events are sent to the state machine after + the operation has been cancelled appropriately. + + Returning an Action: + + Processor functions that are asynchronous must return actions to + allow the calling state machine to cancel the task before completion. + Because some processor functions are reentrant, they can call + back the state machine before the returning from the call that + creates the actions. To handle this case, special values are + returned in place of an action to indicate to the state machine + that the action is already completed. + + - @b ACTION_RESULT_DONE The processor has completed the task + and called the state machine back inline. + - @b ACTION_RESULT_INLINE Not currently used. + - @b ACTION_RESULT_IO_ERROR Not currently used. + + To make matters more complicated, it's possible if the result is + ACTION_RESULT_DONE that state machine deallocated itself on the + reentrant callback. Thus, state machine implementers MUST either + use a scheme to never deallocate their machines on reentrant + callbacks OR immediately check the returned action when creating + an asynchronous task and if it is ACTION_RESULT_DONE neither read + nor write any state variables. With either method, it's imperative + that the returned action always be checked for special values and + the value handled accordingly. + + Allocation policy: + + Actions are allocated by the Processor performing the actions. + It is the processor's responsibility to handle deallocation once + the action is complete or cancelled. A state machine MUST NOT + access an action once the operation that returned the Action has + completed or it has cancelled the Action. + + Action pointer sanity checks must also check whether the lowest + bit of the pointer is 1. If it is 1, then the value must not be + treated as a pointer, and should be used as one of the values + defined below (e.g. ACTION_RESULT_DONE). + */ class Action { public: /** - The Continuation that initiated this Action. - - @par Thread Safety - The owning Processor binds this field (via @c operator= or by - direct assignment) before the Action is made observable to a - canceller, and the value is stable from that point until the - Processor releases the Action. Rebinding (including to nullptr) - is the Processor's responsibility and MUST be serialized against - any concurrent cancel(). + Continuation that initiated this action. + + The reference to the initiating continuation is only used to + verify that the action is being cancelled by the correct + continuation. This field should not be accessed or modified + directly by the state machine. + */ Continuation *continuation = nullptr; /** - A retained reference to the initiating Continuation's @c ProxyMutex. - - Held independently of @c continuation so that @c cancelled remains - accessible under a valid lock even after the initiating Continuation - has been deallocated. - - @par Thread Safety - The owning Processor binds this field (via @c operator= or by - direct assignment) before the Action is made observable to a - canceller, and the value is stable from that point until the - Processor releases the Action. Rebinding (including to nullptr) - is the Processor's responsibility and MUST be serialized against - any concurrent cancel(). The retained reference keeps the - @c ProxyMutex alive for as long as it is held. + Reference to the Continuation's lock. + + Keeps a reference to the Continuation's lock to preserve the + access to the cancelled field valid even when the state machine + has been deallocated. This field should not be accessed or + modified directly by the state machine. + */ Ptr mutex; /** - Set to true after cancel() or cancel_action() is invoked. Initially - false. - - The owning Processor MAY clear this flag back to false when recycling - the Action for a new operation, before the recycled Action is - published to a canceller. - - @par Thread Safety - Plain @c bool. Readers and writers MUST hold @c this->mutex, except - that the owning Processor MAY clear it without the lock while the - Action is not yet published to any canceller. The Processor MUST - inspect this flag under @c this->mutex immediately before invoking - @c continuation, and MUST NOT invoke @c continuation if the flag is - set. + Internal flag used to indicate whether the action has been + cancelled. + + This flag is set after a call to cancel or cancel_action and + it should not be accessed or modified directly by the state + machine. + */ bool cancelled = false; /** - Cancels the asynchronous operation represented by this Action. - - After a successful return, no callback for this operation will be - delivered to @c continuation. Derived Processors may override this - method to release additional resources before flagging the Action as - cancelled. - - @param c The Continuation associated with this Action, or nullptr. - If non-null, MUST equal @c this->continuation. + Cancels the asynchronous operation represented by this action. - @pre @c this->cancelled is false. - @pre Caller MUST hold @c this->mutex. - @pre Caller is the Continuation referenced by @c this->continuation, - i.e. the same Continuation that initiated the Action. - @post @c this->cancelled is true. The Processor will not invoke - @c continuation for this operation. + This method is called by state machines willing to cancel an + ongoing asynchronous operation. Classes derived from Action may + perform additional steps before flagging this action as cancelled. + There are certain rules that must be followed in order to cancel + an action (see the Remarks section). - @par Errors - Cannot fail. Precondition violations are checked by @c ink_assert - in debug builds and produce undefined behavior in release builds. + @param c Continuation associated with this Action. - @par Thread Safety - Caller-synchronized. The caller MUST hold @c this->mutex. Concurrent - cancellation from multiple threads is a precondition violation. */ virtual void cancel(Continuation *c = nullptr) @@ -141,31 +145,15 @@ class Action } /** - Flags the Action as cancelled without invoking any derived-class - cancellation logic. - - Performs only the base cancellation: marks the Action cancelled so - that the Processor will not invoke @c continuation for this - operation. Any cleanup that a derived Processor performs from its - overridden @c cancel() is skipped. - - @param c The Continuation associated with this Action, or nullptr. - If non-null, MUST equal @c this->continuation. - - @pre @c this->cancelled is false. - @pre Caller MUST hold @c this->mutex. - @pre Caller is the Continuation referenced by @c this->continuation, - i.e. the same Continuation that initiated the Action. - @post @c this->cancelled is true. The Processor will not invoke - @c continuation for this operation. - - @par Errors - Cannot fail. Precondition violations are checked by @c ink_assert - in debug builds and produce undefined behavior in release builds. - - @par Thread Safety - Caller-synchronized. The caller MUST hold @c this->mutex. Concurrent - cancellation from multiple threads is a precondition violation. + Cancels the asynchronous operation represented by this action. + + This method is called by state machines willing to cancel an + ongoing asynchronous operation. There are certain rules that + must be followed in order to cancel an action (see the Remarks + section). + + @param c Continuation associated with this Action. + */ void cancel_action(Continuation *c = nullptr) @@ -175,28 +163,6 @@ class Action cancelled = true; } - /** - Binds this Action to a Continuation and retains a reference to that - Continuation's mutex. - - @param acont The Continuation that will cancel and be called back on - this Action. May be nullptr to detach. - @return @p acont, for assignment chaining. - - @pre None. - @post @c this->continuation == @p acont. If @p acont is non-null, - @c this->mutex refers to the same @c ProxyMutex as - @c acont->mutex; otherwise @c this->mutex is null. The - @c cancelled flag is unchanged. - - @par Errors - Cannot fail. - - @par Thread Safety - Intended to be invoked by the Processor that owns the Action, before - the Action is published to other threads. Not safe against concurrent - cancel() once the Action has been published. - */ Continuation * operator=(Continuation *acont) { @@ -210,92 +176,20 @@ class Action } /** - Constructs an Action with no associated Continuation and no retained - mutex. - - @pre None. - @post @c continuation is nullptr; @c mutex is null; - @c cancelled is false. - - @par Errors - Cannot fail. + Constructor of the Action object. Processor implementers are + responsible for associating this action with the proper + Continuation. - @par Thread Safety - Safe to call from any thread. */ Action() {} - - /** - Releases the retained reference to the Continuation's mutex. - - @pre The Action's owning Processor has completed its lifecycle for - this Action (the operation has finished or has been cancelled). - @post The retained @c ProxyMutex reference is dropped; if this was - the last reference, the @c ProxyMutex is destroyed. - - @par Errors - Cannot fail. - - @par Thread Safety - The owning Processor invokes destruction; it MUST guarantee no other - thread accesses the Action concurrently. - */ virtual ~Action() {} }; -/** - Sentinel return value: the Processor completed the request inline and - has already invoked the Continuation. The caller MUST treat the returned - pointer as a tag, not a real @c Action*. Equivalent to - @c MAKE_ACTION_RESULT(1). - - When a caller observes this sentinel, the Continuation may have been - deallocated during the inline callback; the caller MUST NOT touch any - state that the Continuation owned after seeing this value. -*/ #define ACTION_RESULT_DONE MAKE_ACTION_RESULT(1) +#define ACTION_IO_ERROR MAKE_ACTION_RESULT(2) -/** - Sentinel return value: the Processor failed the request inline with an - I/O error and has already invoked the Continuation with a - Processor-specific error event. The caller MUST treat the returned - pointer as a tag, not a real @c Action*. Equivalent to - @c MAKE_ACTION_RESULT(2). - - When a caller observes this sentinel, the Continuation may have been - deallocated during the inline callback; the caller MUST NOT touch any - state that the Continuation owned after seeing this value. -*/ -#define ACTION_IO_ERROR MAKE_ACTION_RESULT(2) - -// Processors that need additional sentinels define them with -// MAKE_ACTION_RESULT, e.g. -// #define MY_PROCESSOR_BASE 3 -// #define ACTION_RESULT_MY_FAILURE MAKE_ACTION_RESULT(MY_PROCESSOR_BASE + 0) - -/** - Constructs a sentinel @c Action* from a small integer. - - The encoding shifts @p _x left by one bit and sets the low bit, so - every sentinel has bit 0 set and is therefore distinguishable from any - validly-aligned @c Action*. A receiver of an @c Action* MUST inspect - @c ((uintptr_t)p & 1) before dereferencing: when set, the value is a - sentinel that MUST be compared against @c ACTION_RESULT_* constants - rather than treated as an @c Action. +// Use these classes by +// #define ACTION_RESULT_HOST_DB_OFFLINE +// MAKE_ACTION_RESULT(ACTION_RESULT_HOST_DB_BASE + 0) - Sentinel values MUST NOT collide; the convention is for each Processor - that defines its own sentinels to reserve a numeric base constant and - define sentinels relative to that base. - - @param _x A non-negative integer expression. After the @c "<<1" shift - and @c "+1", the result MUST fit in @c uintptr_t. Behavior - for values that overflow the shift is undefined. - - @par Errors - Cannot fail. - - @par Thread Safety - Safe to evaluate from any thread; the expansion is a pure expression - with no shared state. -*/ #define MAKE_ACTION_RESULT(_x) (Action *)(((uintptr_t)((_x << 1) + 1))) From 15125dcbb470adda66afc823b971e02a0bf5d2d7 Mon Sep 17 00:00:00 2001 From: Josiah VanderZee Date: Tue, 4 Aug 2026 11:17:10 -0500 Subject: [PATCH 13/14] Check for action sentinels --- include/iocore/eventsystem/EventProcessor.h | 1151 ++++++++++++++--- .../eventsystem/unit_tests/test_Lock.cc | 9 +- 2 files changed, 970 insertions(+), 190 deletions(-) diff --git a/include/iocore/eventsystem/EventProcessor.h b/include/iocore/eventsystem/EventProcessor.h index 82c4f2682a0..a471749c148 100644 --- a/include/iocore/eventsystem/EventProcessor.h +++ b/include/iocore/eventsystem/EventProcessor.h @@ -1,6 +1,6 @@ /** @file - A brief file description + EventProcessor @section license License @@ -19,6 +19,17 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. + + @section details Details + + @c EventProcessor is the singleton @c Processor that owns the Event + System's @c REGULAR EThread pool and a parallel set of @c DEDICATED + EThreads. EThreads are partitioned into named @c EventType groups + (the default group is @c ET_CALL); callers schedule work by group, + which the processor dispatches round-robin across that group's + threads. The library exposes a single global instance, + @c eventProcessor. + */ #pragma once @@ -34,6 +45,19 @@ constexpr int MAX_THREADS_IN_EACH_TYPE = TS_MAX_THREADS_IN_EACH_THREAD_TYPE; constexpr int MAX_THREADS_IN_EACH_TYPE = 3071; #endif +/** + Compile-time upper bound on the number of @c EThreads in a single + @c EventProcessor pool. Applied independently to the @c REGULAR + pool (across all @c EventType groups combined) and to the + @c DEDICATED pool. + + Set from @c TS_MAX_NUMBER_EVENT_THREADS at configure time, or to + 4096 if that macro is not defined. Spawning threads that would + push either pool past this limit aborts the process. + + @par Thread Safety + Compile-time constant; safe to use from any thread. +*/ #ifdef TS_MAX_NUMBER_EVENT_THREADS constexpr int MAX_EVENT_THREADS = TS_MAX_NUMBER_EVENT_THREADS; #else @@ -43,291 +67,979 @@ constexpr int MAX_EVENT_THREADS = 4096; class EThread; /** - Main processor for the Event System. The EventProcessor is the core - component of the Event System. Once started, it is responsible for - creating and managing groups of threads that execute user-defined - tasks asynchronously at a given time or periodically. - - The EventProcessor provides a set of scheduling functions through - which you can specify continuations to be called back by one of its - threads. These function calls do not block. Instead they return an - Event object and schedule the callback to the continuation passed in at - a later or specific time, as soon as possible or at certain intervals. - - Singleton model: - - Every executable that imports and statically links against the - EventSystem library is provided with a global instance of the - EventProcessor called eventProcessor. Therefore, it is not necessary to - create instances of the EventProcessor class because it was designed - as a singleton. It is important to note that none of its functions - are reentrant. - - Thread Groups (Event types): - - When the EventProcessor is started, the first group of threads is spawned and it is assigned the - special id ET_CALL. Depending on the complexity of the state machine or protocol, you may be - interested in creating additional threads and the EventProcessor gives you the ability to create a - single thread or an entire group of threads. In the former case, you call spawn_thread and the - thread is independent of the thread groups and it exists as long as your continuation handle - executes and there are events to process. In the latter, you call @c registerEventType to get an - event type and then @c spawn_event_theads which creates the threads in the group of that - type. Such threads require events to be scheduled on a specific thread in the group or for the - group in general using the event type. Note that between these two calls @c - EThread::schedule_spawn can be used to set up per thread initialization. - - Callback event codes: - - @b UNIX: For all of the scheduling functions, the callback_event - parameter is not used. On a callback, the event code passed in to - the continuation handler is always EVENT_IMMEDIATE. - - @b NT: The value of the event code passed in to the continuation - handler is the value provided in the callback_event parameter. - - Event allocation policy: - - Events are allocated and deallocated by the EventProcessor. A state - machine may access the returned, non-recurring event until it is - cancelled or the callback from the event is complete. For recurring - events, the Event may be accessed until it is cancelled. Once the event - is complete or cancelled, it's the eventProcessor's responsibility to - deallocate it. - + Singleton @c Processor that owns the Event System's thread pools and + dispatches work to them. + + @c EventProcessor::start spawns an initial @c REGULAR EThread group + with @c EventType @c ET_CALL; additional groups are added via + @c register_event_type / @c spawn_event_threads. @c DEDICATED + EThreads are spawned individually by @c spawn_thread. Callers + schedule a Continuation onto a group via @c schedule_imm / + @c schedule_at / @c schedule_in / @c schedule_every (each takes an + @c EventType, defaulting to @c ET_CALL); the processor selects a + thread within that group on each call. + + Allocation: Events handed to a Continuation by the @c schedule_* + family are owned by the framework. A non-recurring Event remains + valid until its single dispatch completes or @c Event::cancel is + called; a recurring Event remains valid until @c Event::cancel is + called. The framework deallocates the Event after that. + + @par Ownership + Singleton; the global @c eventProcessor instance lives for the + entire process lifetime. Direct instantiation is supported but + not the intended usage. + + @par Thread Safety + None of the @c EventProcessor service methods are reentrant on the + same internal state. The @c schedule_* family is safe to call from + any thread; the lifecycle methods (@c start, @c shutdown) are + designed to be called once from the main thread. */ class EventProcessor : public Processor { public: - /** Register an event type with @a name. + /** + Reserves a fresh @c EventType slot and labels it @p name. - This must be called to get an event type to pass to @c spawn_event_threads - @see spawn_event_threads - */ - EventType register_event_type(char const *name); + Subsystems that want a private @c EThread group call this to + obtain an @c EventType, then pass that value to + @c spawn_event_threads to create the actual threads. The + reservation is immediate; the threads are not yet spawned. - /** - Spawn an additional thread for calling back the continuation. Spawns - a dedicated thread (EThread) that calls back the continuation passed - in as soon as possible. + @param name Null-terminated name for the new group; copied. + Stored in the per-group descriptor for + administrative reporting. - @param cont continuation that the spawn thread will call back - immediately. - @return event object representing the start of the thread. + @pre @c n_thread_groups @c < @c MAX_EVENT_TYPES. Calling when + this is not satisfied aborts the process via + @c ink_release_assert. + @post @c n_thread_groups is incremented; the new + @c thread_group[returned] entry has the supplied name and + a zero @c _count until @c spawn_event_threads runs. - */ - Event *spawn_thread(Continuation *cont, const char *thr_name, size_t stacksize = 0); + @return The new @c EventType (zero-based group index). - /** Spawn a group of @a n_threads event dispatching threads. + @par Errors + Aborts the process if the precondition is violated. - The threads run an event loop which dispatches events scheduled for a specific thread or the event type. + @par Thread Safety + Caller-restricted by convention: invoked from the main thread + during process startup. Concurrent calls are not safe. + */ + EventType register_event_type(char const *name); - @return EventType or thread id for the new group of threads (@a ev_type) + /** + Spawns a single @c DEDICATED @c EThread that dispatches @p cont + as its sole task. + + The new thread is created with @c ThreadType @c DEDICATED, given + a single @c start_event whose Continuation is @p cont, and added + to @c all_dthreads. The thread invokes @p cont's handler once + with @c EVENT_IMMEDIATE and exits when the handler returns; it + does not participate in the event-loop dispatch the @c REGULAR + pool runs. As a side effect @p cont 's mutex is overwritten with + the new EThread's mutex. + + @param cont Continuation to dispatch on the new thread. Must + be non-null and remain valid until its handler + returns. The framework allocates and owns the + @c Event passed to the handler. + @param thr_name Null-terminated thread name (truncated to + @c MAX_THREAD_NAME_LENGTH-1 bytes). + @param stacksize Stack size in bytes; zero selects the platform + default (@c DEFAULT_STACKSIZE). + + @pre @p cont is non-null. @c n_dthreads @c < + @c MAX_EVENT_THREADS. + @post A new @c DEDICATED EThread is running. @c all_dthreads + contains the new thread; @c n_dthreads is incremented. + @p cont->mutex points at the new EThread's mutex. + + @return Pointer to the @c Event that will dispatch @p cont on the + new thread. + + @par Errors + Aborts the process via @c ink_release_assert if @c n_dthreads is + already at @c MAX_EVENT_THREADS. + + @par Thread Safety + Safe to call from any thread; the dedicated-thread vector is + serialized by an internal mutex. + */ + Event *spawn_thread(Continuation *cont, const char *thr_name, size_t stacksize = 0); + /** + Spawns @p n_threads @c REGULAR EThreads bound to the @p ev_type + group. + + Each new thread runs the standard event loop and dispatches + events scheduled with @p ev_type. Each Continuation registered + on the group via @c schedule_spawn runs once on every newly + spawned thread before that thread enters its event loop. + + @param ev_type @c EventType obtained from + @c register_event_type. + @param n_threads Number of threads to spawn. MUST be positive + and combined with the existing total MUST NOT + exceed @c MAX_EVENT_THREADS. + @param stacksize Per-thread stack size in bytes. Values below + @c INK_THREAD_STACK_MIN are clamped up to it, + and the result is rounded up to a multiple of + the page size (or huge-page size when huge + pages are enabled). + + @pre @p ev_type was returned by a prior + @c register_event_type and @c spawn_event_threads has not + yet been called for it. @p n_threads is positive and + @p n_threads @c + @c n_ethreads @c <= + @c MAX_EVENT_THREADS. + @post @p n_threads new EThreads have been spawned and bound to + @p ev_type. @c n_ethreads is incremented by @p n_threads; + @c thread_group[ev_type]._count equals @p n_threads and + @c thread_group[ev_type]._thread[0, n_threads) point at + the new threads. + + @return @p ev_type unchanged, for call chaining. + + @par Errors + Aborts the process via @c ink_release_assert if any precondition + is violated. + + @par Thread Safety + Caller-restricted by convention: invoked from the main thread + during process startup. Concurrent calls are not safe. */ EventType spawn_event_threads(EventType ev_type, int n_threads, size_t stacksize = DEFAULT_STACKSIZE); - /// Convenience overload. - /// This registers @a name as an event type using @c registerEventType and then calls the real @c spawn_event_threads + /** + Convenience overload combining @c register_event_type and + @c spawn_event_threads. Registers @p name as a new event type and + immediately spawns @p n_thread threads for it. + + @param name Null-terminated name for the new group; copied. + @param n_thread Number of threads to spawn. MUST be positive. + @param stacksize Per-thread stack size in bytes. Values below + @c INK_THREAD_STACK_MIN are clamped up to it, and + the result is rounded up to a multiple of the + page size (or huge-page size when huge pages are + enabled). + + @pre @c n_thread_groups @c < @c MAX_EVENT_TYPES. @p n_thread is + positive and @p n_thread @c + @c n_ethreads @c <= + @c MAX_EVENT_THREADS. + @post A fresh @c EventType is reserved with @p name and + @p n_thread @c REGULAR EThreads are spawned and bound to it. + @c n_thread_groups is incremented; @c n_ethreads is + incremented by @p n_thread. + + @return The newly registered @c EventType. + + @par Errors + Aborts the process via @c ink_release_assert if any precondition + is violated. + + @par Thread Safety + Caller-restricted by convention: invoked from the main thread + during process startup. Concurrent calls are not safe. + */ EventType spawn_event_threads(const char *name, int n_thread, size_t stacksize = DEFAULT_STACKSIZE); /** - Schedules the continuation on a specific EThread to receive an event - at the given timeout. Requests the EventProcessor to schedule - the callback to the continuation 'c' at the time specified in - 'atimeout_at'. The event is assigned to the specified EThread. - - @param c Continuation to be called back at the time specified in - 'atimeout_at'. - @param atimeout_at time value at which to callback. - @param ethread EThread on which to schedule the event. - @param callback_event code to be passed back to the continuation's - handler. See the Remarks section. - @param cookie user-defined value or pointer to be passed back in - the Event's object cookie field. - @return reference to an Event object representing the scheduling - of this callback. - + Schedules @p c on a thread of group @p event_type for immediate + dispatch. + + Allocates an @c Event and enqueues it on a thread in group + @p event_type. The thread is selected as follows: if @p c has a + thread affinity that belongs to @p event_type 's group, that + thread is used; otherwise if the calling thread is itself in + @p event_type 's group it is used; otherwise a thread is chosen + by the group's round-robin cursor. When @p c had no prior + affinity the chosen thread is recorded as @p c 's affinity. The + Event fires as soon as the dispatch loop reaches it. + + @param c Continuation to dispatch. MUST be non-null + and live until the resulting Event is + delivered or cancelled. + @param event_type @c EventType (group id) on which to + dispatch. Defaults to @c ET_CALL. + @param callback_event @c event_id passed to @c handleEvent on + dispatch. Defaults to @c EVENT_IMMEDIATE. + @param cookie Stored verbatim in @c Event::cookie. + + @pre @p c is non-null. The threads for @p event_type are + spawned. + @post On success, an Event is enqueued on a thread in + @p event_type 's group. If the Event System is in shutdown, + no Event is enqueued. + + @return Pointer to the scheduled Event, or @c nullptr if the + Event System is in shutdown. Use @c Event::cancel to + detach. Framework-owned; do not delete. + + @par Errors + Cannot fail at the contract level. + + @par Thread Safety + Caller-synchronized with respect to @p c: safe to call from any + thread provided no other thread is concurrently scheduling @p c + or otherwise reading or writing @c c->thread_affinity. The + external-queue enqueue itself is thread-safe. */ Event *schedule_imm(Continuation *c, EventType event_type = ET_CALL, int callback_event = EVENT_IMMEDIATE, void *cookie = nullptr); /** - Schedules the continuation on a specific thread group to receive an - event at the given timeout. Requests the EventProcessor to schedule - the callback to the continuation 'c' at the time specified in - 'atimeout_at'. The callback is handled by a thread in the specified - thread group (event_type). - - @param c Continuation to be called back at the time specified in - 'atimeout_at'. - @param atimeout_at Time value at which to callback. - @param event_type thread group id (or event type) specifying the - group of threads on which to schedule the callback. - @param callback_event code to be passed back to the continuation's - handler. See the Remarks section. - @param cookie user-defined value or pointer to be passed back in - the Event's object cookie field. - @return reference to an Event object representing the scheduling of - this callback. - + Schedules @p c on a thread of group @p event_type to be + dispatched at absolute time @p atimeout_at. + + The selected thread is chosen using the same rule as + @c schedule_imm. + + @param c Continuation to dispatch. MUST be non-null + and live until the resulting Event is + delivered or cancelled. + @param atimeout_at Absolute @c ink_hrtime at which to fire. + MUST be strictly positive; a time already + past is legal and fires at the dispatch + loop's next opportunity. + @param event_type @c EventType (group id). Defaults to + @c ET_CALL. + @param callback_event @c event_id passed on dispatch. Defaults + to @c EVENT_INTERVAL. + @param cookie Stored verbatim in @c Event::cookie. + + @pre @p c is non-null. @p atimeout_at @c > 0. The threads for + @p event_type are spawned. + @post On success, an Event is enqueued for delivery at + @p atimeout_at on a thread in @p event_type 's group. If + the Event System is in shutdown, no Event is enqueued. + + @return Pointer to the scheduled Event, or @c nullptr if the + Event System is in shutdown. Use @c Event::cancel to + detach. Framework-owned; do not delete. + + @par Errors + Cannot fail at the contract level. + + @par Thread Safety + Caller-synchronized with respect to @p c: safe to call from any + thread provided no other thread is concurrently scheduling @p c + or otherwise reading or writing @c c->thread_affinity. The + external-queue enqueue itself is thread-safe. */ Event *schedule_at(Continuation *c, ink_hrtime atimeout_at, EventType event_type = ET_CALL, int callback_event = EVENT_INTERVAL, void *cookie = nullptr); /** - Schedules the continuation on a specific thread group to receive an - event after the specified timeout elapses. Requests the EventProcessor - to schedule the callback to the continuation 'c' after the time - specified in 'atimeout_in' elapses. The callback is handled by a - thread in the specified thread group (event_type). - - @param c Continuation to call back aftert the timeout elapses. - @param atimeout_in amount of time after which to callback. - @param event_type Thread group id (or event type) specifying the - group of threads on which to schedule the callback. - @param callback_event code to be passed back to the continuation's - handler. See the Remarks section. - @param cookie user-defined value or pointer to be passed back in - the Event's object cookie field. - @return reference to an Event object representing the scheduling of - this callback. - + Schedules @p c on a thread of group @p event_type to be + dispatched after @p atimeout_in elapses. + + Computes an absolute deadline of @c ink_get_hrtime() @c + + @p atimeout_in and enqueues a one-shot Event for that time. The + selected thread is chosen using the same rule as + @c schedule_imm. + + @param c Continuation to dispatch. MUST be non-null + and live until the resulting Event is + delivered or cancelled. + @param atimeout_in Relative delay in @c ink_hrtime units. Zero + or negative values are legal; they yield a + deadline at or before the current time and + fire at the dispatch loop's next + opportunity. + @param event_type @c EventType (group id). Defaults to + @c ET_CALL. + @param callback_event @c event_id passed on dispatch. Defaults + to @c EVENT_INTERVAL. + @param cookie Stored verbatim in @c Event::cookie. + + @pre @p c is non-null. The threads for @p event_type are + spawned. + @post On success, an Event is enqueued for delivery at the + computed absolute time. If the Event System is in shutdown, + no Event is enqueued. + + @return Pointer to the scheduled Event, or @c nullptr if the + Event System is in shutdown. Use @c Event::cancel to + detach. Framework-owned; do not delete. + + @par Errors + Cannot fail at the contract level. + + @par Thread Safety + Caller-synchronized with respect to @p c: safe to call from any + thread provided no other thread is concurrently scheduling @p c + or otherwise reading or writing @c c->thread_affinity. The + external-queue enqueue itself is thread-safe. */ Event *schedule_in(Continuation *c, ink_hrtime atimeout_in, EventType event_type = ET_CALL, int callback_event = EVENT_INTERVAL, void *cookie = nullptr); /** - Schedules the continuation on a specific thread group to receive - an event periodically. Requests the EventProcessor to schedule the - callback to the continuation 'c' every time 'aperiod' elapses. The - callback is handled by a thread in the specified thread group - (event_type). - - @param c Continuation to call back every time 'aperiod' elapses. - @param aperiod duration of the time period between callbacks. - @param event_type thread group id (or event type) specifying the - group of threads on which to schedule the callback. - @param callback_event code to be passed back to the continuation's - handler. See the Remarks section. - @param cookie user-defined value or pointer to be passed back in - the Event's object cookie field. - @return reference to an Event object representing the scheduling of - this callback. - + Schedules @p c on a thread of group @p event_type to be + dispatched repeatedly every @p aperiod. + + For positive @p aperiod the first dispatch occurs after + @p aperiod elapses and thereafter the Event fires every + @p aperiod until @c Event::cancel is called. For negative + @p aperiod the Event joins the negative-event (poll) rotation + and the handler is dispatched once per event-loop iteration with + @c EVENT_POLL regardless of @p callback_event. The selected + thread is chosen using the same rule as @c schedule_imm. + + @param c Continuation to dispatch. MUST be non-null + and live until the resulting Event is + cancelled. + @param aperiod Period between successive dispatches in + @c ink_hrtime units. MUST be non-zero. + Negative values switch to negative-event + semantics. + @param event_type @c EventType (group id). Defaults to + @c ET_CALL. + @param callback_event @c event_id passed on each positive-period + dispatch. Ignored when @p aperiod is + negative. Defaults to @c EVENT_INTERVAL. + @param cookie Stored verbatim in @c Event::cookie. + + @pre @p c is non-null. @p aperiod is non-zero. The threads for + @p event_type are spawned. + @post On success, a recurring Event is enqueued. If the Event + System is in shutdown, no Event is enqueued. + + @return Pointer to the recurring Event, or @c nullptr if the + Event System is in shutdown. The caller MUST eventually + call @c Event::cancel. + + @par Errors + Cannot fail at the contract level. + + @par Thread Safety + Caller-synchronized with respect to @p c: safe to call from any + thread provided no other thread is concurrently scheduling @p c + or otherwise reading or writing @c c->thread_affinity. The + external-queue enqueue itself is thread-safe. */ Event *schedule_every(Continuation *c, ink_hrtime aperiod, EventType event_type = ET_CALL, int callback_event = EVENT_INTERVAL, void *cookie = nullptr); + /** + Schedules @p c on every thread in group @p event_type. + + For each thread in the group an independent Event is allocated + and enqueued on that thread. Each Event is given a fresh + @c ProxyMutex (rather than sharing @p c 's mutex), so the + per-thread invocations may run concurrently. Used by subsystems + that want a per-thread copy of the same Continuation (e.g., + per-thread initialization or shutdown). + + The timing parameters select one of the following modes; at + least one of @p atimeout and @p aperiod MUST be zero: + - @p atimeout @c == @c 0 and @p aperiod @c == @c 0: each Event + is a one-shot dispatched on the next event-loop iteration. + - @p atimeout @c > @c 0 and @p aperiod @c == @c 0: each Event + is a one-shot dispatched at @c ink_get_hrtime() @c + + @p atimeout. + - @p atimeout @c == @c 0 and @p aperiod @c > @c 0: each Event is + recurring with period @p aperiod; the first dispatch occurs at + @c ink_get_hrtime() @c + @p aperiod. + - @p atimeout @c == @c 0 and @p aperiod @c < @c 0: each Event + joins the negative-event (poll) rotation and the handler is + dispatched once per event-loop iteration. + + @param c Continuation to dispatch. Must remain + valid for the lifetime of every produced + Event. + @param atimeout Relative delay before the (single) one-shot + fire when @p aperiod is zero. Otherwise + MUST be zero. + @param aperiod Period between recurring dispatches; zero + for one-shot. Negative values switch to + negative-event semantics. + @param event_type @c EventType (group id). Defaults to + @c ET_CALL. + @param callback_event @c event_id passed on each positive-period + or one-shot dispatch. Ignored when + @p aperiod is negative. Defaults to + @c EVENT_IMMEDIATE. + @param cookie Stored verbatim in each Event's + @c cookie. + + @pre @p c is non-null. The threads for @p event_type are + spawned. At least one of @p atimeout and @p aperiod is + zero; behavior is undefined when both are non-zero. + @post One Event per thread in the group is enqueued, regardless + of Event-System shutdown state. + + @return Vector of @c TSAction handles, one per thread in the + group, in thread-index order. Each entry is a wrapper + around the per-thread Event; cancel via the wrapper. + + @par Errors + Cannot fail at the contract level. + + @par Thread Safety + Safe to call from any thread. + */ std::vector schedule_entire(Continuation *c, ink_hrtime atimeout, ink_hrtime aperiod, EventType event_type = ET_CALL, int callback_event = EVENT_IMMEDIATE, void *cookie = nullptr); - //////////////////////////////////////////// - // reschedule an already scheduled event. // - // may be called directly or called by // - // schedule_xxx Event member functions. // - // The returned value may be different // - // from the argument e. // - //////////////////////////////////////////// - + // Defect: declared but never defined and never called. Linking against any of + // these will fail; treat as dead declarations pending removal. Event *reschedule_imm(Event *e, int callback_event = EVENT_IMMEDIATE); Event *reschedule_at(Event *e, ink_hrtime atimeout_at, int callback_event = EVENT_INTERVAL); Event *reschedule_in(Event *e, ink_hrtime atimeout_in, int callback_event = EVENT_INTERVAL); Event *reschedule_every(Event *e, ink_hrtime aperiod, int callback_event = EVENT_INTERVAL); - /// Schedule an @a event on continuation @a c when a thread of type @a ev_type is spawned. - /// The @a cookie is attached to the event instance passed to the continuation. - /// @return The scheduled event. + /** + Registers a Continuation to be dispatched once on every thread of + group @p ev_type at thread-spawn time. + + Adds an Event template to @p ev_type 's spawn queue. When each + thread in @p ev_type starts, it walks the queue and invokes + every registered Continuation's handler with a fresh per-thread + Event whose @c ethread points at the spawning thread. Threads + in @p ev_type that have already been spawned do NOT receive the + Event. + + @param c Continuation to dispatch on each newly spawned + thread of @p ev_type. Must remain valid through + every per-thread dispatch. + @param ev_type @c EventType to install on. Use + @c register_event_type to obtain a fresh value. + @param event @c event_id passed to @c handleEvent on each + dispatch. Defaults to @c EVENT_IMMEDIATE. + @param cookie Stored verbatim in each per-thread Event's + @c cookie. + + @pre @p c is non-null. @p ev_type was returned by a prior + @c register_event_type. The threads for @p ev_type have + NOT yet been spawned; for the implicit @c ET_CALL group + this means @c EventProcessor::start has not been called + yet. + @post The Continuation is registered for delivery on each + thread of @p ev_type that the processor subsequently + spawns. + + @return Pointer to the registered template Event. + Framework-owned. + + @par Errors + Cannot fail at the contract level. + + @par Thread Safety + Caller-restricted by convention: invoked from the main thread + during process startup. + */ Event *schedule_spawn(Continuation *c, EventType ev_type, int event = EVENT_IMMEDIATE, void *cookie = nullptr); - /// Schedule the function @a f to be called in a thread of type @a ev_type when it is spawned. + /** + Convenience overload: invokes @p f once on every thread of + @p ev_type at thread-spawn time. + + Wraps @p f in a framework-owned stub Continuation; semantics are + otherwise the same as the Continuation overload above. The + callback event is fixed to @c EVENT_IMMEDIATE. + + @param f Free function called on each newly spawned + thread of @p ev_type. Receives a pointer to the + @c EThread on which it runs. + @param ev_type @c EventType to install on. Use + @c register_event_type to obtain a fresh value. + + @pre @p f is non-null. @p ev_type was returned by a prior + @c register_event_type. The threads for @p ev_type have + NOT yet been spawned. + @post @p f is registered for delivery on each thread of + @p ev_type that the processor subsequently spawns. + + @return Pointer to the registered template Event. + Framework-owned. + + @par Errors + Cannot fail at the contract level. + + @par Thread Safety + Caller-restricted by convention: invoked from the main thread + during process startup. + */ Event *schedule_spawn(void (*f)(EThread *), EventType ev_type); - /// Schedule an @a event on continuation @a c to be called when a thread is spawned by this processor. - /// The @a cookie is attached to the event instance passed to the continuation. - /// @return The scheduled event. // Event *schedule_spawn(Continuation *c, int event, void *cookie = NULL); + /** + Constructs a fresh @c EventProcessor with the @c ET_CALL group + reserved but no threads spawned. + + The global @c eventProcessor singleton is constructed at process + startup; user code does not normally instantiate this class. + + @post @c n_thread_groups @c == @c 1 (the @c ET_CALL group is + reserved with its registered name); @c n_ethreads and + @c n_dthreads are zero; no EThreads exist. The internal + @c thread_initializer Continuation is constructed and the + dedicated-thread mutex is initialized. + + @par Errors + Cannot fail at the contract level. + + @par Thread Safety + Safe to call from any thread. The constructed instance is not + yet observable to others. + */ EventProcessor(); + /** + Destroys an @c EventProcessor. + + Tears down the internal dedicated-thread spawn mutex. Does not + stop, join, or otherwise reclaim any EThread previously spawned + by this @c EventProcessor; those must already be quiesced. + + @pre No EThread spawned by this @c EventProcessor is still + running, and no thread is currently inside + @c spawn_event_threads on this @c EventProcessor. + + @par Errors + Aborts the process if the underlying mutex teardown reports an + error (for example, if the mutex is still locked). + + @par Thread Safety + Safe to call from any thread once the precondition holds. + */ ~EventProcessor() override; EventProcessor(const EventProcessor &) = delete; EventProcessor &operator=(const EventProcessor &) = delete; /** - Initializes the EventProcessor and its associated threads. Spawns the - specified number of threads, initializes their state information and - sets them running. It creates the initial thread group, represented - by the event type ET_CALL. - - @return 0 if successful, and a negative value otherwise. - + Initializes the @c EventProcessor and spawns the @c ET_CALL + thread group. + + Initializes thread-affinity bookkeeping, registers the + Event-System metric stats, prepends a thread-affinity + initializer Continuation to the @c ET_CALL spawn queue (so it + runs first on every @c ET_CALL thread), and spawns + @p n_net_threads @c REGULAR EThreads in @c ET_CALL. Each + Continuation registered for @c ET_CALL via @c schedule_spawn + runs once on every newly spawned thread before that thread + enters its event loop. After this call returns, additional + groups may be created via @c register_event_type / + @c spawn_event_threads. + + @param n_net_threads Number of threads to spawn for the initial + @c ET_CALL group. MUST be positive and not + exceed @c MAX_EVENT_THREADS. + @param stacksize Per-thread stack size in bytes. Values + below @c INK_THREAD_STACK_MIN are clamped + up to it, and the result is rounded up to + a multiple of the page size (or huge-page + size when huge pages are enabled). The + default argument @c DEFAULT_STACKSIZE + selects the platform default. + + @pre Has not been called before on any @c EventProcessor in + this process. @p n_net_threads is positive and does not + exceed @c MAX_EVENT_THREADS. + @post @c ET_CALL group has @p n_net_threads spawned threads; + @c n_ethreads is incremented by @p n_net_threads. + @c n_thread_groups is unchanged. + + @return Zero. The contract retains the negative-on-failure + convention from @c Processor::start, but the current + implementation has no failure path that returns a + value (resource exhaustion aborts). + + @par Errors + Aborts the process via @c ink_release_assert if the + precondition is violated or if a resource-exhaustion failure + occurs during thread spawn. + + @par Thread Safety + Caller-restricted: invoked once from the main thread during + process startup. */ int start(int n_net_threads, size_t stacksize = DEFAULT_STACKSIZE) override; /** - Stop the EventProcessor. Attempts to stop the EventProcessor and - all of the threads in each of the thread groups. + Hook for shutting down the @c EventProcessor subsystem. + + @pre No preconditions. + @post No observable side effects. + @par Errors + Cannot fail. + + @par Thread Safety + Safe to call from any thread. */ + // The current implementation is an empty body: it exists so the + // override resolution from Processor::shutdown works, but there is + // no per-subsystem work to do here. Process shutdown is handled + // elsewhere via TSSystemState. void shutdown() override; /** - Allocates size bytes on the event threads. This function is thread - safe. + Reserves @p size bytes inside every @c EThread's + @c thread_private region and returns the byte offset, measured + from the start of the @c EThread, at which those bytes begin. + + Used by subsystems that want a per-EThread chunk of state. The + returned offset is suitable for use with @c ETHREAD_GET_PTR. + The reserved region's starting address is 16-byte aligned, and + @p size is rounded up to a multiple of 16. - @param size bytes to be allocated. + @param size Number of bytes to reserve. MUST be non-negative. + @pre @p size is non-negative. + @post On success, the running reservation counter is advanced + past the reserved region. On failure, no state is changed. + + @return Byte offset, measured from the start of an @c EThread, + at which the reserved region begins; or @c -1 if the + requested allocation would not fit within + @c PER_THREAD_DATA. + + @par Errors + Returns @c -1 if the requested size cannot be satisfied. + + @par Thread Safety + Safe to call from any thread; concurrent calls are serialized + via a compare-and-swap retry loop. */ off_t allocate(int size); /** - An array of pointers to all of the EThreads handled by the - EventProcessor. An array of pointers to all of the EThreads created - throughout the existence of the EventProcessor instance. + Storage for every @c REGULAR EThread spawned by this processor. + + Indices [0, @c n_ethreads) are valid pointers; remaining slots + are @c nullptr. Consumers iterate via @c active_ethreads (or one + of the per-group accessors) rather than indexing directly. + @par Thread Safety + Written only during thread-pool spawn (single-threaded process + startup). Safe to read concurrently after startup. */ EThread *all_ethreads[MAX_EVENT_THREADS]; - /// Data kept for each thread group. - /// The thread group ID is the index into an array of these and so is not stored explicitly. + /** + Per-group state for a single @c EventType. + + The thread-group id is the index into @c thread_group; it is not + stored on the descriptor itself. Accessed by name from + consumers that want raw access to the per-group thread vector + or count; most consumers prefer @c active_group_threads. + + @par Ownership + Owned by @c EventProcessor; lives for the processor's lifetime. + + @par Thread Safety + Members are written during thread-pool spawn (single-threaded + process startup) except for @c _started (atomic) and + @c _next_round_robin (per-call increment by the dispatch path). + */ struct ThreadGroupDescriptor { - std::string _name; ///< Name for the thread group. - int _count = 0; ///< # of threads of this type. - std::atomic _started = 0; ///< # of started threads of this type. - uint64_t _next_round_robin = 0; ///< Index of thread to use for events assigned to this group. - Que(Event, link) _spawnQueue; ///< Events to dispatch when thread is spawned. - EThread *_thread[MAX_THREADS_IN_EACH_TYPE] = {}; ///< The actual threads in this group. - std::function _afterStartCallback = nullptr; + /** + Name registered with @c register_event_type. Stable for the + lifetime of the descriptor. + + @par Thread Safety + Written once at registration; safe to read concurrently. + */ + std::string _name; + /** + Number of threads in this group. Set by @c spawn_event_threads + before any of those threads start running. + + @par Thread Safety + Plain @c int. Written at thread-pool spawn; safe to read + concurrently after startup. + */ + int _count = 0; + /** + Atomic counter incremented each time a thread in this group + finishes its per-thread initialization and signals readiness. + When @c _started == @c _count, the group is fully running. + + @par Thread Safety + @c std::atomic. Each thread in the group performs one + increment with @c std::memory_order_seq_cst (the default for + @c operator++); readers use @c load with the same default. + Other modules may observe a value bounded by the number of + threads that have completed their per-thread init. + */ + std::atomic _started = 0; + /** + Round-robin cursor used by @c assign_thread to pick the next + thread in the group for a fresh @c schedule_imm / + @c schedule_at / @c schedule_in / @c schedule_every call. + + @par Thread Safety + Plain @c uint64_t. Read and incremented without + synchronization by @c assign_thread. + */ + // Defect: see the data-race note on EventProcessor::assign_thread. + uint64_t _next_round_robin = 0; + /** + Template Events whose continuations are dispatched on each + thread in this group at spawn time. Populated by + @c schedule_spawn before @c spawn_event_threads runs. Each + newly spawned thread iterates this queue read-only, building + a per-thread Event from each entry to invoke its continuation; + the queue itself is not drained. + + @par Thread Safety + Written only from the main thread before the group's threads + spawn; read-only thereafter from each spawning thread. + */ + Que(Event, link) _spawnQueue; + /** + Pointers to the EThreads in this group. Indices [0, + @c _count) are valid; remaining slots are @c nullptr. + + @par Thread Safety + Written during thread-pool spawn; safe to read concurrently + after startup. + */ + EThread *_thread[MAX_THREADS_IN_EACH_TYPE] = {}; + /** + Optional callback invoked once when every thread in the group + has finished its per-thread initialization (i.e., when + @c _started reaches @c _count). May be @c nullptr. + + @par Thread Safety + Written by the main thread before the group's threads spawn; + called once from the last spawning thread to reach the + threshold. + */ + std::function _afterStartCallback = nullptr; }; - /// Storage for per group data. + /** + Per-group descriptors indexed by @c EventType. + + Indices [0, @c n_thread_groups) are populated; the rest are + default-constructed. + + @par Thread Safety + Same as the per-descriptor contracts above. + */ ThreadGroupDescriptor thread_group[MAX_EVENT_TYPES]; - /// Number of defined thread groups. + /** + Number of registered thread groups. + + Bumped by @c register_event_type each time a fresh group is + reserved. + + @par Thread Safety + Plain @c int. Written by @c register_event_type during process + startup; safe to read concurrently after that. + */ int n_thread_groups = 0; /** - Total number of threads controlled by this EventProcessor. This is - the count of all the EThreads spawn by this EventProcessor, excluding - those created by spawn_thread + Total number of @c REGULAR EThreads spawned by this processor. + Equals the sum of @c thread_group[i]._count for valid @c i. + Excludes @c DEDICATED threads created via @c spawn_thread. + + @par Thread Safety + Plain @c int. Written by @c spawn_event_threads during process + startup; safe to read concurrently after that. */ int n_ethreads = 0; + /** + Returns whether every thread in group @p etype has finished its + per-thread initialization. + + @param etype @c EventType (group id) to query. + + @pre @p etype is in the range [0, @c n_thread_groups). + @post No observable side effects. + @return @c true iff every thread in the group has signaled + readiness; @c false otherwise. + + @par Errors + Cannot fail. + + @par Thread Safety + Safe to call from any thread. + */ bool has_tg_started(int etype); /*------------------------------------------------------*\ | Unix & non NT Interface | \*------------------------------------------------------*/ - Event *schedule(Event *e, EventType etype); + /** + Schedules an already-initialized Event @p e onto group @p etype. + + Selects a thread within @p etype 's group as follows: if + @p e 's continuation has a thread affinity that belongs to + @p etype 's group, that thread is used; otherwise if the + calling thread is itself in @p etype 's group it is used; + otherwise the group's round-robin cursor is consulted. When the + continuation had no prior affinity the chosen thread is + recorded as its affinity. If the continuation has a mutex, that + mutex is also installed on @p e. The Event is then enqueued on + the chosen thread's external queue, using the local-enqueue + fast path if the chosen thread is the caller's own thread. + + @param e Event whose @c init has been called. Ownership of + @p e passes to the framework on success. + @param etype @c EventType (group id) to dispatch on. + + @pre @p e is a freshly initialized Event whose @c continuation + is non-null. Threads for @p etype have been spawned. + @post On success, @p e->ethread points at a thread in + @p etype 's group, @p e->mutex equals the continuation's + mutex when the continuation has one, and @p e is enqueued + for delivery on the chosen thread. If the Event System is + in shutdown, @p e is not enqueued and ownership stays + with the caller. + + @return @p e on success, or @c nullptr if the Event System is + in shutdown. + + @par Errors + Cannot fail at the contract level. + + @par Thread Safety + Safe to call from any thread. + */ + // Lower-level entry point used by schedule_imm / schedule_at / + // schedule_in / schedule_every after they allocate and initialize + // the Event. + Event *schedule(Event *e, EventType etype); + /** + Returns a pointer to one of the EThreads in group @p etype, + chosen by the round-robin cursor. + + @param etype @c EventType (group id). + + @pre @p etype is in the range [0, @c MAX_EVENT_TYPES). + Threads for @p etype have been spawned. + @post When the group has more than one thread, the group's + round-robin cursor advances by one and the returned + thread is the one at that cursor position modulo the + group size. Single-thread groups always return that + one thread and leave the cursor unchanged. + + @return Pointer to a thread in the group; never @c nullptr + given the precondition. + + @par Errors + Cannot fail at the contract level. + + @par Thread Safety + Safe to call from any single thread for groups whose + @c _count is at most one. For larger groups, concurrent + calls update the round-robin cursor non-atomically. + */ + // Defect: when a group has more than one thread, the cursor + // increment `++tg->_next_round_robin` is a non-atomic + // read-modify-write on a value concurrently read and written + // by other threads via this routine. That is a C++ data race + // (undefined behavior). Fix is to make `_next_round_robin` + // `std::atomic` and use a relaxed fetch_add. EThread *assign_thread(EventType etype); + /** + Returns an EThread chosen for @p cont in group @p etype using + the affinity rule. + + Selects a thread by the following priority: + 1. If @p cont 's mutex's holding thread is in @p etype 's + group, that thread. + 2. Otherwise if @p cont already has a registered thread + affinity in @p etype 's group, that thread. + 3. Otherwise the group's round-robin cursor via + @c assign_thread. + + If @p cont has no prior thread affinity, the chosen thread is + recorded as its affinity. A prior affinity is never overwritten, + even when it is not in @p etype 's group. + + @param cont Continuation to schedule. Its affinity may be + updated as a side effect when no prior affinity + was set. + @param etype @c EventType (group id) in the range + [0, @c MAX_EVENT_TYPES). + + @pre @p cont is non-null and its @c mutex holds a non-null + @c thread_holding. Threads for @p etype have been + spawned. No other thread concurrently reads or writes + @p cont 's @c thread_affinity. + @post If @p cont had no prior affinity, its affinity is set to + the returned thread. The returned thread is in + @p etype 's group. + + @return Pointer to the chosen thread; never @c nullptr given + the precondition. + + @par Errors + Cannot fail at the contract level. + + @par Thread Safety + The read and write of @p cont 's @c thread_affinity are + unsynchronized; the caller must serialize them against any + other accessor. Round-robin selection inherits the race + characteristics of @c assign_thread. + */ EThread *assign_affinity_by_type(Continuation *cont, EventType etype); + /** + Storage for every @c DEDICATED EThread spawned by + @c spawn_thread. + + Indices [0, @c n_dthreads) are valid pointers; the rest are + @c nullptr. + + @par Thread Safety + Written by @c spawn_thread under + @c dedicated_thread_spawn_mutex. Safe to read concurrently + after the spawn completes; readers MUST first read + @c n_dthreads to know the valid range. + */ EThread *all_dthreads[MAX_EVENT_THREADS]; - int n_dthreads = 0; // No. of dedicated threads - int thread_data_used = 0; + /** + Number of @c DEDICATED EThreads spawned by this processor. + + @par Thread Safety + Plain @c int. Written by @c spawn_thread under + @c dedicated_thread_spawn_mutex; safe to read concurrently + after the spawn completes. + */ + int n_dthreads = 0; + /** + Running tally of bytes reserved out of every EThread's + @c thread_private region by @c allocate. Updated atomically by + @c allocate so multiple subsystems can register from different + threads without corruption. + + @par Thread Safety + Plain @c int updated via @c ink_atomic_cas in @c allocate; + safe for concurrent updates. + */ + int thread_data_used = 0; - /// Provide container style access to just the active threads, not the entire array. + /** + Range view over a contiguous, valid prefix of an EThread + pointer array. + + Provides @c begin and @c end iterators for use in range-for + loops and STL algorithms. Constructed by @c active_ethreads, + @c active_dthreads, and @c active_group_threads; cannot be + constructed directly by consumers. + + @par Ownership + Stateless reference into one of the @c EventProcessor's + pointer arrays; outlives the underlying array only as long as + the @c EventProcessor itself does. + + @par Thread Safety + The view is safe to use after thread-pool startup is complete; + iterating concurrently with thread spawn is not safe. + */ class active_threads_type { using iterator = EThread *const *; ///< Internal iterator type, pointer to array element. @@ -352,19 +1064,69 @@ class EventProcessor : public Processor friend class EventProcessor; }; - // These can be used in container for loops and other range operations. + /** + Returns a range over every @c REGULAR EThread spawned by this + processor. + + @pre No preconditions. + @post No observable side effects. + @return Range whose @c begin / @c end span @c all_ethreads[0, + n_ethreads). + + @par Errors + Cannot fail. + + @par Thread Safety + Safe to call from any thread once @c spawn_event_threads has + finished spawning every group; iterating the returned range + concurrently with thread spawn is not safe. + */ active_threads_type active_ethreads() const { return {all_ethreads, n_ethreads}; } + /** + Returns a range over every @c DEDICATED EThread spawned by + this processor. + + @pre No preconditions. + @post No observable side effects. + @return Range whose @c begin / @c end span @c all_dthreads[0, + n_dthreads). + + @par Errors + Cannot fail. + + @par Thread Safety + Safe to call from any thread once dedicated-thread spawning has + quiesced; iterating the returned range concurrently with + @c spawn_thread is not safe. + */ active_threads_type active_dthreads() const { return {all_dthreads, n_dthreads}; } + /** + Returns a range over every EThread in group @p type. + + @param type @c EventType (group id) to iterate. + + @pre @p type is in the range [0, @c n_thread_groups). + @post No observable side effects. + @return Range whose @c begin / @c end span the group's + @c _thread[0, _count) slice. + + @par Errors + Cannot fail. + + @par Thread Safety + Safe to call from any thread once the group's threads have + been spawned. + */ active_threads_type active_group_threads(int type) const { @@ -399,6 +1161,19 @@ class EventProcessor : public Processor ink_mutex dedicated_thread_spawn_mutex; }; +/** + Global @c EventProcessor singleton. + + Defined in the inkevent library. Every executable that links + inkevent observes this single instance; consumers schedule work + through it (e.g., @c eventProcessor.schedule_imm(cont, ET_TASK)). + + @par Ownership + Static-storage singleton; lives for the entire process lifetime. + + @par Thread Safety + See @c EventProcessor type-level contract. +*/ extern class EventProcessor eventProcessor; void thread_started(EThread *); diff --git a/src/iocore/eventsystem/unit_tests/test_Lock.cc b/src/iocore/eventsystem/unit_tests/test_Lock.cc index f4197580e9d..2194d059382 100644 --- a/src/iocore/eventsystem/unit_tests/test_Lock.cc +++ b/src/iocore/eventsystem/unit_tests/test_Lock.cc @@ -26,6 +26,8 @@ #include +#include + using inkevent_test::AtomicFlag; using inkevent_test::EventProcessorListener; @@ -48,9 +50,8 @@ class HoldOnEThread : public Continuation // it here in order to unfreeze any threads that may be waiting on done. ~HoldOnEThread() { - this->release.set(); this->cancel_callback(); - this->done.wait_until_set(); + this->wait_for_callback_finish(); } bool @@ -101,6 +102,10 @@ class HoldOnEThread : public Continuation bool is_expecting_callback() { + if (reinterpret_cast(this->callback_action) & 1) { + return false; + } + ink_assert(this->mutex->thread_holding == this_ethread()); return !this->held.is_set() && !this->callback_action->cancelled; } From c0dc2e7b57a5ee92f8ee0d20be97de08cdbcbf4f Mon Sep 17 00:00:00 2001 From: Josiah VanderZee Date: Tue, 4 Aug 2026 11:18:03 -0500 Subject: [PATCH 14/14] Remove documentation --- include/iocore/eventsystem/EventProcessor.h | 1151 +++---------------- 1 file changed, 188 insertions(+), 963 deletions(-) diff --git a/include/iocore/eventsystem/EventProcessor.h b/include/iocore/eventsystem/EventProcessor.h index a471749c148..82c4f2682a0 100644 --- a/include/iocore/eventsystem/EventProcessor.h +++ b/include/iocore/eventsystem/EventProcessor.h @@ -1,6 +1,6 @@ /** @file - EventProcessor + A brief file description @section license License @@ -19,17 +19,6 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. - - @section details Details - - @c EventProcessor is the singleton @c Processor that owns the Event - System's @c REGULAR EThread pool and a parallel set of @c DEDICATED - EThreads. EThreads are partitioned into named @c EventType groups - (the default group is @c ET_CALL); callers schedule work by group, - which the processor dispatches round-robin across that group's - threads. The library exposes a single global instance, - @c eventProcessor. - */ #pragma once @@ -45,19 +34,6 @@ constexpr int MAX_THREADS_IN_EACH_TYPE = TS_MAX_THREADS_IN_EACH_THREAD_TYPE; constexpr int MAX_THREADS_IN_EACH_TYPE = 3071; #endif -/** - Compile-time upper bound on the number of @c EThreads in a single - @c EventProcessor pool. Applied independently to the @c REGULAR - pool (across all @c EventType groups combined) and to the - @c DEDICATED pool. - - Set from @c TS_MAX_NUMBER_EVENT_THREADS at configure time, or to - 4096 if that macro is not defined. Spawning threads that would - push either pool past this limit aborts the process. - - @par Thread Safety - Compile-time constant; safe to use from any thread. -*/ #ifdef TS_MAX_NUMBER_EVENT_THREADS constexpr int MAX_EVENT_THREADS = TS_MAX_NUMBER_EVENT_THREADS; #else @@ -67,979 +43,291 @@ constexpr int MAX_EVENT_THREADS = 4096; class EThread; /** - Singleton @c Processor that owns the Event System's thread pools and - dispatches work to them. - - @c EventProcessor::start spawns an initial @c REGULAR EThread group - with @c EventType @c ET_CALL; additional groups are added via - @c register_event_type / @c spawn_event_threads. @c DEDICATED - EThreads are spawned individually by @c spawn_thread. Callers - schedule a Continuation onto a group via @c schedule_imm / - @c schedule_at / @c schedule_in / @c schedule_every (each takes an - @c EventType, defaulting to @c ET_CALL); the processor selects a - thread within that group on each call. - - Allocation: Events handed to a Continuation by the @c schedule_* - family are owned by the framework. A non-recurring Event remains - valid until its single dispatch completes or @c Event::cancel is - called; a recurring Event remains valid until @c Event::cancel is - called. The framework deallocates the Event after that. - - @par Ownership - Singleton; the global @c eventProcessor instance lives for the - entire process lifetime. Direct instantiation is supported but - not the intended usage. - - @par Thread Safety - None of the @c EventProcessor service methods are reentrant on the - same internal state. The @c schedule_* family is safe to call from - any thread; the lifecycle methods (@c start, @c shutdown) are - designed to be called once from the main thread. + Main processor for the Event System. The EventProcessor is the core + component of the Event System. Once started, it is responsible for + creating and managing groups of threads that execute user-defined + tasks asynchronously at a given time or periodically. + + The EventProcessor provides a set of scheduling functions through + which you can specify continuations to be called back by one of its + threads. These function calls do not block. Instead they return an + Event object and schedule the callback to the continuation passed in at + a later or specific time, as soon as possible or at certain intervals. + + Singleton model: + + Every executable that imports and statically links against the + EventSystem library is provided with a global instance of the + EventProcessor called eventProcessor. Therefore, it is not necessary to + create instances of the EventProcessor class because it was designed + as a singleton. It is important to note that none of its functions + are reentrant. + + Thread Groups (Event types): + + When the EventProcessor is started, the first group of threads is spawned and it is assigned the + special id ET_CALL. Depending on the complexity of the state machine or protocol, you may be + interested in creating additional threads and the EventProcessor gives you the ability to create a + single thread or an entire group of threads. In the former case, you call spawn_thread and the + thread is independent of the thread groups and it exists as long as your continuation handle + executes and there are events to process. In the latter, you call @c registerEventType to get an + event type and then @c spawn_event_theads which creates the threads in the group of that + type. Such threads require events to be scheduled on a specific thread in the group or for the + group in general using the event type. Note that between these two calls @c + EThread::schedule_spawn can be used to set up per thread initialization. + + Callback event codes: + + @b UNIX: For all of the scheduling functions, the callback_event + parameter is not used. On a callback, the event code passed in to + the continuation handler is always EVENT_IMMEDIATE. + + @b NT: The value of the event code passed in to the continuation + handler is the value provided in the callback_event parameter. + + Event allocation policy: + + Events are allocated and deallocated by the EventProcessor. A state + machine may access the returned, non-recurring event until it is + cancelled or the callback from the event is complete. For recurring + events, the Event may be accessed until it is cancelled. Once the event + is complete or cancelled, it's the eventProcessor's responsibility to + deallocate it. + */ class EventProcessor : public Processor { public: - /** - Reserves a fresh @c EventType slot and labels it @p name. + /** Register an event type with @a name. - Subsystems that want a private @c EThread group call this to - obtain an @c EventType, then pass that value to - @c spawn_event_threads to create the actual threads. The - reservation is immediate; the threads are not yet spawned. + This must be called to get an event type to pass to @c spawn_event_threads + @see spawn_event_threads + */ + EventType register_event_type(char const *name); - @param name Null-terminated name for the new group; copied. - Stored in the per-group descriptor for - administrative reporting. + /** + Spawn an additional thread for calling back the continuation. Spawns + a dedicated thread (EThread) that calls back the continuation passed + in as soon as possible. - @pre @c n_thread_groups @c < @c MAX_EVENT_TYPES. Calling when - this is not satisfied aborts the process via - @c ink_release_assert. - @post @c n_thread_groups is incremented; the new - @c thread_group[returned] entry has the supplied name and - a zero @c _count until @c spawn_event_threads runs. + @param cont continuation that the spawn thread will call back + immediately. + @return event object representing the start of the thread. - @return The new @c EventType (zero-based group index). + */ + Event *spawn_thread(Continuation *cont, const char *thr_name, size_t stacksize = 0); - @par Errors - Aborts the process if the precondition is violated. + /** Spawn a group of @a n_threads event dispatching threads. - @par Thread Safety - Caller-restricted by convention: invoked from the main thread - during process startup. Concurrent calls are not safe. - */ - EventType register_event_type(char const *name); + The threads run an event loop which dispatches events scheduled for a specific thread or the event type. - /** - Spawns a single @c DEDICATED @c EThread that dispatches @p cont - as its sole task. - - The new thread is created with @c ThreadType @c DEDICATED, given - a single @c start_event whose Continuation is @p cont, and added - to @c all_dthreads. The thread invokes @p cont's handler once - with @c EVENT_IMMEDIATE and exits when the handler returns; it - does not participate in the event-loop dispatch the @c REGULAR - pool runs. As a side effect @p cont 's mutex is overwritten with - the new EThread's mutex. - - @param cont Continuation to dispatch on the new thread. Must - be non-null and remain valid until its handler - returns. The framework allocates and owns the - @c Event passed to the handler. - @param thr_name Null-terminated thread name (truncated to - @c MAX_THREAD_NAME_LENGTH-1 bytes). - @param stacksize Stack size in bytes; zero selects the platform - default (@c DEFAULT_STACKSIZE). - - @pre @p cont is non-null. @c n_dthreads @c < - @c MAX_EVENT_THREADS. - @post A new @c DEDICATED EThread is running. @c all_dthreads - contains the new thread; @c n_dthreads is incremented. - @p cont->mutex points at the new EThread's mutex. - - @return Pointer to the @c Event that will dispatch @p cont on the - new thread. - - @par Errors - Aborts the process via @c ink_release_assert if @c n_dthreads is - already at @c MAX_EVENT_THREADS. - - @par Thread Safety - Safe to call from any thread; the dedicated-thread vector is - serialized by an internal mutex. - */ - Event *spawn_thread(Continuation *cont, const char *thr_name, size_t stacksize = 0); + @return EventType or thread id for the new group of threads (@a ev_type) - /** - Spawns @p n_threads @c REGULAR EThreads bound to the @p ev_type - group. - - Each new thread runs the standard event loop and dispatches - events scheduled with @p ev_type. Each Continuation registered - on the group via @c schedule_spawn runs once on every newly - spawned thread before that thread enters its event loop. - - @param ev_type @c EventType obtained from - @c register_event_type. - @param n_threads Number of threads to spawn. MUST be positive - and combined with the existing total MUST NOT - exceed @c MAX_EVENT_THREADS. - @param stacksize Per-thread stack size in bytes. Values below - @c INK_THREAD_STACK_MIN are clamped up to it, - and the result is rounded up to a multiple of - the page size (or huge-page size when huge - pages are enabled). - - @pre @p ev_type was returned by a prior - @c register_event_type and @c spawn_event_threads has not - yet been called for it. @p n_threads is positive and - @p n_threads @c + @c n_ethreads @c <= - @c MAX_EVENT_THREADS. - @post @p n_threads new EThreads have been spawned and bound to - @p ev_type. @c n_ethreads is incremented by @p n_threads; - @c thread_group[ev_type]._count equals @p n_threads and - @c thread_group[ev_type]._thread[0, n_threads) point at - the new threads. - - @return @p ev_type unchanged, for call chaining. - - @par Errors - Aborts the process via @c ink_release_assert if any precondition - is violated. - - @par Thread Safety - Caller-restricted by convention: invoked from the main thread - during process startup. Concurrent calls are not safe. */ EventType spawn_event_threads(EventType ev_type, int n_threads, size_t stacksize = DEFAULT_STACKSIZE); - /** - Convenience overload combining @c register_event_type and - @c spawn_event_threads. Registers @p name as a new event type and - immediately spawns @p n_thread threads for it. - - @param name Null-terminated name for the new group; copied. - @param n_thread Number of threads to spawn. MUST be positive. - @param stacksize Per-thread stack size in bytes. Values below - @c INK_THREAD_STACK_MIN are clamped up to it, and - the result is rounded up to a multiple of the - page size (or huge-page size when huge pages are - enabled). - - @pre @c n_thread_groups @c < @c MAX_EVENT_TYPES. @p n_thread is - positive and @p n_thread @c + @c n_ethreads @c <= - @c MAX_EVENT_THREADS. - @post A fresh @c EventType is reserved with @p name and - @p n_thread @c REGULAR EThreads are spawned and bound to it. - @c n_thread_groups is incremented; @c n_ethreads is - incremented by @p n_thread. - - @return The newly registered @c EventType. - - @par Errors - Aborts the process via @c ink_release_assert if any precondition - is violated. - - @par Thread Safety - Caller-restricted by convention: invoked from the main thread - during process startup. Concurrent calls are not safe. - */ + /// Convenience overload. + /// This registers @a name as an event type using @c registerEventType and then calls the real @c spawn_event_threads EventType spawn_event_threads(const char *name, int n_thread, size_t stacksize = DEFAULT_STACKSIZE); /** - Schedules @p c on a thread of group @p event_type for immediate - dispatch. - - Allocates an @c Event and enqueues it on a thread in group - @p event_type. The thread is selected as follows: if @p c has a - thread affinity that belongs to @p event_type 's group, that - thread is used; otherwise if the calling thread is itself in - @p event_type 's group it is used; otherwise a thread is chosen - by the group's round-robin cursor. When @p c had no prior - affinity the chosen thread is recorded as @p c 's affinity. The - Event fires as soon as the dispatch loop reaches it. - - @param c Continuation to dispatch. MUST be non-null - and live until the resulting Event is - delivered or cancelled. - @param event_type @c EventType (group id) on which to - dispatch. Defaults to @c ET_CALL. - @param callback_event @c event_id passed to @c handleEvent on - dispatch. Defaults to @c EVENT_IMMEDIATE. - @param cookie Stored verbatim in @c Event::cookie. - - @pre @p c is non-null. The threads for @p event_type are - spawned. - @post On success, an Event is enqueued on a thread in - @p event_type 's group. If the Event System is in shutdown, - no Event is enqueued. - - @return Pointer to the scheduled Event, or @c nullptr if the - Event System is in shutdown. Use @c Event::cancel to - detach. Framework-owned; do not delete. - - @par Errors - Cannot fail at the contract level. - - @par Thread Safety - Caller-synchronized with respect to @p c: safe to call from any - thread provided no other thread is concurrently scheduling @p c - or otherwise reading or writing @c c->thread_affinity. The - external-queue enqueue itself is thread-safe. + Schedules the continuation on a specific EThread to receive an event + at the given timeout. Requests the EventProcessor to schedule + the callback to the continuation 'c' at the time specified in + 'atimeout_at'. The event is assigned to the specified EThread. + + @param c Continuation to be called back at the time specified in + 'atimeout_at'. + @param atimeout_at time value at which to callback. + @param ethread EThread on which to schedule the event. + @param callback_event code to be passed back to the continuation's + handler. See the Remarks section. + @param cookie user-defined value or pointer to be passed back in + the Event's object cookie field. + @return reference to an Event object representing the scheduling + of this callback. + */ Event *schedule_imm(Continuation *c, EventType event_type = ET_CALL, int callback_event = EVENT_IMMEDIATE, void *cookie = nullptr); /** - Schedules @p c on a thread of group @p event_type to be - dispatched at absolute time @p atimeout_at. - - The selected thread is chosen using the same rule as - @c schedule_imm. - - @param c Continuation to dispatch. MUST be non-null - and live until the resulting Event is - delivered or cancelled. - @param atimeout_at Absolute @c ink_hrtime at which to fire. - MUST be strictly positive; a time already - past is legal and fires at the dispatch - loop's next opportunity. - @param event_type @c EventType (group id). Defaults to - @c ET_CALL. - @param callback_event @c event_id passed on dispatch. Defaults - to @c EVENT_INTERVAL. - @param cookie Stored verbatim in @c Event::cookie. - - @pre @p c is non-null. @p atimeout_at @c > 0. The threads for - @p event_type are spawned. - @post On success, an Event is enqueued for delivery at - @p atimeout_at on a thread in @p event_type 's group. If - the Event System is in shutdown, no Event is enqueued. - - @return Pointer to the scheduled Event, or @c nullptr if the - Event System is in shutdown. Use @c Event::cancel to - detach. Framework-owned; do not delete. - - @par Errors - Cannot fail at the contract level. - - @par Thread Safety - Caller-synchronized with respect to @p c: safe to call from any - thread provided no other thread is concurrently scheduling @p c - or otherwise reading or writing @c c->thread_affinity. The - external-queue enqueue itself is thread-safe. + Schedules the continuation on a specific thread group to receive an + event at the given timeout. Requests the EventProcessor to schedule + the callback to the continuation 'c' at the time specified in + 'atimeout_at'. The callback is handled by a thread in the specified + thread group (event_type). + + @param c Continuation to be called back at the time specified in + 'atimeout_at'. + @param atimeout_at Time value at which to callback. + @param event_type thread group id (or event type) specifying the + group of threads on which to schedule the callback. + @param callback_event code to be passed back to the continuation's + handler. See the Remarks section. + @param cookie user-defined value or pointer to be passed back in + the Event's object cookie field. + @return reference to an Event object representing the scheduling of + this callback. + */ Event *schedule_at(Continuation *c, ink_hrtime atimeout_at, EventType event_type = ET_CALL, int callback_event = EVENT_INTERVAL, void *cookie = nullptr); /** - Schedules @p c on a thread of group @p event_type to be - dispatched after @p atimeout_in elapses. - - Computes an absolute deadline of @c ink_get_hrtime() @c + - @p atimeout_in and enqueues a one-shot Event for that time. The - selected thread is chosen using the same rule as - @c schedule_imm. - - @param c Continuation to dispatch. MUST be non-null - and live until the resulting Event is - delivered or cancelled. - @param atimeout_in Relative delay in @c ink_hrtime units. Zero - or negative values are legal; they yield a - deadline at or before the current time and - fire at the dispatch loop's next - opportunity. - @param event_type @c EventType (group id). Defaults to - @c ET_CALL. - @param callback_event @c event_id passed on dispatch. Defaults - to @c EVENT_INTERVAL. - @param cookie Stored verbatim in @c Event::cookie. - - @pre @p c is non-null. The threads for @p event_type are - spawned. - @post On success, an Event is enqueued for delivery at the - computed absolute time. If the Event System is in shutdown, - no Event is enqueued. - - @return Pointer to the scheduled Event, or @c nullptr if the - Event System is in shutdown. Use @c Event::cancel to - detach. Framework-owned; do not delete. - - @par Errors - Cannot fail at the contract level. - - @par Thread Safety - Caller-synchronized with respect to @p c: safe to call from any - thread provided no other thread is concurrently scheduling @p c - or otherwise reading or writing @c c->thread_affinity. The - external-queue enqueue itself is thread-safe. + Schedules the continuation on a specific thread group to receive an + event after the specified timeout elapses. Requests the EventProcessor + to schedule the callback to the continuation 'c' after the time + specified in 'atimeout_in' elapses. The callback is handled by a + thread in the specified thread group (event_type). + + @param c Continuation to call back aftert the timeout elapses. + @param atimeout_in amount of time after which to callback. + @param event_type Thread group id (or event type) specifying the + group of threads on which to schedule the callback. + @param callback_event code to be passed back to the continuation's + handler. See the Remarks section. + @param cookie user-defined value or pointer to be passed back in + the Event's object cookie field. + @return reference to an Event object representing the scheduling of + this callback. + */ Event *schedule_in(Continuation *c, ink_hrtime atimeout_in, EventType event_type = ET_CALL, int callback_event = EVENT_INTERVAL, void *cookie = nullptr); /** - Schedules @p c on a thread of group @p event_type to be - dispatched repeatedly every @p aperiod. - - For positive @p aperiod the first dispatch occurs after - @p aperiod elapses and thereafter the Event fires every - @p aperiod until @c Event::cancel is called. For negative - @p aperiod the Event joins the negative-event (poll) rotation - and the handler is dispatched once per event-loop iteration with - @c EVENT_POLL regardless of @p callback_event. The selected - thread is chosen using the same rule as @c schedule_imm. - - @param c Continuation to dispatch. MUST be non-null - and live until the resulting Event is - cancelled. - @param aperiod Period between successive dispatches in - @c ink_hrtime units. MUST be non-zero. - Negative values switch to negative-event - semantics. - @param event_type @c EventType (group id). Defaults to - @c ET_CALL. - @param callback_event @c event_id passed on each positive-period - dispatch. Ignored when @p aperiod is - negative. Defaults to @c EVENT_INTERVAL. - @param cookie Stored verbatim in @c Event::cookie. - - @pre @p c is non-null. @p aperiod is non-zero. The threads for - @p event_type are spawned. - @post On success, a recurring Event is enqueued. If the Event - System is in shutdown, no Event is enqueued. - - @return Pointer to the recurring Event, or @c nullptr if the - Event System is in shutdown. The caller MUST eventually - call @c Event::cancel. - - @par Errors - Cannot fail at the contract level. - - @par Thread Safety - Caller-synchronized with respect to @p c: safe to call from any - thread provided no other thread is concurrently scheduling @p c - or otherwise reading or writing @c c->thread_affinity. The - external-queue enqueue itself is thread-safe. + Schedules the continuation on a specific thread group to receive + an event periodically. Requests the EventProcessor to schedule the + callback to the continuation 'c' every time 'aperiod' elapses. The + callback is handled by a thread in the specified thread group + (event_type). + + @param c Continuation to call back every time 'aperiod' elapses. + @param aperiod duration of the time period between callbacks. + @param event_type thread group id (or event type) specifying the + group of threads on which to schedule the callback. + @param callback_event code to be passed back to the continuation's + handler. See the Remarks section. + @param cookie user-defined value or pointer to be passed back in + the Event's object cookie field. + @return reference to an Event object representing the scheduling of + this callback. + */ Event *schedule_every(Continuation *c, ink_hrtime aperiod, EventType event_type = ET_CALL, int callback_event = EVENT_INTERVAL, void *cookie = nullptr); - /** - Schedules @p c on every thread in group @p event_type. - - For each thread in the group an independent Event is allocated - and enqueued on that thread. Each Event is given a fresh - @c ProxyMutex (rather than sharing @p c 's mutex), so the - per-thread invocations may run concurrently. Used by subsystems - that want a per-thread copy of the same Continuation (e.g., - per-thread initialization or shutdown). - - The timing parameters select one of the following modes; at - least one of @p atimeout and @p aperiod MUST be zero: - - @p atimeout @c == @c 0 and @p aperiod @c == @c 0: each Event - is a one-shot dispatched on the next event-loop iteration. - - @p atimeout @c > @c 0 and @p aperiod @c == @c 0: each Event - is a one-shot dispatched at @c ink_get_hrtime() @c + - @p atimeout. - - @p atimeout @c == @c 0 and @p aperiod @c > @c 0: each Event is - recurring with period @p aperiod; the first dispatch occurs at - @c ink_get_hrtime() @c + @p aperiod. - - @p atimeout @c == @c 0 and @p aperiod @c < @c 0: each Event - joins the negative-event (poll) rotation and the handler is - dispatched once per event-loop iteration. - - @param c Continuation to dispatch. Must remain - valid for the lifetime of every produced - Event. - @param atimeout Relative delay before the (single) one-shot - fire when @p aperiod is zero. Otherwise - MUST be zero. - @param aperiod Period between recurring dispatches; zero - for one-shot. Negative values switch to - negative-event semantics. - @param event_type @c EventType (group id). Defaults to - @c ET_CALL. - @param callback_event @c event_id passed on each positive-period - or one-shot dispatch. Ignored when - @p aperiod is negative. Defaults to - @c EVENT_IMMEDIATE. - @param cookie Stored verbatim in each Event's - @c cookie. - - @pre @p c is non-null. The threads for @p event_type are - spawned. At least one of @p atimeout and @p aperiod is - zero; behavior is undefined when both are non-zero. - @post One Event per thread in the group is enqueued, regardless - of Event-System shutdown state. - - @return Vector of @c TSAction handles, one per thread in the - group, in thread-index order. Each entry is a wrapper - around the per-thread Event; cancel via the wrapper. - - @par Errors - Cannot fail at the contract level. - - @par Thread Safety - Safe to call from any thread. - */ std::vector schedule_entire(Continuation *c, ink_hrtime atimeout, ink_hrtime aperiod, EventType event_type = ET_CALL, int callback_event = EVENT_IMMEDIATE, void *cookie = nullptr); - // Defect: declared but never defined and never called. Linking against any of - // these will fail; treat as dead declarations pending removal. + //////////////////////////////////////////// + // reschedule an already scheduled event. // + // may be called directly or called by // + // schedule_xxx Event member functions. // + // The returned value may be different // + // from the argument e. // + //////////////////////////////////////////// + Event *reschedule_imm(Event *e, int callback_event = EVENT_IMMEDIATE); Event *reschedule_at(Event *e, ink_hrtime atimeout_at, int callback_event = EVENT_INTERVAL); Event *reschedule_in(Event *e, ink_hrtime atimeout_in, int callback_event = EVENT_INTERVAL); Event *reschedule_every(Event *e, ink_hrtime aperiod, int callback_event = EVENT_INTERVAL); - /** - Registers a Continuation to be dispatched once on every thread of - group @p ev_type at thread-spawn time. - - Adds an Event template to @p ev_type 's spawn queue. When each - thread in @p ev_type starts, it walks the queue and invokes - every registered Continuation's handler with a fresh per-thread - Event whose @c ethread points at the spawning thread. Threads - in @p ev_type that have already been spawned do NOT receive the - Event. - - @param c Continuation to dispatch on each newly spawned - thread of @p ev_type. Must remain valid through - every per-thread dispatch. - @param ev_type @c EventType to install on. Use - @c register_event_type to obtain a fresh value. - @param event @c event_id passed to @c handleEvent on each - dispatch. Defaults to @c EVENT_IMMEDIATE. - @param cookie Stored verbatim in each per-thread Event's - @c cookie. - - @pre @p c is non-null. @p ev_type was returned by a prior - @c register_event_type. The threads for @p ev_type have - NOT yet been spawned; for the implicit @c ET_CALL group - this means @c EventProcessor::start has not been called - yet. - @post The Continuation is registered for delivery on each - thread of @p ev_type that the processor subsequently - spawns. - - @return Pointer to the registered template Event. - Framework-owned. - - @par Errors - Cannot fail at the contract level. - - @par Thread Safety - Caller-restricted by convention: invoked from the main thread - during process startup. - */ + /// Schedule an @a event on continuation @a c when a thread of type @a ev_type is spawned. + /// The @a cookie is attached to the event instance passed to the continuation. + /// @return The scheduled event. Event *schedule_spawn(Continuation *c, EventType ev_type, int event = EVENT_IMMEDIATE, void *cookie = nullptr); - /** - Convenience overload: invokes @p f once on every thread of - @p ev_type at thread-spawn time. - - Wraps @p f in a framework-owned stub Continuation; semantics are - otherwise the same as the Continuation overload above. The - callback event is fixed to @c EVENT_IMMEDIATE. - - @param f Free function called on each newly spawned - thread of @p ev_type. Receives a pointer to the - @c EThread on which it runs. - @param ev_type @c EventType to install on. Use - @c register_event_type to obtain a fresh value. - - @pre @p f is non-null. @p ev_type was returned by a prior - @c register_event_type. The threads for @p ev_type have - NOT yet been spawned. - @post @p f is registered for delivery on each thread of - @p ev_type that the processor subsequently spawns. - - @return Pointer to the registered template Event. - Framework-owned. - - @par Errors - Cannot fail at the contract level. - - @par Thread Safety - Caller-restricted by convention: invoked from the main thread - during process startup. - */ + /// Schedule the function @a f to be called in a thread of type @a ev_type when it is spawned. Event *schedule_spawn(void (*f)(EThread *), EventType ev_type); + /// Schedule an @a event on continuation @a c to be called when a thread is spawned by this processor. + /// The @a cookie is attached to the event instance passed to the continuation. + /// @return The scheduled event. // Event *schedule_spawn(Continuation *c, int event, void *cookie = NULL); - /** - Constructs a fresh @c EventProcessor with the @c ET_CALL group - reserved but no threads spawned. - - The global @c eventProcessor singleton is constructed at process - startup; user code does not normally instantiate this class. - - @post @c n_thread_groups @c == @c 1 (the @c ET_CALL group is - reserved with its registered name); @c n_ethreads and - @c n_dthreads are zero; no EThreads exist. The internal - @c thread_initializer Continuation is constructed and the - dedicated-thread mutex is initialized. - - @par Errors - Cannot fail at the contract level. - - @par Thread Safety - Safe to call from any thread. The constructed instance is not - yet observable to others. - */ EventProcessor(); - /** - Destroys an @c EventProcessor. - - Tears down the internal dedicated-thread spawn mutex. Does not - stop, join, or otherwise reclaim any EThread previously spawned - by this @c EventProcessor; those must already be quiesced. - - @pre No EThread spawned by this @c EventProcessor is still - running, and no thread is currently inside - @c spawn_event_threads on this @c EventProcessor. - - @par Errors - Aborts the process if the underlying mutex teardown reports an - error (for example, if the mutex is still locked). - - @par Thread Safety - Safe to call from any thread once the precondition holds. - */ ~EventProcessor() override; EventProcessor(const EventProcessor &) = delete; EventProcessor &operator=(const EventProcessor &) = delete; /** - Initializes the @c EventProcessor and spawns the @c ET_CALL - thread group. - - Initializes thread-affinity bookkeeping, registers the - Event-System metric stats, prepends a thread-affinity - initializer Continuation to the @c ET_CALL spawn queue (so it - runs first on every @c ET_CALL thread), and spawns - @p n_net_threads @c REGULAR EThreads in @c ET_CALL. Each - Continuation registered for @c ET_CALL via @c schedule_spawn - runs once on every newly spawned thread before that thread - enters its event loop. After this call returns, additional - groups may be created via @c register_event_type / - @c spawn_event_threads. - - @param n_net_threads Number of threads to spawn for the initial - @c ET_CALL group. MUST be positive and not - exceed @c MAX_EVENT_THREADS. - @param stacksize Per-thread stack size in bytes. Values - below @c INK_THREAD_STACK_MIN are clamped - up to it, and the result is rounded up to - a multiple of the page size (or huge-page - size when huge pages are enabled). The - default argument @c DEFAULT_STACKSIZE - selects the platform default. - - @pre Has not been called before on any @c EventProcessor in - this process. @p n_net_threads is positive and does not - exceed @c MAX_EVENT_THREADS. - @post @c ET_CALL group has @p n_net_threads spawned threads; - @c n_ethreads is incremented by @p n_net_threads. - @c n_thread_groups is unchanged. - - @return Zero. The contract retains the negative-on-failure - convention from @c Processor::start, but the current - implementation has no failure path that returns a - value (resource exhaustion aborts). - - @par Errors - Aborts the process via @c ink_release_assert if the - precondition is violated or if a resource-exhaustion failure - occurs during thread spawn. - - @par Thread Safety - Caller-restricted: invoked once from the main thread during - process startup. + Initializes the EventProcessor and its associated threads. Spawns the + specified number of threads, initializes their state information and + sets them running. It creates the initial thread group, represented + by the event type ET_CALL. + + @return 0 if successful, and a negative value otherwise. + */ int start(int n_net_threads, size_t stacksize = DEFAULT_STACKSIZE) override; /** - Hook for shutting down the @c EventProcessor subsystem. - - @pre No preconditions. - @post No observable side effects. + Stop the EventProcessor. Attempts to stop the EventProcessor and + all of the threads in each of the thread groups. - @par Errors - Cannot fail. - - @par Thread Safety - Safe to call from any thread. */ - // The current implementation is an empty body: it exists so the - // override resolution from Processor::shutdown works, but there is - // no per-subsystem work to do here. Process shutdown is handled - // elsewhere via TSSystemState. void shutdown() override; /** - Reserves @p size bytes inside every @c EThread's - @c thread_private region and returns the byte offset, measured - from the start of the @c EThread, at which those bytes begin. - - Used by subsystems that want a per-EThread chunk of state. The - returned offset is suitable for use with @c ETHREAD_GET_PTR. - The reserved region's starting address is 16-byte aligned, and - @p size is rounded up to a multiple of 16. + Allocates size bytes on the event threads. This function is thread + safe. - @param size Number of bytes to reserve. MUST be non-negative. + @param size bytes to be allocated. - @pre @p size is non-negative. - @post On success, the running reservation counter is advanced - past the reserved region. On failure, no state is changed. - - @return Byte offset, measured from the start of an @c EThread, - at which the reserved region begins; or @c -1 if the - requested allocation would not fit within - @c PER_THREAD_DATA. - - @par Errors - Returns @c -1 if the requested size cannot be satisfied. - - @par Thread Safety - Safe to call from any thread; concurrent calls are serialized - via a compare-and-swap retry loop. */ off_t allocate(int size); /** - Storage for every @c REGULAR EThread spawned by this processor. - - Indices [0, @c n_ethreads) are valid pointers; remaining slots - are @c nullptr. Consumers iterate via @c active_ethreads (or one - of the per-group accessors) rather than indexing directly. + An array of pointers to all of the EThreads handled by the + EventProcessor. An array of pointers to all of the EThreads created + throughout the existence of the EventProcessor instance. - @par Thread Safety - Written only during thread-pool spawn (single-threaded process - startup). Safe to read concurrently after startup. */ EThread *all_ethreads[MAX_EVENT_THREADS]; - /** - Per-group state for a single @c EventType. - - The thread-group id is the index into @c thread_group; it is not - stored on the descriptor itself. Accessed by name from - consumers that want raw access to the per-group thread vector - or count; most consumers prefer @c active_group_threads. - - @par Ownership - Owned by @c EventProcessor; lives for the processor's lifetime. - - @par Thread Safety - Members are written during thread-pool spawn (single-threaded - process startup) except for @c _started (atomic) and - @c _next_round_robin (per-call increment by the dispatch path). - */ + /// Data kept for each thread group. + /// The thread group ID is the index into an array of these and so is not stored explicitly. struct ThreadGroupDescriptor { - /** - Name registered with @c register_event_type. Stable for the - lifetime of the descriptor. - - @par Thread Safety - Written once at registration; safe to read concurrently. - */ - std::string _name; - /** - Number of threads in this group. Set by @c spawn_event_threads - before any of those threads start running. - - @par Thread Safety - Plain @c int. Written at thread-pool spawn; safe to read - concurrently after startup. - */ - int _count = 0; - /** - Atomic counter incremented each time a thread in this group - finishes its per-thread initialization and signals readiness. - When @c _started == @c _count, the group is fully running. - - @par Thread Safety - @c std::atomic. Each thread in the group performs one - increment with @c std::memory_order_seq_cst (the default for - @c operator++); readers use @c load with the same default. - Other modules may observe a value bounded by the number of - threads that have completed their per-thread init. - */ - std::atomic _started = 0; - /** - Round-robin cursor used by @c assign_thread to pick the next - thread in the group for a fresh @c schedule_imm / - @c schedule_at / @c schedule_in / @c schedule_every call. - - @par Thread Safety - Plain @c uint64_t. Read and incremented without - synchronization by @c assign_thread. - */ - // Defect: see the data-race note on EventProcessor::assign_thread. - uint64_t _next_round_robin = 0; - /** - Template Events whose continuations are dispatched on each - thread in this group at spawn time. Populated by - @c schedule_spawn before @c spawn_event_threads runs. Each - newly spawned thread iterates this queue read-only, building - a per-thread Event from each entry to invoke its continuation; - the queue itself is not drained. - - @par Thread Safety - Written only from the main thread before the group's threads - spawn; read-only thereafter from each spawning thread. - */ - Que(Event, link) _spawnQueue; - /** - Pointers to the EThreads in this group. Indices [0, - @c _count) are valid; remaining slots are @c nullptr. - - @par Thread Safety - Written during thread-pool spawn; safe to read concurrently - after startup. - */ - EThread *_thread[MAX_THREADS_IN_EACH_TYPE] = {}; - /** - Optional callback invoked once when every thread in the group - has finished its per-thread initialization (i.e., when - @c _started reaches @c _count). May be @c nullptr. - - @par Thread Safety - Written by the main thread before the group's threads spawn; - called once from the last spawning thread to reach the - threshold. - */ - std::function _afterStartCallback = nullptr; + std::string _name; ///< Name for the thread group. + int _count = 0; ///< # of threads of this type. + std::atomic _started = 0; ///< # of started threads of this type. + uint64_t _next_round_robin = 0; ///< Index of thread to use for events assigned to this group. + Que(Event, link) _spawnQueue; ///< Events to dispatch when thread is spawned. + EThread *_thread[MAX_THREADS_IN_EACH_TYPE] = {}; ///< The actual threads in this group. + std::function _afterStartCallback = nullptr; }; - /** - Per-group descriptors indexed by @c EventType. - - Indices [0, @c n_thread_groups) are populated; the rest are - default-constructed. - - @par Thread Safety - Same as the per-descriptor contracts above. - */ + /// Storage for per group data. ThreadGroupDescriptor thread_group[MAX_EVENT_TYPES]; - /** - Number of registered thread groups. - - Bumped by @c register_event_type each time a fresh group is - reserved. - - @par Thread Safety - Plain @c int. Written by @c register_event_type during process - startup; safe to read concurrently after that. - */ + /// Number of defined thread groups. int n_thread_groups = 0; /** - Total number of @c REGULAR EThreads spawned by this processor. + Total number of threads controlled by this EventProcessor. This is + the count of all the EThreads spawn by this EventProcessor, excluding + those created by spawn_thread - Equals the sum of @c thread_group[i]._count for valid @c i. - Excludes @c DEDICATED threads created via @c spawn_thread. - - @par Thread Safety - Plain @c int. Written by @c spawn_event_threads during process - startup; safe to read concurrently after that. */ int n_ethreads = 0; - /** - Returns whether every thread in group @p etype has finished its - per-thread initialization. - - @param etype @c EventType (group id) to query. - - @pre @p etype is in the range [0, @c n_thread_groups). - @post No observable side effects. - @return @c true iff every thread in the group has signaled - readiness; @c false otherwise. - - @par Errors - Cannot fail. - - @par Thread Safety - Safe to call from any thread. - */ bool has_tg_started(int etype); /*------------------------------------------------------*\ | Unix & non NT Interface | \*------------------------------------------------------*/ - /** - Schedules an already-initialized Event @p e onto group @p etype. - - Selects a thread within @p etype 's group as follows: if - @p e 's continuation has a thread affinity that belongs to - @p etype 's group, that thread is used; otherwise if the - calling thread is itself in @p etype 's group it is used; - otherwise the group's round-robin cursor is consulted. When the - continuation had no prior affinity the chosen thread is - recorded as its affinity. If the continuation has a mutex, that - mutex is also installed on @p e. The Event is then enqueued on - the chosen thread's external queue, using the local-enqueue - fast path if the chosen thread is the caller's own thread. - - @param e Event whose @c init has been called. Ownership of - @p e passes to the framework on success. - @param etype @c EventType (group id) to dispatch on. - - @pre @p e is a freshly initialized Event whose @c continuation - is non-null. Threads for @p etype have been spawned. - @post On success, @p e->ethread points at a thread in - @p etype 's group, @p e->mutex equals the continuation's - mutex when the continuation has one, and @p e is enqueued - for delivery on the chosen thread. If the Event System is - in shutdown, @p e is not enqueued and ownership stays - with the caller. - - @return @p e on success, or @c nullptr if the Event System is - in shutdown. - - @par Errors - Cannot fail at the contract level. - - @par Thread Safety - Safe to call from any thread. - */ - // Lower-level entry point used by schedule_imm / schedule_at / - // schedule_in / schedule_every after they allocate and initialize - // the Event. - Event *schedule(Event *e, EventType etype); - /** - Returns a pointer to one of the EThreads in group @p etype, - chosen by the round-robin cursor. - - @param etype @c EventType (group id). - - @pre @p etype is in the range [0, @c MAX_EVENT_TYPES). - Threads for @p etype have been spawned. - @post When the group has more than one thread, the group's - round-robin cursor advances by one and the returned - thread is the one at that cursor position modulo the - group size. Single-thread groups always return that - one thread and leave the cursor unchanged. - - @return Pointer to a thread in the group; never @c nullptr - given the precondition. - - @par Errors - Cannot fail at the contract level. - - @par Thread Safety - Safe to call from any single thread for groups whose - @c _count is at most one. For larger groups, concurrent - calls update the round-robin cursor non-atomically. - */ - // Defect: when a group has more than one thread, the cursor - // increment `++tg->_next_round_robin` is a non-atomic - // read-modify-write on a value concurrently read and written - // by other threads via this routine. That is a C++ data race - // (undefined behavior). Fix is to make `_next_round_robin` - // `std::atomic` and use a relaxed fetch_add. + Event *schedule(Event *e, EventType etype); EThread *assign_thread(EventType etype); - /** - Returns an EThread chosen for @p cont in group @p etype using - the affinity rule. - - Selects a thread by the following priority: - 1. If @p cont 's mutex's holding thread is in @p etype 's - group, that thread. - 2. Otherwise if @p cont already has a registered thread - affinity in @p etype 's group, that thread. - 3. Otherwise the group's round-robin cursor via - @c assign_thread. - - If @p cont has no prior thread affinity, the chosen thread is - recorded as its affinity. A prior affinity is never overwritten, - even when it is not in @p etype 's group. - - @param cont Continuation to schedule. Its affinity may be - updated as a side effect when no prior affinity - was set. - @param etype @c EventType (group id) in the range - [0, @c MAX_EVENT_TYPES). - - @pre @p cont is non-null and its @c mutex holds a non-null - @c thread_holding. Threads for @p etype have been - spawned. No other thread concurrently reads or writes - @p cont 's @c thread_affinity. - @post If @p cont had no prior affinity, its affinity is set to - the returned thread. The returned thread is in - @p etype 's group. - - @return Pointer to the chosen thread; never @c nullptr given - the precondition. - - @par Errors - Cannot fail at the contract level. - - @par Thread Safety - The read and write of @p cont 's @c thread_affinity are - unsynchronized; the caller must serialize them against any - other accessor. Round-robin selection inherits the race - characteristics of @c assign_thread. - */ EThread *assign_affinity_by_type(Continuation *cont, EventType etype); - /** - Storage for every @c DEDICATED EThread spawned by - @c spawn_thread. - - Indices [0, @c n_dthreads) are valid pointers; the rest are - @c nullptr. - - @par Thread Safety - Written by @c spawn_thread under - @c dedicated_thread_spawn_mutex. Safe to read concurrently - after the spawn completes; readers MUST first read - @c n_dthreads to know the valid range. - */ EThread *all_dthreads[MAX_EVENT_THREADS]; - /** - Number of @c DEDICATED EThreads spawned by this processor. - - @par Thread Safety - Plain @c int. Written by @c spawn_thread under - @c dedicated_thread_spawn_mutex; safe to read concurrently - after the spawn completes. - */ - int n_dthreads = 0; - /** - Running tally of bytes reserved out of every EThread's - @c thread_private region by @c allocate. Updated atomically by - @c allocate so multiple subsystems can register from different - threads without corruption. - - @par Thread Safety - Plain @c int updated via @c ink_atomic_cas in @c allocate; - safe for concurrent updates. - */ - int thread_data_used = 0; + int n_dthreads = 0; // No. of dedicated threads + int thread_data_used = 0; - /** - Range view over a contiguous, valid prefix of an EThread - pointer array. - - Provides @c begin and @c end iterators for use in range-for - loops and STL algorithms. Constructed by @c active_ethreads, - @c active_dthreads, and @c active_group_threads; cannot be - constructed directly by consumers. - - @par Ownership - Stateless reference into one of the @c EventProcessor's - pointer arrays; outlives the underlying array only as long as - the @c EventProcessor itself does. - - @par Thread Safety - The view is safe to use after thread-pool startup is complete; - iterating concurrently with thread spawn is not safe. - */ + /// Provide container style access to just the active threads, not the entire array. class active_threads_type { using iterator = EThread *const *; ///< Internal iterator type, pointer to array element. @@ -1064,69 +352,19 @@ class EventProcessor : public Processor friend class EventProcessor; }; - /** - Returns a range over every @c REGULAR EThread spawned by this - processor. - - @pre No preconditions. - @post No observable side effects. - @return Range whose @c begin / @c end span @c all_ethreads[0, - n_ethreads). - - @par Errors - Cannot fail. - - @par Thread Safety - Safe to call from any thread once @c spawn_event_threads has - finished spawning every group; iterating the returned range - concurrently with thread spawn is not safe. - */ + // These can be used in container for loops and other range operations. active_threads_type active_ethreads() const { return {all_ethreads, n_ethreads}; } - /** - Returns a range over every @c DEDICATED EThread spawned by - this processor. - - @pre No preconditions. - @post No observable side effects. - @return Range whose @c begin / @c end span @c all_dthreads[0, - n_dthreads). - - @par Errors - Cannot fail. - - @par Thread Safety - Safe to call from any thread once dedicated-thread spawning has - quiesced; iterating the returned range concurrently with - @c spawn_thread is not safe. - */ active_threads_type active_dthreads() const { return {all_dthreads, n_dthreads}; } - /** - Returns a range over every EThread in group @p type. - - @param type @c EventType (group id) to iterate. - - @pre @p type is in the range [0, @c n_thread_groups). - @post No observable side effects. - @return Range whose @c begin / @c end span the group's - @c _thread[0, _count) slice. - - @par Errors - Cannot fail. - - @par Thread Safety - Safe to call from any thread once the group's threads have - been spawned. - */ active_threads_type active_group_threads(int type) const { @@ -1161,19 +399,6 @@ class EventProcessor : public Processor ink_mutex dedicated_thread_spawn_mutex; }; -/** - Global @c EventProcessor singleton. - - Defined in the inkevent library. Every executable that links - inkevent observes this single instance; consumers schedule work - through it (e.g., @c eventProcessor.schedule_imm(cont, ET_TASK)). - - @par Ownership - Static-storage singleton; lives for the entire process lifetime. - - @par Thread Safety - See @c EventProcessor type-level contract. -*/ extern class EventProcessor eventProcessor; void thread_started(EThread *);