Enterprise

Boost.Fiber Migrating fibers between threads

Migrating fibers between threads

📦 Overview - 개요

각 파이버는 스택을 소유하고 모든 레지스터와 CPU 플래그, 명령어 포인터 및 스택 포인터를 포함한 실행 상태를 관리합니다. 이는 일반적으로 파이버가 특정 스레드에 바인딩되지 않음을 의미합니다.[3],[4]

[3] 각 스레드의 "main(기본)" 파이버, 즉 스레드가 시작되는 파이버는 다른 스레드로 마이그레이션할 수 없습니다. 또한 Boost.Fiber는 각 스레드에 대해 암시적으로 분배(dispatcher) 파이버를 생성합니다. 이 역시 마이그레이션할 수 없습니다.

[4] 물론 스레드 로컬 스토리지에 의존하는 파이버를 마이그레이션하는 것은 문제(problematic)가 될 수 있습니다.

작업부하(workload)가 많은 논리적 CPU에서 작업부하가 적은 다른 논리적 CPU로 파이버를 이동(migrating)하면 전체(overall) 실행 속도가 빨라질 수 있습니다.
NUMA 아키텍처의 경우 스레드 간에 데이터를 마이그레이션하는 것이 항상 권장되는 것은 아닙니다.
파이버 f가 NUMA 노드 node0에 속하는 논리 CPU cpu0에서 실행되고 있다고 가정합니다.
f의 데이터는 node0에 위치한 물리적 메모리에 할당됩니다.
파이버를 cpu0에서 다른 NUMA 노드 nodeX의 일부인 다른 논리 CPU cpuX로 이동(移動)하면 메모리 액세스(access) 지연 시간이 늘어나 애플리케이션 성능이 저하될 수 있습니다.
[NUMA 아키텍처] 참조 https://www.boost.org/doc/libs/1_85_0/libs/fiber/doc/html/fiber/numa.html

알고리즘의 준비 대기열에 포함된 파이버만 스레드 간에 이동할 수 있습니다.
실행 중인 파이버나 차단된 파이버는 이동할 수 없습니다.
context::is_context() 메서드가 'pinned_context'에 대해 true를 반환하는 경우 파이버를 이동할 수 없습니다.

Boost.Fiber에서는 파이버가 이동하는 스레드에서 context::detach()를 호출하고 파이버가 받는 스레드에서 context::attach()를 호출하여 파이버가 이동됩니다.
따라서 파이버 이동은 서로 다른 스레드에서 실행되는 사용자 코딩(user-coded) 알고리즘 구현 인스턴스 간에 상태를 공유함으로써 수행됩니다.
파이버의 원래 스레드는 algorithm::awakened()를 호출하여 파이버의 context*를 전달합니다.
awakened() 구현은 context::detach()를 호출합니다.
나중에 동일하거나 다른 스레드가 algorithm::pick_next()를 호출하면 pick_next() 구현은 준비된 파이버를 선택하고 이를 반환하기 전에 그에 대한 context::attach()를 호출합니다.

위에서 설명한 대로 is_context(pinned_context) == true인 컨텍스트는 context::detach() 또는 context::attach()에 전달되어서는 안 됩니다.
해당 컨텍스트를 awakened()에 전달한 동일한 스레드에 의해 호출된 pick_next()에서만 반환될 수 있습니다.

🔧 제목

📦 Example of work sharing - 작업 공유의 예

work_sharing.cpp 예제에서는 여러 작업자 파이버가 기본(main) 스레드에 생성됩니다.
각 파이버는 생성 시 매개변수로 문자를 얻습니다. 이 문자는 10번 인쇄됩니다.
각 반복 사이에 파이버는 this_fiber::yield()를 호출합니다.
그러면 현재 스레드에서 실행 중인 파이버 스케줄러(fiber-scheduler) 'shared_ready_queue'의 준비 대기열에 파이버가 배치됩니다.
실행 준비가 된 다음 파이버는 공유 준비 대기열에서 제외되고(dequeued) 참여(participating) 스레드에서 실행 중인 'shared_ready_queue'에 의해 재개됩니다.

소스 참조: https://www.boost.org/doc/libs/1_85_0/libs/fiber/examples/work_sharing.cpp

'shared_ready_queue'의 모든 인스턴스는 준비 대기열로 사용되는 하나의 전역 동시 대기열(one global concurrent queue)을 공유합니다.
이 메커니즘은 'shared_ready_queue'의 모든 인스턴스 간에, 즉 참여하는 모든 스레드 간에 모든 작업자 파이버를 공유합니다.

🔧 제목

📦 Setup of threads and fibers - 스레드과 파이버의 설정

main()에는 파이버 스케줄러(fiber-scheduler)가 설치되고 작업자 파이버와 스레드가 실행됩니다. boost::fibers::use_scheduling_algorithm< boost::fibers::algo::shared_work >(); •(1) for ( char c : std::string("abcdefghijklmnopqrstuvwxyz")) { •(2) boost::fibers::fiber([c](){ whatevah( c); }).detach(); ++fiber_count; •(3) } boost::fibers::detail::thread_barrier b( 4); std::thread threads[] = { •(4) std::thread( thread, & b), std::thread( thread, & b), std::thread( thread, & b) }; b.wait(); •(5) { lock_type •(6) lk( mtx_count); cnd_count.wait( lk, [](){ return 0 == fiber_count; } ); •(7) } •(8) BOOST_ASSERT( 0 == fiber_count); for ( std::thread & t : threads) { •(9) t.join(); }

소스 설명

(1) 메인 스레드에도 스케줄링 알고리즘 'boost::fibers::algo::shared_work'를 설치하여 각각의 새로운 파이버가 공유 풀에서 시작되도록 합니다.
(2) 다수의 작업자 파이버를 실행합니다. 각 작업자 파이버는 파이버 기능 'whatevah'에 매개변수로 전달되는 문자를 선택합니다. 각 작업자 파이버가 분리됩니다.
(3) 각각의 새로운 파이버에 대해 파이버 카운터를 증가시킵니다.
(4) 작업 공유에 참여하는 몇 개의 스레드를 시작합니다.
(5) 다른 스레드와 동기화: 처리(processing)를 시작하도록 허용
(6) lock_type은 std::unique_lock< std::mutex >로 형식 정의됩니다.
(7) 그 사이에 메인 파이버를 일시 중단하고 작업자 파이버를 재개하십시오. 모든 작업자 파이버가 완료되면 메인 파이버가 재개됩니다(예: condition_variable_any::wait()에서 반환).
(8) 스레드를 결합하기(joining) 전에 'mtx_count'의 잠금을 해제해야 합니다. 그렇지 않으면 다른 스레드가 condition_variable::wait() 내에서 차단되어 결코 반환되지 않습니다(교착 상태).
(9) 스레드가 종료될 때까지 기다립니다.

⚛ pthread와 std::thread의 공통점과 차이점을 설명 (chatgpt)
⚛ pthread 설명

스레드의 시작은 장벽(barrier)과 동기화됩니다.
-> 참조: Barriers 장벽(랑데뷰) https://www.boost.org/doc/libs/1_85_0/libs/fiber/doc/html/fiber/synchronization/barriers.html
각 스레드의 메인 파이버(메인 스레드 포함)는 모든 작업자 파이버가 완성될 때까지 정지됩니다(suspended).
메인 파이버가 condition_variable::wait()에서 반환되면 스레드가 종료됩니다.
즉, 메인 스레드가 다른 모든 스레드에 합류합니다(joins).
void thread( boost::fibers::detail::thread_barrier * b) { std::ostringstream buffer; buffer << "thread started " << std::this_thread::get_id() << std::endl; std::cout << buffer.str() << std::flush; boost::fibers::use_scheduling_algorithm< boost::fibers::algo::shared_work >(); •(1) b->wait(); •(2) lock_type lk( mtx_count); cnd_count.wait( lk, [](){ return 0 == fiber_count; } ); •(3) BOOST_ASSERT( 0 == fiber_count); }

소스 설명

(1) 작업 공유에 참여하려면 스케줄링 알고리즘 'boost::fibers::algo::shared_work'를 설치하십시오.
(2) 다른 스레드와 동기화: 처리(processing)를 시작하도록 허용
(3) 그 사이에 메인 파이버를 일시 중단하고 작업자 파이버를 재개하십시오. 모든 작업자 파이버가 완료되면 메인 파이버가 재개됩니다(예: condition_variable_any::wait()에서 반환).

각 작업자 파이버는 'me'라는 문자를 매개변수로 사용하여 whatevah() 함수를 실행합니다. 파이버는 루프를 생성하고 다른 스레드로 이동된 경우 메시지를 인쇄합니다. void whatevah( char me) { try { std::thread::id my_thread = std::this_thread::get_id(); •(1) { std::ostringstream buffer; buffer << "fiber " << me << " started on thread " << my_thread << '\n'; std::cout << buffer.str() << std::flush; } for ( unsigned i = 0; i < 10; ++i) { •(2) boost::this_fiber::yield(); •(3) std::thread::id new_thread = std::this_thread::get_id(); •(4) if ( new_thread != my_thread) { •(5) my_thread = new_thread; std::ostringstream buffer; buffer << "fiber " << me << " switched to thread " << my_thread << '\n'; std::cout << buffer.str() << std::flush; } } } catch ( ... ) { } lock_type lk( mtx_count); if ( 0 == --fiber_count) { •(6) lk.unlock(); cnd_count.notify_all(); •(7) } }

소스 설명

(1) 초기 스레드 ID 가져오기
(2) 10번 반복하다
(3) 다른 파이버에 양보
(4) 현재 스레드의 ID를 가져옵니다
(5) 파이버가 다른 스레드로 이동되었는지 확인
(6) 완성된 각 파이버에 대한 파이버 카운터를 줄입니다.
(7) 'cnd_count'에서 대기 중인 모든 파이버를 알립니다.

📦 Scheduling fibers - 파이버 스케줄링

파이버 스케줄러 'shared_ready_queue'는 참여하는 모든 스레드 간에 공통 준비 대기열을 공유한다는 점을 제외하면 round_robin과 같습니다. 스레드는 다른 Boost.Fiber 작업 전에 use_scheduling_algorithm()을 실행하여 이 풀에 참여합니다.

준비 대기열에 대한 중요한 점은 이것이 'shared_ready_queue'의 모든 인스턴스에 공통되는 정적 클래스라는 것입니다. 따라서 algorithm::awakened()(재개 준비가 된 파이버)를 통해 대기열에 추가된 파이버는 모든 스레드에서 사용할 수 있습니다. 스레드의 메인 파이버와 분배(dispatcher) 파이버에 대해 별도의 스케줄러별(scheduler-specific) 대기열을 예약해야 합니다. 이러한 대기열은 스레드 간에 공유될 수 없습니다! 다음 파이버 중 하나를 전달하면 공유 대기열 대신 해당 파이버로 푸시합니다. 스레드 B가 스레드 A의 메인 파이버를 검색하고 실행하려고 시도하는 것은 나쁜 소식(Bad News)입니다.

[awakened_ws]
하나의 스레드 내에서 algorithm::pick_next()가 호출되면 파이버는 'rqueue_'에서 제외되고(dequeued) 해당 스레드에서 다시 시작됩니다.

[pick_next_ws]
위 소스코드는 'work_sharing.cpp'에 있습니다.

⚛ 원문

Email 답글이 올라오면 이메일로 알려드리겠습니다.