Refactor websocket composed ops

This commit is contained in:
Vinnie Falco
2017-06-29 09:48:30 -07:00
parent 4c41db9ca8
commit 4ff3b524c6
3 changed files with 135 additions and 166 deletions

View File

@ -6,6 +6,7 @@ Version 71:
WebSockets: WebSockets:
* Fine tune websocket op asserts * Fine tune websocket op asserts
* Refactor websocket composed ops
-------------------------------------------------------------------------------- --------------------------------------------------------------------------------

View File

@ -29,23 +29,18 @@ template<class NextLayer>
template<class Handler> template<class Handler>
class stream<NextLayer>::close_op class stream<NextLayer>::close_op
{ {
using fb_type = detail::frame_streambuf;
struct data : op struct data : op
{ {
bool cont;
stream<NextLayer>& ws; stream<NextLayer>& ws;
close_reason cr; close_reason cr;
fb_type fb; detail::frame_streambuf fb;
int state = 0; int state = 0;
data(Handler& handler, stream<NextLayer>& ws_, data(Handler&, stream<NextLayer>& ws_,
close_reason const& cr_) close_reason const& cr_)
: ws(ws_) : ws(ws_)
, cr(cr_) , cr(cr_)
{ {
using boost::asio::asio_handler_is_continuation;
cont = asio_handler_is_continuation(std::addressof(handler));
ws.template write_close< ws.template write_close<
static_buffer>(fb, cr); static_buffer>(fb, cr);
} }
@ -63,19 +58,16 @@ public:
: d_(std::forward<DeducedHandler>(h), : d_(std::forward<DeducedHandler>(h),
ws, std::forward<Args>(args)...) ws, std::forward<Args>(args)...)
{ {
(*this)(error_code{}, false);
} }
void operator()() void operator()()
{ {
(*this)(error_code{}); (*this)({});
} }
void void
operator()(error_code ec, std::size_t); operator()(error_code ec,
std::size_t bytes_transferred = 0);
void
operator()(error_code ec, bool again = true);
friend friend
void* asio_handler_allocate( void* asio_handler_allocate(
@ -98,7 +90,9 @@ public:
friend friend
bool asio_handler_is_continuation(close_op* op) bool asio_handler_is_continuation(close_op* op)
{ {
return op->d_->cont; using boost::asio::asio_handler_is_continuation;
return asio_handler_is_continuation(
std::addressof(op->d_.handler()));
} }
template<class Function> template<class Function>
@ -119,85 +113,72 @@ operator()(error_code ec, std::size_t)
{ {
auto& d = *d_; auto& d = *d_;
if(ec) if(ec)
d.ws.failed_ = true;
(*this)(ec);
}
template<class NextLayer>
template<class Handler>
void
stream<NextLayer>::close_op<Handler>::
operator()(error_code ec, bool again)
{
auto& d = *d_;
d.cont = d.cont || again;
if(ec)
goto upcall;
for(;;)
{ {
switch(d.state) BOOST_ASSERT(d.ws.wr_block_ == &d);
d.ws.failed_ = true;
goto upcall;
}
switch(d.state)
{
case 0:
if(d.ws.wr_block_)
{ {
case 0: // suspend
if(d.ws.wr_block_)
{
// suspend
d.state = 2;
d.ws.wr_op_.emplace(std::move(*this));
return;
}
if(d.ws.failed_ || d.ws.wr_close_)
{
// call handler
d.ws.get_io_service().post(
bind_handler(std::move(*this),
boost::asio::error::operation_aborted));
return;
}
d.ws.wr_block_ = &d;
BEAST_FALLTHROUGH;
case 1:
// send close frame
BOOST_ASSERT(d.ws.wr_block_ == &d);
d.state = 99;
d.ws.wr_close_ = true;
boost::asio::async_write(d.ws.stream_,
d.fb.data(), std::move(*this));
return;
case 2:
BOOST_ASSERT(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
d.state = 3;
// The current context is safe but might not be
// the same as the one for this operation (since
// we are being called from a write operation).
// Call post to make sure we are invoked the same
// way as the final handler for this operation.
d.ws.get_io_service().post(
bind_handler(std::move(*this), ec));
return;
case 3:
BOOST_ASSERT(d.ws.wr_block_ == &d);
if(d.ws.failed_ || d.ws.wr_close_)
{
// call handler
ec = boost::asio::error::operation_aborted;
goto upcall;
}
d.state = 1; d.state = 1;
break; d.ws.wr_op_.emplace(std::move(*this));
return;
}
d.ws.wr_block_ = &d;
if(d.ws.failed_ || d.ws.wr_close_)
{
// call handler
d.ws.get_io_service().post(
bind_handler(std::move(*this),
boost::asio::error::operation_aborted));
return;
}
case 99: do_write:
// send close frame
BOOST_ASSERT(d.ws.wr_block_ == &d);
d.state = 3;
d.ws.wr_close_ = true;
boost::asio::async_write(d.ws.stream_,
d.fb.data(), std::move(*this));
return;
case 1:
BOOST_ASSERT(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
d.state = 2;
// The current context is safe but might not be
// the same as the one for this operation (since
// we are being called from a write operation).
// Call post to make sure we are invoked the same
// way as the final handler for this operation.
d.ws.get_io_service().post(
bind_handler(std::move(*this), ec));
return;
case 2:
BOOST_ASSERT(d.ws.wr_block_ == &d);
if(d.ws.failed_ || d.ws.wr_close_)
{
// call handler
ec = boost::asio::error::operation_aborted;
goto upcall; goto upcall;
} }
goto do_write;
case 3:
break;
} }
upcall: upcall:
if(d.ws.wr_block_ == &d) BOOST_ASSERT(d.ws.wr_block_ == &d);
d.ws.wr_block_ = nullptr; d.ws.wr_block_ = nullptr;
d.ws.rd_op_.maybe_invoke() || d.ws.rd_op_.maybe_invoke() ||
d.ws.ping_op_.maybe_invoke(); d.ws.ping_op_.maybe_invoke() ||
d.ws.wr_op_.maybe_invoke();
d_.invoke(ec); d_.invoke(ec);
} }
@ -214,7 +195,7 @@ async_close(close_reason const& cr, CloseHandler&& handler)
void(error_code)> init{handler}; void(error_code)> init{handler};
close_op<handler_type< close_op<handler_type<
CloseHandler, void(error_code)>>{ CloseHandler, void(error_code)>>{
init.completion_handler, *this, cr}; init.completion_handler, *this, cr}({});
return init.result.get(); return init.result.get();
} }
@ -239,6 +220,11 @@ close(close_reason const& cr, error_code& ec)
static_assert(is_sync_stream<next_layer_type>::value, static_assert(is_sync_stream<next_layer_type>::value,
"SyncStream requirements not met"); "SyncStream requirements not met");
BOOST_ASSERT(! wr_close_); BOOST_ASSERT(! wr_close_);
if(wr_close_)
{
ec = boost::asio::error::operation_aborted;
return;
}
wr_close_ = true; wr_close_ = true;
detail::frame_streambuf fb; detail::frame_streambuf fb;
write_close<static_buffer>(fb, cr); write_close<static_buffer>(fb, cr);

View File

@ -32,17 +32,14 @@ class stream<NextLayer>::ping_op
{ {
struct data : op struct data : op
{ {
bool cont;
stream<NextLayer>& ws; stream<NextLayer>& ws;
detail::frame_streambuf fb; detail::frame_streambuf fb;
int state = 0; int state = 0;
data(Handler& handler, stream<NextLayer>& ws_, data(Handler&, stream<NextLayer>& ws_,
detail::opcode op_, ping_data const& payload) detail::opcode op_, ping_data const& payload)
: ws(ws_) : ws(ws_)
{ {
using boost::asio::asio_handler_is_continuation;
cont = asio_handler_is_continuation(std::addressof(handler));
using boost::asio::buffer; using boost::asio::buffer;
using boost::asio::buffer_copy; using boost::asio::buffer_copy;
ws.template write_ping< ws.template write_ping<
@ -62,17 +59,15 @@ public:
: d_(std::forward<DeducedHandler>(h), : d_(std::forward<DeducedHandler>(h),
ws, std::forward<Args>(args)...) ws, std::forward<Args>(args)...)
{ {
(*this)(error_code{}, false);
} }
void operator()() void operator()()
{ {
(*this)(error_code{}); (*this)({});
} }
void operator()(error_code ec, std::size_t); void operator()(error_code ec,
std::size_t bytes_transferred = 0);
void operator()(error_code ec, bool again = true);
friend friend
void* asio_handler_allocate( void* asio_handler_allocate(
@ -95,7 +90,9 @@ public:
friend friend
bool asio_handler_is_continuation(ping_op* op) bool asio_handler_is_continuation(ping_op* op)
{ {
return op->d_->cont; using boost::asio::asio_handler_is_continuation;
return asio_handler_is_continuation(
std::addressof(op->d_.handler()));
} }
template<class Function> template<class Function>
@ -111,85 +108,70 @@ public:
template<class NextLayer> template<class NextLayer>
template<class Handler> template<class Handler>
void void
stream<NextLayer>::ping_op<Handler>:: stream<NextLayer>::
ping_op<Handler>::
operator()(error_code ec, std::size_t) operator()(error_code ec, std::size_t)
{ {
auto& d = *d_; auto& d = *d_;
if(ec) if(ec)
d.ws.failed_ = true;
(*this)(ec);
}
template<class NextLayer>
template<class Handler>
void
stream<NextLayer>::
ping_op<Handler>::
operator()(error_code ec, bool again)
{
auto& d = *d_;
d.cont = d.cont || again;
if(ec)
goto upcall;
for(;;)
{ {
switch(d.state) BOOST_ASSERT(d.ws.wr_block_ == &d);
d.ws.failed_ = true;
goto upcall;
}
switch(d.state)
{
case 0:
if(d.ws.wr_block_)
{ {
case 0: // suspend
if(d.ws.wr_block_)
{
// suspend
d.state = 2;
d.ws.ping_op_.emplace(std::move(*this));
return;
}
if(d.ws.failed_ || d.ws.wr_close_)
{
// call handler
d.state = 99;
d.ws.get_io_service().post(
bind_handler(std::move(*this),
boost::asio::error::operation_aborted));
return;
}
d.ws.wr_block_ = &d;
BEAST_FALLTHROUGH;
case 1:
// send ping frame
BOOST_ASSERT(d.ws.wr_block_ == &d);
d.state = 99;
boost::asio::async_write(d.ws.stream_,
d.fb.data(), std::move(*this));
return;
case 2:
BOOST_ASSERT(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
d.state = 3;
// The current context is safe but might not be
// the same as the one for this operation (since
// we are being called from a write operation).
// Call post to make sure we are invoked the same
// way as the final handler for this operation.
d.ws.get_io_service().post(
bind_handler(std::move(*this), ec));
return;
case 3:
BOOST_ASSERT(d.ws.wr_block_ == &d);
if(d.ws.failed_ || d.ws.wr_close_)
{
// call handler
ec = boost::asio::error::operation_aborted;
goto upcall;
}
d.state = 1; d.state = 1;
break; d.ws.ping_op_.emplace(std::move(*this));
return;
}
d.ws.wr_block_ = &d;
if(d.ws.failed_ || d.ws.wr_close_)
{
// call handler
d.ws.get_io_service().post(
bind_handler(std::move(*this),
boost::asio::error::operation_aborted));
return;
}
case 99: do_write:
// send ping frame
BOOST_ASSERT(d.ws.wr_block_ == &d);
d.state = 3;
boost::asio::async_write(d.ws.stream_,
d.fb.data(), std::move(*this));
return;
case 1:
BOOST_ASSERT(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
d.state = 2;
// The current context is safe but might not be
// the same as the one for this operation (since
// we are being called from a write operation).
// Call post to make sure we are invoked the same
// way as the final handler for this operation.
d.ws.get_io_service().post(
bind_handler(std::move(*this), ec));
return;
case 2:
BOOST_ASSERT(d.ws.wr_block_ == &d);
if(d.ws.failed_ || d.ws.wr_close_)
{
// call handler
ec = boost::asio::error::operation_aborted;
goto upcall; goto upcall;
} }
goto do_write;
case 3:
break;
} }
upcall: upcall:
BOOST_ASSERT(d.ws.wr_block_ == &d); BOOST_ASSERT(d.ws.wr_block_ == &d);
@ -213,7 +195,7 @@ async_ping(ping_data const& payload, WriteHandler&& handler)
ping_op<handler_type< ping_op<handler_type<
WriteHandler, void(error_code)>>{ WriteHandler, void(error_code)>>{
init.completion_handler, *this, init.completion_handler, *this,
detail::opcode::ping, payload}; detail::opcode::ping, payload}({});
return init.result.get(); return init.result.get();
} }
@ -231,7 +213,7 @@ async_pong(ping_data const& payload, WriteHandler&& handler)
ping_op<handler_type< ping_op<handler_type<
WriteHandler, void(error_code)>>{ WriteHandler, void(error_code)>>{
init.completion_handler, *this, init.completion_handler, *this,
detail::opcode::pong, payload}; detail::opcode::pong, payload}({});
return init.result.get(); return init.result.get();
} }