redisgate Gent helio helio_ufd_fiber_interface

Helio fiber_interface

fiber_interface.h

----------------------------------------------------------------------------------------------------
🟢 util/fibers/detail/fiber_interface.h namespace: util/fb2/detail
----------------------------------------------------------------------------------------------------
#include <atomic>
#include <boost/context/fiber.hpp>     ⚛ 설명 (chatgpt)
#include <boost/intrusive/list.hpp>     ⚛ 설명 (chatgpt)
#include <boost/intrusive/set.hpp>     ⚛ 설명 (chatgpt)
#include <chrono>     ⚛ 설명 (chatgpt)
#include "base/mpsc_intrusive_queue.h"
#include "base/pmr/memory_resource.h"
#include "util/fibers/detail/wait_queue.h"

📦 class FiberInterface 요약

class FiberInterface { friend class Scheduler; protected: ::boost::context::fiber_context entry_; // class fiber public: enum Type : uint8_t { MAIN, DISPATCH, WORKER }; // Fiber의 타입을 나타내는 열거형. FI_ListHook list_hook; // ready/terminate 큐에 추가하기 위해 사용. FI_SleepHook sleep_hook; // sleep 큐에 추가하기 위해 사용. FI_ListHook fibers_hook; // For a list of all fibers in the thread. 스레드의 모든 fibers 리스트를 위한 훅. // use_count_(init_count), type_(type), name_(nm) FiberInterface(Type type, uint32_t init_count, std::string_view nm = std::string_view{}); ::boost::context::fiber_context SwitchTo(); // 현재 'fiber'를 중단하고 다른 'fiber'로 전환. SwitchSetup(),resume_with() void SwitchToAndExecute(std::function<void()> fn); // 전환하면서 지정된 함수를 실행. SwitchSetup(),resume_with(),fn() 실행 static void PrintAllFiberStackTraces(); void Start(Launch launch); // fiber 실행을 시작. post: sched->AddReady(this), dispatch: sched->AddReady(fb_init.active); SwitchTo(); void Join(); // 현재 fiber가 종료될 때까지 대기. void Yield(); // 현재 fiber의 실행을 중단하고 다른 fiber로 전환. scheduler_->AddReady(this); scheduler_->Preempt(); bool WaitUntil(std::chrono::steady_clock::time_point tp); // 지정된 시간까지 대기. return scheduler_->WaitUntil(tp, this); fiber_interface_impl.h void Suspend(); // 현재 fiber의 실행을 일시 중지. auto res = scheduler_->Preempt(); void ActivateOther(FiberInterface* other); // 다른 fiber를 실행하지 않고 예약합니다. scheduler_->AddReady(other); void DetachScheduler(); // 스레드 스케줄러에서 자신을 분리합니다. scheduler_->DetachWorker(this); void AttachScheduler(); // 스레드 스케줄러에 자신을 연결하고 실행할 준비를 합니다. scheduler_->Attach(this); scheduler_->AddReady(this); bool IsDefined() const { return bool(entry_); } // DVLOG() friend void intrusive_ptr_add_ref(FiberInterface* ctx) noexcept { ctx->use_count_.fetch_add(1, std::memory_order_relaxed); } friend void intrusive_ptr_release(FiberInterface* ctx) noexcept { if (1 == ctx->use_count_.fetch_sub(1, std::memory_order_release)) { Scheduler* scheduler() { return scheduler_; } Type type() const { return type_; } friend void MPSC_intrusive_store_next(FiberInterface* dest, FiberInterface* next_node) { dest->remote_next_.store(next_node, std::memory_order_relaxed); // 원격 큐에 다음 노드 설정. } friend FiberInterface* MPSC_intrusive_load_next(const FiberInterface& src) { return src.remote_next_.load(std::memory_order_acquire); // 원격 큐에서 다음 노드 로드. } const char* name() const { return name_; } uint32_t DEBUG_use_count() const { return use_count_.load(std::memory_order_relaxed); // 현재 참조 카운트 반환. } Waiter CreateWaiter() { return Waiter{this}; } void PullMyselfFromRemoteReadyQueue(); // 자신을 원격 준비 큐에서 제거. bool IsScheduledRemotely() const { return uint64_t(remote_next_.load(std::memory_order_relaxed)) != kRemoteFree; } uint32_t stack_size() const { return stack_size_; } protected: static constexpr uint16_t kTerminatedBit = 0x1; // 'fiber'가 종료되었음을 나타내는 비트. static constexpr uint16_t kBusyBit = 0x2; // 'fiber'가 바쁨 상태임을 나타내는 비트. static constexpr uint16_t kScheduleRemote = 0x4; // 다른 스레드에서 'fiber'를 예약할 때 사용되는 플래그. std::atomic<uint16_t> flags_{0}; // 상태 플래그. Join(), Terminate()에서 사용. Type type_; // 'fiber'의 타입. enum Type : uint8_t { MAIN, DISPATCH, WORKER }; char name_[24]; // 'fiber'의 이름. std::atomic<uint32_t> use_count_; // intrusive_ptr 참조 카운팅에 사용. uint64_t cpu_tsc_ = 0; // 이 'fiber'가 준비되거나 활성화된 시점의 tsc (사이클 단위). std::chrono::steady_clock::time_point tp_; // 타임아웃과 함께 대기할 때 사용됩니다. 이 'fiber'가 깨워져야 하는 시간을 지정합니다. uint32_t stack_size_ = 0; // 스택 크기. WaitQueue join_q_; // 이 'fiber'가 종료될 때까지 'join'하는 다른 'FiberInterface'들을 위한 큐. Scheduler* scheduler_ = nullptr; // 'fiber'가 속한 스케줄러. TL_FiberInitializer sched; std::atomic<FiberInterface*> remote_next_{nullptr}; // 원격 큐에서 다음 'fiber'를 가리킴. ::boost::context::fiber_context Terminate(); // 'fiber'를 종료하는 함수. private: FiberInterface* SwitchSetup(); }; void FiberInterface::Start(Launch launch) { auto& fb_init = detail::FbInitializer(); // struct TL_FiberInitializer {} fb_init.sched->Attach(this); switch (launch) { case Launch::post: fb_init.sched->AddReady(this); // Activate but do not switch to it. break; case Launch::dispatch: fb_init.sched->AddReady(fb_init.active); // Add the active fiber to the ready queue and switch to the new fiber. { auto fc = SwitchTo(); DCHECK(!fc); } break; // fc=fiber_context=fiber 클래스 } } void FiberInterface::ActivateOther(FiberInterface* other) { // 같은 스레드안에 있으면 if (other->scheduler_ == scheduler_) { // Scheduler* scheduler_ if (!other->list_hook.is_linked()) scheduler_->AddReady(other); } else { // 다른 스레드에 있으면 other->scheduler_->ScheduleFromRemote(other); } } class WorkerFiberImpl : public FiberInterface { using FbCntx = ::boost::context::fiber_context; // class fiber public: WorkerFiberImpl(...): FiberInterface(WORKER, 1, name) { entry_ = FbCntx(...); } private: FbCntx run_(FbCntx&& c) { ... } typename std::decay<Fn>::type fn_; std::tuple<std::decay_t<Arg>...> arg_; }; boost::context::preallocated MakePreallocated(const boost::context::stack_context& sctx) { // util/fibers/fibers.h에서 호출 static WorkerFiberImpl<Fn, Arg...>* MakeWorkerFiberImpl(std::string_view name, StackAlloc&& salloc, Fn&& fn, Arg&&... arg) { boost::context::stack_context sctx = salloc.allocate(); using WorkerImpl = WorkerFiberImpl<Fn, Arg...>; boost::context::preallocated palloc = MakePreallocated<WorkerImpl>(sctx); void* obj_ptr = palloc.sp; // copy because we move palloc. WorkerImpl* fctx = new (obj_ptr) WorkerImpl{name, std::move(palloc), std::forward<StackAlloc>(salloc), std::forward<Fn>(fn), std::forward<Arg>(arg)...}; return fctx; } // TL_FiberInitializer()에서 호출 class MainFiberImpl final : public FiberInterface { public: MainFiberImpl() noexcept : FiberInterface{MAIN, 1, "main"} { } ~MainFiberImpl() { use_count_.fetch_sub(1, memory_order_relaxed); // std::atomic<uint32_t> use_count_; intrusive_ptr 참조 카운팅에 사용. } protected: void Terminate() { } }; struct TL_FiberInitializer { TL_FiberInitializer* next = nullptr; FiberInterface* active; // 현재 활성화된 fiber Scheduler* sched; // 스레드별 스케줄러 인스턴스. uint64_t epoch = 0; // 현재 에포크 번호입니다. 특정한 fiber 실행 상태를 나타낼 수 있습니다. uint64_t switch_delay_cycles = 0; // fiber 전환 지연 시간을 주기 단위로 나타냅니다. uint64_t long_runtime_cnt = 0; // 실행 시간이 1ms 이상 걸린 fiber의 실행 횟수를 추적합니다. uint64_t long_runtime_usec = 0; // 1ms 이상의 실행 시간이 걸린 fiber의 총 실행 시간을 마이크로초 단위로 추적합니다. uint32_t atomic_section = 0; // 원자적 섹션을 나타내는 카운터입니다. 원자적 섹션의 진입 및 종료를 관리합니다. TL_FiberInitializer(const TL_FiberInitializer&) = delete; // 복사 생성자는 사용 불가하게 설정합니다. TL_FiberInitializer() noexcept; // 기본 생성자입니다. ~TL_FiberInitializer(); // 소멸자입니다. }; TL_FiberInitializer::TL_FiberInitializer() noexcept : sched(nullptr) { FiberInterface* main_ctx = new MainFiberImpl{}; // 메인 fiber 구현을 위한 객체를 생성합니다. active = main_ctx; // 현재 활성화된 fiber로 설정합니다. sched = new Scheduler(main_ctx); // 새 스케줄러를 생성하고 메인 컨텍스트를 전달합니다. } TL_FiberInitializer::~TL_FiberInitializer() { FiberInterface* main_cntx = sched->main_context(); if (main_cntx == active) { delete sched; delete main_cntx; } } TL_FiberInitializer& FbInitializer() noexcept { thread_local static TL_FiberInitializer fb_initializer; return fb_initializer; } FiberInterface* FiberActive() noexcept { return FbInitializer().active; } void EnterFiberAtomicSection() noexcept { ++FbInitializer().atomic_section; } void LeaveFiberAtomicSection() noexcept { --FbInitializer().atomic_section; } bool IsFiberAtomicSection() noexcept { return FbInitializer().atomic_section > 0; } void PrintAllFiberStackTraces() { FbInitializer().sched->PrintAllFiberStackTraces(); } void ExecuteOnAllFiberStacks(FiberInterface::PrintFn fn) { FbInitializer().sched->ExecuteOnAllFiberStacks(std::move(fn)); } void SetCustomDispatcher(DispatchPolicy* policy) { detail::TL_FiberInitializer& fb_init = detail::FbInitializer(); fb_init.sched->AttachCustomPolicy(policy); } uint64_t FiberSwitchEpoch() noexcept { return detail::FbInitializer().epoch; } uint64_t FiberSwitchDelayUsec() noexcept { return detail::FbInitializer().switch_delay_cycles * 1000 / detail::g_tsc_cycles_per_ms; } uint64_t FiberLongRunCnt() noexcept { return detail::FbInitializer().long_runtime_cnt; } uint64_t FiberLongRunSumUsec() noexcept { return detail::FbInitializer().long_runtime_usec; } size_t WorkerFibersStackSize() { return detail::FbInitializer().sched->worker_stack_size(); } size_t WorkerFibersCount() { return detail::FbInitializer().sched->num_worker_fibers(); }

📦 enum class Launch

namespace: util/fb2
// ex) void Start(Launch launch); fiber 실행을 시작. enum class Launch { dispatch, // switch to the fiber immediately. 해당 fiber를 즉시 활성화합니다. // fb_init.sched->AddReady(fb_init.active); -> ready_queue_.push_back(*fibi); // auto fc = SwitchTo(); resume_with() post // enqueue the fiber for activation but continue with the current fiber. // 해당 fiber를 활성화 큐에 넣지만 현재 fiber의 실행을 계속할 것을 나타냅니다. // fb_init.sched->AddReady(this); };

📦 class FixedStackAllocator

namespace: util/fb2
// based on boost::context::fixedsize_stack but uses pmr::memory_resource for allocation. // 이 클래스는 boost::context::fixedsize_stack을 기반으로 하며, // 메모리 할당을 위해 pmr::memory_resource를 사용합니다. class FixedStackAllocator { public: using stack_context = boost::context::stack_context; // 생성자: 메모리 리소스 포인터와 스택 크기를 인자로 받아 초기화합니다. FixedStackAllocator(PMR_NS::memory_resource* mr, std::size_t size = 64 * 1024) : mr_(mr), size_(size) { } // fiber에 사용할 스택 메모리를 할당하고, 스택 컨텍스트(stack_context)를 반환합니다. stack_context allocate() { void* vp = mr_->allocate(size_); // 메모리 리소스에서 스택 크기만큼 메모리를 할당합니다. stack_context sctx; sctx.size = size_; // 스택 컨텍스트의 크기를 설정합니다. sctx.sp = static_cast<char*>(vp) + sctx.size; // 스택 포인터(sp)를 메모리의 끝으로 설정합니다. return sctx; } // 할당된 스택 메모리를 해제합니다. void deallocate(stack_context& sctx) BOOST_NOEXCEPT_OR_NOTHROW { // 스택 포인터로부터 메모리의 시작 위치를 계산합니다. void* vp = static_cast<char*>(sctx.sp) - sctx.size; mr_->deallocate(vp, sctx.size); // 메모리 리소스에서 해당 메모리를 해제합니다. } private: PMR_NS::memory_resource* mr_; // 메모리 리소스 포인터 std::size_t size_; // 할당할 스택의 크기 };

📦 'FI_ListHook', 'FI_SleepHook' 타입 정의

namespace: util/fb2/defail
🟢 원본 ex) FI_ListHook list_hook; // used to add to ready/terminate queues. ready/terminate 큐에 추가하기 위해 사용. FI_ListHook fibers_hook; // For a list of all fibers in the thread. 스레드의 모든 fibers 리스트를 위한 훅. using FI_ListHook = boost::intrusive::list_member_hook<boost::intrusive::link_mode<boost::intrusive::safe_link>>; ex) FI_SleepHook sleep_hook; // sleep 큐(set)에 추가하기 위해 사용. using FI_SleepHook = boost::intrusive::set_member_hook<boost::intrusive::link_mode<boost::intrusive::safe_link>>; 🟢 해설 ⬛️ 'FI_ListHook', 'FI_SleepHook' 타입 정의 using FI_ListHook = boost::intrusive::list_member_hook<boost::intrusive::link_mode<boost::intrusive::safe_link>>; using FI_SleepHook = boost::intrusive::set_member_hook<boost::intrusive::link_mode<boost::intrusive::safe_link>>; - 'FI_ListHook': 'boost::intrusive::list_member_hook'의 타입 정의입니다. 'safe_link' 모드를 사용하여 안전하게 연결된 리스트에 'fiber'를 삽입할 수 있습니다. - 'FI_SleepHook': 'boost::intrusive::set_member_hook'의 타입 정의입니다. 'fiber'가 대기 중일 때 사용됩니다. ⚛ 설명 (chatgpt)
⚛ set_member_hook 설명 (chatgpt)

fiber_interface.cc

📦 제목