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>
 
// C++.
#include <unordered_map>
#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 () = default;
 
        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;
        }
 
        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;
 
        Reactor* _reactor;
    };
}
 
#endif