file_io.cpp

//
// 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
//

//------------------------------------------------------------------------------
//
// Example: HTTP server, coroutine
//
//------------------------------------------------------------------------------

#include <boost/asio/any_io_executor.hpp>
#include <boost/asio/executor.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <boost/system/detail/error_code.hpp>

#include <algorithm>
#include <cstdlib>
#include <iostream>
#include <memory>
#include <string>
#include <system_error>
#include <thread>
#include <vector>

#include <boost/asio/this_coro.hpp>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/signal_set.hpp>
#include <boost/asio/write.hpp>
#include <boost/asio/experimental/as_tuple.hpp>
#include <boost/asio/redirect_error.hpp>
#include <boost/asio/stream_file.hpp>
#include <boost/asio/steady_timer.hpp>
#include <boost/asio/experimental/awaitable_operators.hpp>

using namespace std::chrono_literals;

using boost::asio::awaitable;
using boost::asio::co_spawn;
using boost::asio::detached;
using boost::asio::use_awaitable_t;

namespace this_coro = boost::asio::this_coro;

using boost::asio::use_awaitable;
using boost::asio::redirect_error;

// https://github.com/chriskohlhoff/talking-async/blob/master/episode1/step_7.cpp
constexpr auto use_nothrow_awaitable = boost::asio::experimental::as_tuple(use_awaitable);

using namespace boost::asio::experimental::awaitable_operators;

// static const std::string path = "test.dat";
static const std::string path = "/dev/null";

awaitable<void> reader()
{
    auto executor = co_await this_coro::executor;

    boost::asio::steady_timer timer(executor);
    timer.expires_after(250ms);
    co_await(timer.async_wait(use_awaitable));

    boost::system::error_code ec;
    boost::asio::stream_file body(executor);
    body.open(path.c_str(), boost::asio::stream_file::read_only, ec);

    std::cout << "file opened for reading: " << ec << std::endl;

    size_t total = 0, count = 0;
    char data[1024*10];
    for (;;)
    {
        timer.expires_after(10ms);
        auto result = co_await (timer.async_wait(use_awaitable) || body.async_read_some(boost::asio::buffer(data), use_awaitable));
        if (result.index() == 0)
        {
            // std::cout << "timeout" << std::endl;
            continue;
        }

        total += std::get<size_t>(result);
        if (count++ % 10000 == 0)
            std::cout << "read: " << total/1024/1024 << " MB" << std::endl;
    }
}

constexpr size_t BS = 16*1024;
constexpr size_t TOTAL = 1024*1024*1024;

awaitable<void> writer()
{
    auto executor = co_await this_coro::executor;

    boost::system::error_code ec;
    boost::asio::stream_file stream(executor);
    stream.open(path.c_str(), boost::asio::stream_file::write_only|boost::asio::stream_file::create, ec);

    std::cout << "file opened for writing: " << ec << std::endl;

    // co_spawn(executor, reader(), detached);

    boost::asio::steady_timer timer(executor);
    std::array<char, BS> data;
    size_t i = 0, total = 0, bytes_left = TOTAL;
    while (bytes_left)
    {
        // timer.expires_after(1ms);
        // co_await timer.async_wait(use_awaitable);
        // co_await body.async_write_some(boost::asio::buffer(data), use_awaitable);
        auto written =  co_await boost::asio::async_write(stream, boost::asio::buffer(data, bytes_left), use_awaitable);        
        total += written;
        bytes_left -= written;
        if (i++ % 10000 == 0)
            std::cout << "coro    write: " << total/1024/1024 << " MB" << std::endl;
    }

    std::cout << "done writing, total=" << total << " bytes" << std::endl;
}

struct State
{
    boost::asio::stream_file stream;
    size_t bytes_left;
    size_t i = 0;
    size_t total = 0;
    std::array<char, BS> data;
};

void do_write(std::unique_ptr<State> state)
{
    if (state->bytes_left == 0)
    {
        std::cout << "done writing, total=" << state->total << " bytes" << std::endl;
        return;
    }

    // if (state->i % 10000 == 0)
    //    std::cout << "write: " << state->total/1024/1024 << " MB" << std::endl;

    auto temp = state.get();
    boost::asio::async_write(temp->stream, boost::asio::buffer(temp->data, temp->bytes_left),
        [state = std::move(state)](boost::system::error_code ec, size_t written) mutable {
        if (state->i++ % 10000 == 0)
            std::cout << "classic write: " << state->total/1024/1024 << " MB" << std::endl;

        if (ec)
            std::cerr << ec.what() << std::endl;
        else {
            state->bytes_left -= written;
            state->total += written;
            do_write(std::move(state));
        }
    });
}

void start_write(boost::asio::any_io_executor executor)
{
    auto state = std::make_unique<State>(State{boost::asio::stream_file(executor), TOTAL});
    boost::system::error_code ec;
    state->stream.open(path.c_str(), boost::asio::stream_file::write_only|boost::asio::stream_file::create, ec);
    do_write(std::move(state));
}


int main()
{
    try
    {
        boost::asio::io_context io_context(1);

#if 1
        boost::asio::signal_set signals(io_context, SIGINT, SIGTERM);
        signals.async_wait([&](auto, auto) {
            std::cout << "interrupted" << std::endl;
            io_context.stop();
        });
#endif

        // co_spawn(io_context, reader(), detached);
        for (size_t i = 0; i < 8; ++i)
        {
            co_spawn(io_context, writer(), detached);
            start_write(io_context.get_executor());
        }

        std::array<std::optional<std::thread>, 0> threads;
        for (auto& thread: threads)
        {
            thread.emplace([&](){ io_context.run(); });
        }

        io_context.run();
        io_context.stop();
        
        for (auto& thread: threads)
        {
            thread->join();
            thread.reset();
        }
    }
    catch (std::exception& e)
    {
        std::printf("Exception: %s\n", e.what());
    }
}