aboutsummaryrefslogtreecommitdiffstats
path: root/library/cpp/actors/interconnect/poller_tcp_unit.cpp
blob: 994d9070048679aad505fd6fd624b92466227888 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
#include "poller_tcp_unit.h"

#if !defined(_win_) && !defined(_darwin_)
#include "poller_tcp_unit_epoll.h"
#endif

#include "poller_tcp_unit_select.h"
#include "poller.h"

#include <library/cpp/actors/prof/tag.h>
#include <library/cpp/actors/util/intrinsics.h>

#if defined _linux_
#include <pthread.h>
#endif

namespace NInterconnect {
    TPollerUnit::TPtr
    TPollerUnit::Make(bool useSelect) {
#if defined(_win_) || defined(_darwin_)
        Y_UNUSED(useSelect);
        return TPtr(new TPollerUnitSelect);
#else
        return useSelect ? TPtr(new TPollerUnitSelect) : TPtr(new TPollerUnitEpoll);
#endif
    }

    TPollerUnit::TPollerUnit()
        : StopFlag(true)
        , ReadLoop(TThread::TParams(IdleThread<false>, this).SetName("network read"))
        , WriteLoop(TThread::TParams(IdleThread<true>, this).SetName("network write"))
    {
    }

    TPollerUnit::~TPollerUnit() {
        if (!AtomicLoad(&StopFlag))
            Stop();
    }

    void
    TPollerUnit::Start() {
        AtomicStore(&StopFlag, false);
        ReadLoop.Start();
        WriteLoop.Start();
    }

    void
    TPollerUnit::Stop() {
        AtomicStore(&StopFlag, true);
        ReadLoop.Join();
        WriteLoop.Join();
    }

    template <>
    TPollerUnit::TSide&
    TPollerUnit::GetSide<false>() {
        return Read;
    }

    template <>
    TPollerUnit::TSide&
    TPollerUnit::GetSide<true>() {
        return Write;
    }

    void
    TPollerUnit::StartReadOperation(
        const TIntrusivePtr<TSharedDescriptor>& stream,
        TFDDelegate&& operation) {
        Y_DEBUG_ABORT_UNLESS(stream);
        if (AtomicLoad(&StopFlag))
            return;
        GetSide<false>().InputQueue.Push(TSide::TItem(stream, std::move(operation)));
    }

    void
    TPollerUnit::StartWriteOperation(
        const TIntrusivePtr<TSharedDescriptor>& stream,
        TFDDelegate&& operation) {
        Y_DEBUG_ABORT_UNLESS(stream);
        if (AtomicLoad(&StopFlag))
            return;
        GetSide<true>().InputQueue.Push(TSide::TItem(stream, std::move(operation)));
    }

    template <bool IsWrite>
    void*
    TPollerUnit::IdleThread(void* param) {
    // TODO: musl-libc version of `sched_param` struct is for some reason different from pthread
    // version in Ubuntu 12.04
#if defined(_linux_) && !defined(_musl_)
        pthread_t threadSelf = pthread_self();
        sched_param sparam = {20};
        pthread_setschedparam(threadSelf, SCHED_FIFO, &sparam);
#endif

        static_cast<TPollerUnit*>(param)->RunLoop<IsWrite>();
        return nullptr;
    }

    template <>
    void
    TPollerUnit::RunLoop<false>() {
        NProfiling::TMemoryTagScope tag("INTERCONNECT_RECEIVED_DATA");
        while (!AtomicLoad(&StopFlag))
            ProcessRead();
    }

    template <>
    void
    TPollerUnit::RunLoop<true>() {
        NProfiling::TMemoryTagScope tag("INTERCONNECT_SEND_DATA");
        while (!AtomicLoad(&StopFlag))
            ProcessWrite();
    }

    void
    TPollerUnit::TSide::ProcessInput() {
        if (!InputQueue.IsEmpty())
            do {
                auto sock = InputQueue.Top().first->GetDescriptor();
                if (!Operations.emplace(sock, std::move(InputQueue.Top())).second)
                    Y_ABORT("Descriptor is already in pooler.");
            } while (InputQueue.Pop());
    }
}