feedc0de/boost_beast
1
0
Fork 0
forked from boostorg/beast
Code Pull Requests Activity

basic_stream subsumes stranded_stream:

Browse Source 
* All functionality of stranded_stream is folded into basic_stream
* tcp_stream is an alias for basic_stream with tcp
* The tests are expanded to produce full coverage
* Timeout implementation is simplified
This commit is contained in:
Vinnie Falco
2019-02-10 11:42:01 -08:00
parent 0ce8ebbefd
commit 9b14774ada
20 changed files with 3151 additions and 3920 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/boost/beast/core/impl/basic_timeout_stream.hpp → include/boost/beast/core/impl/basic_stream.hpp
+308 -262
View File
@@ -7,11 +7,13 @@
// Official repository: https://github.com/boostorg/beast
//
#ifndef BOOST_BEAST_CORE_IMPL_BASIC_TIMEOUT_STREAM_HPP
#define BOOST_BEAST_CORE_IMPL_BASIC_TIMEOUT_STREAM_HPP
#ifndef BOOST_BEAST_CORE_IMPL_BASIC_STREAM_HPP
#define BOOST_BEAST_CORE_IMPL_BASIC_STREAM_HPP
#include <boost/beast/core/async_op_base.hpp>
#include <boost/beast/core/buffers_prefix.hpp>
#include <boost/beast/core/buffer_size.hpp>
#include <boost/beast/core/detail/type_traits.hpp>
#include <boost/beast/websocket/teardown.hpp>
#include <boost/asio/bind_executor.hpp>
#include <boost/asio/coroutine.hpp>
#include <boost/assert.hpp>
@@ -23,8 +25,96 @@
namespace boost {
namespace beast {
//------------------------------------------------------------------------------
template<class Protocol, class Executor>
struct basic_timeout_stream<
template<class... Args>
basic_stream<Protocol, Executor>::
impl_type::
impl_type(
Executor const& ex_,
Args&&... args)
: boost::empty_value<Executor>(
boost::empty_init_t{}, ex_)
, read(ex().context())
, write(ex().context())
, socket(std::forward<Args>(args)...)
{
reset();
}
template<class Protocol, class Executor>
template<class OtherProtocol>
basic_stream<Protocol, Executor>::
impl_type::
impl_type(net::basic_stream_socket<OtherProtocol>&& socket_,
std::true_type)
: boost::empty_value<Executor>(
boost::empty_init_t{}, socket_.get_executor())
, read(ex().context())
, write(ex().context())
, socket(std::move(socket_))
{
reset();
}
template<class Protocol, class Executor>
template<class OtherProtocol>
basic_stream<Protocol, Executor>::
impl_type::
impl_type(net::basic_stream_socket<OtherProtocol>&& socket_,
std::false_type)
: boost::empty_value<Executor>(boost::empty_init_t{},
socket_.get_executor().context())
, read(ex().context())
, write(ex().context())
, socket(std::move(socket_))
{
reset();
}
template<class Protocol, class Executor>
void
basic_stream<Protocol, Executor>::
impl_type::
reset()
{
// If assert goes off, it means that there are
// already read or write (or connect) operations
// outstanding, so there is nothing to apply
// the expiration time to!
//
BOOST_ASSERT(! read.pending || ! write.pending);
if(! read.pending)
BOOST_VERIFY(
read.timer.expires_at(never()) == 0);
if(! write.pending)
BOOST_VERIFY(
write.timer.expires_at(never()) == 0);
}
template<class Protocol, class Executor>
void
basic_stream<Protocol, Executor>::
impl_type::
close()
{
socket.close();
// have to let the read/write ops cancel the timer,
// otherwise we will get error::timeout on close when
// we actually want net::error::operation_aborted.
//
//read.timer.cancel();
//write.timer.cancel();
}
//------------------------------------------------------------------------------
template<class Protocol, class Executor>
struct basic_stream<
Protocol, Executor>::timeout_handler
{
op_state& state;
@@ -50,14 +140,8 @@ struct basic_timeout_stream<
return;
BOOST_ASSERT(tick == state.tick);
// late completion
if(state.timeout)
{
state.timeout = false;
return;
}
// timeout
BOOST_ASSERT(! state.timeout);
sp->close();
state.timeout = true;
}
@@ -75,7 +159,7 @@ struct basic_timeout_stream<
template<class Protocol, class Executor>
template<bool isRead, class Buffers, class Handler>
class basic_timeout_stream<Protocol, Executor>::async_op
class basic_stream<Protocol, Executor>::async_op
: public async_op_base<Handler, Executor>
, public boost::asio::coroutine
{
@@ -119,7 +203,7 @@ class basic_timeout_stream<Protocol, Executor>::async_op
public:
template<class Handler_>
async_op(
basic_timeout_stream& s,
basic_stream& s,
Buffers const& b,
Handler_&& h)
: async_op_base<Handler, Executor>(
@@ -138,44 +222,55 @@ public:
{
BOOST_ASIO_CORO_REENTER(*this)
{
// VFALCO TODO handle buffer size == 0
if(detail::buffers_empty(b_))
{
BOOST_ASIO_CORO_YIELD
async_perform(
std::integral_constant<bool, isRead>{});
if(state().timer.expiry() <= clock_type::now())
ec = beast::error::timeout;
goto upcall;
}
state().timer.async_wait(
net::bind_executor(
this->get_executor(),
timeout_handler{
state(),
impl_,
state().tick
}));
if(state().timer.expiry() != never())
state().timer.async_wait(
net::bind_executor(
this->get_executor(),
timeout_handler{
state(),
impl_,
state().tick
}));
BOOST_ASIO_CORO_YIELD
async_perform(
std::integral_constant<bool, isRead>{});
++state().tick;
// try cancelling timer
auto const n =
state().timer.cancel();
if(n == 0)
if(state().timer.expiry() != never())
{
if(state().timeout)
++state().tick;
// try cancelling timer
auto const n =
state().timer.cancel();
if(n == 0)
{
// timeout handler invoked
ec = beast::error::timeout;
state().timeout = false;
// timeout handler invoked?
if(state().timeout)
{
// yes, socket already closed
ec = beast::error::timeout;
state().timeout = false;
}
}
else
{
// timeout handler queued, stale
BOOST_ASSERT(n == 1);
BOOST_ASSERT(! state().timeout);
}
}
else
{
BOOST_ASSERT(n == 1);
BOOST_ASSERT(! state().timeout);
}
upcall:
pg_.reset();
this->invoke(ec, bytes_transferred);
}
@@ -188,17 +283,17 @@ namespace detail {
template<
class Protocol, class Executor, class Handler>
class timeout_stream_connect_op
class basic_stream_connect_op
: public async_op_base<Handler, Executor>
{
using stream_type =
beast::basic_timeout_stream<Protocol, Executor>;
beast::basic_stream<Protocol, Executor>;
using timeout_handler =
typename stream_type::timeout_handler;
std::shared_ptr<typename basic_timeout_stream<
Protocol, Executor>::impl_type> impl_;
std::shared_ptr<typename
stream_type::impl_type> impl_;
typename stream_type::pending_guard pg0_;
typename stream_type::pending_guard pg1_;
@@ -209,64 +304,8 @@ class timeout_stream_connect_op
}
public:
template<
class Endpoints, class Condition,
class Handler_>
timeout_stream_connect_op(
stream_type& s,
Endpoints const& eps,
Condition cond,
Handler_&& h)
: async_op_base<Handler, Executor>(
std::forward<Handler_>(h), s.get_executor())
, impl_(s.impl_)
, pg0_(impl_->read.pending)
, pg1_(impl_->write.pending)
{
// must come first
impl_->write.timer.async_wait(
net::bind_executor(
this->get_executor(),
timeout_handler{
state(),
impl_,
state().tick}));
net::async_connect(impl_->socket,
eps, cond, std::move(*this));
// *this is now moved-from
}
template<
class Iterator, class Condition,
class Handler_>
timeout_stream_connect_op(
stream_type& s,
Iterator begin, Iterator end,
Condition cond,
Handler_&& h)
: async_op_base<Handler, Executor>(
std::forward<Handler_>(h), s.get_executor())
, impl_(s.impl_)
, pg0_(impl_->read.pending)
, pg1_(impl_->write.pending)
{
// must come first
impl_->write.timer.async_wait(
net::bind_executor(
this->get_executor(),
timeout_handler{
state(),
impl_,
state().tick}));
net::async_connect(impl_->socket,
begin, end, cond, std::move(*this));
// *this is now moved-from
}
template<class Handler_>
timeout_stream_connect_op(
basic_stream_connect_op(
stream_type& s,
typename stream_type::endpoint_type ep,
Handler_&& h)
@@ -276,48 +315,104 @@ public:
, pg0_(impl_->read.pending)
, pg1_(impl_->write.pending)
{
// must come first
impl_->write.timer.async_wait(
net::bind_executor(
this->get_executor(),
timeout_handler{
state(),
impl_,
state().tick}));
if(state().timer.expiry() != stream_base::never())
impl_->write.timer.async_wait(
net::bind_executor(
this->get_executor(),
timeout_handler{
state(),
impl_,
state().tick}));
impl_->socket.async_connect(
ep, std::move(*this));
// *this is now moved-from
}
template<
class Endpoints, class Condition,
class Handler_>
basic_stream_connect_op(
stream_type& s,
Endpoints const& eps,
Condition const& cond,
Handler_&& h)
: async_op_base<Handler, Executor>(
std::forward<Handler_>(h), s.get_executor())
, impl_(s.impl_)
, pg0_(impl_->read.pending)
, pg1_(impl_->write.pending)
{
if(state().timer.expiry() != stream_base::never())
impl_->write.timer.async_wait(
net::bind_executor(
this->get_executor(),
timeout_handler{
state(),
impl_,
state().tick}));
net::async_connect(impl_->socket,
eps, cond, std::move(*this));
// *this is now moved-from
}
template<
class Iterator, class Condition,
class Handler_>
basic_stream_connect_op(
stream_type& s,
Iterator begin, Iterator end,
Condition const& cond,
Handler_&& h)
: async_op_base<Handler, Executor>(
std::forward<Handler_>(h), s.get_executor())
, impl_(s.impl_)
, pg0_(impl_->read.pending)
, pg1_(impl_->write.pending)
{
if(state().timer.expiry() != stream_base::never())
impl_->write.timer.async_wait(
net::bind_executor(
this->get_executor(),
timeout_handler{
state(),
impl_,
state().tick}));
net::async_connect(impl_->socket,
begin, end, cond, std::move(*this));
// *this is now moved-from
}
template<class... Args>
void
operator()(error_code ec, Args&&... args)
{
++state().tick;
// try cancelling timer
auto const n =
impl_->write.timer.cancel();
if(n == 0)
if(state().timer.expiry() != stream_base::never())
{
if(state().timeout)
++state().tick;
// try cancelling timer
auto const n =
impl_->write.timer.cancel();
if(n == 0)
{
// timeout handler invoked
ec = beast::error::timeout;
state().timeout = false;
// timeout handler invoked?
if(state().timeout)
{
// yes, socket already closed
ec = beast::error::timeout;
state().timeout = false;
}
}
else
{
// timeout handler queued, stale
BOOST_ASSERT(n == 1);
BOOST_ASSERT(! state().timeout);
}
}
else
{
BOOST_ASSERT(n == 1);
BOOST_ASSERT(! state().timeout);
}
pg0_.reset();
pg1_.reset();
this->invoke(ec, std::forward<Args>(args)...);
@@ -329,82 +424,8 @@ public:
//------------------------------------------------------------------------------
template<class Protocol, class Executor>
template<class... Args>
basic_timeout_stream<Protocol, Executor>::
impl_type::
impl_type(
Executor const& ex_,
Args&&... args)
: ex(ex_)
, read(ex_.context())
, write(ex_.context())
, socket(std::forward<Args>(args)...)
{
reset();
}
template<class Protocol, class Executor>
auto
basic_timeout_stream<Protocol, Executor>::
impl_type::
operator=(impl_type&& other) -> impl_type&
{
// VFALCO This hack is because legacy io_context::strand
// doesn't support operator=. Don't worry, constructing
// an executor cannot throw.
ex.~Executor();
::new(&ex) Executor(other.ex);
socket = std::move(other.socket);
read = std::move(other.read);
write = std::move(other.write);
return *this;
}
template<class Protocol, class Executor>
void
basic_timeout_stream<Protocol, Executor>::
impl_type::
reset()
{
// If assert goes off, it means that there are
// already read or write (or connect) operations
// outstanding, so there is nothing to apply
// the expiration time to!
//
BOOST_ASSERT(! read.pending || ! write.pending);
if(! read.pending)
BOOST_VERIFY(
read.timer.expires_at(never()) == 0);
if(! write.pending)
BOOST_VERIFY(
write.timer.expires_at(never()) == 0);
}
template<class Protocol, class Executor>
void
basic_timeout_stream<Protocol, Executor>::
impl_type::
close()
{
socket.close();
// have to let the read/write ops cancel the timer,
// otherwise we will get error::timeout on close when
// we actually want net::error::operation_aborted.
//
//read.timer.cancel();
//write.timer.cancel();
}
//------------------------------------------------------------------------------
template<class Protocol, class Executor>
basic_timeout_stream<Protocol, Executor>::
~basic_timeout_stream()
basic_stream<Protocol, Executor>::
~basic_stream()
{
// the shared object can outlive *this,
// cancel any operations so the shared
@@ -413,12 +434,12 @@ basic_timeout_stream<Protocol, Executor>::
}
template<class Protocol, class Executor>
template<class ExecutionContext, class>
basic_timeout_stream<Protocol, Executor>::
basic_timeout_stream(ExecutionContext& ctx)
template<class ExecutionContext, class... Args, class>
basic_stream<Protocol, Executor>::
basic_stream(ExecutionContext& ctx, Args&&... args)
: impl_(std::make_shared<impl_type>(
ctx.get_executor(),
ctx))
ctx, std::forward<Args>(args)...))
{
// Restriction is necessary until Asio fully supports P1322R0
static_assert(
@@ -427,67 +448,52 @@ basic_timeout_stream(ExecutionContext& ctx)
}
template<class Protocol, class Executor>
basic_timeout_stream<Protocol, Executor>::
basic_timeout_stream(executor_type const& ex)
template<class... Args>
basic_stream<Protocol, Executor>::
basic_stream(
executor_type const& ex, Args&&... args)
: impl_(std::make_shared<impl_type>(
ex, ex.context()))
ex,
ex.context(), std::forward<Args>(args)...))
{
}
template<class Protocol, class Executor>
basic_timeout_stream<Protocol, Executor>::
basic_timeout_stream(
net::basic_stream_socket<Protocol>&& socket)
template<class OtherProtocol, class>
basic_stream<Protocol, Executor>::
basic_stream(net::basic_stream_socket<OtherProtocol>&& socket)
: impl_(std::make_shared<impl_type>(
socket.get_executor(), std::move(socket)))
std::move(socket),
std::is_constructible<Executor,
decltype(std::declval<net::basic_stream_socket<
Protocol>&>().get_executor())>{}))
{
}
template<class Protocol, class Executor>
basic_timeout_stream<Protocol, Executor>::
basic_timeout_stream(
executor_type const& ex,
net::basic_stream_socket<Protocol>&& socket)
: impl_(std::make_shared<impl_type>(
ex, std::move(socket)))
{
// Restriction is necessary until Asio fully supports P1322R0
if(ex.context().get_executor() != socket.get_executor())
throw std::invalid_argument(
"basic_timeout_stream currently requires ctx.get_executor() == socket.get_executor()");
}
template<class Protocol, class Executor>
basic_timeout_stream<Protocol, Executor>::
basic_timeout_stream(basic_timeout_stream&& other)
basic_stream<Protocol, Executor>::
basic_stream(basic_stream&& other)
: impl_(std::make_shared<impl_type>(
std::move(*other.impl_)))
{
// Can't move while operations are pending!
BOOST_ASSERT(! impl_->read.pending);
BOOST_ASSERT(! impl_->write.pending);
}
template<class Protocol, class Executor>
auto
basic_timeout_stream<Protocol, Executor>::
operator=(basic_timeout_stream&& other) ->
basic_timeout_stream&
{
// Can't move while operations are pending!
BOOST_ASSERT(! impl_->read.pending);
BOOST_ASSERT(! impl_->write.pending);
BOOST_ASSERT(! other.impl_->read.pending);
BOOST_ASSERT(! other.impl_->write.pending);
*impl_ = std::move(*other.impl_);
return *this;
// VFALCO I'm not sure this implementation is correct...
}
//------------------------------------------------------------------------------
template<class Protocol, class Executor>
auto
basic_stream<Protocol, Executor>::
release_socket() ->
socket_type
{
this->cancel();
return std::move(impl_->socket);
}
template<class Protocol, class Executor>
void
basic_timeout_stream<Protocol, Executor>::
basic_stream<Protocol, Executor>::
expires_after(std::chrono::nanoseconds expiry_time)
{
// If assert goes off, it means that there are
@@ -512,7 +518,7 @@ expires_after(std::chrono::nanoseconds expiry_time)
template<class Protocol, class Executor>
void
basic_timeout_stream<Protocol, Executor>::
basic_stream<Protocol, Executor>::
expires_at(
net::steady_timer::time_point expiry_time)
{
@@ -538,7 +544,7 @@ expires_at(
template<class Protocol, class Executor>
void
basic_timeout_stream<Protocol, Executor>::
basic_stream<Protocol, Executor>::
expires_never()
{
impl_->reset();
@@ -546,7 +552,7 @@ expires_never()
template<class Protocol, class Executor>
void
basic_timeout_stream<Protocol, Executor>::
basic_stream<Protocol, Executor>::
cancel()
{
error_code ec;
@@ -555,7 +561,7 @@ cancel()
template<class Protocol, class Executor>
void
basic_timeout_stream<Protocol, Executor>::
basic_stream<Protocol, Executor>::
close()
{
impl_->close();
@@ -565,14 +571,14 @@ template<class Protocol, class Executor>
template<class ConnectHandler>
BOOST_ASIO_INITFN_RESULT_TYPE(ConnectHandler,
void(error_code))
basic_timeout_stream<Protocol, Executor>::
basic_stream<Protocol, Executor>::
async_connect(
endpoint_type ep,
endpoint_type const& ep,
ConnectHandler&& handler)
{
BOOST_BEAST_HANDLER_INIT(
ConnectHandler, void(error_code));
detail::timeout_stream_connect_op<
detail::basic_stream_connect_op<
Protocol, Executor, BOOST_ASIO_HANDLER_TYPE(
ConnectHandler, void(error_code))>(*this,
ep, std::forward<ConnectHandler>(handler));
@@ -583,7 +589,7 @@ template<class Protocol, class Executor>
template<class MutableBufferSequence, class ReadHandler>
BOOST_ASIO_INITFN_RESULT_TYPE(ReadHandler,
void(error_code, std::size_t))
basic_timeout_stream<Protocol, Executor>::
basic_stream<Protocol, Executor>::
async_read_some(
MutableBufferSequence const& buffers,
ReadHandler&& handler)
@@ -603,7 +609,7 @@ template<class Protocol, class Executor>
template<class ConstBufferSequence, class WriteHandler>
BOOST_ASIO_INITFN_RESULT_TYPE(WriteHandler,
void(error_code, std::size_t))
basic_timeout_stream<Protocol, Executor>::
basic_stream<Protocol, Executor>::
async_write_some(
ConstBufferSequence const& buffers,
WriteHandler&& handler)
@@ -629,13 +635,13 @@ template<
BOOST_ASIO_INITFN_RESULT_TYPE(RangeConnectHandler,
void(error_code, typename Protocol::endpoint))
async_connect(
basic_timeout_stream<Protocol, Executor>& stream,
basic_stream<Protocol, Executor>& stream,
EndpointSequence const& endpoints,
RangeConnectHandler&& handler)
{
BOOST_BEAST_HANDLER_INIT(RangeConnectHandler,
void(error_code, typename Protocol::endpoint));
detail::timeout_stream_connect_op<Protocol, Executor,
detail::basic_stream_connect_op<Protocol, Executor,
BOOST_ASIO_HANDLER_TYPE(RangeConnectHandler,
void(error_code, typename Protocol::endpoint))>(
stream, endpoints, detail::any_endpoint{},
@@ -652,14 +658,14 @@ template<
BOOST_ASIO_INITFN_RESULT_TYPE(RangeConnectHandler,
void (error_code, typename Protocol::endpoint))
async_connect(
basic_timeout_stream<Protocol, Executor>& stream,
basic_stream<Protocol, Executor>& stream,
EndpointSequence const& endpoints,
ConnectCondition connect_condition,
RangeConnectHandler&& handler)
{
BOOST_BEAST_HANDLER_INIT(RangeConnectHandler,
void(error_code, typename Protocol::endpoint));
detail::timeout_stream_connect_op<Protocol, Executor,
detail::basic_stream_connect_op<Protocol, Executor,
BOOST_ASIO_HANDLER_TYPE(RangeConnectHandler,
void(error_code, typename Protocol::endpoint))>(
stream, endpoints, connect_condition,
@@ -674,13 +680,13 @@ template<
BOOST_ASIO_INITFN_RESULT_TYPE(IteratorConnectHandler,
void (error_code, Iterator))
async_connect(
basic_timeout_stream<Protocol, Executor>& stream,
basic_stream<Protocol, Executor>& stream,
Iterator begin, Iterator end,
IteratorConnectHandler&& handler)
{
BOOST_BEAST_HANDLER_INIT(IteratorConnectHandler,
void(error_code, Iterator));
detail::timeout_stream_connect_op<Protocol, Executor,
detail::basic_stream_connect_op<Protocol, Executor,
BOOST_ASIO_HANDLER_TYPE(IteratorConnectHandler,
void(error_code, Iterator))>(
stream, begin, end, detail::any_endpoint{},
@@ -696,14 +702,14 @@ template<
BOOST_ASIO_INITFN_RESULT_TYPE(IteratorConnectHandler,
void (error_code, Iterator))
async_connect(
basic_timeout_stream<Protocol, Executor>& stream,
basic_stream<Protocol, Executor>& stream,
Iterator begin, Iterator end,
ConnectCondition connect_condition,
IteratorConnectHandler&& handler)
{
BOOST_BEAST_HANDLER_INIT(IteratorConnectHandler,
void(error_code, Iterator));
detail::timeout_stream_connect_op<Protocol, Executor,
detail::basic_stream_connect_op<Protocol, Executor,
BOOST_ASIO_HANDLER_TYPE(IteratorConnectHandler,
void(error_code, Iterator))>(
stream, begin, end, connect_condition,
@@ -711,6 +717,46 @@ async_connect(
return init.result.get();
}
//------------------------------------------------------------------------------
#if ! BOOST_BEAST_DOXYGEN
template<class Protocol, class Executor>
void
beast_close_socket(
basic_stream<Protocol, Executor>& stream)
{
error_code ec;
stream.socket().close(ec);
}
template<class Protocol, class Executor>
void
teardown(
websocket::role_type role,
basic_stream<Protocol, Executor>& stream,
error_code& ec)
{
using beast::websocket::teardown;
teardown(role, stream.socket(), ec);
}
template<
class Protocol, class Executor,
class TeardownHandler>
void
async_teardown(
websocket::role_type role,
basic_stream<Protocol, Executor>& stream,
TeardownHandler&& handler)
{
using beast::websocket::async_teardown;
async_teardown(role, stream.socket(),
std::forward<TeardownHandler>(handler));
}
#endif
} // beast
} // boost