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

Refactor docs

Browse Source
This commit is contained in:
Vinnie Falco
2019-02-24 18:46:27 -08:00
parent 50be985254
commit d9400ce618
64 changed files with 1357 additions and 383 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/doc/core_3_timeouts.cpp
+647
View File
@@ -0,0 +1,647 @@
//
// Copyright (c) 2016-2019 Vinnie Falco (vinnie dot falco at gmail dot com)
//
// Distributed under the Boost Software License, Version 1.0. (See accompanying
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//
// Official repository: https://github.com/boostorg/beast
//
#include "snippets.hpp"
#include <boost/beast/_experimental/unit_test/suite.hpp>
#include <boost/beast/_experimental/test/stream.hpp>
#include <boost/beast/core/async_op_base.hpp>
#include <boost/beast/core/buffers_prefix.hpp>
#include <boost/beast/core/error.hpp>
#include <boost/beast/core/flat_buffer.hpp>
#include <boost/beast/core/stream_traits.hpp>
#include <boost/beast/core/tcp_stream.hpp>
#include <boost/beast/http.hpp>
#include <boost/beast/ssl/ssl_stream.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/asio/read.hpp>
#include <cstdlib>
#include <utility>
namespace boost {
namespace beast {
namespace {
struct handler_type
{
template<class... Args>
void operator()(Args&&...)
{
}
};
void
core_3_timeouts_snippets()
{
handler_type handler;
#include "snippets.ipp"
{
//[code_core_3_timeouts_1
// `ioc` will be used to dispatch completion handlers
tcp_stream stream(ioc);
//]
}
{
//[code_core_3_timeouts_2
// The resolver is used to look up the IP addresses for a domain name
net::ip::tcp::resolver resolver(ioc);
// The stream will use the same executor as the resolver
tcp_stream stream(resolver.get_executor());
//]
}
{
//[code_core_3_timeouts_3
// The strand will be used to invoke all completion handlers
tcp_stream stream(make_strand(ioc));
//]
net::ip::tcp::resolver resolver(ioc);
//[code_core_3_timeouts_4
// Set the logical operation timer to 30 seconds
stream.expires_after (std::chrono::seconds(30));
// If the connection is not established within 30 seconds,
// the operation will be canceled and the handler will receive
// error::timeout as the error code.
stream.async_connect(resolver.resolve("www.example.com", "http"),
[](error_code ec, net::ip::tcp::endpoint ep)
{
if(ec == error::timeout)
std::cerr << "async_connect took too long\n";
else if(! ec)
std::cout << "Connected to " << ep << "\n";
}
);
// The timer is still running. If we don't want the next
// operation to time out 30 seconds relative to the previous
// call to `expires_after`, we need to turn it off before
// starting another asynchronous operation.
stream.expires_never();
//]
}
{
//[code_core_3_timeouts_5
// The acceptor is used to listen and accept incoming connections.
// We construct the acceptor to use a new strand, and listen
// on the loopback address with an operating-system assigned port.
net::ip::tcp::acceptor acceptor(make_strand(ioc));
acceptor.bind(net::ip::tcp::endpoint(net::ip::make_address_v4("127.0.0.1"), 0));
acceptor.listen(0);
// This blocks until a new incoming connection is established.
// Upon success, the function returns a new socket which is
// connected to the peer. The socket will have its own executor,
// which in the call below is a new strand for the I/O context.
net::ip::tcp::socket s = acceptor.accept(make_strand(ioc));
// Construct a new tcp_stream from the connected socket.
// The stream will use the strand created when the connection
// was accepted.
tcp_stream stream(std::move(s));
//]
//[code_core_3_timeouts_6
flat_buffer b;
// Set the logical operation timer to 30 seconds.
stream.expires_after (std::chrono::seconds(30));
// Read a line from the stream into our dynamic buffer.
// The function dynamic_buffer_ref is used because Asio
// treats these buffers as non-owning references, but
// Beast uses them as first-class containers.
net::async_read_until(stream, dynamic_buffer_ref(b), '\n',
[&b, &stream](error_code ec, std::size_t bytes_transferred)
{
if(ec)
return;
// read_until can read past the '\n', these will end up in
// our buffer but we don't want to echo those extra received
// bytes. `bytes_transferred` will be the number of bytes
// up to and including the '\n'. We use `buffers_prefix` so
// that extra data is not written.
net::async_write(stream, buffers_prefix(bytes_transferred, b.data()),
[&b](error_code ec, std::size_t bytes_transferred)
{
// Consume the line from the buffer
b.consume(bytes_transferred);
if(ec)
std::cerr << "Error: " << ec.message() << "\n";
});
});
//]
//[code_core_3_timeouts_7
flat_buffer b2;
// Set the logical operation timer to 15 seconds.
stream.expires_after (std::chrono::seconds(15));
// Read another line from the stream into our dynamic buffer.
// The operation will time out after 15 seconds.
net::async_read_until(stream, dynamic_buffer_ref(b2), '\n', handler);
// Set the logical operation timer to 30 seconds.
stream.expires_after (std::chrono::seconds(30));
// Write the contents of the other buffer.
// This operation will time out after 30 seconds.
net::async_write(stream, b.data(), handler);
//]
}
{
//[code_core_3_timeouts_8
// To declare a stream with a rate policy, it is necessary to
// write out all of the template parameter types.
//
// `simple_rate_policy` is default constructible, but
// if the choice of RatePolicy is not DefaultConstructible,
// an instance of the type may be passed to the constructor.
basic_stream<net::ip::tcp, net::executor, simple_rate_policy> stream(ioc);
// The policy object, which is default constructed, or
// decay-copied upon construction, is attached to the stream
// and may be accessed through the function `rate_policy`.
//
// Here we set individual rate limits for reading and writing
stream.rate_policy().read_limit(10000); // bytes per second
stream.rate_policy().write_limit(850000); // bytes per second
//]
}
}
//[code_core_3_timeouts_1f
/** This function echoes back received lines from a peer, with a timeout.
The algorithm terminates upon any error (including timeout).
*/
template <class Protocol, class Executor>
void do_async_echo (basic_stream<Protocol, Executor>& stream)
{
// This object will hold our state when reading the line.
struct echo_line
{
basic_stream<Protocol, Executor>& stream;
// The shared pointer is used to extend the lifetime of the
// buffer until the last asynchronous operation completes.
std::shared_ptr<flat_buffer> buffer;
// This starts a new operation to read and echo a line
void operator()()
{
// If a line is not sent and received within 30 seconds, then
// the connection will be closed and this algorithm will terminate.
stream.expires_after(std::chrono::seconds(30));
// Read a line from the stream into our dynamic buffer, with a timeout
net::async_read_until(stream, dynamic_buffer_ref(*buffer), '\n', std::move(*this));
}
// This function is called when the read completes
void operator()(error_code ec, std::size_t bytes_transferred)
{
if(ec)
return;
net::async_write(stream, buffers_prefix(bytes_transferred, buffer->data()),
[this](error_code ec, std::size_t bytes_transferred)
{
buffer->consume(bytes_transferred);
if(! ec)
{
// Run this algorithm again
echo_line{stream, std::move(buffer)}();
}
else
{
std::cerr << "Error: " << ec.message() << "\n";
}
});
}
};
// Create the operation and run it
echo_line{stream, std::make_shared<flat_buffer>()}();
}
//]
//[code_core_3_timeouts_2f
/** Request an HTTP resource from a TLS host and return it as a string, with a timeout.
This example uses fibers (stackful coroutines) and its own I/O context.
*/
std::string
https_get (std::string const& host, std::string const& target, error_code& ec)
{
// It is the responsibility of the algorithm to clear the error first.
ec = {};
// We use our own I/O context, to make this function blocking.
net::io_context ioc;
// This context is used to hold client and server certificates.
// We do not perform certificate verification in this example.
net::ssl::context ctx(net::ssl::context::sslv23);
// This string will hold the body of the HTTP response, if any.
std::string result;
// Note that Networking TS does not come with spawn. This function
// launches a "fiber" which is a coroutine that has its own separately
// allocated stack.
boost::asio::spawn(ioc,
[&](boost::asio::yield_context yield)
{
// We use the Beast ssl_stream wrapped around a beast tcp_stream.
ssl_stream<tcp_stream> stream(ioc, ctx);
// The resolver will be used to look up the IP addresses for the host name
net::ip::tcp::resolver resolver(ioc);
// First, look up the name. Networking has its own timeout for this.
// The `yield` object is a CompletionToken which specializes the
// `net::async_result` customization point to make the fiber work.
//
// This call will appear to "block" until the operation completes.
// It isn't really blocking. Instead, the fiber implementation saves
// the call stack and suspends the function until the asynchronous
// operation is complete. Then it restores the call stack, and resumes
// the function to the statement following the async_resolve. This
// allows an asynchronous algorithm to be expressed synchronously.
auto const endpoints = resolver.async_resolve(host, "https", {}, yield[ec]);
if(ec)
return;
// The function `get_lowest_layer` retrieves the "bottom most" object
// in the stack of stream layers. In this case it will be the tcp_stream.
// This timeout will apply to all subsequent operations collectively.
// That is to say, they must all complete within the same 30 second
// window.
get_lowest_layer(stream).expires_after(std::chrono::seconds(30));
// `tcp_stream` range connect algorithms are member functions, unlike net::
get_lowest_layer(stream).async_connect(endpoints, yield[ec]);
if(ec)
return;
// Perform the TLS handshake
stream.async_handshake(net::ssl::stream_base::client, yield[ec]);
if(ec)
return;
// Send an HTTP GET request for the target
{
http::request<http::empty_body> req;
req.method(http::verb::get);
req.target(target);
req.version(11);
req.set(http::field::server, host);
req.set(http::field::user_agent, "Beast");
http::async_write(stream, req, yield[ec]);
if(ec)
return;
}
// Now read the response
flat_buffer buffer;
http::response<http::string_body> res;
http::async_read(stream, buffer, res, yield[ec]);
if(ec)
return;
// Try to perform the TLS shutdown handshake
stream.async_shutdown(yield[ec]);
// `net::ssl::error::stream_truncated`, also known as an SSL "short read",
// indicates the peer closed the connection without performing the
// required closing handshake (for example, Google does this to
// improve performance). Generally this can be a security issue,
// but if your communication protocol is self-terminated (as
// it is with both HTTP and WebSocket) then you may simply
// ignore the lack of close_notify.
//
// https://github.com/boostorg/beast/issues/38
//
// https://security.stackexchange.com/questions/91435/how-to-handle-a-malicious-ssl-tls-shutdown
//
// When a short read would cut off the end of an HTTP message,
// Beast returns the error beast::http::error::partial_message.
// Therefore, if we see a short read here, it has occurred
// after the message has been completed, so it is safe to ignore it.
if(ec == net::ssl::error::stream_truncated)
ec = {};
else if(ec)
return;
// Set the string to return to the caller
result = std::move(res.body());
});
// `run` will dispatch completion handlers, and block until there is
// no more "work" remaining. When this call returns, the operations
// are complete and we can give the caller the result.
ioc.run();
return result;
}
//]
//[code_core_3_timeouts_3f
class window
{
std::size_t value_ = 0;
// The size of the exponential window, in seconds.
// This should be a power of two.
static std::size_t constexpr Window = 4;
public:
/** Returns the number of elapsed seconds since the given time, and adjusts the time.
This function returns the number of elapsed seconds since the
specified time point, rounding down. It also moves the specified
time point forward by the number of elapsed seconds.
@param since The time point from which to calculate elapsed time.
The function will modify the value, by adding the number of elapsed
seconds to it.
@return The number of elapsed seconds.
*/
template<class Clock, class Duration>
static
std::chrono::seconds
get_elapsed(std::chrono::time_point<Clock, Duration>& since) noexcept
{
auto const elapsed = std::chrono::duration_cast<
std::chrono::seconds>(Clock::now() - since);
since += elapsed;
return elapsed;
}
/// Returns the current value, after adding the given sample.
std::size_t
update(std::size_t sample, std::chrono::seconds elapsed) noexcept
{
// Apply exponential decay.
//
// This formula is fast (no division or multiplication) but inaccurate.
// It overshoots by `n*(1-a)/(1-a^n), where a=(window-1)/window`.
// Could be good enough for a rough approximation, but if relying
// on this for production please perform tests!
auto count = elapsed.count();
while(count--)
value_ -= (value_ + Window - 1) / Window;
value_ += sample;
return value_ / Window;
}
/// Returns the current value
std::size_t
value() const noexcept
{
return value_ / Window;
}
};
//]
//[code_core_3_timeouts_4f
/** A RatePolicy to measure instantaneous throughput.
This measures the rate of transfer for reading and writing
using a simple exponential decay function.
*/
class rate_gauge
{
// The clock used to measure elapsed time
using clock_type = std::chrono::steady_clock;
// This implements an exponential smoothing window function.
// The value `Seconds` is the size of the window in seconds.
clock_type::time_point when_;
std::size_t read_bytes_ = 0;
std::size_t write_bytes_ = 0;
window read_window_;
window write_window_;
// Friending this type allows us to mark the
// member functions required by RatePolicy as private.
friend class rate_policy_access;
// Returns the number of bytes available to read currently
// Required by RatePolicy
std::size_t
available_read_bytes() const noexcept
{
// no limit
return (std::numeric_limits<std::size_t>::max)();
}
// Returns the number of bytes available to write currently
// Required by RatePolicy
std::size_t
available_write_bytes() const noexcept
{
// no limit
return (std::numeric_limits<std::size_t>::max)();
}
// Called every time bytes are read
// Required by RatePolicy
void
transfer_read_bytes(std::size_t n) noexcept
{
// Add this to our running total of bytes read
read_bytes_ += n;
}
// Called every time bytes are written
// Required by RatePolicy
void
transfer_write_bytes(std::size_t n) noexcept
{
// Add this to our running total of bytes written
write_bytes_ += n;
}
// Called approximately once per second
// Required by RatePolicy
void
on_timer()
{
// Calculate elapsed time in seconds, and adjust our time point
auto const elapsed = window::get_elapsed(when_);
// Skip the update when elapsed==0,
// otherwise the measurement will have jitter
if(elapsed.count() == 0)
return;
// Add our samples and apply exponential decay
read_window_.update(read_bytes_, elapsed);
write_window_.update(write_bytes_, elapsed);
// Reset our counts of bytes transferred
read_bytes_ = 0;
write_bytes_ = 0;
}
public:
rate_gauge()
: when_(clock_type::now())
{
}
/// Returns the current rate of reading in bytes per second
std::size_t
read_bytes_per_second() const noexcept
{
return read_window_.value();
}
/// Returns the current rate of writing in bytes per second
std::size_t
write_bytes_per_second() const noexcept
{
return write_window_.value();
}
};
//]
void
core_3_timeouts_snippets2()
{
#include "snippets.ipp"
{
//[code_core_3_timeouts_9
// This stream will use our new rate_gauge policy
basic_stream<net::ip::tcp, net::executor, rate_gauge> stream(ioc);
//...
// Print the current rates
std::cout <<
stream.rate_policy().read_bytes_per_second() << " bytes/second read\n" <<
stream.rate_policy().write_bytes_per_second() << " bytes/second written\n";
//]
}
}
} // (anon)
template class basic_stream<net::ip::tcp, net::executor, rate_gauge>;
struct core_3_timeouts_test
: public beast::unit_test::suite
{
void
testWindow()
{
window w;
std::size_t v0 = w.value();
std::size_t const N = 100000;
for(std::size_t n = 1; n <= 2; ++n)
{
for(std::size_t i = 0;;++i)
{
auto const v = w.update(n * N, std::chrono::seconds(n));
if(v == v0)
{
BEAST_PASS();
#if 0
log <<
"update(" << n*N << ", " << n <<
") converged to " << w.value() <<
" in " << i << std::endl;
#endif
break;
}
if(i > 1000)
{
BEAST_FAIL();
break;
}
v0 = v;
}
}
}
void
run() override
{
testWindow();
BEAST_EXPECT(&core_3_timeouts_snippets);
BEAST_EXPECT(&core_3_timeouts_snippets2);
BEAST_EXPECT((&do_async_echo<net::ip::tcp, net::io_context::executor_type>));
BEAST_EXPECT(&https_get);
}
};
BEAST_DEFINE_TESTSUITE(beast,doc,core_3_timeouts);
} // beast
} // boost