poller.h

Go to the documentation of this file.

#ifndef ZTH_POLLER_H
#define ZTH_POLLER_H
/*
 * SPDX-FileCopyrightText: 2019-2026 Jochem Rutgers
 *
 * SPDX-License-Identifier: MPL-2.0
 */
 
#include <libzth/macros.h>
 
#ifdef __cplusplus
 
#  include <libzth/allocator.h>
#  include <libzth/waiter.h>
#  include <libzth/worker.h>
 
#  include <bitset>
#  include <memory>
#  include <stdexcept>
#  include <vector>
 
#  if __cplusplus >= 201103L
#    include <functional>
#    include <initializer_list>
#  endif
 
#  if defined(ZTH_HAVE_POLL) || defined(ZTH_HAVE_LIBZMQ)
#    define ZTH_HAVE_POLLER
#  endif
 
#  if defined(ZTH_HAVE_LIBZMQ)
#    include <zmq.h>
#  elif defined(ZTH_HAVE_POLL)
#    include <poll.h>
#  endif
 
namespace zth {
 
// Pollable
//
 
struct Pollable {
        enum EventsFlags {
                PollInIndex,
                PollOutIndex,
                PollErrIndex,
                PollPriIndex,
                PollHupIndex,
                FlagCount,
        };
 
        typedef std::bitset<FlagCount> Events;
 
        // unsigned long can be converted implicitly to bitset.
        static const unsigned long PollIn = 1UL << PollInIndex;
        static const unsigned long PollOut = 1UL << PollOutIndex;
        static const unsigned long PollErr = 1UL << PollErrIndex;
        static const unsigned long PollPri = 1UL << PollPriIndex;
        static const unsigned long PollHup = 1UL << PollHupIndex;
 
        constexpr explicit Pollable(Events const& e, void* user = nullptr) noexcept
                : user_data(user)
                , events(e)
                , revents()
        {}
 
        void* user_data;
 
        Events events;
 
        Events revents;
};
 
struct PollableFd : public Pollable {
        constexpr PollableFd(int f, Events const& e, void* user = nullptr) noexcept
                : Pollable(e, user)
#  ifdef ZTH_HAVE_LIBZMQ
                , socket()
#  endif // ZTH_HAVE_LIBZMQ
                , fd(f)
        {}
 
#  ifdef ZTH_HAVE_LIBZMQ
        constexpr PollableFd(void* s, Events const& e, void* user = nullptr) noexcept
                : Pollable(e, user)
                , socket(s)
                , fd()
        {}
 
        void* socket;
#  endif // ZTH_HAVE_LIBZMQ
 
        int fd;
};
 
 
 
// Poller base classes
//
 
class PollerInterface : public UniqueID<PollerInterface> {
public:
        virtual ~PollerInterface() override is_default
 
        virtual int add(Pollable& p) noexcept = 0;
 
#  if __cplusplus >= 201103L
        int add(std::initializer_list<std::reference_wrapper<Pollable>> l) noexcept;
#  endif
 
        virtual int remove(Pollable& p) noexcept = 0;
 
        virtual void reserve(size_t more) = 0;
 
        virtual bool empty() const noexcept = 0;
};
 
class PollerClientBase : public PollerInterface {
public:
        virtual ~PollerClientBase() override is_default
 
        typedef small_vector<Pollable*> Result;
 
        virtual Result const& poll(int timeout_ms = -1) noexcept = 0;
 
        virtual void event(Pollable& p) noexcept = 0;
};
 
class PollerServerBase : public PollerInterface {
public:
        typedef PollerClientBase Client;
 
        virtual ~PollerServerBase() override is_default
 
        virtual int poll(int timeout_ms) noexcept = 0;
 
        virtual int migrateTo(PollerServerBase& p) noexcept = 0;
 
        virtual int add(Pollable& p, Client* client) noexcept = 0;
 
        using PollerInterface::add;
 
        virtual int remove(Pollable& p, Client* client) noexcept = 0;
 
        using PollerInterface::remove;
};
 
 
 
// PollerServers
//
 
template <typename PollItem_>
class PollerServer : public PollerServerBase {
public:
        typedef PollerServerBase base;
        using base::Client;
 
        PollerServer() is_default
 
#  if __cplusplus >= 201103L
        PollerServer(PollerServer const&) = delete;
        void operator=(PollerServer const&) = delete;
        PollerServer(PollerServer&& p) = delete;
        void operator=(PollerServer&& p) = delete;
#  else
private:
        PollerServer(PollerServer const&);
        PollerServer& operator=(PollerServer const&);
 
public:
#  endif
 
        virtual ~PollerServer() override
        {
                // Call clear() in the subclass. The virtual deinit() will be called.
                zth_assert(empty());
        }
 
        virtual int migrateTo(PollerServerBase& p) noexcept override
        {
                try {
                        p.reserve(m_metaItems.size());
                } catch(...) {
                        return ENOMEM;
                }
 
                for(size_t i = 0; i < m_metaItems.size(); i++) {
                        MetaItem const& m = m_metaItems[i];
 
                        int res = p.add(*m.pollable, m.client);
                        if(res) {
                                // Rollback
                                for(size_t j = i; j > 0; j--)
                                        p.remove(
                                                *m_metaItems[j - 1U].pollable,
                                                m_metaItems[j - 1U].client);
 
                                return res;
                        }
                }
 
                clear();
 
                // Really release all memory. This object is probably not used
                // anymore.
                m_metaItems.clear_and_release();
                m_pollItems.clear_and_release();
 
                return 0;
        }
 
        virtual void reserve(size_t more) override
        {
                m_metaItems.reserve(m_metaItems.size() + more);
                m_pollItems.reserve(m_metaItems.capacity());
        }
 
        int add(Pollable& p, Client* client) noexcept final
        {
                try {
                        MetaItem m = {&p, client};
                        m_metaItems.push_back(m);
                } catch(...) {
                        return ENOMEM;
                }
 
                try {
                        // Reserve space, but do not initialize yet.
                        m_pollItems.reserve(m_metaItems.size());
                } catch(...) {
                        // Rollback.
                        m_metaItems.pop_back();
                        return ENOMEM;
                }
 
                if(client)
                        zth_dbg(io, "[%s] added pollable %p for client %s", this->id_str(), &p,
                                client->id_str());
                else
                        zth_dbg(io, "[%s] added pollable %p", this->id_str(), &p);
 
                zth_assert(m_metaItems.size() >= m_pollItems.size());
                currentWorker().waiter().wakeup();
                return 0;
        }
 
        int add(Pollable& p) noexcept final
        {
                return add(p, nullptr);
        }
 
        // cppcheck-suppress[constParameter,constParameterPointer]
        int remove(Pollable& p, Client* client) noexcept final
        {
                size_t count = m_metaItems.size();
 
                size_t i;
                for(i = count; i > 0; i--) {
                        MetaItem const& m = m_metaItems[i - 1U];
                        if(m.pollable == &p && m.client == client)
                                break;
                }
 
                if(i == 0U)
                        return ESRCH;
 
                i--;
 
                if(i < count - 1U) {
                        // Not removing the last element, fill the gap.
                        if(i < m_pollItems.size()) {
                                deinit(*m_metaItems[i].pollable, m_pollItems[i]);
                                if(i < m_pollItems.size() - 1U) {
#  if __cplusplus >= 201103L
                                        m_pollItems[i] = std::move(m_pollItems.back());
#  else
                                        m_pollItems[i] = m_pollItems.back();
#  endif
                                }
                                m_pollItems.pop_back();
                        }
 
#  if __cplusplus >= 201103L
                        m_metaItems[i] = std::move(m_metaItems.back());
#  else
                        m_metaItems[i] = m_metaItems.back();
#  endif
                        m_metaItems.pop_back();
                } else {
                        // Drop the last element.
                        remove(1U);
                }
 
                zth_dbg(io, "[%s] removed pollable %p", this->id_str(), &p);
                zth_assert(m_metaItems.size() >= m_pollItems.size());
                return 0;
        }
 
        void remove(size_t last) noexcept
        {
                if(last >= m_metaItems.size()) {
                        clear();
                } else {
                        for(size_t i = m_metaItems.size() - last; i < m_pollItems.size(); i++)
                                deinit(*m_metaItems[i].pollable, m_pollItems[i]);
 
                        m_metaItems.resize(m_metaItems.size() - last);
                        if(m_pollItems.size() > m_metaItems.size())
                                m_pollItems.resize(m_metaItems.size());
                }
        }
 
        int remove(Pollable& p) noexcept final
        {
                return remove(p, nullptr);
        }
 
        void clear() noexcept
        {
                for(size_t i = 0; i < m_pollItems.size(); i++)
                        deinit(*m_metaItems[i].pollable, m_pollItems[i]);
 
                m_metaItems.clear();
                m_pollItems.clear();
        }
 
        bool empty() const noexcept final
        {
                return m_metaItems.empty();
        }
 
        virtual int poll(int timeout_ms) noexcept override
        {
                int res = 0;
 
                if((res = mirror()))
                        return res;
 
                if(m_metaItems.empty()) {
                        return EINVAL;
                }
 
                zth_dbg(io, "[%s] polling %u items for %d ms", this->id_str(),
                        (unsigned)m_metaItems.size(), timeout_ms);
                return doPoll(timeout_ms, m_pollItems);
        }
 
 
protected:
        typedef PollItem_ PollItem;
 
        typedef small_vector<PollItem, 4> PollItemList;
 
        // cppcheck-suppress passedByValue
        void event(Pollable::Events revents, size_t index) noexcept
        {
                zth_assert(index < m_metaItems.size());
 
                MetaItem const& m = m_metaItems[index];
                m.pollable->revents = revents;
 
                if(revents.any()) {
                        zth_dbg(io, "[%s] pollable %p got event 0x%lx", this->id_str(), m.pollable,
                                revents.to_ulong());
                        if(m.client)
                                m.client->event(*m.pollable);
                }
        }
 
private:
        virtual int init(Pollable const& p, PollItem& item) noexcept = 0;
 
        virtual void deinit(Pollable const& UNUSED_PAR(p), PollItem& UNUSED_PAR(item)) noexcept {}
 
        virtual int doPoll(int timeout_ms, PollItemList& items) noexcept = 0;
 
        int mirror() noexcept
        {
                int res = 0;
 
                size_t doInit = m_pollItems.size();
                m_pollItems.resize(m_metaItems.size());
 
                for(size_t i = doInit; i < m_metaItems.size(); i++)
                        if((res = init(*m_metaItems[i].pollable, m_pollItems[i]))) {
                                m_pollItems.resize(i);
                                break;
                        }
 
                zth_assert(m_metaItems.size() == m_pollItems.size());
                return res;
        }
 
private:
        struct MetaItem {
                Pollable* pollable;
                Client* client;
        };
 
        typedef small_vector<MetaItem, PollItemList::prealloc> MetaItemList;
 
        MetaItemList m_metaItems;
 
        PollItemList m_pollItems;
};
 
 
#  ifdef ZTH_HAVE_LIBZMQ
// By default, use zmq_poll(). It allows polling everything poll() can do,
// including all ZeroMQ sockets.
 
class ZmqPoller : public PollerServer<zmq_pollitem_t> {
        ZTH_CLASS_NEW_DELETE(ZmqPoller)
public:
        typedef PollerServer<zmq_pollitem_t> base;
 
        ZmqPoller();
        virtual ~ZmqPoller() override;
 
private:
        virtual int init(Pollable const& p, zmq_pollitem_t& item) noexcept override;
        virtual int doPoll(int timeout_ms, base::PollItemList& items) noexcept override;
};
 
typedef ZmqPoller DefaultPollerServer;
 
#  elif defined(ZTH_HAVE_POLL)
// If we don't have ZeroMQ, use the OS's poll() instead.
 
class PollPoller : public PollerServer<struct pollfd> {
        ZTH_CLASS_NEW_DELETE(PollPoller)
public:
        typedef PollerServer<struct pollfd> base;
 
        PollPoller();
        virtual ~PollPoller() override;
 
private:
        virtual int init(Pollable const& p, struct pollfd& item) noexcept override;
        virtual int doPoll(int timeout_ms, base::PollItemList& items) noexcept override;
};
 
typedef PollPoller DefaultPollerServer;
 
#  else
// No poller available.
// Do provide a PollerServer, but it can only return an error upon a poll.
 
class NoPoller final : public PollerServer<int> {
        ZTH_CLASS_NEW_DELETE(NoPoller)
public:
        typedef PollerServer<int> base;
 
        NoPoller();
        ~NoPoller() final;
 
protected:
        int init(Pollable const& p, int& item) noexcept final;
        int doPoll(int timeout_ms, base::PollItemList& items) noexcept final;
};
 
typedef NoPoller DefaultPollerServer;
#  endif // No poller
 
// PollerClient
//
 
class PollerClient : public PollerClientBase {
        ZTH_CLASS_NEW_DELETE(PollerClient)
 
        typedef small_vector<Pollable*> Pollables;
 
public:
        typedef PollerClientBase base;
        using base::Result;
 
        PollerClient();
 
#  if __cplusplus >= 201103L
        // cppcheck-suppress noExplicitConstructor
        PollerClient(std::initializer_list<std::reference_wrapper<Pollable>> l);
#  endif
 
        virtual ~PollerClient() override;
 
        virtual void reserve(size_t more) override;
        virtual int add(Pollable& p) noexcept override;
        using base::add;
        virtual int remove(Pollable& p) noexcept override;
 
        virtual Result const& poll(int timeout_ms = -1) noexcept override;
        virtual void event(Pollable& p) noexcept override;
 
        bool empty() const noexcept final;
 
private:
        TimedWaitable m_wait;
 
        Pollables m_pollables;
 
        Result m_result;
};
 
typedef PollerClient Poller;
 
template <typename P>
int poll(P pollable, int timeout_ms = -1)
{
        try {
                Poller poller;
                poller.add(pollable);
 
                while(true) {
                        Poller::Result const& result = poller.poll(timeout_ms);
 
                        if(result.empty()) {
                                switch(errno) {
                                case EINTR:
                                case EAGAIN:
                                        if(timeout_ms == -1)
                                                // Retry.
                                                continue;
                                        ZTH_FALLTHROUGH
                                default:
                                        // Error.
                                        return errno;
                                }
                        }
 
                        if((result[0]->revents & result[0]->events).any()) {
                                // Got it.
                                return 0;
                        }
 
                        // Hmm, there is another reason we returned...
                        return EIO;
                }
        } catch(...) {
        }
        return ENOMEM;
}
 
} // namespace zth
#endif // __cplusplus
#endif // ZTH_POLLER_H