boost::asio::io_context::run_one_for() fails to send large buffer

Question

I need to get synchronous I/O but with following features:

1. interruption by other thread
2. support for timeouts

So, I use async I/O from Boost.Asio and execute it by boost::asio::io_context::run_one_for(). There is an example of such implementation of resolve, connect, read and write.

long const DefaultTimeout = 50;

namespace asio = boost::asio;
using boost::asio::ip::tcp;

template <typename T>
void PerformIO(T &Object, long Timeout)
{
    auto &Context = Object.get_executor().context();
    Context.reset();
    if (!Context.run_one_for(std::chrono::seconds(Timeout)))
        throw std::exception("I/O operation was timed out");
}

template <typename T, typename TSuccessFlag>
void PerformIO(T &Object, long Timeout, TSuccessFlag &Success)
{
    PerformIO(Object, Timeout);
    boost::this_thread::interruption_point();
    if (!Success)
        throw std::exception("I/O operation was not successful");
}

tcp::resolver::results_type Resolve(tcp::resolver &Resolver, std::string const &Host, std::string const &Port)
{
    bool Resolved = false;
    tcp::resolver::results_type Endpoints;
    Resolver.async_resolve(Host, Port,
        [&](boost::system::error_code const &Error, boost::asio::ip::tcp::resolver::results_type Results){ Endpoints = Results; Resolved = true; });
    PerformIO(Resolver, DefaultTimeout, Resolved);
    if (Endpoints.begin() == Endpoints.end())
        throw std::exception("Not resolved");
    return Endpoints;
}

template <typename T>
void Connect(tcp::socket &Socket, T const &Endpoints)
{
    bool Connected = false;
    asio::async_connect(Socket, Endpoints,
        [&](boost::system::error_code const &Error, boost::asio::ip::tcp::endpoint const &Endpoint){ Connected = true; });
    PerformIO(Socket, DefaultTimeout, Connected);
}

template <typename T>
size_t ReadSome(tcp::socket &Socket, T &Buffers)
{
    size_t Bytes = 0;
    asio::async_read(Socket, Buffers, asio::transfer_at_least(1),
        [&](boost::system::error_code const &Error, std::size_t BytesTransferred){ Bytes += BytesTransferred; });
    PerformIO(Socket, DefaultTimeout, Bytes);
    return Bytes;
}

template <typename T>
size_t Write(tcp::socket &Socket, T &Buffers)
{
    size_t Bytes = 0;
    asio::async_write(Socket, Buffers,
        [&](boost::system::error_code const &Error, std::size_t BytesTransferred){ Bytes += BytesTransferred; });
    PerformIO(Socket, DefaultTimeout, Bytes);
    return Bytes;
}

It works fine for small data, but it fails if I try to send large buffer of data like this:

// tcp::socket Socket;
// char const *Buffer;
// size_t BufferSize = 1000000;
Write(Socket, asio::buffer(Buffer, BufferSize))

I think the reason is the implementation of boost::asio::io_context::run_one_for(). There it is (boost 1.67):

template <typename Rep, typename Period>
std::size_t io_context::run_one_for(
    const chrono::duration<Rep, Period>& rel_time)
{
  return this->run_one_until(chrono::steady_clock::now() + rel_time);
}

template <typename Clock, typename Duration>
std::size_t io_context::run_one_until(
    const chrono::time_point<Clock, Duration>& abs_time)
{
  typename Clock::time_point now = Clock::now();
  while (now < abs_time)
  {
    typename Clock::duration rel_time = abs_time - now;
    if (rel_time > chrono::seconds(1))
      rel_time = chrono::seconds(1);

    boost::system::error_code ec;
    std::size_t s = impl_.wait_one(
        static_cast<long>(chrono::duration_cast<
          chrono::microseconds>(rel_time).count()), ec);
    boost::asio::detail::throw_error(ec);

    if (s || impl_.stopped())
      return s;

    now = Clock::now();
  }

  return 0;
}

Notice the limit of 1 second. I think sending large buffer taking more than 1 second and execution stops. I can execute Write() sequentially by 64k-bytes and it works. But I think it will fail if connection will unable to send 64k-bytes per second. So, I need a better solution.

Answer 1

I think the reason is the implementation of boost::asio::io_context::run_one_for(). There it is (boost 1.67):

No. The reason is that boost::asio::[async_][read|write][_.*] are composed operations. run_one[_*] only runs a single handler so it wouldn't complete a composed async operation.

Note also, that boost::asio::write is a synchronous operation, regardless of any of the surrounding logic, so your timeout doesn't even come into play.

You could just wrap an async operation in a local io_context and run it to completion using a deadline (so, run_for instead of run_one_for).

You can glean a similar concept from How to simulate boost::asio::write with a timeout - though it is strictly (a lot) better to use run_for(...) instead of the run_one() (for the reason explained above. At the time I wrote the other answer, run_for() didn't exist, and I might have to update that answer anyways because I might have missed the problem with composed operations back then).