avoid deadlock in push thread in case of errors
Costa Tsaousis (ktsaou) committed
Feb 21, 2017 at 10:02 UTC
4613856eafe1e928f81887886565662d92d38430
1 file changed
+32
-13
src/rrdpush.c
+32
-13
@@ -7,7 +7,8 @@ int rrdpush_pipe[2];
7
8
static BUFFER *rrdpush_buffer = NULL;
9
static pthread_mutex_t rrdpush_mutex = PTHREAD_MUTEX_INITIALIZER;
10
-static RRDHOST *last_host = NULL;
10
+static volatile RRDHOST *last_host = NULL;
11
+static volatile int rrdpush_connected = 0;
12
13
static inline void rrdpush_lock() {
14
pthread_mutex_lock(&rrdpush_mutex);
@@ -98,10 +99,20 @@ static void reset_all_charts(void) {
99
}
100
101
void rrdset_done_push(RRDSET *st) {
102
+ static int error_shown = 0;
103
102
- if(unlikely(!rrdset_flag_check(st, RRDSET_FLAG_ENABLED) || !rrdpush_buffer))
104
+ if(unlikely(!rrdset_flag_check(st, RRDSET_FLAG_ENABLED)))
105
return;
106
107
+ if(unlikely(!rrdpush_buffer || !rrdpush_connected)) {
108
+ if(!error_shown)
109
+ error("PUSH: not ready - discarding collected metrics.");
110
+
111
+ error_shown = 1;
112
+ return;
113
+ }
114
+ error_shown = 0;
115
+
116
rrdpush_lock();
117
rrdset_rdlock(st);
118
@@ -123,6 +134,17 @@ void rrdset_done_push(RRDSET *st) {
134
rrdpush_unlock();
135
}
136
137
+static inline void rrdpush_flush(void) {
138
+ rrdpush_lock();
139
+ if(buffer_strlen(rrdpush_buffer))
140
+ error("PUSH: discarding %zu bytes of metrics data already in the buffer.", buffer_strlen(rrdpush_buffer));
141
+
142
+ buffer_flush(rrdpush_buffer);
143
+ reset_all_charts();
144
+ last_host = NULL;
145
+ rrdpush_unlock();
146
+}
147
+
148
void *central_netdata_push_thread(void *ptr) {
149
struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr;
150
@@ -145,13 +167,13 @@ void *central_netdata_push_thread(void *ptr) {
167
.tv_usec = 0
168
};
169
170
+ rrdpush_connected = 0;
171
size_t begin = 0;
172
size_t max_size = 1024 * 1024;
173
size_t reconnects_counter = 0;
174
size_t sent_bytes = 0;
175
size_t sent_connection = 0;
176
int sock = -1;
154
- char buffer[1000 + 1];
177
178
struct pollfd fds[2], *ifd, *ofd;
179
@@ -162,12 +184,13 @@ void *central_netdata_push_thread(void *ptr) {
184
if(netdata_exit) break;
185
186
if(unlikely(sock == -1)) {
187
+ rrdpush_connected = 0;
188
+
189
info("PUSH: connecting to central netdata at: %s", central_netdata_to_push_data);
190
sock = connect_to_one_of(central_netdata_to_push_data, 19999, &tv, &reconnects_counter);
191
192
if(unlikely(sock == -1)) {
193
error("PUSH: failed to connect to central netdata at: %s", central_netdata_to_push_data);
170
- sleep(5);
194
continue;
195
}
196
@@ -202,15 +225,9 @@ void *central_netdata_push_thread(void *ptr) {
225
if(fcntl(sock, F_SETFL, O_NONBLOCK) < 0)
226
error("PUSH: cannot set non-blocking mode for socket.");
227
205
- rrdpush_lock();
206
- if(buffer_strlen(rrdpush_buffer))
207
- error("PUSH: discarding %zu bytes of metrics data already in the buffer.", buffer_strlen(rrdpush_buffer));
208
-
209
- buffer_flush(rrdpush_buffer);
210
- reset_all_charts();
211
- last_host = NULL;
212
- rrdpush_unlock();
228
+ rrdpush_flush();
229
sent_connection = 0;
230
+ rrdpush_connected = 1;
231
}
232
233
ifd->fd = rrdpush_pipe[PIPE_READ];
@@ -247,6 +264,7 @@ void *central_netdata_push_thread(void *ptr) {
264
}
265
266
if(ifd->revents & POLLIN) {
267
+ char buffer[1000 + 1];
268
if(read(rrdpush_pipe[PIPE_READ], buffer, 1000) == -1)
269
error("PUSH: Cannot read from internal pipe.");
270
}
@@ -291,8 +309,9 @@ void *central_netdata_push_thread(void *ptr) {
309
}
310
311
debug(D_WEB_CLIENT, "Central netdata push thread exits.");
294
- if(sock != -1)
312
+ if(sock != -1) {
313
close(sock);
314
+ }
315
316
static_thread->enabled = 0;
317
pthread_exit(NULL);