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
|
#if defined(OS_LINUX)
#include "Epoll.h"
#include <Common/Exception.h>
#include <Common/Stopwatch.h>
#include <base/defines.h>
#include <unistd.h>
namespace DB
{
namespace ErrorCodes
{
extern const int EPOLL_ERROR;
extern const int LOGICAL_ERROR;
}
Epoll::Epoll() : events_count(0)
{
epoll_fd = epoll_create1(0);
if (epoll_fd == -1)
throwFromErrno("Cannot open epoll descriptor", DB::ErrorCodes::EPOLL_ERROR);
}
Epoll::Epoll(Epoll && other) noexcept : epoll_fd(other.epoll_fd), events_count(other.events_count.load())
{
other.epoll_fd = -1;
}
Epoll & Epoll::operator=(Epoll && other) noexcept
{
epoll_fd = other.epoll_fd;
other.epoll_fd = -1;
events_count.store(other.events_count.load());
return *this;
}
void Epoll::add(int fd, void * ptr, uint32_t events)
{
epoll_event event;
event.events = events | EPOLLPRI;
if (ptr)
event.data.ptr = ptr;
else
event.data.fd = fd;
++events_count;
if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd, &event) == -1)
throwFromErrno("Cannot add new descriptor to epoll", DB::ErrorCodes::EPOLL_ERROR);
}
void Epoll::remove(int fd)
{
--events_count;
if (epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, nullptr) == -1)
throwFromErrno("Cannot remove descriptor from epoll", DB::ErrorCodes::EPOLL_ERROR);
}
size_t Epoll::getManyReady(int max_events, epoll_event * events_out, int timeout) const
{
if (events_count == 0)
throw Exception(ErrorCodes::LOGICAL_ERROR, "There are no events in epoll");
Stopwatch watch;
int ready_size;
while (true)
{
ready_size = epoll_wait(epoll_fd, events_out, max_events, timeout);
/// If `ready_size` = 0, it's timeout.
if (ready_size < 0)
{
if (errno == EINTR)
{
if (timeout >= 0)
{
timeout = std::max(0, static_cast<int>(timeout - watch.elapsedMilliseconds()));
watch.restart();
}
continue;
}
else
throwFromErrno("Error in epoll_wait", DB::ErrorCodes::EPOLL_ERROR);
}
else
break;
}
return ready_size;
}
Epoll::~Epoll()
{
if (epoll_fd != -1)
{
int err = close(epoll_fd);
chassert(!err || errno == EINTR);
}
}
}
#endif
|
