nd_poll() fairness (#19298)
* nd_poll() fairness * ensure we dont return events the user may have deleted * fixed warnings * fixed test compilation * minor fixes * statsd fix * track writer tid in rw spinlock * log replication counters on sender disconnect
Costa Tsaousis committed
Dec 30, 2024 at 10:51 UTC
d3b09d814096654fcafd4fa5207133d44b164676
10 files changed
+245
-131
CMakeLists.txt
+1
@@ -562,6 +562,7 @@ void my_function() { ; }
562
" HAVE_FUNC_ATTRIBUTE_NOINLINE)
563
564
check_c_source_compiles("
565
+#include <stdlib.h>
566
void my_exit_function() __attribute__((noreturn));
567
int main() {
568
my_exit_function(); // Call the noreturn function
src/collectors/debugfs.plugin/module-libsensors.c
+3
-3
@@ -664,7 +664,7 @@ static inline void check_value_greater_than_zero(SENSOR *s, SENSOR_SUBFEATURE_TY
664
s->state = state;
665
666
string_freez(s->log_msg);
667
- char buf[100];
667
+ char buf[1024];
668
snprintf(buf, sizeof(buf), "%s == %f (kernel driver generated)",
669
SENSOR_SUBFEATURE_TYPE_2str(*config), status);
670
s->log_msg = string_strdupz(buf);
@@ -674,7 +674,7 @@ static inline void check_value_greater_than_zero(SENSOR *s, SENSOR_SUBFEATURE_TY
674
675
static void userspace_evaluation_log_msg(SENSOR *s, const char *reading_txt, const char *condition, const char *threshold_txt, double reading, double threshold) {
676
string_freez(s->log_msg);
677
- char buf[200];
677
+ char buf[1024];
678
snprintf(buf, sizeof(buf), "%s %f %s %s %f (userspace evaluation using kernel provided thresholds)",
679
reading_txt, reading, condition, threshold_txt, threshold);
680
s->log_msg = string_strdupz(buf);
@@ -1210,7 +1210,7 @@ static int sensors_collect_data(void) {
1210
static bool libsensors_running = false;
1211
static int libsensors_update_every = 1;
1212
1213
-void *libsensors_thread(void *ptr) {
1213
+void *libsensors_thread(void *ptr __maybe_unused) {
1214
int update_every = libsensors_update_every;
1215
1216
// first try the default directory for libsensors
src/libnetdata/locks/rw-spinlock.c
+20
-12
@@ -2,10 +2,13 @@
2
3
#include "libnetdata/libnetdata.h"
4
5
+#define WRITER_LOCKED (-65536)
6
+
7
// ----------------------------------------------------------------------------
8
// rw_spinlock implementation
9
10
void rw_spinlock_init_with_trace(RW_SPINLOCK *rw_spinlock, const char *func __maybe_unused) {
11
+ rw_spinlock->writer = 0;
12
rw_spinlock->counter = 0;
13
}
14
@@ -14,10 +17,13 @@ bool rw_spinlock_tryread_lock_with_trace(RW_SPINLOCK *rw_spinlock, const char *f
17
18
REFCOUNT expected = rw_spinlock->counter;
19
while (true) {
17
- if(expected < 0)
20
+ if(expected == WRITER_LOCKED)
21
// writer is active
22
return false;
23
24
+ if(expected < 0)
25
+ fatal("RW_SPINLOCK: refcount found negative, on %s(), called from %s()", __FUNCTION__, func);
26
+
27
// increment reader count
28
if (__atomic_compare_exchange_n(
29
&rw_spinlock->counter,
@@ -43,9 +49,12 @@ void rw_spinlock_read_lock_with_trace(RW_SPINLOCK *rw_spinlock, const char *func
49
REFCOUNT expected = rw_spinlock->counter;
50
51
// we should not increase it if it is negative (a writer holds the lock)
46
- if(expected < 0) expected = 0;
52
+ if(expected == WRITER_LOCKED) expected = 0;
53
54
while (true) {
55
+ if(expected < 0)
56
+ fatal("RW_SPINLOCK: refcount found negative, on %s(), called from %s()", __FUNCTION__, func);
57
+
58
// Attempt to increment reader count
59
if (__atomic_compare_exchange_n(
60
&rw_spinlock->counter,
@@ -59,7 +68,7 @@ void rw_spinlock_read_lock_with_trace(RW_SPINLOCK *rw_spinlock, const char *func
68
69
spins++;
70
62
- if (expected < 0) {
71
+ if (expected == WRITER_LOCKED) {
72
// writer is active
73
74
// we should not increase it if it is negative (a writer holds the lock)
@@ -76,13 +85,9 @@ void rw_spinlock_read_lock_with_trace(RW_SPINLOCK *rw_spinlock, const char *func
85
}
86
87
void rw_spinlock_read_unlock_with_trace(RW_SPINLOCK *rw_spinlock, const char *func __maybe_unused) {
79
-#ifndef NETDATA_INTERNAL_CHECKS
80
- __atomic_sub_fetch(&rw_spinlock->counter, 1, __ATOMIC_RELEASE);
81
-#else
88
REFCOUNT x = __atomic_sub_fetch(&rw_spinlock->counter, 1, __ATOMIC_RELEASE);
89
if (x < 0)
84
- fatal("RW_SPINLOCK: readers is negative %d", x);
85
-#endif
90
+ fatal("RW_SPINLOCK: readers is negative %d, on %s called from %s()", x, __FUNCTION__, func);
91
92
nd_thread_rwspinlock_read_unlocked();
93
}
@@ -94,7 +99,7 @@ bool rw_spinlock_trywrite_lock_with_trace(RW_SPINLOCK *rw_spinlock, const char *
99
if (!__atomic_compare_exchange_n(
100
&rw_spinlock->counter,
101
&expected,
97
- -1,
102
+ WRITER_LOCKED,
103
false, // Strong CAS
104
__ATOMIC_ACQUIRE, // Success memory order
105
__ATOMIC_RELAXED // Failure memory order
@@ -102,6 +107,7 @@ bool rw_spinlock_trywrite_lock_with_trace(RW_SPINLOCK *rw_spinlock, const char *
107
return false;
108
}
109
110
+ __atomic_store_n(&rw_spinlock->writer, gettid_cached(), __ATOMIC_RELAXED);
111
worker_spinlock_contention(func, 0);
112
nd_thread_rwspinlock_write_locked();
113
return true;
@@ -117,7 +123,7 @@ void rw_spinlock_write_lock_with_trace(RW_SPINLOCK *rw_spinlock, const char *fun
123
if (__atomic_compare_exchange_n(
124
&rw_spinlock->counter,
125
&expected,
120
- -1,
126
+ WRITER_LOCKED,
127
false, // Strong CAS
128
__ATOMIC_ACQUIRE, // Success memory order
129
__ATOMIC_RELAXED // Failure memory order
@@ -129,6 +135,7 @@ void rw_spinlock_write_lock_with_trace(RW_SPINLOCK *rw_spinlock, const char *fun
135
tinysleep();
136
}
137
138
+ __atomic_store_n(&rw_spinlock->writer, gettid_cached(), __ATOMIC_RELAXED);
139
worker_spinlock_contention(func, spins);
140
nd_thread_rwspinlock_write_locked();
141
}
@@ -136,10 +143,11 @@ void rw_spinlock_write_lock_with_trace(RW_SPINLOCK *rw_spinlock, const char *fun
143
void rw_spinlock_write_unlock_with_trace(RW_SPINLOCK *rw_spinlock, const char *func __maybe_unused) {
144
#ifdef NETDATA_INTERNAL_CHECKS
145
int32_t x = __atomic_load_n(&rw_spinlock->counter, __ATOMIC_RELAXED);
139
- if (x != -1)
140
- fatal("RW_SPINLOCK: writer unlock encountered unexpected state: %d", x);
146
+ if (x != WRITER_LOCKED)
147
+ fatal("RW_SPINLOCK: writer unlock encountered unexpected state: %d, on %s() called from %s()", x, __FUNCTION__, func);
148
#endif
149
150
+ __atomic_store_n(&rw_spinlock->writer, 0, __ATOMIC_RELAXED);
151
__atomic_store_n(&rw_spinlock->counter, 0, __ATOMIC_RELEASE); // Release writer lock
152
nd_thread_rwspinlock_write_unlocked();
153
}
src/libnetdata/locks/rw-spinlock.h
+2
-1
@@ -7,10 +7,11 @@
7
#include "spinlock.h"
8
9
typedef struct netdata_rw_spinlock {
10
+ pid_t writer;
11
REFCOUNT counter; // positive is readers, negative is a writer
12
} RW_SPINLOCK;
13
13
-#define RW_SPINLOCK_INITIALIZER { .counter = 0, }
14
+#define RW_SPINLOCK_INITIALIZER { .counter = 0, .writer = 0, }
15
16
void rw_spinlock_init_with_trace(RW_SPINLOCK *rw_spinlock, const char *func);
17
void rw_spinlock_read_lock_with_trace(RW_SPINLOCK *rw_spinlock, const char *func);
src/libnetdata/socket/nd-poll.c
+176
-87
@@ -9,15 +9,28 @@
9
#if defined(OS_LINUX)
10
#include <sys/epoll.h>
11
12
+struct fd_info {
13
+ uint32_t events;
14
+ uint32_t last_served;
15
+ const void *data;
16
+};
17
+
18
+DEFINE_JUDYL_TYPED(POINTERS, struct fd_info *);
19
+
20
#define MAX_EVENTS_PER_CALL 100
21
22
// Event poll context
23
struct nd_poll_t {
24
int epoll_fd;
25
+
26
struct epoll_event ev[MAX_EVENTS_PER_CALL];
27
size_t last_pos;
28
size_t used;
20
- size_t nfds;
29
+
30
+ POINTERS_JudyLSet pointers; // Judy array to store user data
31
+
32
+ uint32_t nfds; // the number of sockets we have
33
+ uint32_t iteration_counter;
34
};
35
36
// Initialize the event poll context
@@ -33,97 +46,161 @@ nd_poll_t *nd_poll_create() {
46
return ndpl;
47
}
48
36
-static inline void nd_poll_replace_data_on_loaded_events(nd_poll_t *ndpl, int fd __maybe_unused, void *old_data, void *new_data) {
37
- for(size_t i = ndpl->last_pos; i < ndpl->used; i++) {
38
- if(ndpl->ev[i].data.ptr == old_data)
39
- ndpl->ev[i].data.ptr = new_data;
40
- }
49
+static inline uint32_t nd_poll_events_to_epoll_events(nd_poll_event_t events) {
50
+ uint32_t pevents = EPOLLERR | EPOLLHUP;
51
+ if (events & ND_POLL_READ) pevents |= EPOLLIN;
52
+ if (events & ND_POLL_WRITE) pevents |= EPOLLOUT;
53
+ return pevents;
54
+}
55
+
56
+static inline nd_poll_event_t nd_poll_events_from_epoll_events(uint32_t events) {
57
+ nd_poll_event_t nd_poll_events = ND_POLL_NONE;
58
+
59
+ if (events & (EPOLLIN|EPOLLPRI|EPOLLRDNORM|EPOLLRDBAND))
60
+ nd_poll_events |= ND_POLL_READ;
61
+
62
+ if (events & (EPOLLOUT|EPOLLWRNORM|EPOLLWRBAND))
63
+ nd_poll_events |= ND_POLL_WRITE;
64
+
65
+ if (events & EPOLLERR)
66
+ nd_poll_events |= ND_POLL_ERROR;
67
+
68
+ if (events & (EPOLLHUP|EPOLLRDHUP))
69
+ nd_poll_events |= ND_POLL_HUP;
70
+
71
+ return nd_poll_events;
72
}
73
74
// Add a file descriptor to the event poll
44
-bool nd_poll_add(nd_poll_t *ndpl, int fd, nd_poll_event_t events, void *data) {
75
+bool nd_poll_add(nd_poll_t *ndpl, int fd, nd_poll_event_t events, const void *data) {
76
internal_fatal(!data, "nd_poll() does not support NULL data pointers");
77
78
+ struct fd_info *fdi = mallocz(sizeof(*fdi));
79
+ fdi->data = data;
80
+ fdi->last_served = 0;
81
+ fdi->events = nd_poll_events_to_epoll_events(events);
82
+
83
+ if(POINTERS_GET(&ndpl->pointers, fd) || !POINTERS_SET(&ndpl->pointers, fd, fdi)) {
84
+ freez(fdi);
85
+ return false;
86
+ }
87
+
88
struct epoll_event ev = {
48
- .events = (events & ND_POLL_READ ? EPOLLIN : 0) | (events & ND_POLL_WRITE ? EPOLLOUT : 0),
49
- .data.ptr = data,
89
+ .events = fdi->events,
90
+ .data.fd = fd,
91
};
92
+
93
bool rc = epoll_ctl(ndpl->epoll_fd, EPOLL_CTL_ADD, fd, &ev) == 0;
52
- if(rc) ndpl->nfds++;
94
+ if(rc)
95
+ ndpl->nfds++;
96
+ else {
97
+ POINTERS_DEL(&ndpl->pointers, fd);
98
+ freez(fdi);
99
+ }
100
+
101
internal_fatal(!rc, "epoll_ctl() failed");
102
+
103
return rc;
104
}
105
106
// Remove a file descriptor from the event poll
58
-bool nd_poll_del(nd_poll_t *ndpl, int fd, void *data) {
59
- internal_fatal(!data, "nd_poll() does not support NULL data pointers");
107
+bool nd_poll_del(nd_poll_t *ndpl, int fd) {
108
+ struct fd_info *fdi = POINTERS_GET(&ndpl->pointers, fd);
109
+ if(!fdi) return false;
110
+
111
+ POINTERS_DEL(&ndpl->pointers, fd);
112
+ freez(fdi);
113
114
ndpl->nfds--; // we can't check for success/failure here, because epoll() removes fds when they are closed
115
bool rc = epoll_ctl(ndpl->epoll_fd, EPOLL_CTL_DEL, fd, NULL) == 0;
116
internal_error(!rc, "epoll_ctl() failed (is the socket already closed)"); // this is ok if the socket is already closed
64
-
65
- // we may have an event pending for this fd.
66
- // but epoll() does not give us fd in the events,
67
- // so we use the data pointer to invalidate it
68
- nd_poll_replace_data_on_loaded_events(ndpl, fd, data, NULL);
117
return rc;
118
}
119
120
// Update an existing file descriptor in the event poll
73
-bool nd_poll_upd(nd_poll_t *ndpl, int fd, nd_poll_event_t events, void *data) {
74
- internal_fatal(!data, "nd_poll() does not support NULL data pointers - you should also NEVER change the pointer with an update");
121
+bool nd_poll_upd(nd_poll_t *ndpl, int fd, nd_poll_event_t events) {
122
+ struct fd_info *fdi = POINTERS_GET(&ndpl->pointers, fd);
123
+ if(!fdi) return false;
124
+
125
+ fdi->events = nd_poll_events_to_epoll_events(events);
126
127
struct epoll_event ev = {
77
- .events = (events & ND_POLL_READ ? EPOLLIN : 0) | (events & ND_POLL_WRITE ? EPOLLOUT : 0),
78
- .data.ptr = data,
128
+ .events = fdi->events,
129
+ .data.fd = fd,
130
};
131
bool rc = epoll_ctl(ndpl->epoll_fd, EPOLL_CTL_MOD, fd, &ev) == 0;
81
- internal_fatal(!rc, "epoll_ctl() failed");
132
+ internal_fatal(!rc, "epoll_ctl() failed"); // this may happen if fd is closed
133
return rc;
134
}
135
85
-static inline nd_poll_event_t nd_poll_events_from_epoll_events(uint32_t events) {
86
- nd_poll_event_t nd_poll_events = ND_POLL_NONE;
87
-
88
- if (events & (EPOLLIN|EPOLLPRI|EPOLLRDNORM|EPOLLRDBAND))
89
- nd_poll_events |= ND_POLL_READ;
90
-
91
- if (events & (EPOLLOUT|EPOLLWRNORM|EPOLLWRBAND))
92
- nd_poll_events |= ND_POLL_WRITE;
93
-
94
- if (events & EPOLLERR)
95
- nd_poll_events |= ND_POLL_ERROR;
96
-
97
- if (events & (EPOLLHUP|EPOLLRDHUP))
98
- nd_poll_events |= ND_POLL_HUP;
99
-
100
- return nd_poll_events;
101
-}
102
-
136
static inline bool nd_poll_get_next_event(nd_poll_t *ndpl, nd_poll_result_t *result) {
137
while(ndpl->last_pos < ndpl->used) {
105
- void *data = ndpl->ev[ndpl->last_pos].data.ptr;
138
+ struct fd_info *fdi = POINTERS_GET(&ndpl->pointers, ndpl->ev[ndpl->last_pos].data.fd);
139
140
// Skip events that have been invalidated by nd_poll_del()
108
- if(!data) {
141
+ if(!fdi || !fdi->data) {
142
ndpl->last_pos++;
143
continue;
144
}
145
146
*result = (nd_poll_result_t){
114
- .events = nd_poll_events_from_epoll_events(ndpl->ev[ndpl->last_pos].events),
115
- .data = data,
147
+ .events = nd_poll_events_from_epoll_events(ndpl->ev[ndpl->last_pos].events & fdi->events),
148
+ .data = fdi->data,
149
};
150
151
ndpl->last_pos++;
152
+
153
+ if(!result->events)
154
+ // nd_poll_upd() may have removed some flags since we got this
155
+ continue;
156
+
157
+ fdi->last_served = ndpl->iteration_counter;
158
return true;
159
}
160
161
return false;
162
}
163
164
+typedef struct {
165
+ struct epoll_event event;
166
+ uint32_t last_served;
167
+} sortable_event_t;
168
+
169
+static int compare_last_served(const void *a, const void *b) {
170
+ const sortable_event_t *ev_a = (const sortable_event_t *)a;
171
+ const sortable_event_t *ev_b = (const sortable_event_t *)b;
172
+
173
+ if (ev_a->last_served < ev_b->last_served)
174
+ return -1;
175
+ if (ev_a->last_served > ev_b->last_served)
176
+ return 1;
177
+
178
+ return 0;
179
+}
180
+
181
+static void sort_events(nd_poll_t *ndpl) {
182
+ if(ndpl->used <= 1) return;
183
+
184
+ sortable_event_t sortable_array[ndpl->used];
185
+ for (size_t i = 0; i < ndpl->used; ++i) {
186
+ struct fd_info *fdi = POINTERS_GET(&ndpl->pointers, ndpl->ev[i].data.fd);
187
+ sortable_array[i] = (sortable_event_t){
188
+ .event = ndpl->ev[i],
189
+ .last_served = fdi ? fdi->last_served : UINT32_MAX,
190
+ };
191
+ }
192
+
193
+ qsort(sortable_array, ndpl->used, sizeof(sortable_event_t), compare_last_served);
194
+
195
+ // Reorder `ndpl->ev` based on the sorted order
196
+ for (size_t i = 0; i < ndpl->used; ++i)
197
+ ndpl->ev[i] = sortable_array[i].event;
198
+}
199
+
200
// Wait for events
201
int nd_poll_wait(nd_poll_t *ndpl, int timeout_ms, nd_poll_result_t *result) {
202
+ ndpl->iteration_counter++;
203
+
204
if(nd_poll_get_next_event(ndpl, result))
205
return 1;
206
@@ -132,11 +209,7 @@ int nd_poll_wait(nd_poll_t *ndpl, int timeout_ms, nd_poll_result_t *result) {
209
ndpl->last_pos = 0;
210
ndpl->used = 0;
211
135
- int maxevents = ndpl->nfds / 2;
136
- if(maxevents > (int)_countof(ndpl->ev)) maxevents = (int)_countof(ndpl->ev);
137
- if(maxevents < 2) maxevents = 2;
138
-
139
- int n = epoll_wait(ndpl->epoll_fd, &ndpl->ev[0], maxevents, timeout_ms);
212
+ int n = epoll_wait(ndpl->epoll_fd, &ndpl->ev[0], _countof(ndpl->ev), timeout_ms);
213
214
if(unlikely(n <= 0)) {
215
if(n == 0) {
@@ -155,6 +228,7 @@ int nd_poll_wait(nd_poll_t *ndpl, int timeout_ms, nd_poll_result_t *result) {
228
229
ndpl->used = n;
230
ndpl->last_pos = 0;
231
+ sort_events(ndpl);
232
if (nd_poll_get_next_event(ndpl, result))
233
return 1;
234
@@ -162,16 +236,21 @@ int nd_poll_wait(nd_poll_t *ndpl, int timeout_ms, nd_poll_result_t *result) {
236
} while(true);
237
}
238
239
+static void nd_poll_free_callback(Word_t fd __maybe_unused, struct fd_info *fdi) {
240
+ freez(fdi);
241
+}
242
+
243
// Destroy the event poll context
244
void nd_poll_destroy(nd_poll_t *ndpl) {
245
if (ndpl) {
246
close(ndpl->epoll_fd);
247
+ POINTERS_FREE(&ndpl->pointers, nd_poll_free_callback);
248
freez(ndpl);
249
}
250
}
251
#else
252
174
-DEFINE_JUDYL_TYPED(POINTERS, void *);
253
+DEFINE_JUDYL_TYPED(POINTERS, const void *);
254
255
struct nd_poll_t {
256
struct pollfd *fds; // Array of file descriptors
@@ -206,31 +285,60 @@ static void ensure_capacity(nd_poll_t *ndpl) {
285
ndpl->capacity = new_capacity;
286
}
287
209
-bool nd_poll_add(nd_poll_t *ndpl, int fd, nd_poll_event_t events, void *data) {
288
+static inline short int nd_poll_events_to_poll_events(nd_poll_event_t events) {
289
+ short int pevents = POLLERR | POLLHUP | POLLNVAL;
290
+ if (events & ND_POLL_READ) pevents |= POLLIN;
291
+ if (events & ND_POLL_WRITE) pevents |= POLLOUT;
292
+ return pevents;
293
+}
294
+
295
+static inline nd_poll_event_t nd_poll_events_from_poll_revents(short int events) {
296
+ nd_poll_event_t nd_poll_events = ND_POLL_NONE;
297
+
298
+ if (events & (POLLIN|POLLPRI|POLLRDNORM|POLLRDBAND))
299
+ nd_poll_events |= ND_POLL_READ;
300
+
301
+ if (events & (POLLOUT|POLLWRNORM|POLLWRBAND))
302
+ nd_poll_events |= ND_POLL_WRITE;
303
+
304
+ if (events & POLLERR)
305
+ nd_poll_events |= ND_POLL_ERROR;
306
+
307
+ if (events & (POLLHUP|POLLRDHUP))
308
+ nd_poll_events |= ND_POLL_HUP;
309
+
310
+ if (events & (POLLNVAL))
311
+ nd_poll_events |= ND_POLL_INVALID;
312
+
313
+ return nd_poll_events;
314
+}
315
+
316
+bool nd_poll_add(nd_poll_t *ndpl, int fd, nd_poll_event_t events, const void *data) {
317
internal_fatal(POINTERS_GET(&ndpl->pointers, fd) != NULL, "File descriptor %d is already served - cannot add", fd);
318
319
+ if(POINTERS_GET(&ndpl->pointers, fd) || !POINTERS_SET(&ndpl->pointers, fd, data))
320
+ return false;
321
+
322
ensure_capacity(ndpl);
323
struct pollfd *pfd = &ndpl->fds[ndpl->nfds++];
324
pfd->fd = fd;
215
- pfd->events = 0;
216
- if (events & ND_POLL_READ) pfd->events |= POLLIN;
217
- if (events & ND_POLL_WRITE) pfd->events |= POLLOUT;
325
+ pfd->events = nd_poll_events_to_poll_events(events);
326
pfd->revents = 0;
327
220
- POINTERS_SET(&ndpl->pointers, fd, data);
221
-
328
return true;
329
}
330
331
// Remove a file descriptor from the event poll
226
-bool nd_poll_del(nd_poll_t *ndpl, int fd, void *data __maybe_unused) {
332
+bool nd_poll_del(nd_poll_t *ndpl, int fd) {
333
+ if(!POINTERS_DEL(&ndpl->pointers, fd))
334
+ return false;
335
+
336
for (nfds_t i = 0; i < ndpl->nfds; i++) {
337
if (ndpl->fds[i].fd == fd) {
338
339
// Remove the file descriptor by shifting the array
340
memmove(&ndpl->fds[i], &ndpl->fds[i + 1], (ndpl->nfds - i - 1) * sizeof(struct pollfd));
341
ndpl->nfds--;
233
- POINTERS_DEL(&ndpl->pointers, fd);
342
343
if(i < ndpl->last_pos)
344
ndpl->last_pos--;
@@ -243,16 +351,11 @@ bool nd_poll_del(nd_poll_t *ndpl, int fd, void *data __maybe_unused) {
351
}
352
353
// Update an existing file descriptor in the event poll
246
-bool nd_poll_upd(nd_poll_t *ndpl, int fd, nd_poll_event_t events, void *data) {
247
- internal_fatal(POINTERS_GET(&ndpl->pointers, fd) == NULL, "File descriptor %d is not found - cannot modify", fd);
248
-
354
+bool nd_poll_upd(nd_poll_t *ndpl, int fd, nd_poll_event_t events) {
355
for (nfds_t i = 0; i < ndpl->nfds; i++) {
356
if (ndpl->fds[i].fd == fd) {
357
struct pollfd *pfd = &ndpl->fds[i];
252
- pfd->events = 0;
253
- if (events & ND_POLL_READ) pfd->events |= POLLIN;
254
- if (events & ND_POLL_WRITE) pfd->events |= POLLOUT;
255
- POINTERS_SET(&ndpl->pointers, fd, data);
358
+ pfd->events = nd_poll_events_to_poll_events(events);
359
return true;
360
}
361
}
@@ -261,33 +364,19 @@ bool nd_poll_upd(nd_poll_t *ndpl, int fd, nd_poll_event_t events, void *data) {
364
return false;
365
}
366
264
-static inline nd_poll_event_t nd_poll_events_from_poll_revents(short int events) {
265
- nd_poll_event_t nd_poll_events = ND_POLL_NONE;
266
-
267
- if (events & (POLLIN|POLLPRI|POLLRDNORM|POLLRDBAND))
268
- nd_poll_events |= ND_POLL_READ;
269
-
270
- if (events & (POLLOUT|POLLWRNORM|POLLWRBAND))
271
- nd_poll_events |= ND_POLL_WRITE;
272
-
273
- if (events & POLLERR)
274
- nd_poll_events |= ND_POLL_ERROR;
275
-
276
- if (events & (POLLHUP|POLLRDHUP))
277
- nd_poll_events |= ND_POLL_HUP;
278
-
279
- if (events & (POLLNVAL))
280
- nd_poll_events |= ND_POLL_INVALID;
281
-
282
- return nd_poll_events;
283
-}
284
-
367
static inline bool nd_poll_get_next_event(nd_poll_t *ndpl, nd_poll_result_t *result) {
368
for (nfds_t i = ndpl->last_pos; i < ndpl->nfds; i++) {
369
if (ndpl->fds[i].revents != 0) {
370
371
result->data = POINTERS_GET(&ndpl->pointers, ndpl->fds[i].fd);
290
- result->events = nd_poll_events_from_poll_revents(ndpl->fds[i].revents);
372
+ if(!result->data)
373
+ continue;
374
+
375
+ result->events = nd_poll_events_from_poll_revents(ndpl->fds[i].revents & ndpl->fds[i].events);
376
+ if(!result->events)
377
+ // nd_poll_upd() may have removed some flags since we got this
378
+ continue;
379
+
380
ndpl->fds[i].revents = 0;
381
382
ndpl->last_pos = i + 1;
src/libnetdata/socket/nd-poll.h
+7
-10
@@ -20,26 +20,23 @@ typedef enum __attribute__((packed)) {
20
21
typedef struct {
22
nd_poll_event_t events;
23
- void *data;
23
+ const void *data;
24
} nd_poll_result_t;
25
26
typedef struct nd_poll_t nd_poll_t;
27
28
-nd_poll_t *nd_poll_create();
28
+nd_poll_t *nd_poll_create() WARNUNUSED;
29
void nd_poll_destroy(nd_poll_t *ndpl);
30
31
// the events can be updated with nd_poll_upd
32
// the data pointer SHOULD NEVER be changed and cannot be NULL
33
-bool nd_poll_add(nd_poll_t *ndpl, int fd, nd_poll_event_t events, void *data);
33
+bool nd_poll_add(nd_poll_t *ndpl, int fd, nd_poll_event_t events, const void *data);
34
35
-// give the same data pointer used in nd_poll_add()
36
-// otherwise, you may receive back invalid events
37
-bool nd_poll_del(nd_poll_t *ndpl, int fd, void *data);
35
+// delete an fd
36
+bool nd_poll_del(nd_poll_t *ndpl, int fd);
37
39
-// this is for updating events
40
-// the data pointer must be the same used in nd_poll_add()
41
-// to change the data pointer, delete and add the same fd again
42
-bool nd_poll_upd(nd_poll_t *ndpl, int fd, nd_poll_event_t events, void *data);
38
+// update the expected events on an fd
39
+bool nd_poll_upd(nd_poll_t *ndpl, int fd, nd_poll_event_t events);
40
41
// returns -1 = error, 0 = timeout, 1 = event in result
42
int nd_poll_wait(nd_poll_t *ndpl, int timeout_ms, nd_poll_result_t *result);
src/libnetdata/socket/poll-events.c
+23
-7
@@ -4,7 +4,7 @@
4
5
static inline void poll_process_updated_events(POLLINFO *pi) {
6
if(pi->events != pi->events_we_wait_for) {
7
- if(!nd_poll_upd(pi->p->ndpl, pi->fd, pi->events, pi))
7
+ if(!nd_poll_upd(pi->p->ndpl, pi->fd, pi->events))
8
nd_log(NDLS_DAEMON, NDLP_ERR, "Failed to update socket %d to nd_poll", pi->fd);
9
pi->events_we_wait_for = pi->events;
10
}
@@ -76,7 +76,7 @@ static inline void poll_close_fd(POLLINFO *pi, const char *func) {
76
POLLJOB *p = pi->p;
77
78
DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(p->ll, pi, prev, next);
79
- if(!nd_poll_del(p->ndpl, pi->fd, pi))
79
+ if(!nd_poll_del(p->ndpl, pi->fd))
80
// this is ok, if the socket is already closed
81
nd_log(NDLS_DAEMON, NDLP_DEBUG,
82
"Failed to delete socket %d from nd_poll() - called from %s() - is the socket already closed?",
@@ -438,7 +438,7 @@ void poll_events(LISTEN_SOCKETS *sockets
438
;
439
}
440
else {
441
- POLLINFO *pi = result.data;
441
+ POLLINFO *pi = (POLLINFO *)result.data;
442
443
if(result.events & (ND_POLL_HUP | ND_POLL_INVALID | ND_POLL_ERROR))
444
poll_process_error(pi, result.events);
@@ -466,8 +466,24 @@ void poll_events(LISTEN_SOCKETS *sockets
466
}
467
}
468
else if (pi->flags & POLLINFO_FLAG_SERVER_SOCKET) {
469
- if(!p.limit || p.used < p.limit)
470
- poll_process_new_tcp_connection(pi, now);
469
+ if(pi->socktype == SOCK_DGRAM)
470
+ poll_process_udp_read(pi, now);
471
+
472
+ else if(pi->socktype == SOCK_STREAM) {
473
+ if (!p.limit || p.used < p.limit)
474
+ poll_process_new_tcp_connection(pi, now);
475
+ }
476
+ else {
477
+ nd_log(NDLS_DAEMON, NDLP_ERR,
478
+ "POLLFD: LISTENER: server slot %zu (fd %d) connection from %s port %s using unhandled socket type %d.",
479
+ i,
480
+ pi->fd,
481
+ pi->client_ip ? pi->client_ip : "<undefined-ip>",
482
+ pi->client_port ? pi->client_port : "<undefined-port>",
483
+ pi->socktype);
484
+
485
+ poll_close_fd(pi, "poll_events2");
486
+ }
487
}
488
else {
489
nd_log(NDLS_DAEMON, NDLP_ERR,
@@ -479,7 +495,7 @@ void poll_events(LISTEN_SOCKETS *sockets
495
, pi->flags
496
);
497
482
- poll_close_fd(pi, "poll_events2");
498
+ poll_close_fd(pi, "poll_events3");
499
}
500
}
501
else {
@@ -492,7 +508,7 @@ void poll_events(LISTEN_SOCKETS *sockets
508
, (int)result.events
509
);
510
495
- poll_close_fd(pi, "poll_events3");
511
+ poll_close_fd(pi, "poll_events4");
512
}
513
}
514
src/streaming/stream-receiver.c
+3
-3
@@ -562,7 +562,7 @@ static void stream_receiver_remove(struct stream_thread *sth, struct receiver_st
562
META_DEL(&sth->run.meta, (Word_t)&rpt->thread.meta);
563
564
rpt->thread.wanted = 0;
565
- if(!nd_poll_del(sth->run.ndpl, rpt->sock.fd, &rpt->thread.meta))
565
+ if(!nd_poll_del(sth->run.ndpl, rpt->sock.fd))
566
nd_log(NDLS_DAEMON, NDLP_ERR, "Failed to delete receiver socket from nd_poll()");
567
568
rpt->host->stream.rcv.status.tid = 0;
@@ -727,7 +727,7 @@ bool stream_receiver_send_data(struct stream_thread *sth, struct receiver_state
727
stream_circular_buffer_del_unsafe(scb, rc, now_ut);
728
if (!stats->bytes_outstanding) {
729
rpt->thread.wanted = ND_POLL_READ;
730
- if (!nd_poll_upd(sth->run.ndpl, rpt->sock.fd, rpt->thread.wanted, &rpt->thread.meta))
730
+ if (!nd_poll_upd(sth->run.ndpl, rpt->sock.fd, rpt->thread.wanted))
731
nd_log(NDLS_DAEMON, NDLP_ERR,
732
"STREAM RCV[%zu] '%s' [from [%s]:%s]: cannot update nd_poll()",
733
sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port);
@@ -983,7 +983,7 @@ void stream_receiver_check_all_nodes_from_poll(struct stream_thread *sth, usec_t
983
}
984
985
rpt->thread.wanted = ND_POLL_READ | (stats.bytes_outstanding ? ND_POLL_WRITE : 0);
986
- if(!nd_poll_upd(sth->run.ndpl, rpt->sock.fd, rpt->thread.wanted, &rpt->thread.meta))
986
+ if(!nd_poll_upd(sth->run.ndpl, rpt->sock.fd, rpt->thread.wanted))
987
nd_log(NDLS_DAEMON, NDLP_ERR,
988
"STREAM RCV[%zu] '%s' [from %s]: failed to update nd_poll().",
989
sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip);
src/streaming/stream-sender.c
+7
-5
@@ -371,8 +371,10 @@ static void stream_sender_log_disconnection(struct stream_thread *sth, struct se
371
ND_LOG_STACK_PUSH(lgs);
372
373
nd_log(NDLS_DAEMON, NDLP_NOTICE,
374
- "STREAM SND[%zu] '%s' [to %s]: sender disconnected from parent, reason: %s",
375
- sth->id, rrdhost_hostname(s->host), s->remote_ip, stream_handshake_error_to_string(reason));
374
+ "STREAM SND[%zu] '%s' [to %s]: sender disconnected from parent, reason: %s (replication in: %u, out: %u, pending: %u)",
375
+ sth->id, rrdhost_hostname(s->host), s->remote_ip, stream_handshake_error_to_string(reason),
376
+ s->host->stream.snd.status.replication.counter_in, s->host->stream.snd.status.replication.counter_out,
377
+ dictionary_entries(s->replication.requests));
378
}
379
380
static void stream_sender_move_running_to_connector_or_remove(struct stream_thread *sth, struct sender_state *s, STREAM_HANDSHAKE reason, bool reconnect) {
@@ -392,7 +394,7 @@ static void stream_sender_move_running_to_connector_or_remove(struct stream_thre
394
META_DEL(&sth->run.meta, (Word_t)&s->thread.meta);
395
396
s->thread.wanted = 0;
395
- if(!nd_poll_del(sth->run.ndpl, s->sock.fd, &s->thread.meta))
397
+ if(!nd_poll_del(sth->run.ndpl, s->sock.fd))
398
nd_log(NDLS_DAEMON, NDLP_ERR,
399
"STREAM SND[%zu] '%s' [to %s]: failed to delete sender socket from nd_poll()",
400
sth->id, rrdhost_hostname(s->host), s->remote_ip);
@@ -488,7 +490,7 @@ void stream_sender_check_all_nodes_from_poll(struct stream_thread *sth, usec_t n
490
bytes_uncompressed += stats.bytes_uncompressed;
491
492
s->thread.wanted = ND_POLL_READ | (stats.bytes_outstanding ? ND_POLL_WRITE : 0);
491
- if(!nd_poll_upd(sth->run.ndpl, s->sock.fd, s->thread.wanted, &s->thread.meta))
493
+ if(!nd_poll_upd(sth->run.ndpl, s->sock.fd, s->thread.wanted))
494
nd_log(NDLS_DAEMON, NDLP_ERR,
495
"STREAM SND[%zu] '%s' [to %s]: failed to update nd_poll().",
496
sth->id, rrdhost_hostname(s->host), s->remote_ip);
@@ -617,7 +619,7 @@ bool stream_sender_send_data(struct stream_thread *sth, struct sender_state *s,
619
if (!stats->bytes_outstanding) {
620
// we sent them all - remove ND_POLL_WRITE
621
s->thread.wanted = ND_POLL_READ;
620
- if (!nd_poll_upd(sth->run.ndpl, s->sock.fd, s->thread.wanted, &s->thread.meta))
622
+ if (!nd_poll_upd(sth->run.ndpl, s->sock.fd, s->thread.wanted))
623
nd_log(NDLS_DAEMON, NDLP_ERR,
624
"STREAM SND[%zu] '%s' [to %s]: failed to update nd_poll().",
625
sth->id, rrdhost_hostname(s->host), s->remote_ip);
src/streaming/stream-thread.c
+3
-3
@@ -30,7 +30,7 @@ static void stream_thread_handle_op(struct stream_thread *sth, struct stream_opc
30
if(m->type == POLLFD_TYPE_SENDER) {
31
if(msg->opcode & STREAM_OPCODE_SENDER_POLLOUT) {
32
m->s->thread.wanted = ND_POLL_READ | ND_POLL_WRITE;
33
- if(!nd_poll_upd(sth->run.ndpl, m->s->sock.fd, m->s->thread.wanted, m)) {
33
+ if(!nd_poll_upd(sth->run.ndpl, m->s->sock.fd, m->s->thread.wanted)) {
34
nd_log_limit_static_global_var(erl, 1, 0);
35
nd_log_limit(&erl, NDLS_DAEMON, NDLP_DEBUG,
36
"STREAM SND[%zu] '%s' [to %s]: cannot enable output on sender socket %d.",
@@ -50,7 +50,7 @@ static void stream_thread_handle_op(struct stream_thread *sth, struct stream_opc
50
else if(m->type == POLLFD_TYPE_RECEIVER) {
51
if (msg->opcode & STREAM_OPCODE_RECEIVER_POLLOUT) {
52
m->rpt->thread.wanted = ND_POLL_READ | ND_POLL_WRITE;
53
- if (!nd_poll_upd(sth->run.ndpl, m->rpt->sock.fd, m->rpt->thread.wanted, m)) {
53
+ if (!nd_poll_upd(sth->run.ndpl, m->rpt->sock.fd, m->rpt->thread.wanted)) {
54
nd_log_limit_static_global_var(erl, 1, 0);
55
nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR,
56
"STREAM RCV[%zu] '%s' [from [%s]:%s]: cannot enable output on receiver socket %d.",
@@ -328,7 +328,7 @@ static void stream_thread_messages_resize_unsafe(struct stream_thread *sth) {
328
static bool stream_thread_process_poll_slot(struct stream_thread *sth, nd_poll_result_t *ev, usec_t now_ut, size_t *replay_entries) {
329
internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ );
330
331
- struct pollfd_meta *m = ev->data;
331
+ struct pollfd_meta *m = (struct pollfd_meta *)ev->data;
332
if(!m) {
333
nd_log(NDLS_DAEMON, NDLP_ERR,
334
"STREAM THREAD[%zu]: cannot get meta from nd_poll() event. Ignoring event.", sth->id);