Re: Need cancel() on async_resolve() to callback the resolve handler immediately

From: Richard Hodges (hodges.r_at_[hidden])
Date: 2021-08-25 17:08:44

> >> At present the "cleanest" way is to fork a child process to do the
> >> resolve
> >> call for you and async_wait on a pipe from that child. If you want to
> >> cancel it, send the child a SIGKILL which will result in the wait on the
> >> pipe completing with an error.
> >>
> >> Ghastly I know...
>
>
Here is a possible workaround that I've hacked together. Uses C++20
coroutines and latest boost.asio.

example output:
$ resolve
timeout: 1ms : exception Connection timed out
timeout: 5000ms : timeout: 5000ms : 142.250.184.14:80,
[2a00:1450:4003:80f::200e]:80
timeout: 20000ms : timeout: 20000ms : 142.250.184.14:80,
[2a00:1450:4003:80f::200e]:80

Here's the code:

//
// Copyright (c) 2021 Richard Hodges (hodges.r_at_[hidden])
//
// 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)
//
//

#include <boost/asio.hpp>
#include <boost/asio/experimental/as_tuple.hpp>
#include <boost/asio/experimental/awaitable_operators.hpp>
#include <sys/wait.h>

#include <array>
#include <iostream>
#include <string>
#include <vector>

namespace asio   = boost::asio;
namespace asioex = asio::experimental;
using boost::system::error_code;
using boost::system::generic_category;
using boost::system::system_error;

/// Serialise enough of a resolver's results into a null-separarted
sequence of string segments
std::string
serialise(asio::ip::tcp::resolver::results_type const &results)
{
    std::string result;
    auto        host    = results->host_name();
    auto        service = results->service_name();
    result.insert(result.end(), host.c_str(), host.c_str() + host.size() +
1);
    result.insert(result.end(), service.c_str(), service.c_str() +
service.size() + 1);

    result += std::to_string(results.size());
    result.append(1, '\0');
    for(auto& e : results)
    {
        result += e.endpoint().address().to_string();
        result.append(1, '\0');
        result += std::to_string(e.endpoint().port());
        result.append(1, '\0');
    }

    return result;
}

/// Synchronous function executed in the child process to perform the
resolve,
/// serialise the resulting results obejct, and
/// send the serialised string to the write end of the pipe
void
do_child(asio::local::stream_protocol::socket sock, std::string hostname,
std::string service)
{
    auto resolver = asio::ip::tcp::resolver(sock.get_executor());
    auto results  = resolver.resolve(hostname, service);
    auto buf     = serialise(results);
    auto written = asio::write(sock, asio::buffer(buf));
    sock.shutdown(asio::socket_base::shutdown_send);
}

/// Coroutine to asynchronously wait for the given child to exit
asio::awaitable< void >
wait_child(int childpid)
{
    auto sigs = asio::signal_set(co_await asio::this_coro::executor,
SIGCHLD);
    for (;;)
    {
        int  state = 0;
        auto ret   = ::waitpid(childpid, &state, WNOHANG);
        if (ret != 0)
            break;
        auto sig = co_await sigs.async_wait(asio::use_awaitable);
        if (sig != SIGCHLD)
            std::cout << "strange signal: " << sig << "\n";
    }
}

/// Consume an individual string segment from the read end of the pipe.
asio::awaitable<std::string>
consume_token(asio::local::stream_protocol::socket& s, std::string&
rxbuffer)
{
    auto size = co_await asio::async_read_until(s,
asio::dynamic_buffer(rxbuffer), '\0', asio::use_awaitable);
    auto result = rxbuffer.substr(0, size - 1);
    rxbuffer.erase(0, size);
    co_return result;
}

/// Coroutine to deserialise the resolve results from the read end of the
socket.
asio::awaitable<std::vector<asio::ip::tcp::endpoint >>
collect_endpoints(asio::local::stream_protocol::socket& from_child)
{
    std::vector<asio::ip::tcp::endpoint> endpoints;
    std::string rxbuf;

    // rxhost and rxservice unused for now
    auto rxhost = co_await consume_token(from_child, rxbuf);
    auto rxservice = co_await consume_token(from_child, rxbuf);

    auto n_str = co_await consume_token(from_child, rxbuf);
    auto entries = ::atoi(n_str.c_str());
    endpoints.reserve(entries);
    while(entries--)
    {
        auto addr = asio::ip::make_address(co_await
consume_token(from_child, rxbuf));
        auto port = static_cast<unsigned short>(::atol((co_await
consume_token(from_child, rxbuf)).c_str()));
        endpoints.push_back(asio::ip::tcp::endpoint(addr, port));
    }
    co_return endpoints;
}

/// Asyncronous resolve expressed in terms of a child process
asio::awaitable<std::vector<asio::ip::tcp::endpoint >>
resolve(std::string hostname, std::string service,
std::chrono::milliseconds timeout)
{
    // make a pipe
    asio::local::stream_protocol::socket
        parent { co_await asio::this_coro::executor },
        child { co_await asio::this_coro::executor };
    asio::local::connect_pair(parent, child);

    // query the current executor for the execution context
    auto& ctx = asio::query(co_await asio::this_coro::executor,
asio::execution::context);

    // notify the execution context that we are about to fork
    ctx.notify_fork(asio::execution_context::fork_prepare);
    auto childpid = ::fork();
    if (childpid == 0)
    {
        // in the child we close the read end of the pipe and write to the
other end
        ctx.notify_fork(asio::execution_context::fork_child);
        parent.close();
        do_child(std::move(child), hostname, service);
        std::exit(0);
    }
    else if (childpid == -1)
    {
        // error case - failed to fork
        auto ec = error_code(errno, generic_category());
        ctx.notify_fork(asio::execution_context::fork_parent);
        throw system_error(ec);
    }

    // parent - close the write end of the pipe
    ctx.notify_fork(asio::execution_context::fork_parent);
    child.close();

    std::vector<asio::ip::tcp::endpoint> endpoints;
    std::exception_ptr except = nullptr;
    try
    {
        using namespace asioex::awaitable_operators;
        auto timer = asio::steady_timer(co_await asio::this_coro::executor,
timeout);

        // wait for either collect_endpoints or timeout
        auto t0 = std::chrono::steady_clock::now();
        auto which = co_await (
            collect_endpoints(parent) ||
            timer.async_wait(asio::use_awaitable));
        auto t1 = std::chrono::steady_clock::now();

        // whichever finishes first, kill the child
        ::kill(childpid, SIGTERM);

        if (which.index() == 0)
            endpoints = std::get<0>(std::move(which));
        else {
            throw system_error(asio::error::timed_out);
        }
    }
    catch (std::exception &e)
    {
        // catch exception and place into an exception pointer
        // because we need to call a coroutine in the cleanup
        // and you can't call coroutines from exception handlers
        except = std::current_exception();
    }

    // wait for the child to finish
    co_await wait_child(childpid);

    if (except)
        std::rethrow_exception(except);

    co_return endpoints;
}

asio::awaitable< void >
resolve_test()
{
    using namespace std::literals;

    auto hostname = "google.com";
    auto service  = "http";

    auto timeouts = std::array { 1ms, 5'000ms, 20'000ms };

    for(auto t : timeouts)
    {
        std::cout << "timeout: " << t.count() << "ms : ";
        try
        {
            auto endpoints = co_await resolve(hostname, service, t);
            const char*  sep = "";
            for(auto&& e : endpoints) {
                std::cout << sep << e;
                sep = ", ";
            }
        }
        catch(std::exception& e)
        {
            std::cout << "exception " << e.what();
        }
        std::cout << "\n";
    }
}

int
main()
{
    auto ioc = asio::io_context();

    asio::co_spawn(ioc, resolve_test(), asio::detached);

    ioc.run();
}

If Chris K is watching, I'd value his critique. This is probably much more
convoluted than it needs to be.

_______________________________________________
> Unsubscribe & other changes:
> http://lists.boost.org/mailman/listinfo.cgi/boost
>