1
\$\begingroup\$

I need to implement async_read_some from Boost Asio. The reason is here https://stackoverflow.com/questions/59223064/implementing-behaviour-of-boostasio-for-a-class but I don't think it's relevant. Mainly because my code does not use I/O, but I want to fool OpenVPN3 into thinking it's using I/O.

async_read_some passes a buffer and a handler and returns immediately. When Asio fills the buffer it calls the handler, which delivers the data.

I created a class called AsyncChannel where, on one side, it receives an async_read_some call (stored in a Reader) and in the other it receives data (called Buffer) to be written to Reader. The job of the class is to match a read with a write. That is, fill the asio buffer with data when data is available. Since sometimes Reader's buffer can be smaller than Buffer, it becomes a little difficult to do these matches.

#include <queue>
#include <mutex>
#include <future>
#include <atomic>
#include <condition_variable>
template <class Buffer, class Reader>
class AsyncChannel
{
public:
    typedef std::function<void()> Task;
    AsyncChannel()
    {
        shouldContinue.store(true);
    }
    void emplace_reader(Reader &&reader)
    {
        {
            std::unique_lock<std::mutex> lock{readerMutex};
            readerFifo.emplace(std::forward(reader));
        }
        match();
    }

    void emplace_buffer(Buffer &&buffer)
    {
        {
            std::unique_lock<std::mutex> lock{bufferMutex};
            bufferFifo.emplace(std::forward(buffer));
        }
        match();
    }
    /*
        If both FIFOs are not empty, it means we can match a read with a write
    */
    void match()
    {
        std::unique_lock<std::mutex> readerLock{readerMutex};
        std::unique_lock<std::mutex> bufferLock{bufferMutex};
        if (!reader.empty() && !buffer.empty())
        {
            auto r = readerFifo.front();
            auto w = bufferFifo.front();
            /*
                If the read buffer has sufficient size 
                we can simply copy the write buffer
            */
            if (r->size() <= w.size())
            {
                readerFifo.pop();
                bufferFifo.pop();
                readerLock.unlock();
                bufferLock.unlock();
                {
                    std::unique_lock<std::mutex> lock{tasksMutex};
                    tasks.emplace([r, w]() {
                        //Copies all bytes from w into r (not cheap)
                        r.receive(w);
                        //Finally calls the Reader handler
                        r.deliver();
                    });
                }
                tasksConditionVariable.notify_all();
            }
            /*
                Otherwise, we can copy only up to r.size() of w for each r.
                Do not pop buffer since it still has data to be read on next
                iteration
            */
            else
            {
                readerFifo.pop();
                readerLock.unlock();
                bufferLock.unlock();
                {
                    std::unique_lock<std::mutex> lock{tasksMutex};
                    tasks.emplace([r, w]() {
                        //Copies r.size() bytes from w into r (not cheap)
                        r.receive(w, r.size());
                        //Makes w point to the non consumed part of its buffer
                        w.consumed(r.size());
                        //Finally calls the Reader handler
                        r.deliver();
                    });
                }
            }
        }
    }

    void stop()
    {
        shouldContinue.store(false);
    }

    void taskExecutor()
    {
        while (shouldContinue.load())
        {
            std::unique_lock<std::mutex> lock{tasksMutex};
            //Only unblocks when tasks is not empty
            tasksConditionVariable.wait(lock, [] { return !tasks.empty() });
            {
                //Check again because it could have had an element in between here and wait
                if (!tasks.empty())
                {
                    auto &task = tasks.front();
                    task();
                }
            }
        }
    }

    ~AsyncChannel()
    {
        //Blocks until taskExecutor loop stops. It's not good do block but what can I do?
        stop();
    }

private:
    //TODO: add size control to these queues so it won't eat all RAM in case reader or writer stop emplacing
    std::queue<Buffer> bufferFifo;
    std::queue<Reader> readerFifo;
    std::queue<Task> tasks;
    std::mutex tasksMutex;
    std::condition_variable tasksConditionVariable;
    std::mutex bufferMutex;
    std::mutex readerMutex;
    std::atomic<bool> shouldContinue;
};

I'd like to know if this code is efficient in doing the matches of a Reader with a Buffer and if anyone can think of improvements. Note that when AsyncChannel is destructed, it first needs to wait for a copy to finish. This could be prevented by, instead of running a task executor, spawning std::asyncs, but I've heard that spawning lots of std::asyncs is much more costly than a single-threaded task executor.

Note that I have to use a not cheap memory copy. I don't think it's possible to exchange Asio's buffer with a Buffer that came from the other side, so a not cheap memory copy is needed.

\$\endgroup\$
  • 1
    \$\begingroup\$ It is hard to review - you need examples of the Buffer and Reader. Also codereview is for reviewing a working code. You didn't test this code. You never pop elements from tasks queue - you only stack em up. Either you perform only first task repeatedly or you miss tasks from time to time and repeat certain tasks at random. At any rate this is a memory leak. \$\endgroup\$ – ALX23z Dec 10 '19 at 4:13

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.