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
|
#include <Processors/Executors/PollingQueue.h>
#if defined(OS_LINUX)
#include <Common/Exception.h>
#include <base/defines.h>
#include <sys/epoll.h>
#include <unistd.h>
#include <fcntl.h>
#include <IO/WriteBufferFromString.h>
#include <IO/Operators.h>
namespace DB
{
namespace ErrorCodes
{
extern const int CANNOT_OPEN_FILE;
extern const int CANNOT_READ_FROM_SOCKET;
extern const int LOGICAL_ERROR;
}
PollingQueue::PollingQueue()
{
if (-1 == pipe2(pipe_fd, O_NONBLOCK))
throwFromErrno("Cannot create pipe", ErrorCodes::CANNOT_OPEN_FILE);
epoll.add(pipe_fd[0], pipe_fd);
}
PollingQueue::~PollingQueue()
{
int err;
err = close(pipe_fd[0]);
chassert(!err || errno == EINTR);
err = close(pipe_fd[1]);
chassert(!err || errno == EINTR);
}
void PollingQueue::addTask(size_t thread_number, void * data, int fd)
{
std::uintptr_t key = reinterpret_cast<uintptr_t>(data);
if (tasks.contains(key))
throw Exception(ErrorCodes::LOGICAL_ERROR, "Task {} was already added to task queue", key);
tasks[key] = TaskData{thread_number, data, fd};
epoll.add(fd, data);
}
static std::string dumpTasks(const std::unordered_map<std::uintptr_t, PollingQueue::TaskData> & tasks)
{
WriteBufferFromOwnString res;
res << "Tasks = [";
for (const auto & task : tasks)
{
res << "(id " << task.first << " thread " << task.second.thread_num << " ptr ";
writePointerHex(task.second.data, res);
res << " fd " << task.second.fd << ")";
}
res << "]";
return res.str();
}
PollingQueue::TaskData PollingQueue::wait(std::unique_lock<std::mutex> & lock)
{
if (is_finished)
return {};
lock.unlock();
epoll_event event;
event.data.ptr = nullptr;
epoll.getManyReady(1, &event, -1);
lock.lock();
if (event.data.ptr == pipe_fd)
return {};
void * ptr = event.data.ptr;
std::uintptr_t key = reinterpret_cast<uintptr_t>(ptr);
auto it = tasks.find(key);
if (it == tasks.end())
{
throw Exception(ErrorCodes::LOGICAL_ERROR, "Task {} ({}) was not found in task queue: {}",
key, ptr, dumpTasks(tasks));
}
auto res = it->second;
tasks.erase(it);
epoll.remove(res.fd);
return res;
}
void PollingQueue::finish()
{
is_finished = true;
uint64_t buf = 0;
while (-1 == write(pipe_fd[1], &buf, sizeof(buf)))
{
if (errno == EAGAIN)
break;
if (errno != EINTR)
throwFromErrno("Cannot write to pipe", ErrorCodes::CANNOT_READ_FROM_SOCKET);
}
}
}
#endif
|
