![]() |
Home | Libraries | People | FAQ | More |
Since the simplest form of Boost.Asio asynchronous operation completion token
is a callback function, we could apply the same tactics for Asio as for our
hypothetical AsyncAPI
asynchronous
operations.
Fortunately we need not. Boost.Asio incorporates a mechanism[5] by which the caller can customize the notification behavior of any async operation. Therefore we can construct a completion token which, when passed to a Boost.Asio async operation, requests blocking for the calling fiber.
A typical Asio async function might look something like this:[6]
template < ..., class CompletionToken > deduced_return_type async_something( ... , CompletionToken&& token) { // construct handler_type instance from CompletionToken handler_type<CompletionToken, ...>::typehandler(token)
; // construct async_result instance from handler_type async_result<decltype(handler)>result(handler)
; // ... arrange to call handler on completion ... // ... initiate actual I/O operation ... returnresult.get()
; }
We will engage that mechanism, which is based on specializing Asio’s handler_type<>
template for the CompletionToken
type and the signature of the specific callback. The remainder of this discussion
will refer back to async_something()
as the Asio async function under consideration.
The implementation described below uses lower-level facilities than promise
and future
because the promise
mechanism
interacts badly with io_service::stop()
.
It produces broken_promise
exceptions.
boost::fibers::asio::yield
is a completion token of this kind.
yield
is an instance of
yield_t
:
class yield_t { public: yield_t() = default; /** * @code * static yield_t yield; * boost::system::error_code myec; * func(yield[myec]); * @endcode * @c yield[myec] returns an instance of @c yield_t whose @c ec_ points * to @c myec. The expression @c yield[myec] "binds" @c myec to that * (anonymous) @c yield_t instance, instructing @c func() to store any * @c error_code it might produce into @c myec rather than throwing @c * boost::system::system_error. */ yield_t operator[]( boost::system::error_code & ec) const { yield_t tmp; tmp.ec_ = & ec; return tmp; } //private: // ptr to bound error_code instance if any boost::system::error_code * ec_{ nullptr }; };
yield_t
is in fact only a
placeholder, a way to trigger Boost.Asio customization. It can bind a boost::system::error_code
for use by the actual
handler.
yield
is declared as:
// canonical instance thread_local yield_t yield{};
Asio customization is engaged by specializing boost::asio::handler_type<>
for yield_t
:
// Handler type specialisation for fibers::asio::yield. // When 'yield' is passed as a completion handler which accepts only // error_code, use yield_handler<void>. yield_handler will take care of the // error_code one way or another. template< typename ReturnType > struct handler_type< fibers::asio::yield_t, ReturnType( boost::system::error_code) > { typedef fibers::asio::detail::yield_handler< void > type; };
(There are actually four different specializations in detail/yield.hpp, one for each of the four Asio async callback signatures we expect.)
The above directs Asio to use yield_handler
as the actual handler for an async operation to which yield
is passed. There’s a generic yield_handler<T>
implementation and a yield_handler<void>
specialization. Let’s start with the <void>
specialization:
// yield_handler<void> is like yield_handler<T> without value_. In fact it's // just like yield_handler_base. template<> class yield_handler< void >: public yield_handler_base { public: explicit yield_handler( yield_t const& y) : yield_handler_base{ y } { } // nullary completion callback void operator()() { ( * this)( boost::system::error_code() ); } // inherit operator()(error_code) overload from base class using yield_handler_base::operator(); };
async_something()
,
having consulted the handler_type<>
traits specialization, instantiates
a yield_handler<void>
to
be passed as the actual callback for the async operation. yield_handler
’s
constructor accepts the yield_t
instance (the yield
object
passed to the async function) and passes it along to yield_handler_base
:
// This class encapsulates common elements between yield_handler<T> (capturing // a value to return from asio async function) and yield_handler<void> (no // such value). See yield_handler<T> and its <void> specialization below. Both // yield_handler<T> and yield_handler<void> are passed by value through // various layers of asio functions. In other words, they're potentially // copied multiple times. So key data such as the yield_completion instance // must be stored in our async_result<yield_handler<>> specialization, which // should be instantiated only once. class yield_handler_base { public: yield_handler_base( yield_t const& y) : // capture the context* associated with the running fiber ctx_{ boost::fibers::context::active() }, // capture the passed yield_t yt_( y ) { } // completion callback passing only (error_code) void operator()( boost::system::error_code const& ec) { BOOST_ASSERT_MSG( ycomp_, "Must inject yield_completion* " "before calling yield_handler_base::operator()()"); BOOST_ASSERT_MSG( yt_.ec_, "Must inject boost::system::error_code* " "before calling yield_handler_base::operator()()"); // If originating fiber is busy testing state_ flag, wait until it // has observed (completed != state_). yield_completion::lock_t lk{ ycomp_->mtx_ }; yield_completion::state_t state = ycomp_->state_; // Notify a subsequent yield_completion::wait() call that it need not // suspend. ycomp_->state_ = yield_completion::complete; // set the error_code bound by yield_t * yt_.ec_ = ec; // unlock the lock that protects state_ lk.unlock(); // If ctx_ is still active, e.g. because the async operation // immediately called its callback (this method!) before the asio // async function called async_result_base::get(), we must not set it // ready. if ( yield_completion::waiting == state) { // wake the fiber fibers::context::active()->schedule( ctx_); } } //private: boost::fibers::context * ctx_; yield_t yt_; // We depend on this pointer to yield_completion, which will be injected // by async_result. yield_completion::ptr_t ycomp_{}; };
yield_handler_base
stores
a copy of the yield_t
instance
— which, as shown above, contains only an error_code*
. It also captures the context
*
for the currently-running fiber by calling context::active()
.
You will notice that yield_handler_base
has one more data member (ycomp_
)
that is initialized to nullptr
by its constructor — though its operator()()
method relies on ycomp_
being non-null. More on this in a moment.
Having constructed the yield_handler<void>
instance, async_something()
goes on to construct an async_result
specialized for the handler_type<>::type
:
in this case, async_result<yield_handler<void>>
.
It passes the yield_handler<void>
instance to the new async_result
instance.
// Without the need to handle a passed value, our yield_handler<void> // specialization is just like async_result_base. template<> class async_result< boost::fibers::asio::detail::yield_handler< void > > : public boost::fibers::asio::detail::async_result_base { public: typedef void type; explicit async_result( boost::fibers::asio::detail::yield_handler< void > & h): boost::fibers::asio::detail::async_result_base{ h } { } };
Naturally that leads us straight to async_result_base
:
// Factor out commonality between async_result<yield_handler<T>> and // async_result<yield_handler<void>> class async_result_base { public: explicit async_result_base( yield_handler_base & h) : ycomp_{ new yield_completion{} } { // Inject ptr to our yield_completion instance into this // yield_handler<>. h.ycomp_ = this->ycomp_; // if yield_t didn't bind an error_code, make yield_handler_base's // error_code* point to an error_code local to this object so // yield_handler_base::operator() can unconditionally store through // its error_code* if ( ! h.yt_.ec_) { h.yt_.ec_ = & ec_; } } void get() { // Unless yield_handler_base::operator() has already been called, // suspend the calling fiber until that call. ycomp_->wait(); // The only way our own ec_ member could have a non-default value is // if our yield_handler did not have a bound error_code AND the // completion callback passed a non-default error_code. if ( ec_) { throw_exception( boost::system::system_error{ ec_ } ); } } private: // If yield_t does not bind an error_code instance, store into here. boost::system::error_code ec_{}; yield_completion::ptr_t ycomp_; };
This is how yield_handler_base::ycomp_
becomes non-null: async_result_base
’s
constructor injects a pointer back to its own yield_completion
member.
Recall that the canonical yield_t
instance yield
initializes
its error_code*
member ec_
to nullptr
. If this instance is passed to async_something()
(ec_
is still nullptr
), the copy stored in yield_handler_base
will likewise have null
ec_
. async_result_base
’s
constructor sets yield_handler_base
’s
yield_t
’s ec_
member to point to its own error_code
member.
The stage is now set. async_something()
initiates the actual async operation, arranging
to call its yield_handler<void>
instance on completion. Let’s say, for the sake of argument, that the actual
async operation’s callback has signature void(error_code)
.
But since it’s an async operation, control returns at once to async_something()
.
async_something()
calls async_result<yield_handler<void>>::get()
,
and will return its return value.
async_result<yield_handler<void>>::get()
inherits
async_result_base::get()
.
async_result_base::get()
immediately
calls yield_completion::wait()
.
// Bundle a completion bool flag with a spinlock to protect it. struct yield_completion { enum state_t { init, waiting, complete }; typedef fibers::detail::spinlock mutex_t; typedef std::unique_lock< mutex_t > lock_t; typedef boost::intrusive_ptr< yield_completion > ptr_t; std::atomic< std::size_t > use_count_{ 0 }; mutex_t mtx_{}; state_t state_{ init }; void wait() { // yield_handler_base::operator()() will set state_ `complete` and // attempt to wake a suspended fiber. It would be Bad if that call // happened between our detecting (complete != state_) and suspending. lock_t lk{ mtx_ }; // If state_ is already set, we're done here: don't suspend. if ( complete != state_) { state_ = waiting; // suspend(unique_lock<spinlock>) unlocks the lock in the act of // resuming another fiber fibers::context::active()->suspend( lk); } } friend void intrusive_ptr_add_ref( yield_completion * yc) noexcept { BOOST_ASSERT( nullptr != yc); yc->use_count_.fetch_add( 1, std::memory_order_relaxed); } friend void intrusive_ptr_release( yield_completion * yc) noexcept { BOOST_ASSERT( nullptr != yc); if ( 1 == yc->use_count_.fetch_sub( 1, std::memory_order_release) ) { std::atomic_thread_fence( std::memory_order_acquire); delete yc; } } };
Supposing that the pending async operation has not yet completed, yield_completion::completed_
will still be false
, and wait()
will call context::suspend()
on
the currently-running fiber.
Other fibers will now have a chance to run.
Some time later, the async operation completes. It calls yield_handler<void>::operator()(error_code const&)
with an error_code
indicating either success or failure. We’ll consider both cases.
yield_handler<void>
explicitly
inherits operator()(error_code const&)
from yield_handler_base
.
yield_handler_base::operator()(error_code const&)
first sets yield_completion::completed_
true
. This way, if async_something()
’s
async operation completes immediately — if yield_handler_base::operator()
is called even before async_result_base::get()
— the calling fiber will not suspend.
The actual error_code
produced
by the async operation is then stored through the stored yield_t::ec_
pointer.
If async_something()
’s
caller used (e.g.) yield[my_ec]
to bind a local error_code
instance, the actual error_code
value is stored into the caller’s variable. Otherwise, it is stored into
async_result_base::ec_
.
If the stored fiber context yield_handler_base::ctx_
is not already running, it is marked as ready to run by passing it to context::schedule()
.
Control then returns from yield_handler_base::operator()
: the callback is done.
In due course, that fiber is resumed. Control returns from context::suspend()
to
yield_completion::wait()
,
which returns to async_result_base::get()
.
yield[my_ec]
to async_something()
to bind a local error_code
instance, then yield_handler_base::operator()
stored its error_code
to the caller’s my_ec
instance, leaving async_result_base::ec_
initialized to success.
yield
to async_something()
without binding a local error_code
variable, then yield_handler_base::operator()
stored its error_code
into async_result_base::ec_
.
If in fact that error_code
is success, then all is well.
error_code
and yield_handler_base::operator()
was called with an error_code
indicating error — async_result_base::get()
throws system_error
with that error_code
.
The case in which async_something()
’s completion callback has signature void()
is
similar. yield_handler<void>::operator()()
invokes the machinery above with a “success” error_code
.
A completion callback with signature void(error_code, T)
(that is: in addition to error_code
,
callback receives some data item) is handled somewhat differently. For this
kind of signature, handler_type<>::type
specifies yield_handler<T>
(for
T
other than void
).
A yield_handler<T>
reserves
a value_
pointer to a value
of type T
:
// asio uses handler_type<completion token type, signature>::type to decide // what to instantiate as the actual handler. Below, we specialize // handler_type< yield_t, ... > to indicate yield_handler<>. So when you pass // an instance of yield_t as an asio completion token, asio selects // yield_handler<> as the actual handler class. template< typename T > class yield_handler: public yield_handler_base { public: // asio passes the completion token to the handler constructor explicit yield_handler( yield_t const& y) : yield_handler_base{ y } { } // completion callback passing only value (T) void operator()( T t) { // just like callback passing success error_code (*this)( boost::system::error_code(), std::move(t) ); } // completion callback passing (error_code, T) void operator()( boost::system::error_code const& ec, T t) { BOOST_ASSERT_MSG( value_, "Must inject value ptr " "before caling yield_handler<T>::operator()()"); // move the value to async_result<> instance BEFORE waking up a // suspended fiber * value_ = std::move( t); // forward the call to base-class completion handler yield_handler_base::operator()( ec); } //private: // pointer to destination for eventual value // this must be injected by async_result before operator()() is called T * value_{ nullptr }; };
This pointer is initialized to nullptr
.
When async_something()
instantiates async_result<yield_handler<T>>
:
// asio constructs an async_result<> instance from the yield_handler specified // by handler_type<>::type. A particular asio async method constructs the // yield_handler, constructs this async_result specialization from it, then // returns the result of calling its get() method. template< typename T > class async_result< boost::fibers::asio::detail::yield_handler< T > > : public boost::fibers::asio::detail::async_result_base { public: // type returned by get() typedef T type; explicit async_result( boost::fibers::asio::detail::yield_handler< T > & h) : boost::fibers::asio::detail::async_result_base{ h } { // Inject ptr to our value_ member into yield_handler<>: result will // be stored here. h.value_ = & value_; } // asio async method returns result of calling get() type get() { boost::fibers::asio::detail::async_result_base::get(); return std::move( value_); } private: type value_{}; };
this async_result<>
specialization reserves a member of type T
to receive the passed data item, and sets yield_handler<T>::value_
to point to its own data member.
async_result<yield_handler<T>>
overrides get()
.
The override calls async_result_base::get()
,
so the calling fiber suspends as described above.
yield_handler<T>::operator()(error_code, T)
stores
its passed T
value into
async_result<yield_handler<T>>::value_
.
Then it passes control to yield_handler_base::operator()(error_code)
to deal with waking the original fiber as
described above.
When async_result<yield_handler<T>>::get()
resumes,
it returns the stored value_
to async_something()
and ultimately to async_something()
’s caller.
The case of a callback signature void(T)
is handled by having yield_handler<T>::operator()(T)
engage
the void(error_code, T)
machinery,
passing a “success” error_code
.
The source code above is found in yield.hpp and detail/yield.hpp.