netlinkmanager.hpp

Go to the documentation of this file.

 
#ifndef JOIN_FABRIC_NETLINKMANAGER_HPP
#define JOIN_FABRIC_NETLINKMANAGER_HPP
 
// libjoin.
#include <join/condition.hpp>
#include <join/socket.hpp>
#include <join/queue.hpp>
 
// C++.
#include <unordered_map>
#include <functional>
#include <memory>
#include <atomic>
 
// C.
#include <linux/rtnetlink.h>
#include <linux/netlink.h>
#include <cstdint>
#include <cstddef>
 
namespace join
{
    class NetlinkManager : private Netlink::Socket
    {
    public:
        NetlinkManager (uint32_t groups, Reactor* reactor = nullptr);
 
        NetlinkManager (const NetlinkManager&) = delete;
 
        NetlinkManager (NetlinkManager&&) = delete;
 
        NetlinkManager& operator= (const NetlinkManager&) = delete;
 
        NetlinkManager& operator= (NetlinkManager&&) = delete;
 
        virtual ~NetlinkManager ();
 
        Reactor* reactor () const noexcept;
 
    protected:
        void start ();
 
        void stop ();
 
        int sendRequest (struct nlmsghdr* nlh, bool sync, std::chrono::milliseconds timeout = std::chrono::seconds (5));
 
        template <class Rep, class Period>
        int waitResponse (ScopedLock<Mutex>& lock, uint32_t seq, std::chrono::duration<Rep, Period> timeout)
        {
            auto inserted = _pending.emplace (seq, std::make_unique<PendingRequest> ());
            if (!inserted.second)
            {
                lastError = make_error_code (Errc::OperationFailed);
                return -1;
            }
 
            if (!inserted.first->second->cond.timedWait (lock, timeout))
            {
                _pending.erase (inserted.first);
                lastError = make_error_code (Errc::TimedOut);
                return -1;
            }
 
            if (inserted.first->second->error != 0)
            {
                int err = inserted.first->second->error;
                _pending.erase (inserted.first);
                lastError = std::error_code (err, std::generic_category ());
                return -1;
            }
 
            _pending.erase (inserted.first);
            return 0;
        }
 
        template <typename Func, typename... Args>
        void pushJob (Func&& func, Args&&... args) noexcept
        {
            Job job;
            job.func = std::bind (std::forward<Func> (func), std::forward<Args> (args)...);
 
            if (_reactor->isReactorThread ())
            {
                job.func ();
                return;
            }
 
            _jobs.push (&job);
 
            uint64_t v = 1;
            [[maybe_unused]] ssize_t bytes = ::write (_wakeup, &v, sizeof (v));
 
            Backoff backoff;
            while (!job.done.load (std::memory_order_acquire))
            {
                backoff ();
            }
        }
 
        virtual void onReceive (int fd) override final;
 
        virtual void onMessage (struct nlmsghdr* nlh) = 0;
 
        void notifyRequest (uint32_t seq, int error = 0);
 
        static void addAttributes (struct nlmsghdr* nlh, int type, const void* data, int alen);
 
        static struct rtattr* startNestedAttributes (struct nlmsghdr* nlh, int type);
 
        static int stopNestedAttributes (struct nlmsghdr* nlh, struct rtattr* nested);
 
        template <typename T, typename Flag>
        static Flag updateValue (T& oldVal, const T& newVal, Flag changed)
        {
            if (oldVal != newVal)
            {
                oldVal = newVal;
                return changed;
            }
            return static_cast<Flag> (0);
        }
 
        static constexpr size_t _bufferSize = 16384;
 
        std::unique_ptr<char[]> _buffer;
 
        std::atomic<uint32_t> _seq;
 
        struct PendingRequest
        {
            Condition cond;
            int error = 0;
        };
 
        std::unordered_map<uint32_t, std::unique_ptr<PendingRequest>> _pending;
 
        Mutex _syncMutex;
 
        struct Job
        {
            std::function<void ()> func;
 
            std::atomic<bool> done{false};
        };
 
        static constexpr size_t _jobQueueSize = 256;
 
        LocalMem::Mpsc::Queue<Job*> _jobs;
 
        int _wakeup = -1;
 
        Reactor* _reactor;
    };
}
 
#endif