ARCHIVE/rippled
1
0
Fork 0
mirror of https://github.com/XRPLF/rippled.git synced 2025-12-06 17:27:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

Support Boost 1.70:

Browse Source 
This patch removes calls to several deprecated asio functions.

* `io_service::post` becomes `post` (free function)
* `io_service::work` becomes `executor_work_guard`
* `io_service::wrap` becomes `bind_executor`
* `get_io_context`   becomes `get_executor` or `get_executor().context()`

This patch was tested with boost 1.69 and 1.70. The functions
`ripple::get_lowest_layer` and `beast::create_waitable_timer` are required to
handle a breaking difference between these versions. When rippled no longer
needs to support pre 1.70 boost versions, both of these functions may be
removed, and the waitable timer injections may also be removed.
This commit is contained in:
seelabs
2019-04-11 20:39:53 -04:00
committed by Manoj doshi
parent b39b0fef39
commit 6f9e8dc720
21 changed files with 727 additions and 451 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
src/ripple/beast/asio/waitable_timer.h Normal file
View File

@@ -0,0 +1,45 @@
//------------------------------------------------------------------------------
/*
This file is part of rippled: https://github.com/ripple/rippled
Copyright (c) 2019 Ripple Labs Inc.
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL , DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#ifndef RIPPLE_BEAST_ASIO_WAITABLETIMER_H_INCLUDED
#define RIPPLE_BEAST_ASIO_WAITABLETIMER_H_INCLUDED
#include <boost/asio/ip/tcp.hpp>
namespace beast {
// Pre boost 1.70, a waitable timer may only be created from an io_context.
// However, post 1.70, the `get_io_service` function is deleted. Post
// boost 1.70, a waitable time may be created from an executor. This functions
// allows a waitable timer to be created from a socket in both pre and post
// boost 1.70 versions
template <class T>
T
create_waitable_timer(boost::asio::ip::tcp::socket& socket)
{
#if BOOST_VERSION >= 107000
return T{socket.get_executor()};
#else
return T{socket.get_io_service()};
#endif
}
} // namespace ripple
#endif

15
src/ripple/net/AutoSocket.h
View File

@@ -64,11 +64,6 @@ public:
{
}
boost::asio::io_service& get_io_service () noexcept
{
return mSocket->get_io_service ();
}
bool isSecure ()
{
return mSecure;
@@ -167,8 +162,9 @@ public:
{
// must be plain
mSecure = false;
mSocket->get_io_service ().post (
boost::beast::bind_handler (cbFunc, error_code()));
post(
mSocket->get_executor(),
boost::beast::bind_handler(cbFunc, error_code()));
}
else
{
@@ -200,8 +196,9 @@ public:
{
ec = e.code();
}
mSocket->get_io_service ().post (
boost::beast::bind_handler (handler, ec));
post(
mSocket->get_executor(),
boost::beast::bind_handler(handler, ec));
}
}

4
src/ripple/net/impl/SSLHTTPDownloader.cpp
View File

@@ -132,14 +132,14 @@ SSLHTTPDownloader::do_session(
using namespace boost::beast;
boost::system::error_code ec;
ip::tcp::resolver resolver {strand_.get_io_context()};
ip::tcp::resolver resolver {strand_.context()};
auto const results = resolver.async_resolve(host, port, yield[ec]);
if (ec)
return fail(dstPath, complete, ec, "async_resolve");
try
{
stream_.emplace(strand_.get_io_service(), ctx_);
stream_.emplace(strand_.context(), ctx_);
}
catch (std::exception const& e)
{

97
src/ripple/overlay/impl/PeerImp.cpp
View File

@@ -66,8 +66,8 @@ PeerImp::PeerImp (Application& app, id_t id, endpoint_type remote_endpoint,
, ssl_bundle_(std::move(ssl_bundle))
, socket_ (ssl_bundle_->socket)
, stream_ (ssl_bundle_->stream)
, strand_ (socket_.get_io_service())
, timer_ (socket_.get_io_service())
, strand_ (socket_.get_executor())
, timer_ (beast::create_waitable_timer<waitable_timer>(socket_))
, remote_address_ (
beast::IPAddressConversion::from_asio(remote_endpoint))
, overlay_ (overlay)
@@ -102,7 +102,7 @@ void
PeerImp::run()
{
if(! strand_.running_in_this_thread())
return strand_.post(std::bind (
return post(strand_, std::bind (
&PeerImp::run, shared_from_this()));
if (m_inbound)
{
@@ -145,8 +145,7 @@ void
PeerImp::stop()
{
if(! strand_.running_in_this_thread())
return strand_.post(std::bind (
&PeerImp::stop, shared_from_this()));
return post(strand_, std::bind(&PeerImp::stop, shared_from_this()));
if (socket_.is_open())
{
// The rationale for using different severity levels is that
@@ -172,8 +171,7 @@ void
PeerImp::send (Message::pointer const& m)
{
if (! strand_.running_in_this_thread())
return strand_.post(std::bind (
&PeerImp::send, shared_from_this(), m));
return post(strand_, std::bind(&PeerImp::send, shared_from_this(), m));
if(gracefulClose_)
return;
if(detaching_)
@@ -204,11 +202,16 @@ PeerImp::send (Message::pointer const& m)
if(sendq_size != 0)
return;
boost::asio::async_write (stream_, boost::asio::buffer(
send_queue_.front()->getBuffer()), strand_.wrap(std::bind(
&PeerImp::onWriteMessage, shared_from_this(),
boost::asio::async_write(
stream_,
boost::asio::buffer(send_queue_.front()->getBuffer()),
bind_executor(
strand_,
std::bind(
&PeerImp::onWriteMessage,
shared_from_this(),
std::placeholders::_1,
std::placeholders::_2)));
std::placeholders::_2)));
}
void
@@ -438,9 +441,12 @@ void
PeerImp::fail(std::string const& reason)
{
if(! strand_.running_in_this_thread())
return strand_.post(std::bind (
(void(Peer::*)(std::string const&))&PeerImp::fail,
shared_from_this(), reason));
return post(
strand_,
std::bind(
(void (Peer::*)(std::string const&)) & PeerImp::fail,
shared_from_this(),
reason));
if (socket_.is_open())
{
JLOG (journal_.warn()) <<
@@ -498,8 +504,10 @@ PeerImp::gracefulClose()
if (send_queue_.size() > 0)
return;
setTimer();
stream_.async_shutdown(strand_.wrap(std::bind(&PeerImp::onShutdown,
shared_from_this(), std::placeholders::_1)));
stream_.async_shutdown(bind_executor(
strand_,
std::bind(
&PeerImp::onShutdown, shared_from_this(), std::placeholders::_1)));
}
void
@@ -514,8 +522,10 @@ PeerImp::setTimer()
JLOG(journal_.error()) << "setTimer: " << ec.message();
return;
}
timer_.async_wait(strand_.wrap(std::bind(&PeerImp::onTimer,
shared_from_this(), std::placeholders::_1)));
timer_.async_wait(bind_executor(
strand_,
std::bind(
&PeerImp::onTimer, shared_from_this(), std::placeholders::_1)));
}
// convenience for ignoring the error code
@@ -709,9 +719,14 @@ PeerImp::onWriteResponse (error_code ec, std::size_t bytes_transferred)
if (write_buffer_.size() == 0)
return doProtocolStart();
stream_.async_write_some (write_buffer_.data(),
strand_.wrap (std::bind (&PeerImp::onWriteResponse,
shared_from_this(), std::placeholders::_1,
stream_.async_write_some(
write_buffer_.data(),
bind_executor(
strand_,
std::bind(
&PeerImp::onWriteResponse,
shared_from_this(),
std::placeholders::_1,
std::placeholders::_2)));
}
@@ -785,9 +800,14 @@ PeerImp::onReadMessage (error_code ec, std::size_t bytes_transferred)
read_buffer_.consume (bytes_consumed);
}
// Timeout on writes only
stream_.async_read_some (read_buffer_.prepare (Tuning::readBufferBytes),
strand_.wrap (std::bind (&PeerImp::onReadMessage,
shared_from_this(), std::placeholders::_1,
stream_.async_read_some(
read_buffer_.prepare(Tuning::readBufferBytes),
bind_executor(
strand_,
std::bind(
&PeerImp::onReadMessage,
shared_from_this(),
std::placeholders::_1,
std::placeholders::_2)));
}
@@ -814,17 +834,25 @@ PeerImp::onWriteMessage (error_code ec, std::size_t bytes_transferred)
if (! send_queue_.empty())
{
// Timeout on writes only
return boost::asio::async_write (stream_, boost::asio::buffer(
send_queue_.front()->getBuffer()), strand_.wrap(std::bind(
&PeerImp::onWriteMessage, shared_from_this(),
return boost::asio::async_write(
stream_,
boost::asio::buffer(send_queue_.front()->getBuffer()),
bind_executor(
strand_,
std::bind(
&PeerImp::onWriteMessage,
shared_from_this(),
std::placeholders::_1,
std::placeholders::_2)));
std::placeholders::_2)));
}
if (gracefulClose_)
{
return stream_.async_shutdown(strand_.wrap(std::bind(
&PeerImp::onShutdown, shared_from_this(),
return stream_.async_shutdown(bind_executor(
strand_,
std::bind(
&PeerImp::onShutdown,
shared_from_this(),
std::placeholders::_1)));
}
}
@@ -1816,9 +1844,12 @@ void PeerImp::check ()
if (reject)
{
overlay_.peerFinder().on_failure (slot_);
strand_.post (std::bind (
(void (PeerImp::*)(std::string const&)) &PeerImp::fail,
shared_from_this(), "Not useful"));
post(
strand_,
std::bind(
(void (PeerImp::*)(std::string const&)) & PeerImp::fail,
shared_from_this(),
"Not useful"));
}
}

12
src/ripple/overlay/impl/PeerImp.h
View File

@@ -23,6 +23,7 @@
#include <ripple/app/consensus/RCLCxPeerPos.h>
#include <ripple/basics/Log.h>
#include <ripple/basics/RangeSet.h>
#include <ripple/beast/asio/waitable_timer.h>
#include <ripple/beast/utility/WrappedSink.h>
#include <ripple/overlay/impl/ProtocolMessage.h>
#include <ripple/overlay/impl/OverlayImpl.h>
@@ -94,6 +95,7 @@ private:
using stream_type = boost::asio::ssl::stream <socket_type&>;
using address_type = boost::asio::ip::address;
using endpoint_type = boost::asio::ip::tcp::endpoint;
using waitable_timer = boost::asio::basic_waitable_timer<std::chrono::steady_clock>;
// The length of the smallest valid finished message
static const size_t sslMinimumFinishedLength = 12;
@@ -107,9 +109,8 @@ private:
std::unique_ptr<beast::asio::ssl_bundle> ssl_bundle_;
socket_type& socket_;
stream_type& stream_;
boost::asio::io_service::strand strand_;
boost::asio::basic_waitable_timer<
std::chrono::steady_clock> timer_;
boost::asio::strand<boost::asio::executor> strand_;
waitable_timer timer_;
//Type type_ = Type::legacy;
@@ -479,6 +480,7 @@ private:
beast::Journal journal);
};
//------------------------------------------------------------------------------
template <class Buffers>
@@ -498,8 +500,8 @@ PeerImp::PeerImp (Application& app, std::unique_ptr<beast::asio::ssl_bundle>&& s
, ssl_bundle_(std::move(ssl_bundle))
, socket_ (ssl_bundle_->socket)
, stream_ (ssl_bundle_->stream)
, strand_ (socket_.get_io_service())
, timer_ (socket_.get_io_service())
, strand_ (socket_.get_executor())
, timer_ (beast::create_waitable_timer<waitable_timer>(socket_))
, remote_address_ (slot->remote_endpoint())
, overlay_ (overlay)
, m_inbound (false)

110
src/ripple/server/impl/BaseHTTPPeer.h
View File

@@ -84,8 +84,8 @@ protected:
Port const& port_;
Handler& handler_;
boost::asio::io_service::work work_;
boost::asio::io_service::strand strand_;
boost::asio::executor_work_guard<boost::asio::executor> work_;
boost::asio::strand<boost::asio::executor> strand_;
waitable_timer timer_;
endpoint_type remote_address_;
beast::Journal journal_;
@@ -109,10 +109,15 @@ protected:
//--------------------------------------------------------------------------
public:
template<class ConstBufferSequence>
BaseHTTPPeer(Port const& port, Handler& handler,
boost::asio::io_service& io_service, beast::Journal journal,
endpoint_type remote_address, ConstBufferSequence const& buffers);
template <class ConstBufferSequence>
BaseHTTPPeer(
Port const& port,
Handler& handler,
boost::asio::executor const& executor,
waitable_timer timer,
beast::Journal journal,
endpoint_type remote_address,
ConstBufferSequence const& buffers);
virtual
~BaseHTTPPeer();
@@ -208,18 +213,21 @@ protected:
//------------------------------------------------------------------------------
template<class Handler, class Impl>
template<class ConstBufferSequence>
BaseHTTPPeer<Handler, Impl>::
BaseHTTPPeer(Port const& port, Handler& handler,
boost::asio::io_service& io_service, beast::Journal journal,
endpoint_type remote_address,
ConstBufferSequence const& buffers)
template <class Handler, class Impl>
template <class ConstBufferSequence>
BaseHTTPPeer<Handler, Impl>::BaseHTTPPeer(
Port const& port,
Handler& handler,
boost::asio::executor const& executor,
waitable_timer timer,
beast::Journal journal,
endpoint_type remote_address,
ConstBufferSequence const& buffers)
: port_(port)
, handler_(handler)
, work_(io_service)
, strand_(io_service)
, timer_(io_service)
, work_(executor)
, strand_(executor)
, timer_(std::move(timer))
, remote_address_(remote_address)
, journal_(journal)
{
@@ -248,8 +256,10 @@ BaseHTTPPeer<Handler, Impl>::
close()
{
if(! strand_.running_in_this_thread())
return strand_.post(std::bind(
(void(BaseHTTPPeer::*)(void))&BaseHTTPPeer::close,
return post(
strand_,
std::bind(
(void (BaseHTTPPeer::*)(void)) & BaseHTTPPeer::close,
impl().shared_from_this()));
error_code ec;
impl().stream_.lowest_layer().close(ec);
@@ -285,8 +295,11 @@ start_timer()
ec);
if(ec)
return fail(ec, "start_timer");
timer_.async_wait(strand_.wrap(std::bind(
&BaseHTTPPeer<Handler, Impl>::on_timer, impl().shared_from_this(),
timer_.async_wait(bind_executor(
strand_,
std::bind(
&BaseHTTPPeer<Handler, Impl>::on_timer,
impl().shared_from_this(),
std::placeholders::_1)));
}
@@ -359,9 +372,15 @@ on_write(error_code const& ec,
for(auto const& b : wq2_)
v.emplace_back(b.data.get(), b.bytes);
start_timer();
return boost::asio::async_write(impl().stream_, v,
strand_.wrap(std::bind(&BaseHTTPPeer::on_write,
impl().shared_from_this(), std::placeholders::_1,
return boost::asio::async_write(
impl().stream_,
v,
bind_executor(
strand_,
std::bind(
&BaseHTTPPeer::on_write,
impl().shared_from_this(),
std::placeholders::_1,
std::placeholders::_2)));
}
if(! complete_)
@@ -432,10 +451,13 @@ write(
}())
{
if(! strand_.running_in_this_thread())
return strand_.post(std::bind(
&BaseHTTPPeer::on_write,
return post(
strand_,
std::bind(
&BaseHTTPPeer::on_write,
impl().shared_from_this(),
error_code{}, 0));
error_code{},
0));
else
return on_write(error_code{}, 0);
}
@@ -447,9 +469,14 @@ BaseHTTPPeer<Handler, Impl>::
write(std::shared_ptr <Writer> const& writer,
bool keep_alive)
{
boost::asio::spawn(strand_, std::bind(
&BaseHTTPPeer<Handler, Impl>::do_writer, impl().shared_from_this(),
writer, keep_alive, std::placeholders::_1));
boost::asio::spawn(bind_executor(
strand_,
std::bind(
&BaseHTTPPeer<Handler, Impl>::do_writer,
impl().shared_from_this(),
writer,
keep_alive,
std::placeholders::_1)));
}
// DEPRECATED
@@ -470,8 +497,11 @@ BaseHTTPPeer<Handler, Impl>::
complete()
{
if(! strand_.running_in_this_thread())
return strand_.post(std::bind(&BaseHTTPPeer<Handler, Impl>::complete,
impl().shared_from_this()));
return post(
strand_,
std::bind(
&BaseHTTPPeer<Handler, Impl>::complete,
impl().shared_from_this()));
message_ = {};
complete_ = true;
@@ -483,8 +513,12 @@ complete()
}
// keep-alive
boost::asio::spawn(strand_, std::bind(&BaseHTTPPeer<Handler, Impl>::do_read,
impl().shared_from_this(), std::placeholders::_1));
boost::asio::spawn(bind_executor(
strand_,
std::bind(
&BaseHTTPPeer<Handler, Impl>::do_read,
impl().shared_from_this(),
std::placeholders::_1)));
}
// DEPRECATED
@@ -495,9 +529,13 @@ BaseHTTPPeer<Handler, Impl>::
close(bool graceful)
{
if(! strand_.running_in_this_thread())
return strand_.post(std::bind(
(void(BaseHTTPPeer::*)(bool))&BaseHTTPPeer<Handler, Impl>::close,
impl().shared_from_this(), graceful));
return post(
strand_,
std::bind(
(void (BaseHTTPPeer::*)(bool)) &
BaseHTTPPeer<Handler, Impl>::close,
impl().shared_from_this(),
graceful));
complete_ = true;
if(graceful)

47
src/ripple/server/impl/BasePeer.h
View File

@@ -22,6 +22,7 @@
#include <ripple/server/Port.h>
#include <ripple/server/impl/io_list.h>
#include <ripple/server/impl/LowestLayer.h>
#include <ripple/beast/utility/WrappedSink.h>
#include <boost/asio.hpp>
#include <atomic>
@@ -48,14 +49,15 @@ protected:
beast::WrappedSink sink_;
beast::Journal j_;
boost::asio::io_service::work work_;
boost::asio::io_service::strand strand_;
boost::asio::executor_work_guard<boost::asio::executor> work_;
boost::asio::strand<boost::asio::executor> strand_;
public:
BasePeer(Port const& port, Handler& handler,
BasePeer(
Port const& port,
Handler& handler,
boost::asio::executor const& executor,
endpoint_type remote_address,
boost::asio::io_service& io_service,
beast::Journal journal);
beast::Journal journal);
void
close() override;
@@ -70,24 +72,25 @@ private:
//------------------------------------------------------------------------------
template<class Handler, class Impl>
BasePeer<Handler, Impl>::
BasePeer(Port const& port, Handler& handler,
template <class Handler, class Impl>
BasePeer<Handler, Impl>::BasePeer(
Port const& port,
Handler& handler,
boost::asio::executor const& executor,
endpoint_type remote_address,
boost::asio::io_service& io_service,
beast::Journal journal)
beast::Journal journal)
: port_(port)
, handler_(handler)
, remote_address_(remote_address)
, sink_(journal.sink(),
[]
{
static std::atomic<unsigned> id{0};
return "##" + std::to_string(++id) + " ";
}())
, sink_(
journal.sink(),
[] {
static std::atomic<unsigned> id{0};
return "##" + std::to_string(++id) + " ";
}())
, j_(sink_)
, work_(io_service)
, strand_(io_service)
, work_(executor)
, strand_(executor)
{
}
@@ -97,10 +100,10 @@ BasePeer<Handler, Impl>::
close()
{
if (! strand_.running_in_this_thread())
return strand_.post(std::bind(
&BasePeer::close, impl().shared_from_this()));
return post(
strand_, std::bind(&BasePeer::close, impl().shared_from_this()));
error_code ec;
impl().ws_.lowest_layer().close(ec);
ripple::get_lowest_layer(impl().ws_).close(ec);
}
} // ripple

143
src/ripple/server/impl/BaseWSPeer.h
View File

@@ -22,6 +22,7 @@
#include <ripple/basics/safe_cast.h>
#include <ripple/server/impl/BasePeer.h>
#include <ripple/server/impl/LowestLayer.h>
#include <ripple/protocol/BuildInfo.h>
#include <ripple/beast/utility/rngfill.h>
#include <ripple/crypto/csprng.h>
@@ -66,9 +67,10 @@ public:
BaseWSPeer(
Port const& port,
Handler& handler,
boost::asio::executor const& executor,
waitable_timer timer,
endpoint_type remote_address,
boost::beast::http::request<Body, Headers>&& request,
boost::asio::io_service& io_service,
beast::Journal journal);
void
@@ -156,21 +158,25 @@ protected:
//------------------------------------------------------------------------------
template<class Handler, class Impl>
template<class Body, class Headers>
BaseWSPeer<Handler, Impl>::
BaseWSPeer(
template <class Handler, class Impl>
template <class Body, class Headers>
BaseWSPeer<Handler, Impl>::BaseWSPeer(
Port const& port,
Handler& handler,
boost::asio::executor const& executor,
waitable_timer timer,
endpoint_type remote_address,
boost::beast::http::request<Body, Headers>&& request,
boost::asio::io_service& io_service,
beast::Journal journal)
: BasePeer<Handler, Impl>(port, handler, remote_address,
io_service, journal)
: BasePeer<Handler, Impl>(
port,
handler,
executor,
remote_address,
journal)
, request_(std::move(request))
, timer_(io_service)
, payload_ ("12345678") // ensures size is 8 bytes
, timer_(std::move(timer))
, payload_("12345678") // ensures size is 8 bytes
{
}
@@ -180,8 +186,8 @@ BaseWSPeer<Handler, Impl>::
run()
{
if(! strand_.running_in_this_thread())
return strand_.post(std::bind(
&BaseWSPeer::run, impl().shared_from_this()));
return post(
strand_, std::bind(&BaseWSPeer::run, impl().shared_from_this()));
impl().ws_.set_option(port().pmd_options);
// Must manage the control callback memory outside of the `control_callback` function
control_callback_ = std::bind(
@@ -190,14 +196,18 @@ run()
impl().ws_.control_callback(control_callback_);
start_timer();
close_on_timer_ = true;
impl().ws_.async_accept_ex(request_,
[](auto & res)
{
impl().ws_.async_accept_ex(
request_,
[](auto& res) {
res.set(boost::beast::http::field::server,
BuildInfo::getFullVersionString());
},
strand_.wrap(std::bind(&BaseWSPeer::on_ws_handshake,
impl().shared_from_this(), std::placeholders::_1)));
bind_executor(
strand_,
std::bind(
&BaseWSPeer::on_ws_handshake,
impl().shared_from_this(),
std::placeholders::_1)));
}
template<class Handler, class Impl>
@@ -206,9 +216,10 @@ BaseWSPeer<Handler, Impl>::
send(std::shared_ptr<WSMsg> w)
{
if(! strand_.running_in_this_thread())
return strand_.post(std::bind(
&BaseWSPeer::send, impl().shared_from_this(),
std::move(w)));
return post(
strand_,
std::bind(
&BaseWSPeer::send, impl().shared_from_this(), std::move(w)));
if(do_close_)
return;
if(wq_.size() > port().ws_queue_limit)
@@ -233,13 +244,18 @@ BaseWSPeer<Handler, Impl>::
close()
{
if(! strand_.running_in_this_thread())
return strand_.post(std::bind(
&BaseWSPeer::close, impl().shared_from_this()));
return post(
strand_, std::bind(&BaseWSPeer::close, impl().shared_from_this()));
do_close_ = true;
if(wq_.empty())
impl().ws_.async_close({}, strand_.wrap(std::bind(
&BaseWSPeer::on_close, impl().shared_from_this(),
std::placeholders::_1)));
impl().ws_.async_close(
{},
bind_executor(
strand_,
std::bind(
&BaseWSPeer::on_close,
impl().shared_from_this(),
std::placeholders::_1)));
}
template<class Handler, class Impl>
@@ -248,8 +264,9 @@ BaseWSPeer<Handler, Impl>::
complete()
{
if(! strand_.running_in_this_thread())
return strand_.post(std::bind(
&BaseWSPeer::complete, impl().shared_from_this()));
return post(
strand_,
std::bind(&BaseWSPeer::complete, impl().shared_from_this()));
do_read();
}
@@ -270,8 +287,9 @@ BaseWSPeer<Handler, Impl>::
do_write()
{
if(! strand_.running_in_this_thread())
return strand_.post(std::bind(
&BaseWSPeer::do_write, impl().shared_from_this()));
return post(
strand_,
std::bind(&BaseWSPeer::do_write, impl().shared_from_this()));
on_write({});
}
@@ -293,18 +311,22 @@ on_write(error_code const& ec)
impl().ws_.async_write_some(
static_cast<bool>(result.first),
result.second,
strand_.wrap(std::bind(
&BaseWSPeer::on_write,
impl().shared_from_this(),
std::placeholders::_1)));
bind_executor(
strand_,
std::bind(
&BaseWSPeer::on_write,
impl().shared_from_this(),
std::placeholders::_1)));
else
impl().ws_.async_write_some(
static_cast<bool>(result.first),
result.second,
strand_.wrap(std::bind(
&BaseWSPeer::on_write_fin,
impl().shared_from_this(),
std::placeholders::_1)));
bind_executor(
strand_,
std::bind(
&BaseWSPeer::on_write_fin,
impl().shared_from_this(),
std::placeholders::_1)));
}
template<class Handler, class Impl>
@@ -316,9 +338,14 @@ on_write_fin(error_code const& ec)
return fail(ec, "write_fin");
wq_.pop_front();
if(do_close_)
impl().ws_.async_close(cr_, strand_.wrap(std::bind(
&BaseWSPeer::on_close, impl().shared_from_this(),
std::placeholders::_1)));
impl().ws_.async_close(
cr_,
bind_executor(
strand_,
std::bind(
&BaseWSPeer::on_close,
impl().shared_from_this(),
std::placeholders::_1)));
else if(! wq_.empty())
on_write({});
}
@@ -329,11 +356,17 @@ BaseWSPeer<Handler, Impl>::
do_read()
{
if(! strand_.running_in_this_thread())
return strand_.post(std::bind(
&BaseWSPeer::do_read, impl().shared_from_this()));
impl().ws_.async_read(rb_, strand_.wrap(
std::bind(&BaseWSPeer::on_read,
impl().shared_from_this(), std::placeholders::_1)));
return post(
strand_,
std::bind(&BaseWSPeer::do_read, impl().shared_from_this()));
impl().ws_.async_read(
rb_,
bind_executor(
strand_,
std::bind(
&BaseWSPeer::on_read,
impl().shared_from_this(),
std::placeholders::_1)));
}
template<class Handler, class Impl>
@@ -376,8 +409,11 @@ start_timer()
ec);
if(ec)
return fail(ec, "start_timer");
timer_.async_wait(strand_.wrap(std::bind(
&BaseWSPeer<Handler, Impl>::on_timer, impl().shared_from_this(),
timer_.async_wait(bind_executor(
strand_,
std::bind(
&BaseWSPeer<Handler, Impl>::on_timer,
impl().shared_from_this(),
std::placeholders::_1)));
}
@@ -444,11 +480,14 @@ on_timer(error_code ec)
// cryptographic is probably overkill..
beast::rngfill(payload_.begin(),
payload_.size(), crypto_prng());
impl().ws_.async_ping(payload_,
strand_.wrap(std::bind(
&BaseWSPeer::on_ping,
impl().ws_.async_ping(
payload_,
bind_executor(
strand_,
std::bind(
&BaseWSPeer::on_ping,
impl().shared_from_this(),
std::placeholders::_1)));
std::placeholders::_1)));
JLOG(this->j_.trace()) <<
"sent ping";
return;
@@ -474,7 +513,7 @@ fail(error_code ec, String const& what)
ec_ = ec;
JLOG(this->j_.trace()) <<
what << ": " << ec.message();
impl().ws_.lowest_layer().close(ec);
ripple::get_lowest_layer(impl().ws_).close(ec);
}
}

77
src/ripple/server/impl/Door.h
View File

@@ -29,7 +29,7 @@
#include <boost/beast/core/multi_buffer.hpp>
#include <boost/asio/basic_waitable_timer.hpp>
#include <boost/asio/buffer.hpp>
#include <boost/asio/io_service.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/container/flat_map.hpp>
@@ -65,16 +65,21 @@ private:
private:
Port const& port_;
Handler& handler_;
boost::asio::io_context& ioc_;
socket_type socket_;
timer_type timer_;
endpoint_type remote_address_;
boost::asio::io_service::strand strand_;
boost::asio::io_context::strand strand_;
beast::Journal j_;
public:
Detector (Port const& port, Handler& handler,
socket_type&& socket, endpoint_type remote_address,
beast::Journal j);
Detector(
Port const& port,
Handler& handler,
boost::asio::io_context& ioc,
socket_type&& socket,
endpoint_type remote_address,
beast::Journal j);
void run();
void close() override;
@@ -86,13 +91,14 @@ private:
beast::Journal j_;
Port const& port_;
Handler& handler_;
boost::asio::io_context& ioc_;
acceptor_type acceptor_;
boost::asio::io_service::strand strand_;
boost::asio::io_context::strand strand_;
bool ssl_;
bool plain_;
public:
Door(Handler& handler, boost::asio::io_service& io_service,
Door(Handler& handler, boost::asio::io_context& io_context,
Port const& port, beast::Journal j);
// Work-around because we can't call shared_from_this in ctor
@@ -169,17 +175,21 @@ detect_ssl (Socket& socket, StreamBuf& buf, Yield do_yield)
return result;
}
template<class Handler>
Door<Handler>::Detector::
Detector(Port const& port,
Handler& handler, socket_type&& socket,
endpoint_type remote_address, beast::Journal j)
template <class Handler>
Door<Handler>::Detector::Detector(
Port const& port,
Handler& handler,
boost::asio::io_context& ioc,
socket_type&& socket,
endpoint_type remote_address,
beast::Journal j)
: port_(port)
, handler_(handler)
, ioc_(ioc)
, socket_(std::move(socket))
, timer_(socket_.get_io_service())
, timer_(ioc_)
, remote_address_(remote_address)
, strand_(socket_.get_io_service())
, strand_(ioc_)
, j_(j)
{
}
@@ -238,15 +248,15 @@ do_detect(boost::asio::yield_context do_yield)
{
if (ssl)
{
if(auto sp = ios().template emplace<SSLHTTPPeer<Handler>>(
port_, handler_, j_, remote_address_,
buf.data(), std::move(socket_)))
if (auto sp = ios().template emplace<SSLHTTPPeer<Handler>>(
port_, handler_, ioc_, j_, remote_address_,
buf.data(), std::move(socket_)))
sp->run();
return;
}
if(auto sp = ios().template emplace<PlainHTTPPeer<Handler>>(
port_, handler_, j_, remote_address_,
buf.data(), std::move(socket_)))
if (auto sp = ios().template emplace<PlainHTTPPeer<Handler>>(
port_, handler_, ioc_, j_, remote_address_,
buf.data(), std::move(socket_)))
sp->run();
return;
}
@@ -262,13 +272,14 @@ do_detect(boost::asio::yield_context do_yield)
template<class Handler>
Door<Handler>::
Door(Handler& handler, boost::asio::io_service& io_service,
Door(Handler& handler, boost::asio::io_context& io_context,
Port const& port, beast::Journal j)
: j_(j)
, port_(port)
, handler_(handler)
, acceptor_(io_service)
, strand_(io_service)
, ioc_(io_context)
, acceptor_(io_context)
, strand_(io_context)
, ssl_(
port_.protocol.count("https") > 0 ||
port_.protocol.count("wss") > 0 ||
@@ -352,15 +363,15 @@ create(bool ssl, ConstBufferSequence const& buffers,
{
if (ssl)
{
if(auto sp = ios().template emplace<SSLHTTPPeer<Handler>>(
port_, handler_, j_, remote_address,
buffers, std::move(socket)))
if (auto sp = ios().template emplace<SSLHTTPPeer<Handler>>(
port_, handler_, ioc_, j_, remote_address,
buffers, std::move(socket)))
sp->run();
return;
}
if(auto sp = ios().template emplace<PlainHTTPPeer<Handler>>(
port_, handler_, j_, remote_address,
buffers, std::move(socket)))
if (auto sp = ios().template emplace<PlainHTTPPeer<Handler>>(
port_, handler_, ioc_, j_, remote_address,
buffers, std::move(socket)))
sp->run();
}
@@ -373,7 +384,7 @@ do_accept(boost::asio::yield_context do_yield)
{
error_code ec;
endpoint_type remote_address;
socket_type socket (acceptor_.get_io_service());
socket_type socket (ioc_);
acceptor_.async_accept (socket, remote_address, do_yield[ec]);
if (ec && ec != boost::asio::error::operation_aborted)
{
@@ -387,9 +398,9 @@ do_accept(boost::asio::yield_context do_yield)
if (ssl_ && plain_)
{
if(auto sp = ios().template emplace<Detector>(
port_, handler_, std::move(socket),
remote_address, j_))
if (auto sp = ios().template emplace<Detector>(
port_, handler_, ioc_, std::move(socket),
remote_address, j_))
sp->run();
}
else if (ssl_ || plain_)

46
src/ripple/server/impl/LowestLayer.h Normal file
View File

@@ -0,0 +1,46 @@
//------------------------------------------------------------------------------
/*
This file is part of rippled: https://github.com/ripple/rippled
Copyright (c) 2019 Ripple Labs Inc.
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL , DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#ifndef RIPPLE_SERVER_LOWESTLAYER_H_INCLUDED
#define RIPPLE_SERVER_LOWESTLAYER_H_INCLUDED
#if BOOST_VERSION >= 107000
#include <boost/beast/core/stream_traits.hpp>
#else
#include <boost/beast/core/type_traits.hpp>
#endif
namespace ripple {
// Before boost 1.70, get_lowest_layer required an explicit templat parameter
template <class T>
decltype(auto)
get_lowest_layer(T& t) noexcept
{
#if BOOST_VERSION >= 107000
return boost::beast::get_lowest_layer(t);
#else
return t.lowest_layer();
#endif
}
} // namespace ripple
#endif

43
src/ripple/server/impl/PlainHTTPPeer.h
View File

@@ -34,19 +34,24 @@ class PlainHTTPPeer
{
private:
friend class BaseHTTPPeer<Handler, PlainHTTPPeer>;
using waitable_timer = typename BaseHTTPPeer<Handler, PlainHTTPPeer>::waitable_timer;
using socket_type = boost::asio::ip::tcp::socket;
using endpoint_type = boost::asio::ip::tcp::endpoint;
socket_type stream_;
public:
template<class ConstBufferSequence>
PlainHTTPPeer(Port const& port, Handler& handler,
beast::Journal journal, endpoint_type remote_address,
ConstBufferSequence const& buffers, socket_type&& socket);
template <class ConstBufferSequence>
PlainHTTPPeer(
Port const& port,
Handler& handler,
boost::asio::io_context& ioc,
beast::Journal journal,
endpoint_type remote_address,
ConstBufferSequence const& buffers,
socket_type&& socket);
void
run();
void run();
std::shared_ptr<WSSession>
websocketUpgrade() override;
@@ -61,14 +66,24 @@ private:
//------------------------------------------------------------------------------
template<class Handler>
template<class ConstBufferSequence>
PlainHTTPPeer<Handler>::
PlainHTTPPeer(Port const& port, Handler& handler,
beast::Journal journal, endpoint_type remote_endpoint,
ConstBufferSequence const& buffers, socket_type&& socket)
: BaseHTTPPeer<Handler, PlainHTTPPeer>(port, handler,
socket.get_io_service(), journal, remote_endpoint, buffers)
template <class Handler>
template <class ConstBufferSequence>
PlainHTTPPeer<Handler>::PlainHTTPPeer(
Port const& port,
Handler& handler,
boost::asio::io_context& ioc,
beast::Journal journal,
endpoint_type remote_endpoint,
ConstBufferSequence const& buffers,
socket_type&& socket)
: BaseHTTPPeer<Handler, PlainHTTPPeer>(
port,
handler,
ioc.get_executor(),
waitable_timer{ioc},
journal,
remote_endpoint,
buffers)
, stream_(std::move(socket))
{
// Set TCP_NODELAY on loopback interfaces,

18
src/ripple/server/impl/PlainWSPeer.h
View File

@@ -21,6 +21,7 @@
#define RIPPLE_SERVER_PLAINWSPEER_H_INCLUDED
#include <ripple/server/impl/BaseWSPeer.h>
#include <ripple/beast/asio/waitable_timer.h>
#include <memory>
namespace ripple {
@@ -54,18 +55,23 @@ public:
//------------------------------------------------------------------------------
template<class Handler>
template<class Body, class Headers>
PlainWSPeer<Handler>::
PlainWSPeer(
template <class Handler>
template <class Body, class Headers>
PlainWSPeer<Handler>::PlainWSPeer(
Port const& port,
Handler& handler,
endpoint_type remote_address,
boost::beast::http::request<Body, Headers>&& request,
socket_type&& socket,
beast::Journal journal)
: BaseWSPeer<Handler, PlainWSPeer>(port, handler, remote_address,
std::move(request), socket.get_io_service(), journal)
: BaseWSPeer<Handler, PlainWSPeer>(
port,
handler,
socket.get_executor(),
beast::create_waitable_timer<waitable_timer>(socket),
remote_address,
std::move(request),
journal)
, ws_(std::move(socket))
{
}

48
src/ripple/server/impl/SSLHTTPPeer.h
View File

@@ -23,6 +23,7 @@
#include <ripple/server/impl/BaseHTTPPeer.h>
#include <ripple/server/impl/SSLWSPeer.h>
#include <ripple/beast/asio/ssl_bundle.h>
#include <ripple/beast/asio/waitable_timer.h>
#include <memory>
namespace ripple {
@@ -34,6 +35,7 @@ class SSLHTTPPeer
{
private:
friend class BaseHTTPPeer<Handler, SSLHTTPPeer>;
using waitable_timer = typename BaseHTTPPeer<Handler, SSLHTTPPeer>::waitable_timer;
using socket_type = boost::asio::ip::tcp::socket;
using stream_type = boost::asio::ssl::stream <socket_type&>;
using endpoint_type = boost::asio::ip::tcp::endpoint;
@@ -44,10 +46,15 @@ private:
stream_type& stream_;
public:
template<class ConstBufferSequence>
SSLHTTPPeer(Port const& port, Handler& handler,
beast::Journal journal, endpoint_type remote_address,
ConstBufferSequence const& buffers, socket_type&& socket);
template <class ConstBufferSequence>
SSLHTTPPeer(
Port const& port,
Handler& handler,
boost::asio::io_context& ioc,
beast::Journal journal,
endpoint_type remote_address,
ConstBufferSequence const& buffers,
socket_type&& socket);
void
run();
@@ -71,14 +78,24 @@ private:
//------------------------------------------------------------------------------
template<class Handler>
template<class ConstBufferSequence>
SSLHTTPPeer<Handler>::
SSLHTTPPeer(Port const& port, Handler& handler,
beast::Journal journal, endpoint_type remote_address,
ConstBufferSequence const& buffers, socket_type&& socket)
: BaseHTTPPeer<Handler, SSLHTTPPeer>(port, handler,
socket.get_io_service(), journal, remote_address, buffers)
template <class Handler>
template <class ConstBufferSequence>
SSLHTTPPeer<Handler>::SSLHTTPPeer(
Port const& port,
Handler& handler,
boost::asio::io_context& ioc,
beast::Journal journal,
endpoint_type remote_address,
ConstBufferSequence const& buffers,
socket_type&& socket)
: BaseHTTPPeer<Handler, SSLHTTPPeer>(
port,
handler,
ioc.get_executor(),
waitable_timer{ioc},
journal,
remote_address,
buffers)
, ssl_bundle_(std::make_unique<beast::asio::ssl_bundle>(
port.context, std::move(socket)))
, stream_(ssl_bundle_->stream)
@@ -168,8 +185,11 @@ SSLHTTPPeer<Handler>::
do_close()
{
this->start_timer();
stream_.async_shutdown(this->strand_.wrap(std::bind (
&SSLHTTPPeer::on_shutdown, this->shared_from_this(),
stream_.async_shutdown(bind_executor(
this->strand_,
std::bind(
&SSLHTTPPeer::on_shutdown,
this->shared_from_this(),
std::placeholders::_1)));
}

22
src/ripple/server/impl/SSLWSPeer.h
View File

@@ -23,6 +23,7 @@
#include <ripple/server/impl/BaseHTTPPeer.h>
#include <ripple/server/WSSession.h>
#include <ripple/beast/asio/ssl_bundle.h>
#include <ripple/beast/asio/waitable_timer.h>
#include <boost/beast/websocket/ssl.hpp>
#include <memory>
@@ -60,20 +61,23 @@ public:
//------------------------------------------------------------------------------
template<class Handler>
template<class Body, class Headers>
SSLWSPeer<Handler>::
SSLWSPeer(
template <class Handler>
template <class Body, class Headers>
SSLWSPeer<Handler>::SSLWSPeer(
Port const& port,
Handler& handler,
endpoint_type remote_endpoint,
boost::beast::http::request<Body, Headers>&& request,
std::unique_ptr<
beast::asio::ssl_bundle>&& ssl_bundle,
std::unique_ptr<beast::asio::ssl_bundle>&& ssl_bundle,
beast::Journal journal)
: BaseWSPeer<Handler, SSLWSPeer>(port, handler,
remote_endpoint, std::move(request),
ssl_bundle->socket.get_io_service(), journal)
: BaseWSPeer<Handler, SSLWSPeer>(
port,
handler,
ssl_bundle->socket.get_executor(),
beast::create_waitable_timer<waitable_timer>(ssl_bundle->socket),
remote_endpoint,
std::move(request),
journal)
, ssl_bundle_(std::move(ssl_bundle))
, ws_(ssl_bundle_->stream)
{

8
src/test/jtx/TrustedPublisherServer.h
View File

@@ -58,14 +58,14 @@ public:
};
TrustedPublisherServer(
boost::asio::io_service& ios,
boost::asio::io_context& ioc,
std::pair<PublicKey, SecretKey> keys,
std::string const& manifest,
int sequence,
NetClock::time_point expiration,
int version,
std::vector<Validator> const& validators)
: sock_(ios), acceptor_(ios)
: sock_(ioc), acceptor_(ioc)
{
endpoint_type const& ep {
beast::IP::Address::from_string (ripple::test::getEnvLocalhostAddr()),
@@ -124,13 +124,13 @@ private:
int id;
TrustedPublisherServer& self;
socket_type sock;
boost::asio::io_service::work work;
boost::asio::executor_work_guard<boost::asio::executor> work;
lambda(int id_, TrustedPublisherServer& self_, socket_type&& sock_)
: id(id_)
, self(self_)
, sock(std::move(sock_))
, work(sock.get_io_service())
, work(sock_.get_executor())
{
}

196
src/test/overlay/short_read_test.cpp
View File

@@ -48,8 +48,8 @@ will complete with eof.
class short_read_test : public beast::unit_test::suite
{
private:
using io_service_type = boost::asio::io_service;
using strand_type = io_service_type::strand;
using io_context_type = boost::asio::io_context;
using strand_type = boost::asio::io_context::strand;
using timer_type = boost::asio::basic_waitable_timer<
std::chrono::steady_clock>;
using acceptor_type = boost::asio::ip::tcp::acceptor;
@@ -59,8 +59,8 @@ private:
using endpoint_type = boost::asio::ip::tcp::endpoint;
using address_type = boost::asio::ip::address;
io_service_type io_service_;
boost::optional<io_service_type::work> work_;
io_context_type io_context_;
boost::optional<boost::asio::executor_work_guard<boost::asio::executor>> work_;
std::thread thread_;
std::shared_ptr<boost::asio::ssl::context> context_;
@@ -183,11 +183,11 @@ private:
: Child(server)
, server_(server)
, test_(server_.test_)
, acceptor_(test_.io_service_,
, acceptor_(test_.io_context_,
endpoint_type(beast::IP::Address::from_string(
test::getEnvLocalhostAddr()), 0))
, socket_(test_.io_service_)
, strand_(socket_.get_io_service())
, socket_(test_.io_context_)
, strand_(test_.io_context_)
{
acceptor_.listen();
server_.endpoint_ = acceptor_.local_endpoint();
@@ -199,7 +199,7 @@ private:
close() override
{
if(! strand_.running_in_this_thread())
return strand_.post(std::bind(&Acceptor::close,
return post(strand_, std::bind(&Acceptor::close,
shared_from_this()));
acceptor_.close();
}
@@ -207,9 +207,14 @@ private:
void
run()
{
acceptor_.async_accept(socket_, strand_.wrap(std::bind(
&Acceptor::on_accept, shared_from_this(),
std::placeholders::_1)));
acceptor_.async_accept(
socket_,
bind_executor(
strand_,
std::bind(
&Acceptor::on_accept,
shared_from_this(),
std::placeholders::_1)));
}
void
@@ -233,9 +238,14 @@ private:
server_, std::move(socket_));
server_.add(p);
p->run();
acceptor_.async_accept(socket_, strand_.wrap(std::bind(
&Acceptor::on_accept, shared_from_this(),
std::placeholders::_1)));
acceptor_.async_accept(
socket_,
bind_executor(
strand_,
std::bind(
&Acceptor::on_accept,
shared_from_this(),
std::placeholders::_1)));
}
};
@@ -256,8 +266,8 @@ private:
, test_(server_.test_)
, socket_(std::move(socket))
, stream_(socket_, *test_.context_)
, strand_(socket_.get_io_service())
, timer_(socket_.get_io_service())
, strand_(test_.io_context_)
, timer_(test_.io_context_)
{
}
@@ -265,7 +275,7 @@ private:
close() override
{
if(! strand_.running_in_this_thread())
return strand_.post(std::bind(&Connection::close,
return post(strand_, std::bind(&Connection::close,
shared_from_this()));
if (socket_.is_open())
{
@@ -278,11 +288,20 @@ private:
run()
{
timer_.expires_from_now(std::chrono::seconds(3));
timer_.async_wait(strand_.wrap(std::bind(&Connection::on_timer,
shared_from_this(), std::placeholders::_1)));
stream_.async_handshake(stream_type::server, strand_.wrap(
std::bind(&Connection::on_handshake, shared_from_this(),
timer_.async_wait(bind_executor(
strand_,
std::bind(
&Connection::on_timer,
shared_from_this(),
std::placeholders::_1)));
stream_.async_handshake(
stream_type::server,
bind_executor(
strand_,
std::bind(
&Connection::on_handshake,
shared_from_this(),
std::placeholders::_1)));
}
void
@@ -315,9 +334,16 @@ private:
if (ec)
return fail("handshake", ec);
#if 1
boost::asio::async_read_until(stream_, buf_, "\n", strand_.wrap(
std::bind(&Connection::on_read, shared_from_this(),
std::placeholders::_1,
boost::asio::async_read_until(
stream_,
buf_,
"\n",
bind_executor(
strand_,
std::bind(
&Connection::on_read,
shared_from_this(),
std::placeholders::_1,
std::placeholders::_2)));
#else
close();
@@ -330,8 +356,11 @@ private:
if (ec == boost::asio::error::eof)
{
server_.test_.log << "[server] read: EOF" << std::endl;
return stream_.async_shutdown(strand_.wrap(std::bind(
&Connection::on_shutdown, shared_from_this(),
return stream_.async_shutdown(bind_executor(
strand_,
std::bind(
&Connection::on_shutdown,
shared_from_this(),
std::placeholders::_1)));
}
if (ec)
@@ -340,9 +369,15 @@ private:
buf_.commit(bytes_transferred);
buf_.consume(bytes_transferred);
write(buf_, "BYE\n");
boost::asio::async_write(stream_, buf_.data(), strand_.wrap(
std::bind(&Connection::on_write, shared_from_this(),
std::placeholders::_1,
boost::asio::async_write(
stream_,
buf_.data(),
bind_executor(
strand_,
std::bind(
&Connection::on_write,
shared_from_this(),
std::placeholders::_1,
std::placeholders::_2)));
}
@@ -352,8 +387,11 @@ private:
buf_.consume(bytes_transferred);
if (ec)
return fail("write", ec);
stream_.async_shutdown(strand_.wrap(std::bind(
&Connection::on_shutdown, shared_from_this(),
stream_.async_shutdown(bind_executor(
strand_,
std::bind(
&Connection::on_shutdown,
shared_from_this(),
std::placeholders::_1)));
}
@@ -412,10 +450,10 @@ private:
: Child(client)
, client_(client)
, test_(client_.test_)
, socket_(test_.io_service_)
, socket_(test_.io_context_)
, stream_(socket_, *test_.context_)
, strand_(socket_.get_io_service())
, timer_(socket_.get_io_service())
, strand_(test_.io_context_)
, timer_(test_.io_context_)
, ep_(ep)
{
}
@@ -424,8 +462,9 @@ private:
close() override
{
if(! strand_.running_in_this_thread())
return strand_.post(std::bind(&Connection::close,
shared_from_this()));
return post(
strand_,
std::bind(&Connection::close, shared_from_this()));
if (socket_.is_open())
{
socket_.close();
@@ -437,11 +476,20 @@ private:
run(endpoint_type const& ep)
{
timer_.expires_from_now(std::chrono::seconds(3));
timer_.async_wait(strand_.wrap(std::bind(&Connection::on_timer,
shared_from_this(), std::placeholders::_1)));
socket_.async_connect(ep, strand_.wrap(std::bind(
&Connection::on_connect, shared_from_this(),
timer_.async_wait(bind_executor(
strand_,
std::bind(
&Connection::on_timer,
shared_from_this(),
std::placeholders::_1)));
socket_.async_connect(
ep,
bind_executor(
strand_,
std::bind(
&Connection::on_connect,
shared_from_this(),
std::placeholders::_1)));
}
void
@@ -473,9 +521,14 @@ private:
{
if (ec)
return fail("connect", ec);
stream_.async_handshake(stream_type::client, strand_.wrap(
std::bind(&Connection::on_handshake, shared_from_this(),
std::placeholders::_1)));
stream_.async_handshake(
stream_type::client,
bind_executor(
strand_,
std::bind(
&Connection::on_handshake,
shared_from_this(),
std::placeholders::_1)));
}
void
@@ -486,13 +539,22 @@ private:
write(buf_, "HELLO\n");
#if 1
boost::asio::async_write(stream_, buf_.data(), strand_.wrap(
std::bind(&Connection::on_write, shared_from_this(),
std::placeholders::_1,
boost::asio::async_write(
stream_,
buf_.data(),
bind_executor(
strand_,
std::bind(
&Connection::on_write,
shared_from_this(),
std::placeholders::_1,
std::placeholders::_2)));
#else
stream_.async_shutdown(strand_.wrap(std::bind(
&Connection::on_shutdown, shared_from_this(),
stream_.async_shutdown(bind_executor(
strand_,
std::bind(
&Connection::on_shutdown,
shared_from_this(),
std::placeholders::_1)));
#endif
}
@@ -504,13 +566,23 @@ private:
if (ec)
return fail("write", ec);
#if 1
boost::asio::async_read_until(stream_, buf_, "\n", strand_.wrap(
std::bind(&Connection::on_read, shared_from_this(),
std::placeholders::_1,
boost::asio::async_read_until(
stream_,
buf_,
"\n",
bind_executor(
strand_,
std::bind(
&Connection::on_read,
shared_from_this(),
std::placeholders::_1,
std::placeholders::_2)));
#else
stream_.async_shutdown(strand_.wrap(std::bind(
&Connection::on_shutdown, shared_from_this(),
stream_.async_shutdown(bind_executor(
strand_,
std::bind(
&Connection::on_shutdown,
shared_from_this(),
std::placeholders::_1)));
#endif
}
@@ -521,8 +593,11 @@ private:
if (ec)
return fail("read", ec);
buf_.commit(bytes_transferred);
stream_.async_shutdown(strand_.wrap(std::bind(
&Connection::on_shutdown, shared_from_this(),
stream_.async_shutdown(bind_executor(
strand_,
std::bind(
&Connection::on_shutdown,
shared_from_this(),
std::placeholders::_1)));
}
@@ -555,12 +630,11 @@ private:
public:
short_read_test()
: work_(boost::in_place(std::ref(io_service_)))
, thread_(std::thread([this]()
{
beast::setCurrentThreadName("io_service");
this->io_service_.run();
}))
: work_(io_context_.get_executor())
, thread_(std::thread([this]() {
beast::setCurrentThreadName("io_context");
this->io_context_.run();
}))
, context_(make_SSLContext(""))
{
}