redisgate Gent helio helio_uf_proactor_pool

Helio proactor_pool

proactor_pool.h

⦿ 위치: util/proactor_pool.h     namespace: util
⦿ proactor_pool.h를 포함하는 파일들(h/cc)
    util/sliding_counter.h
    util/fibers/pool.h
    util/fibers/proactor_pool.cc
    util/fibers/accept_server.cc
    util/fibers/listener_interface.cc
    util/metrics/family.cc
    util/metrics/metrics.cc
⦿ proactor_pool.h가 include하는 파일들
    #include <absl/container/flat_hash_set.h>
    #include <string_view>
    #include "base/RWSpinLock.h"
    #include "base/pmr/memory_resource.h"
    #include "base/type_traits.h"
    #include "util/fibers/proactor_base.h"

📦 요약

⬛️ 클래스 멤버 및 메소드 설명 1. 타입 별칭 및 템플릿 메타프로그래밍 template <typename Func, typename... Args> using AcceptArgsCheck = typename std::enable_if<base::is_invocable<Func, Args...>::value, int>::type; // 'AcceptArgsCheck'는 주어진 함수 타입 'Func'가 주어진 인수 타입 'Args...'와 함께 호출 가능할 때만 활성화되는 SFINAE(SFINAE: Substitution Failure Is Not An Error) 메타프로그래밍 기법입니다. 2. 생성자 및 소멸자 ProactorPool(const ProactorPool&) = delete; // 복사 생성자와 할당 연산자를 삭제하여 'ProactorPool' 객체의 복사를 방지합니다. void operator=(const ProactorPool&) = delete; explicit ProactorPool(std::size_t pool_size); // 'ProactorPool' 생성자는 스레드 풀의 크기를 설정합니다. // 만약 'pool_size'가 0이면 시스템의 코어 수에 맞춰 풀 크기를 자동으로 설정합니다. // auto num_pthreads = absl::GetFlag(FLAGS_proactor_threads); // pool_size = num_pthreads > 0 ? num_pthreads : NumOnlineCpus(); virtual ~ProactorPool(); // 가상 소멸자입니다. 3. Run 및 Stop 메소드 void Run(); // 'Run'은 모든 'Proactor' 객체를 실행 상태로 만듭니다. // SetupProactors(); void Stop(); // 'Stop'은 모든 IO 컨텍스트를 중지시키고 모든 스레드가 종료될 때까지 현재 스레드를 블록합니다. // proactor_[i]->Stop(); // pthread_join(proactor_[i]->thread_id(), nullptr); 4. Proactor 접근 메소드 ProactorBase* GetNextProactor(); // 'GetNextProactor'는 사용할 다음 'Proactor' 객체를 반환합니다. 이 메소드는 스레드-세이프(thread-safe)합니다. // ProactorBase* proactor = at(index++); ProactorBase& operator[](size_t i); // 'operator[]'와 'at'은 인덱스로 'Proactor' 객체에 접근합니다. ProactorBase* at(size_t i); size_t size() const; // 'size'는 풀의 크기를 반환합니다. 5. 비동기 함수 디스패치 template <typename Func, AcceptArgsCheck<Func, ProactorBase*> = 0> void DispatchBrief(Func&& func); template <typename Func, AcceptArgsCheck<Func, unsigned, ProactorBase*> = 0> void DispatchBrief(Func&& func); - 주어진 함수 'func'를 모든 IO 스레드에서 비동기적으로 실행합니다. 함수는 IO-차단을 하지 않는 CPU 전용 코드여야 합니다. - 첫 번째 템플릿은 'ProactorBase*'를 인수로 받는 함수를 실행합니다. - 두 번째 템플릿은 'unsigned' (IO 컨텍스트 인덱스)와 'ProactorBase*'를 인수로 받는 함수를 실행합니다. 6. 함수 실행 및 대기 template <typename Func, AcceptArgsCheck<Func, ProactorBase*> = 0> void Await(Func&& func); template <typename Func, AcceptArgsCheck<Func, unsigned, ProactorBase*> = 0> void AwaitBrief(Func&& func); - 'Await'는 주어진 함수를 모든 IO 스레드에서 비동기적으로 실행하고, 모든 비동기 호출이 완료될 때까지 대기합니다. - 'AwaitBrief'는 IO 컨텍스트 인덱스와 함께 주어진 함수를 실행하고 모든 호출이 완료될 때까지 대기합니다. 7. Deprecated Await 메소드 template <typename Func, AcceptArgsCheck<Func, unsigned, ProactorBase*> = 0> [[deprecated]] void Await(Func&& func); - 'AwaitBrief'를 대신 사용하도록 권장합니다. 8. Fiber 디스패치 template <typename Func, AcceptArgsCheck<Func, unsigned, ProactorBase*> = 0> void DispatchOnAll(Func&& func); template <typename Func, AcceptArgsCheck<Func, ProactorBase*> = 0> void DispatchOnAll(Func&& func); - 주어진 함수를 모든 IO 스레드에서 비동기적으로 실행하며, 함수는 fiber 내에서 실행됩니다. - 첫 번째 템플릿은 'unsigned' (IO 컨텍스트 인덱스)와 'ProactorBase*'를 인수로 받는 함수를 실행합니다. - 두 번째 템플릿은 'ProactorBase*'를 인수로 받는 함수를 실행합니다. 9. Fiber Await template <typename Func, AcceptArgsCheck<Func, unsigned, ProactorBase*> = 0> void AwaitFiberOnAll(Func&& func); template <typename Func, AcceptArgsCheck<Func, ProactorBase*> = 0> void AwaitFiberOnAll(Func&& func); - 주어진 함수를 모든 IO 스레드에서 비동기적으로 실행하고 모든 콜백이 완료될 때까지 대기합니다. - 함수는 fiber 내에서 실행됩니다. - RunServer@echo_server.cc for server - main@echo_server.cc for client 10. CPU 매핑 const std::vector<unsigned>& MapCpuToThreads(unsigned cpu_id) const; - 특정 'cpu_id'에 매핑된 스레드 인덱스의 벡터를 반환합니다. 매핑된 스레드가 없으면 빈 벡터를 반환합니다. 11. 보조 함수 std::string_view GetString(std::string_view source); // 'GetString'은 문자열을 풀의 전역 스토리지에 저장하고 반환합니다. 내부적으로 RW 스핀락을 사용합니다. const std::vector<std::vector<unsigned>>& cpu_threads() const; // 'cpu_threads'는 'cpu_id'와 매핑된 스레드 배열의 벡터를 반환합니다. 12. 가상 함수 virtual ProactorBase* CreateProactor() = 0; // 'CreateProactor'는 'Proactor' 객체를 생성합니다. 파생 클래스에서 구현되어야 합니다. virtual void InitInThread(unsigned index) = 0; // 'InitInThread'는 특정 인덱스의 스레드에서 초기화를 수행합니다. 파생 클래스에서 구현되어야 합니다. 13. 비공개 멤버 변수 및 함수 void SetupProactors(); // 'SetupProactors'는 'Proactor' 객체를 설정합니다. void WrapLoop(size_t index, BlockingCounter* bc); // 'WrapLoop'는 주어진 인덱스의 루프를 래핑합니다. void CheckRunningState(); // 'CheckRunningState'는 실행 상태를 확인합니다. ⬛️ 멤버 변수 - 'std::unique_ptr<ProactorBase*[]> proactor_' : 'ProactorBase' 객체의 배열입니다. - 'std::atomic_uint_fast32_t next_io_context_{0}' : 다음에 사용할 IO 컨텍스트를 나타내는 원자적 변수입니다. - 'uint32_t pool_size_' : 풀의 크기입니다. - 'folly::RWSpinLock str_lock_' : 문자열 접근을 위한 스핀락입니다. - 'absl::flat_hash_set<std::string_view> str_set_' : 문자열 집합입니다. - 'enum State { STOPPED, RUN } state_ = STOPPED' : 현재 상태를 나타내는 열거형입니다. - 'std::vector<std::vector<unsigned>> cpu_threads_' : 'cpu_id'와 매핑된 스레드 배열의 벡터입니다.

proactor_pool.cc

📦 OnlineCpus() 함수

⦿ 원본 #if defined(__linux__) || defined(__FreeBSD__) // /usr/include/bits/cpu-set.h:#define __CPU_SETSIZE 1024 constexpr int kTotalCpus = CPU_SETSIZE; // struct cpu_set_t static cpu_set_t OnlineCpus() { cpu_set_t online_cpus; CPU_ZERO(&online_cpus); CHECK_EQ(0, sched_getaffinity(0, sizeof(online_cpus), &online_cpus)); return online_cpus; } #elif defined(__APPLE__) #define SYSCTL_CORE_COUNT "machdep.cpu.core_count" constexpr unsigned kTotalCpus = 128; typedef struct { uint64_t __bits[2]; } cpu_set_t; void CPU_ZERO(cpu_set_t* cs) { cs->__bits[0] = 0; cs->__bits[1] = 0; } inline void CPU_SET(unsigned num, cpu_set_t* cs) { unsigned index = num / 64; unsigned rem = num & 63; cs->__bits[index] |= (1 << rem); } inline unsigned CPU_COUNT(const cpu_set_t* cs) { unsigned res = 0; for (auto v : cs->__bits) { res += __builtin_popcount(v); } return res; } static inline int CPU_ISSET(unsigned num, const cpu_set_t* cs) { unsigned index = num / 64; unsigned rem = num & 63; return cs->__bits[index] & (1 << rem); } static cpu_set_t OnlineCpus() { cpu_set_t online_cpus; CPU_ZERO(&online_cpus); int32_t core_count = 0; size_t len = sizeof(core_count); int ret = sysctlbyname(SYSCTL_CORE_COUNT, &core_count, &len, 0, 0); CHECK_EQ(0, ret); for (int i = 0; i < core_count; ++i) { CPU_SET(i, &online_cpus); } return online_cpus; } #else #error "unsupported architecture " #endif static unsigned NumOnlineCpus() { cpu_set_t cpus = OnlineCpus(); return CPU_COUNT(&cpus); }

⦿ 설명 이 코드는 'proactor_pool.cc' 파일에서 CPU의 온라인 상태를 감지하고 사용 가능한 CPU의 수를 반환하는 기능을 제공합니다. 각 운영체제에 맞게 구현되어 있으며, Linux, FreeBSD, macOS를 지원합니다. 자세한 내용은 다음과 같습니다. ⬛️ 리눅스 및 FreeBSD 리눅스와 FreeBSD에서는 '<sched.h>' 라이브러리의 'sched_getaffinity' 함수를 사용하여 현재 프로세스가 실행될 수 있는 CPU 집합을 가져옵니다. 1. 상수 정의 constexpr int kTotalCpus = CPU_SETSIZE; 'CPU_SETSIZE'는 시스템에서 사용할 수 있는 최대 CPU 수를 나타냅니다. 2. OnlineCpus 함수 static cpu_set_t OnlineCpus() { cpu_set_t online_cpus; CPU_ZERO(&online_cpus); CHECK_EQ(0, sched_getaffinity(0, sizeof(online_cpus), &online_cpus)); return online_cpus; } - 'cpu_set_t online_cpus': CPU 집합을 저장할 변수입니다. - 'CPU_ZERO(&online_cpus)': CPU 집합을 초기화합니다. - 'sched_getaffinity(0, sizeof(online_cpus), &online_cpus)': 현재 프로세스가 실행될 수 있는 CPU 집합을 가져옵니다. - 'CHECK_EQ(0, ...)': 'sched_getaffinity'의 반환값이 0인지 확인합니다. 0이 아니면 프로그램이 중단됩니다. ⬛️ macOS macOS에서는 'sysctl' 함수를 사용하여 CPU 정보를 가져옵니다. macOS는 '<sys/sysctl.h>' 헤더 파일에서 'sysctlbyname' 함수를 통해 시스템 정보를 쿼리할 수 있습니다. 1. 상수 및 구조체 정의 #define SYSCTL_CORE_COUNT "machdep.cpu.core_count" constexpr unsigned kTotalCpus = 128; typedef struct { uint64_t __bits[2]; } cpu_set_t; - 'SYSCTL_CORE_COUNT': CPU 코어 수를 가져오기 위한 sysctl 쿼리 문자열입니다. - 'kTotalCpus': macOS에서 최대 128개의 CPU를 사용할 수 있다고 가정합니다. - 'cpu_set_t': CPU 집합을 저장하기 위한 구조체입니다. 2. CPU 집합 조작 함수들 void CPU_ZERO(cpu_set_t* cs) { cs->__bits[0] = 0; cs->__bits[1] = 0; } inline void CPU_SET(unsigned num, cpu_set_t* cs) { unsigned index = num / 64; unsigned rem = num & 63; cs->__bits[index] |= (1 << rem); } inline unsigned CPU_COUNT(const cpu_set_t* cs) { unsigned res = 0; for (auto v : cs->__bits) { res += __builtin_popcount(v); } return res; } static inline int CPU_ISSET(unsigned num, const cpu_set_t* cs) { unsigned index = num / 64; unsigned rem = num & 63; return cs->__bits[index] & (1 << rem); } - 'CPU_ZERO': CPU 집합을 초기화합니다. - 'CPU_SET': 지정된 CPU 번호를 CPU 집합에 추가합니다. - 'CPU_COUNT': CPU 집합에 포함된 CPU의 수를 반환합니다. - 'CPU_ISSET': 특정 CPU 번호가 CPU 집합에 포함되어 있는지 확인합니다. 3. OnlineCpus 함수 static cpu_set_t OnlineCpus() { cpu_set_t online_cpus; CPU_ZERO(&online_cpus); int32_t core_count = 0; size_t len = sizeof(core_count); int ret = sysctlbyname(SYSCTL_CORE_COUNT, &core_count, &len, 0, 0); CHECK_EQ(0, ret); for (int i = 0; i < core_count; ++i) { CPU_SET(i, &online_cpus); } return online_cpus; } - 'sysctlbyname' 함수를 사용하여 시스템의 CPU 코어 수를 가져옵니다. - 가져온 코어 수에 따라 각 코어를 'CPU_SET' 함수를 통해 'online_cpus' 집합에 추가합니다. ⬛️ 공통 함수 NumOnlineCpus 함수 static unsigned NumOnlineCpus() { cpu_set_t cpus = OnlineCpus(); return CPU_COUNT(&cpus); } 이 함수는 'OnlineCpus' 함수를 호출하여 온라인 상태의 CPU 집합을 가져오고, 'CPU_COUNT' 함수를 통해 집합에 포함된 CPU의 수를 반환합니다. ⬛️ 요약 이 코드는 각 운영체제별로 현재 사용 가능한 CPU 집합을 가져오는 방법을 정의하고, 이를 기반으로 온라인 상태의 CPU 수를 계산하여 반환합니다. 리눅스 및 FreeBSD에서는 'sched_getaffinity' 함수를 사용하고, macOS에서는 'sysctlbyname' 함수를 사용하여 CPU 정보를 얻습니다. ⚛ 설명 (chatgpt)

📦 ProactorPool 클래스 생성자와 소멸자

ProactorPool::ProactorPool(std::size_t pool_size) { if (pool_size == 0) { auto num_pthreads = absl::GetFlag(FLAGS_proactor_threads); // thread::hardware_concurrency() returns number of online cpus but ignores taskset. pool_size = num_pthreads > 0 ? num_pthreads : NumOnlineCpus(); VLOG(1) << "Setting pool size to " << pool_size; } pool_size_ = pool_size; proactor_.reset(new ProactorBase*[pool_size]); std::fill(proactor_.get(), proactor_.get() + pool_size, nullptr); } ProactorPool::~ProactorPool() { Stop(); for (size_t i = 0; i < pool_size_; ++i) { delete proactor_[i]; } }

⦿ 설명 'ProactorPool' 클래스의 생성자와 소멸자는 다음과 같이 동작합니다: ⬛️ 생성자: 'ProactorPool::ProactorPool(std::size_t pool_size)' • 입력된 'pool_size' 값을 사용하여 'ProactorPool' 객체를 초기화합니다. • 만약 'pool_size'가 0이라면, 다음 단계를 통해 'pool_size' 값을 설정합니다: - 'absl::GetFlag(FLAGS_proactor_threads)'를 호출하여 'FLAGS_proactor_threads' 플래그의 값을 가져옵니다. 이 값이 'num_pthreads'에 저장됩니다. - 'num_pthreads' 값이 0보다 크면 'pool_size'는 'num_pthreads'로 설정됩니다. 그렇지 않으면 'NumOnlineCpus()' 함수를 호출하여 시스템에서 사용 가능한 온라인 CPU의 수를 가져와서 'pool_size'를 설정합니다. - 'VLOG(1)'을 사용하여 로그 메시지를 출력하고, 설정된 'pool_size' 값을 기록합니다. • 'pool_size_' 멤버 변수를 'pool_size'로 설정합니다. • 'proactor_' 멤버 변수에 'ProactorBase*' 타입의 포인터 배열을 동적 할당합니다. 배열의 크기는 'pool_size'입니다. • 'std::fill'을 사용하여 할당된 배열의 모든 요소를 'nullptr'로 초기화합니다. ⬛️ 소멸자: 'ProactorPool::~ProactorPool()' • 'Stop()' 함수를 호출하여 풀 내의 모든 작업을 중지합니다. • 'proactor_' 배열의 각 요소를 반복하여 순회합니다: - 각 요소가 가리키는 'ProactorBase' 객체를 삭제합니다. (동적으로 할당된 메모리를 해제) • 이는 리소스 누수를 방지하고 동적 할당된 메모리를 적절히 해제합니다. ⬛️ 코드 분석 - 생성자는 입력된 풀 크기에 따라 동적으로 'ProactorBase' 포인터 배열을 초기화합니다. 만약 풀 크기가 0이라면, 시스템의 설정이나 하드웨어의 상태에 따라 적절한 크기를 자동으로 설정합니다. - 소멸자는 생성자에서 동적으로 할당된 메모리를 안전하게 해제하며, 이로 인해 리소스 누수가 발생하지 않도록 합니다. - 'Stop()' 함수는 풀 내의 작업을 중지하는 역할을 합니다. 이는 소멸자가 메모리를 해제하기 전에 실행됩니다. 이 클래스는 동적 메모리 할당 및 해제와 관련된 작업을 안전하게 처리하여 효율적인 리소스 관리를 목표로 하고 있습니다. ⚛ 설명 (chatgpt)

📦 제목