path: root/library/cpp/coroutine/engine/poller.cpp

#include "poller.h" 
#include "sockmap.h" 
 
#include <util/memory/smallobj.h> 
#include <util/generic/intrlist.h> 
#include <util/generic/singleton.h> 
#include <util/system/env.h>
#include <util/string/cast.h>
 
namespace { 
    using TChange = IPollerFace::TChange;
    using TEvent = IPollerFace::TEvent;
    using TEvents = IPollerFace::TEvents;
 
    template <class T> 
    class TUnsafeBuf { 
    public: 
        TUnsafeBuf() noexcept
            : L_(0) 
        { 
        } 
 
        T* operator~() const noexcept {
            return B_.Get(); 
        } 
 
        size_t operator+() const noexcept {
            return L_; 
        } 
 
        void Reserve(size_t len) {
            len = FastClp2(len); 
 
            if (len > L_) { 
                B_.Reset(new T[len]); 
                L_ = len; 
            } 
        } 
 
    private: 
        TArrayHolder<T> B_; 
        size_t L_; 
    }; 
 

    template <class T> 
    class TVirtualize: public IPollerFace { 
    public: 
        TVirtualize(EContPoller pollerEngine)
            : PollerEngine_(pollerEngine)
        {
        }

        void Set(const TChange& c) override {
            P_.Set(c); 
        } 
 
        void Wait(TEvents& events, TInstant deadLine) override {
            P_.Wait(events, deadLine); 
        } 
 
        EContPoller PollEngine() const override {
            return PollerEngine_;
        }
    private: 
        T P_; 
        const EContPoller PollerEngine_;
    }; 
 

    template <class T> 
    class TPoller { 
        using TInternalEvent = typename T::TEvent;
 
    public: 
        TPoller() {
            E_.Reserve(1); 
        } 
 
        void Set(const TChange& c) {
            P_.Set(c.Data, c.Fd, c.Flags); 
        } 
 
        void Reserve(size_t size) {
            E_.Reserve(size);
        }

        void Wait(TEvents& events, TInstant deadLine) {
            const size_t ret = P_.WaitD(~E_, +E_, deadLine); 
 
            events.reserve(ret);

            for (size_t i = 0; i < ret; ++i) { 
                const TInternalEvent* ie = ~E_ + i; 
 
                const TEvent e = { 
                    T::ExtractEvent(ie), 
                    T::ExtractStatus(ie), 
                    (ui16)T::ExtractFilter(ie), 
                }; 
 
                events.push_back(e); 
            } 
 
            E_.Reserve(ret + 1); 
        } 
 
    private: 
        T P_; 
        TUnsafeBuf<TInternalEvent> E_; 
    }; 
 

    template <class T> 
    class TIndexedArray { 
        struct TVal:
            public T,
            public TIntrusiveListItem<TVal>,
            public TObjectFromPool<TVal>
        {
            // NOTE Constructor must be user-defined (and not =default) here
            // because TVal objects are created in the UB-capable placement
            // TObjectFromPool::new operator that stores data in a memory
            // allocated for the object. Without user defined constructor
            // zero-initialization takes place in TVal() expression and the
            // data is overwritten.
            TVal() {
            } 
        }; 
 
        typedef TIntrusiveList<TVal> TListType;
 
    public: 
        typedef typename TListType::TIterator TIterator;
        typedef typename TListType::TConstIterator TConstIterator;
 
        TIndexedArray()
            : P_(TMemoryPool::TExpGrow::Instance(), TDefaultAllocator::Instance()) 
        { 
        } 
 
        TIterator Begin() noexcept {
            return I_.Begin(); 
        } 
 
        TIterator End() noexcept {
            return I_.End(); 
        } 
 
        TConstIterator Begin() const noexcept {
            return I_.Begin(); 
        } 
 
        TConstIterator End() const noexcept {
            return I_.End(); 
        } 
 
        T& operator[](size_t i) {
            return *Get(i); 
        } 
 
        T* Get(size_t i) {
            TValRef& v = V_.Get(i); 
 
            if (Y_UNLIKELY(!v)) {
                v.Reset(new (&P_) TVal()); 
                I_.PushFront(v.Get()); 
            } 
 
            Y_PREFETCH_WRITE(v.Get(), 1); 
 
            return v.Get(); 
        } 
 
        void Erase(size_t i) noexcept {
            V_.Get(i).Destroy(); 
        } 
 
        size_t Size() const noexcept {
            return I_.Size(); 
        } 
 
    private: 
        using TValRef = THolder<TVal>;
        typename TVal::TPool P_; 
        TSocketMap<TValRef> V_; 
        TListType I_;
    }; 
 

    inline short PollFlags(ui16 flags) noexcept {
        short ret = 0; 
 
        if (flags & CONT_POLL_READ) { 
            ret |= POLLIN; 
        } 
 
        if (flags & CONT_POLL_WRITE) { 
            ret |= POLLOUT; 
        } 
 
#if defined(_linux_)
        if (flags & CONT_POLL_RDHUP) {
            ret |= POLLRDHUP;
        }
#endif

        return ret; 
    } 
 

    class TPollPoller { 
    public: 
        size_t Size() const noexcept {
            return S_.Size(); 
        } 
 
        template <class T> 
        void Build(T& t) const {
            for (TFds::TConstIterator it = S_.Begin(); it != S_.End(); ++it) { 
                t.Set(*it); 
            } 

            t.Reserve(Size());
        } 
 
        void Set(const TChange& c) {
            if (c.Flags) { 
                S_[c.Fd] = c; 
            } else { 
                S_.Erase(c.Fd); 
            } 
        } 
 
        void Wait(TEvents& events, TInstant deadLine) {
            T_.clear(); 
            T_.reserve(Size()); 
 
            for (TFds::TConstIterator it = S_.Begin(); it != S_.End(); ++it) { 
                const pollfd pfd = { 
                    it->Fd, 
                    PollFlags(it->Flags), 
                    0, 
                }; 
 
                T_.push_back(pfd); 
            } 
 
            const ssize_t ret = PollD(T_.data(), (nfds_t) T_.size(), deadLine);
 
            if (ret <= 0) {
                return; 
            } 
 
            events.reserve(T_.size());

            for (size_t i = 0; i < T_.size(); ++i) {
                const pollfd& pfd = T_[i]; 
                const short ev = pfd.revents; 
 
                if (!ev) { 
                    continue; 
                } 
 
                int status = 0; 
                ui16 filter = 0; 
 
                // We are perfectly fine with an EOF while reading a pipe or a unix socket
                if ((ev & POLLIN) || (ev & POLLHUP) && (pfd.events & POLLIN)) {
                    filter |= CONT_POLL_READ; 
                } 
 
                if (ev & POLLOUT) { 
                    filter |= CONT_POLL_WRITE; 
                } 
 
#if defined(_linux_)
                if (ev & POLLRDHUP) {
                    filter |= CONT_POLL_RDHUP;
                }
#endif

                if (ev & POLLERR) { 
                    status = EIO; 
                } else if (ev & POLLHUP && pfd.events & POLLOUT) {
                    // Only write operations may cause EPIPE
                    status = EPIPE; 
                } else if (ev & POLLNVAL) { 
                    status = EINVAL; 
                } 
 
                if (status) { 
                    filter = CONT_POLL_READ | CONT_POLL_WRITE | CONT_POLL_RDHUP;
                } 
 
                const TEvent res = { 
                    S_[pfd.fd].Data, 
                    status, 
                    filter, 
                }; 
 
                events.push_back(res); 
            } 
        } 
 
    private: 
        typedef TIndexedArray<TChange> TFds; 
        TFds S_; 
        typedef TVector<pollfd> TPollVec;
        TPollVec T_; 
    }; 
 

    class TCombinedPoller { 
        typedef TPoller<TPollerImpl<TWithoutLocking>> TDefaultPoller; 
 
    public: 
        TCombinedPoller() {
            P_.Reset(new TPollPoller()); 
        } 
 
        void Set(const TChange& c) {
            if (!P_) { 
                D_->Set(c); 
            } else { 
                P_->Set(c); 
            } 
        } 
 
        void Wait(TEvents& events, TInstant deadLine) {
            if (!P_) { 
                D_->Wait(events, deadLine); 
            } else { 
                if (P_->Size() > 200) { 
                    D_.Reset(new TDefaultPoller()); 
                    P_->Build(*D_); 
                    P_.Destroy(); 
                    D_->Wait(events, deadLine); 
                } else { 
                    P_->Wait(events, deadLine); 
                } 
            } 
        } 
 
    private: 
        THolder<TPollPoller> P_;
        THolder<TDefaultPoller> D_;
    }; 
 
    struct TUserPoller: public TString { 
        TUserPoller()
            : TString(GetEnv("USER_POLLER")) 
        { 
        } 
    }; 
} 
 
THolder<IPollerFace> IPollerFace::Default() {
    return Construct(*SingletonWithPriority<TUserPoller, 0>());
} 
 
THolder<IPollerFace> IPollerFace::Construct(TStringBuf name) {
    return Construct(name ? FromString<EContPoller>(name) : EContPoller::Default);
}
 
THolder<IPollerFace> IPollerFace::Construct(EContPoller poller) {
    switch (poller) {
    case EContPoller::Default:
    case EContPoller::Combined:
        return MakeHolder<TVirtualize<TCombinedPoller>>(EContPoller::Combined);
    case EContPoller::Select:
        return MakeHolder<TVirtualize<TPoller<TGenericPoller<TSelectPoller<TWithoutLocking>>>>>(poller);
    case EContPoller::Poll:
        return MakeHolder<TVirtualize<TPollPoller>>(poller);
    case EContPoller::Epoll:
#if defined(HAVE_EPOLL_POLLER) 
        return MakeHolder<TVirtualize<TPoller<TGenericPoller<TEpollPoller<TWithoutLocking>>>>>(poller);
#else
        return nullptr;
#endif 
    case EContPoller::Kqueue:
#if defined(HAVE_KQUEUE_POLLER)
        return MakeHolder<TVirtualize<TPoller<TGenericPoller<TKqueuePoller<TWithoutLocking>>>>>(poller);
#else
        return nullptr;
#endif 
    default:
        Y_FAIL("bad poller type");
    } 
}