理解Boost.Fiber库中的代码示例时出现问题

Question

我目前正在尝试(为了学习)理解Boost.Fiber库中的代码示例：https://www.boost.org/doc/libs/1_71_0/libs/fiber/examples/work_sharing.cpp

//          Copyright Nat Goodspeed + Oliver Kowalke 2015.
// Distributed under the Boost Software License, Version 1.0.
//    (See accompanying file LICENSE_1_0.txt or copy at
//          http://www.boost.org/LICENSE_1_0.txt)

#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <deque>
#include <iomanip>
#include <iostream>
#include <mutex>
#include <sstream>
#include <string>
#include <thread>

#include <boost/assert.hpp>

#include <boost/fiber/all.hpp>

#include <boost/fiber/detail/thread_barrier.hpp>

static std::size_t fiber_count{ 0 };
static std::mutex mtx_count{};
static boost::fibers::condition_variable_any cnd_count{};
typedef std::unique_lock< std::mutex > lock_type;

/*****************************************************************************
 *   example fiber function
 *****************************************************************************/
//[fiber_fn_ws
void whatevah( char me) {
    try {
        std::thread::id my_thread = std::this_thread::get_id(); /**< get ID of initial thread >*/
        {
            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) { /**< loop ten times >*/
            boost::this_fiber::yield(); /**< yield to other fibers >*/
            std::thread::id new_thread = std::this_thread::get_id(); /**< get ID of current thread >*/
            if ( new_thread != my_thread) { /**< test if fiber was migrated to another thread >*/
                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) { /**< Decrement fiber counter for each completed fiber. >*/
        lk.unlock();
        cnd_count.notify_all(); /**< Notify all fibers waiting on `cnd_count`. >*/
    }
}
//]

/*****************************************************************************
 *   example thread function
 *****************************************************************************/
//[thread_fn_ws
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 >(); /**<
                                                                                    Install the scheduling algorithm `boost::fibers::algo::shared_work` in order to
                                                                                    join the work sharing.
                                                                                    >*/
    b->wait(); /**< sync with other threads: allow them to start processing >*/
    lock_type lk( mtx_count);
    cnd_count.wait( lk, [](){ return 0 == fiber_count; } ); /**<
                                                             Suspend main fiber and resume worker fibers in the meanwhile.
                                                             Main fiber gets resumed (e.g returns from `condition_variable_any::wait()`)
                                                             if all worker fibers are complete.
                                                             >*/
    BOOST_ASSERT( 0 == fiber_count);
}
//]

/*****************************************************************************
 *   main()
 *****************************************************************************/
int main( int argc, char *argv[]) {
    std::cout << "main thread started " << std::this_thread::get_id() << std::endl;
    //[main_ws
    boost::fibers::use_scheduling_algorithm< boost::fibers::algo::shared_work >(); /*<
                                                                                    Install the scheduling algorithm `boost::fibers::algo::shared_work` in the main thread
                                                                                    too, so each new fiber gets launched into the shared pool.
                                                                                    >*/

    for ( char c : std::string("abcdefghijklmnopqrstuvwxyz")) { /*<
                                                                 Launch a number of worker fibers; each worker fiber picks up a character
                                                                 that is passed as parameter to fiber-function `whatevah`.
                                                                 Each worker fiber gets detached.
                                                                 >*/
        boost::fibers::fiber([c](){ whatevah( c); }).detach();
        ++fiber_count; /*< Increment fiber counter for each new fiber. >*/
    }
    boost::fibers::detail::thread_barrier b( 4);
    std::thread threads[] = { /*<
                               Launch a couple of threads that join the work sharing.
                               >*/
        std::thread( thread, & b),
        std::thread( thread, & b),
        std::thread( thread, & b)
    };
    b.wait(); /*< sync with other threads: allow them to start processing >*/
    {
        lock_type/*< `lock_type` is typedef'ed as __unique_lock__< [@http://en.cppreference.com/w/cpp/thread/mutex `std::mutex`] > >*/ lk( mtx_count);
        cnd_count.wait( lk, [](){ return 0 == fiber_count; } ); /*<
                                                                 Suspend main fiber and resume worker fibers in the meanwhile.
                                                                 Main fiber gets resumed (e.g returns from `condition_variable_any::wait()`)
                                                                 if all worker fibers are complete.
                                                                 >*/
    } /*<
       Releasing lock of mtx_count is required before joining the threads, otherwise
       the other threads would be blocked inside condition_variable::wait() and
       would never return (deadlock).
       >*/
    BOOST_ASSERT( 0 == fiber_count);
    for ( std::thread & t : threads) { /*< wait for threads to terminate >*/
        t.join();
    }
    //]
    std::cout << "done." << std::endl;
    return EXIT_SUCCESS;
}

我有一个问题，那就是当所有线程都在等待notify_all()时，在不同的线程中，纤程如何继续执行，因为当所有线程都因为wait函数：cnd_count.wait( lk, [](){ return 0 == fiber_count; } )而阻塞时。所以如果都被wait函数阻塞了，纤程怎么可能继续执行，我认为纤程是由线程在它们自己的调度管理器的帮助下执行的。我读到线程只是执行单元，纤程使用它来运行它们自己的可调用或函数。那么，当所有线程都阻塞时，为什么所有的纤程都继续运行呢？

Answer 1

请参阅：https://www.boost.org/doc/libs/1_71_0/libs/fiber/doc/html/fiber/scheduling.html

每当线程挂起时，调度器都会运行下一个就绪纤程，因此在调用cnd_count.wait时会执行whatevah。

您可以尝试删除thread_fn_ws上的行boost::fibers::use_scheduling_algorithm< boost::fibers::algo::shared_work >();，您会发现所有纤程都在主线程中运行，因为没有安装调度程序，因此纤程不会运行。