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_FAIL("Descriptor is already in pooler.");
} while (InputQueue.Pop());
}
}
|
