@cryptotaxi247 / netdata-1 / commits / 1c846d568

Remove broken optimization in the sender thread (#9703)

The sender thread avoided locking the circular buffer to check if there was outstanding data on the connection. The condition it needs (unsent data) grows monotonically w.r.t. other threads as the collectors can add data but only this thread can remove it. However, it cached the pointer into the buffer as a side-effect and then reused it later during the transmission. This fails if the buffer is resized by a collector thread. Peeking at the buffer sizes without locking could fail in the same situation. The optimization is removed and the sender thread now locks the mutex before checking the buffer, throws away the data buffer pointer and releases the mutex over the poll() operation. It then reacquires the mutex and checks the buffer size and data pointer again when it performs the send.

Andrew Moss committed Aug 11, 2020 at 16:16 UTC 1c846d568d8ad075fbc3444d35f9c93b8aaf028c

1 file changed +13 -8

streaming/sender.c

+13 -8

@@ -418,15 +418,17 @@ static void attempt_to_connect(struct sender_state *state)

418 }

419

420 // TCP window is open and we have data to transmit.

421 -void attempt_to_send(struct sender_state *s, char *chunk, size_t outstanding) {

421 +void attempt_to_send(struct sender_state *s) {

422 +

423 rrdpush_send_labels(s->host);

424

424 - struct circular_buffer *cb = s->host->sender->buffer;

425 - debug(D_STREAM, "STREAM: Sending data. Buffer r=%zu w=%zu s=%zu, next chunk=%zu", cb->read, cb->write, cb->size, outstanding);

425 + struct circular_buffer *cb = s->buffer;

426

427 netdata_thread_disable_cancelability();

428 - netdata_mutex_lock(&s->host->sender->mutex);

429 -

428 + netdata_mutex_lock(&s->mutex);

429 + char *chunk;

430 + size_t outstanding = cbuffer_next_unsafe(s->buffer, &chunk);

431 + debug(D_STREAM, "STREAM: Sending data. Buffer r=%zu w=%zu s=%zu, next chunk=%zu", cb->read, cb->write, cb->size, outstanding);

432 ssize_t ret;

433 #ifdef ENABLE_HTTPS

434 SSL *conn = s->host->ssl.conn ;

@@ -439,7 +441,7 @@ void attempt_to_send(struct sender_state *s, char *chunk, size_t outstanding) {

441 ret = send(s->host->rrdpush_sender_socket, chunk, outstanding, MSG_DONTWAIT);

442 #endif

443 if (likely(ret > 0)) {

442 - cbuffer_remove_unsafe(s->host->sender->buffer, ret);

444 + cbuffer_remove_unsafe(s->buffer, ret);

445 s->sent_bytes_on_this_connection += ret;

446 s->sent_bytes += ret;

447 debug(D_STREAM, "STREAM %s [send to %s]: Sent %zd bytes", s->host->hostname, s->connected_to, ret);

@@ -457,7 +459,7 @@ void attempt_to_send(struct sender_state *s, char *chunk, size_t outstanding) {

459 debug(D_STREAM, "STREAM: send() returned 0 -> no error but no transmission");

460 }

461

460 - netdata_mutex_unlock(&s->host->sender->mutex);

462 + netdata_mutex_unlock(&s->mutex);

463 netdata_thread_enable_cancelability();

464 }

465

@@ -635,8 +637,11 @@ void *rrdpush_sender_thread(void *ptr) {

637 fds[Socket].revents = 0;

638 fds[Socket].fd = s->host->rrdpush_sender_socket;

639

640 + netdata_mutex_lock(&s->mutex);

641 char *chunk;

642 size_t outstanding = cbuffer_next_unsafe(s->host->sender->buffer, &chunk);

643 + chunk = NULL; // Do not cache pointer outside of region - could be invalidated

644 + netdata_mutex_unlock(&s->mutex);

645 if(outstanding) {

646 s->send_attempts++;

647 fds[Socket].events = POLLIN | POLLOUT;

@@ -679,7 +684,7 @@ void *rrdpush_sender_thread(void *ptr) {

684

685 // If we have data and have seen the TCP window open then try to close it by a transmission.

686 if (outstanding && fds[Socket].revents & POLLOUT)

682 - attempt_to_send(s, chunk, outstanding);

687 + attempt_to_send(s);

688

689 // TODO-GAPS - why do we only check this on the socket, not the pipe?

690 if (outstanding) {