redisgate Gent boost_fiber bfiber_1000

Boost.Fiber when_all, return values

when_all, return values - 반환 값

모든 작업 함수에서 반환 값을 수집하자마자 wait_first_value()의 queue를 해당 목적에 맞게 재사용하는 방법을 바로 확인할 수 있습니다.   우리가 해야 할 일은 첫 번째 값 이후에 닫히지 않는 것입니다!

그러나 실제로 여러 값을 수집하면 흥미로운 질문이 제기됩니다.   우리는 그들 중 가장 느린 것이 도착할 때까지 정말로 기다리고 싶습니까?   결과가 나오는 대로 각 결과를 처리하지 않겠습니까?

다행히 두 가지 API를 모두 제공할 수 있습니다.   shared_ptr>을 반환하도록 wait_all_values_source()를 정의해 보겠습니다.

wait_all_values_source()가 주어지면 wait_all_values()를 구현하는 것이 간단합니다. template< typename Fn, typename ... Fns > std::vector< typename std::result_of< Fn() >::type > wait_all_values( Fn && function, Fns && ... functions) { std::size_t count( 1 + sizeof ... ( functions) ); typedef typename std::result_of< Fn() >::type return_t; typedef std::vector< return_t > vector_t; vector_t results; results.reserve( count); // get channel std::shared_ptr< boost::fibers::buffered_channel< return_t > > chan = wait_all_values_source( std::forward< Fn >( function), std::forward< Fns >( functions) ... ); // fill results vector return_t value; while ( boost::fibers::channel_op_status::success == chan->pop(value) ) { results.push_back( value); } // return vector to caller return results; } 다음과 같이 호출될 수 있습니다. std::vector< std::string > values = wait_all_values( [](){ return sleeper("wav_late", 150); }, [](){ return sleeper("wav_middle", 100); }, [](){ return sleeper("wav_early", 50); }); wait_all_values()의 루프에서 볼 수 있듯이 호출자에게 값을 계산하도록 요구하는 대신 완료되면 대기열을 buffered_channel::close()로 wait_all_values_source()를 정의합니다.   그런데 어떻게 그렇게 할 수 있나요? 각 생산자 파이버는 독립적입니다.   buffered_channel::push() 값이 마지막인지 여부는 알 수 없습니다.

queue<>에 대한 계산 파사드(façade)를 사용하여 이 문제를 해결할 수 있습니다.   실제로 우리의 파사드는 대기열의 생산자 측만 지원하면 됩니다.

[wait_nqueue]

nqueue<>로 무장하면 wait_all_values_source()를 구현할 수 있습니다.   wait_first_value()처럼 시작됩니다.   차이점은 queue를 nqueue로 래핑하여 생산자 파이버에 전달한다는 것입니다.

그런 다음, 물론 첫 번째 값을 팝하고 큐를 닫고 반환하는 대신 단순히 shared_ptr>를 반환합니다. // 호출자가 'closed' 때까지 새로운 결과가 도착할 때마다 이를 검색할 수 있는 // shared_ptr<buffered_channel<T>>을 반환합니다. template< typename Fn, typename ... Fns > std::shared_ptr< boost::fibers::buffered_channel< typename std::result_of< Fn() >::type > > wait_all_values_source( Fn && function, Fns && ... functions) { std::size_t count( 1 + sizeof ... ( functions) ); typedef typename std::result_of< Fn() >::type return_t; typedef boost::fibers::buffered_channel< return_t > channel_t; // make the channel auto chanp( std::make_shared< channel_t >( 64) ); // 항목을 'count'한 후에 닫히도록 nchannel 파사드(facade)를 만듭니다. auto ncp( std::make_shared< nchannel< return_t > >( chanp, count) ); // 해당 nchannel 파사드를 모든 관련 파이버에 전달합니다. wait_all_values_impl< return_t >( ncp, std::forward< Fn >( function), std::forward< Fns >( functions) ... ); // then return the channel for consumer return chanp; } 예를 들어: std::shared_ptr< boost::fibers::buffered_channel< std::string > > chan = wait_all_values_source( [](){ return sleeper("wavs_third", 150); }, [](){ return sleeper("wavs_second", 100); }, [](){ return sleeper("wavs_first", 50); }); std::string value; while ( boost::fibers::channel_op_status::success == chan->pop(value) ) { std::cout << "wait_all_values_source() => '" << value << "'" << std::endl; } wait_all_values_impl()은 실제로 queue 대신 nqueue를 사용한다는 점을 제외하면 wait_first_value_impl()과 같습니다. template< typename T, typename Fn > void wait_all_values_impl( std::shared_ptr< nchannel< T > > chan, Fn && function) { boost::fibers::fiber( [chan, function](){ chan->push(function()); }).detach(); }