9 IoContextPool

IoContextPool manages a fixed size pool of boost asio io_context objects where each io_context is run by a single worker thread. The number of io_contexts (and threads) is passed in the constructor and cannot be changed over time.

A thread-safe function is provided to get the "next" io_context (in round robin).

A boost::asio::io_context::work object is used to avoid having a thread exit from run() when there are no more jobs to process.

Public API for application programmers

cxThread.h provides the following API to create and close an IoContextPool and to access the default IoContextPool (which is a singleton provided by cxThread).

class IoContextPool;

// numThreads must be nonnegative.  
// If 0 then the number of threads is set to std::thread::hardware_concurrency()
cxThread_API IoContextPool* CreateIoContextPool(int numThreads);
cxThread_API void Close(IoContextPool*);

// If this is not called then the default IoContextPool uses std::thread::hardware_concurrency() threads.
// If this function is used to set a different number of threads then it must be called at most once and before the 
// first call to GetTheDefaultIoContextPool()
cxThread_API void SeTheDefaultIoContextPoolSize(int numThreads);

cxThread_API IoContextPool* GetTheDefaultIoContextPool();

Note that these functions may sometimes be used by CEDA application developers. They avoid a dependency on boost, which otherwise complicates the packaging of ceda-core.

Singleton for the IoContextPool?

cxThread provides a singleton IoContextPool which is obtained using the following free function:

cxThread_API IoContextPool* GetTheDefaultIoContextPool();

This typically uses std::thread::hardware_concurrency() threads. It is only created on demand, so if GetTheDefaultIoContextPool() is never called, no threads and io_contexts are created.

The singleton is a static variable which automatically destructs after main() completes. This is when join() is called on the threads used by the default IoContextPool. But that shouldn't be a problem!

However, singletons are often problematic so we have to consider any negative repercussions:

• Testing may not be as good, for example you can't easily test for a race condition on first access to a singleton because that only happens once when a process runs.
• When the IoContextPool is closed it stops all the threads. This is important, how do we ensure that happens with a singleton?
• What happens with a singleton when there are multiple libraries independently static linking cxThread? See Singleton class in a static library on Stack Overflow. But is this really a problem? We already use singletons in cxObject for things like the reflected class registry, so we have that problem anyway. Indeed for this reason using DLLs instead of static libraries is preferable, unless you're creating a single self-contained executable with no DLLs at all.

Passing a pointer to an IoContextPool into functions

In classes that need an IoContextPool, we will provide it using a pointer which defaults to nullptr. If a IoContextPool is not provided then the singleton IoContextPool is used.

For example the function to open an LSS takes a pointer to a IoContextPool which defaults to nullptr:

cxLss_API ILogStructuredStore* CreateOrOpenLSS(
    ConstStringZ lssPath,
    ConstStringZ deltasDirPath,
    bool& createdNew,
    EOpenMode openMode,
    const LssSettings& settings,
    IoContextPool* ioContextPool = nullptr);

IoContextPool.h

// IoContextPool.h
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2020

#pragma once
#ifndef Ceda_cxThread_IoContextPool_H
#define Ceda_cxThread_IoContextPool_H

#include "cxThread.h"
#include <boost/asio.hpp>
#include <memory>
#include <atomic>
#include <functional>

namespace ceda
{
// class IoContextPool2;
// cxThread_API IoContextPool2* CreateIoContextPool2(int numThreads);
// cxThread_API void Close(IoContextPool2* pool);
// cxThread_API boost::asio::io_context& GetRoundRobinIoContext2(IoContextPool2* pool);




cxThread_API boost::asio::io_context& GetDefaultIoContext();

cxThread_API boost::asio::io_context& GetRoundRobinIoContext(IoContextPool* pool);

class AsyncTaskExecuter;
cxThread_API std::shared_ptr<AsyncTaskExecuter> MakeAsyncTaskExecuter(IoContextPool* pool);
cxThread_API const std::atomic<bool>& GetAbortFlag(const AsyncTaskExecuter& e);
cxThread_API void Stop(AsyncTaskExecuter& e);
cxThread_API void Post(AsyncTaskExecuter& e, std::function<void()> task);

// A bit of a hack to allows emscripten to be used without threads
cxThread_API void DisableThreads(boost::asio::io_context* ioContext);

} // namespace ceda

#endif // include guard

IoContextPool.cpp

// IoContextPool.cpp
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2020

#include "IoContextPool.h"
#include "cxThread.h"
#include "ThreadName.h"
#include "ManualResetEvent.h"
#include "Ceda/cxUtils/StringStream.h"
#include "Ceda/cxUtils/Tracer.h"
#include <stdexcept>
#include <iostream>
#include <list>
#include <vector>
#include <memory>
#include <atomic>
#include <algorithm>

namespace ceda
{
#if 0
struct SingleThreadedIoContext2
{
    SingleThreadedIoContext2() : workGuard(io_context.get_executor())
    {
        thread = std::thread([this]() { io_context.run(); });
    }
    ~SingleThreadedIoContext2()
    {
        workGuard.reset();
        thread.join();
    }
    boost::asio::io_context io_context;
    boost::asio::executor_work_guard<boost::asio::io_context::executor_type> workGuard;
    std::thread thread;
};

class IoContextPool2
{
public:
    explicit IoContextPool2(int size) : contexts_(size) {}
    boost::asio::io_context& get_round_robin_io_context()
    {
        return contexts_[(next_++) % contexts_.size()].io_context;
    }
private:
    std::vector<SingleThreadedIoContext2> contexts_;
    std::atomic_ullong next_;
};

cxThread_API IoContextPool2* CreateIoContextPool2(int numThreads)
{
    return new IoContextPool2(numThreads);
}

cxThread_API void Close(IoContextPool2* pool)
{
    delete pool;
}

cxThread_API boost::asio::io_context& GetRoundRobinIoContext2(IoContextPool2* pool)
{
    return pool->get_round_robin_io_context();
}
#endif



#if 1
struct SingleThreadedIoContext
{
    SingleThreadedIoContext() :
        io_context(1),
        workGuard(io_context.get_executor())
    {
        //thread = std::thread([this]() { io_context.run(); });
    }

    SingleThreadedIoContext( const SingleThreadedIoContext& ) = delete; // non construction-copyable
    SingleThreadedIoContext& operator=( const SingleThreadedIoContext& ) = delete; // non copyable

    void Start(int i)
    {
        this->index = i;
        thread = std::thread(
            [this]()
            {
                SetThreadName(cxMakeString2("IoContextPool thread " << index));
                Tracer() << "-- Started thread #" << index << "  " << std::hash<std::thread::id>()(std::this_thread::get_id()) << '\n';
                io_context.run();
            });
    }

    ~SingleThreadedIoContext()
    {
        workGuard.reset();
        thread.join();
    }

    int index;
    boost::asio::io_context io_context;
    boost::asio::executor_work_guard<boost::asio::io_context::executor_type> workGuard;
    std::thread thread;
};

class IoContextPool
{
public:
    explicit IoContextPool(int poolSize) : io_contexts_(poolSize) 
    {
        if (poolSize <= 0)
            throw std::runtime_error("IoContextPool size must be > 0");

        Tracer() << "Creating IoContextPool with " << poolSize << " threads\n";

        for (int i=0 ; i < poolSize ; ++i)
        {
            io_contexts_[i].Start(i);
        }
    }

    ~IoContextPool()
    {
        Tracer() << "Destroyed IoContextPool\n";
    }

    IoContextPool( const IoContextPool& ) = delete; // non construction-copyable
    IoContextPool& operator=( const IoContextPool& ) = delete; // non copyable

    ssize_t size() const { return io_contexts_.size(); }
    SingleThreadedIoContext& operator[](ssize_t i)
    {
        cxAssert(0 <= i && i < io_contexts_.size());
        return io_contexts_[i];
    }

    // Get an io_context to use (these are provided round robin from the pool).
    // This function is threadsafe
    boost::asio::io_context& get_round_robin_io_context()
    {
        // Use a round-robin scheme to choose the next io_context to use.
        // The increment is atomic so this function is thread-safe
        return io_contexts_[(next_++) % io_contexts_.size()].io_context;
    }

private:
    std::vector<SingleThreadedIoContext> io_contexts_;

    // The next io_context to use for a connection.
    std::atomic_ullong next_;
};
#endif

static bool s_enableThreads = true;
static boost::asio::io_context* s_ioContext = nullptr;        // boost::asio::io_context to be used when threads are not enabled

cxThread_API bool ThreadsAreEnabled()
{
    return s_enableThreads;
}

cxThread_API void DisableThreads(boost::asio::io_context* ioContext)
{
    s_enableThreads = false;
    s_ioContext = ioContext;
}

cxThread_API IoContextPool* CreateIoContextPool(int numThreads)
{
    if (numThreads == 0)
    {
        // This is part of the C++11 standard.
        numThreads = std::thread::hardware_concurrency();
    }

    return new IoContextPool(numThreads);
}

cxThread_API void Close(IoContextPool* pool)
{
    cxAssert(pool);
    delete pool;
}

cxThread_API boost::asio::io_context& GetRoundRobinIoContext(IoContextPool* pool)
{
    if (s_ioContext)
    {
        return *s_ioContext;
    }

    cxAssert(pool);
    return pool->get_round_robin_io_context();
}

int s_defaultIoContextPoolSize = 0;     // means use std::thread::hardware_concurrency()

cxThread_API void SetTheDefaultIoContextPoolSize(int poolSize)
{
    cxAssert(poolSize > 0);
    cxAssert(s_defaultIoContextPoolSize == 0);
    s_defaultIoContextPoolSize = poolSize;
}

int GetDefaultIoContextPoolSize()
{
    if (s_defaultIoContextPoolSize == 0)
    {
        s_defaultIoContextPoolSize = std::min(4, (int)std::thread::hardware_concurrency());
    }
    return s_defaultIoContextPoolSize;
}

// The default, singleton IoContextPool
cxThread_API IoContextPool* GetTheDefaultIoContextPool()
{
    //Tracer() << "GetTheDefaultIoContextPool()\n";
    static IoContextPool instance(GetDefaultIoContextPoolSize());
    return &instance;
}

cxThread_API boost::asio::io_context& GetDefaultIoContext()
{
    if (s_ioContext)
    {
        return *s_ioContext;
    }
    
    return GetTheDefaultIoContextPool()->get_round_robin_io_context();
}

///////////////////////////////////////////////////////////////////////////////////////////////////
// AsyncTaskExecuter

/*
template<typename Task>
void PostTask(boost::asio::io_context& context, std::atomic<bool>& abort, Task task)
{
    boost::asio::post(context, [&abort, task]()
        {
            if (!abort)
            {
                task();
            }
        });    
}

void Stop(IoContextPool& pool, std::atomic<bool>& abort)
{
    abort = true;

    std::atomic<int> count = (int)pool.size();
    ManualResetEvent evt;
        
    // Post tasks to every context and signal evt once they have all been processed
    auto n = pool.size();
    for (int i=0 ; i < n ; ++i)
    {
        boost::asio::post(pool[i].io_context, [&count, &evt]()
            {
                if (--count == 0)
                {
                    evt.Signal();
                }
            });        
    }
    evt.Wait();
}
*/

class AsyncTaskExecuter final :  public std::enable_shared_from_this<AsyncTaskExecuter>
{
public:
    AsyncTaskExecuter(IoContextPool& pool) :
        pool_(pool)
    {
    }

    const std::atomic<bool>& Abort() const { return abort_; }

    template<typename Task>
    void PostTask(Task task)
    {
        auto self(shared_from_this());
        boost::asio::post(pool_.get_round_robin_io_context(), [this, self, task]()
            {
                if (!abort_)
                {
                    task();
                }
            });    
    }

    void Stop()
    {
        abort_ = true;

        int n = (int)pool_.size();
        std::atomic<int> count = n;
        ManualResetEvent evt;
        
        // Post tasks to every context and signal evt once they have all processed the task.
        // Since each context is run by a single thread, this implies that none of the contexts
        // are still running any of the tasks posted to PostTask().
        for (int i=0 ; i < n ; ++i)
        {
            auto self(shared_from_this());
            boost::asio::post(pool_[i].io_context, [self, &count, &evt]()
                {
                    if (--count == 0)
                    {
                        evt.Signal();
                    }
                });        
        }
        evt.Wait();
    }

private:
    IoContextPool& pool_;
    std::atomic<bool> abort_ = false;
};

cxThread_API std::shared_ptr<AsyncTaskExecuter> MakeAsyncTaskExecuter(IoContextPool* pool)
{
    cxAssert(pool);
    return std::make_shared<AsyncTaskExecuter>(*pool);
}

cxThread_API const std::atomic<bool>& GetAbortFlag(const AsyncTaskExecuter& e)
{
    return e.Abort();
}

cxThread_API void Stop(AsyncTaskExecuter& e)
{
    return e.Stop();
}

cxThread_API void Post(AsyncTaskExecuter& e, std::function<void()> task)
{
    e.PostTask(task);
}

} // namespace ceda