wake.cc

#include <unistd.h>
#include <list>
#include <thread>
#include <cstdint>
#include <iostream>
#include <mutex>
#include <atomic>

#include <condition_variable>

#include <sys/eventfd.h>
#include <sys/types.h>
#include <sys/socket.h>


class IPair {
public:
    ~IPair()
    {
        close(In), close(Ou);
    }

    void Push(uint64_t val)
    {
        auto wr = write(Ou, &val, sizeof(val));

        if (wr != sizeof(val)) throw "Cannot put bytes to evd";
    }

    uint64_t Pop() const
    {
        uint64_t val = 0;

        auto rd = read(In, &val, sizeof(val));

        if (rd != sizeof(val)) throw "Cannot get bytes from evd";

        return val;
    }

protected:
    int     In  =    -1;
    int     Ou  =   -1;
};


class TPipes: public IPair {
public:
    TPipes()
    {
        int fd[2];

        if (auto rv = pipe(fd)) {

            throw "Cannot create pipes pair";
        }

        In = fd[0], Ou = fd[1];
    }
};


class TSocks: public IPair {
public:
    TSocks()
    {
        int fd[2];

        if (auto rv = socketpair(AF_UNIX, SOCK_STREAM, 0, fd)) {

            throw "Cannot create pipes pair";
        }

        In = fd[0], Ou = fd[1];
    }
};


class TEvd {
public:
    TEvd(): Fd(eventfd(0, 0)) { }

    ~TEvd()
    {
        close(Fd), Fd = -1;
    }

    void Push(uint64_t val) const
    {
        auto wr = write(Fd, &val, sizeof(val));

        if (wr != sizeof(val)) throw "Cannot put bytes to evd";
    }

    uint64_t Pop() const
    {
        uint64_t val = 0;

        auto rd = read(Fd, &val, sizeof(val));

        if (rd != sizeof(val)) throw "Cannot get bytes from evd";

        return val;
    }

private:
    int     Fd      = -1;    
};


class TCond {
    using TVar = std::condition_variable;

public:
    void Push(uint64_t val)
    {
        {
            std::lock_guard<std::mutex> g(Loc);

            Vol += val;
        }

        Var.notify_one();
    }

    uint64_t Pop() noexcept
    {
        uint64_t val = 0;

        {
            std::unique_lock<std::mutex> g(Loc);

            if (val == 0) {

                Var.wait(g, [&] { return Vol > 0;});
            }

            return std::swap(val, Vol), val;
        }
    }

private:
    std::mutex  Loc;
    TVar        Var;

    uint64_t    Vol =   0;
};


template<typename TLock>
void Feed(TLock *lock, size_t num) noexcept
{
    for (size_t z = 0; z < num; z++) lock->Push(1);
}

template<typename TLock>
int Execute(size_t thr, size_t num) noexcept
{
    TLock sync;
    
    std::list<std::thread>  all;

    for (size_t z = 0; z < thr; z++) {

        all.emplace_back(Feed<TLock>, &sync, num);
    }

    for (size_t z = 0; z < num * thr; ) z += sync.Pop();

    while (!all.empty()) all.back().join(), all.pop_back();

    return 0;
}

int main(int argc, char *argv[])
{
    if (argc != 4) {

        std::cerr
            << "Usage: " << argv[0] 
            << " (cond|pipe|sock|evfd) thr num"
            << std::endl;

    } else {
        size_t thr = atoi(argv[2]), num = atoi(argv[3]);

        const std::string kind(argv[1]);

        if (kind == "cond") return Execute<TCond>(thr, num);
        if (kind == "pipe") return Execute<TPipes>(thr, num);
        if (kind == "sock") return Execute<TSocks>(thr, num);
        if (kind == "evfd") return Execute<TEvd>(thr, num);

        std::cerr << "Unknown wake mode " << kind << std::endl;

        return 1;
    }
}