@cryptotaxi247 / netdata-1 / commits / b53dccd58

replication fixes 9 (#14079)

* replication fixes 9 * no room metric is now absolute * decrement senders full and flip the actions on sender full and sender available * finer timings * execute all requests, unconditionally, once they are in the replication queue; worker charts are now sorted * remove left-over debug message * log verification of replication completion only when there are no pending requests any more * use up to 50% of the sender buffer for replication responses

Costa Tsaousis committed Dec 2, 2022 at 11:03 UTC b53dccd58881f2c6f060e50c5414564d3cde8d9c

4 files changed +136 -99

daemon/global_statistics.c

+4 -4

@@ -2322,7 +2322,7 @@ static void workers_utilization_update_chart(struct worker_utilization *wu) {

2322 snprintf(context, RRD_ID_LENGTH_MAX, "netdata.workers.%s.value.%s", wu->name_lowercase, job_name_sanitized);

2323

2324 char title[1000 + 1];

2325 - snprintf(title, 1000, "Netdata Workers %s Value of %s", wu->name_lowercase, string2str(wu->per_job_type[i].name));

2325 + snprintf(title, 1000, "Netdata Workers %s value of %s", wu->name_lowercase, string2str(wu->per_job_type[i].name));

2326

2327 wu->per_job_type[i].st = rrdset_create_localhost(

2328 "netdata"

@@ -2334,7 +2334,7 @@ static void workers_utilization_update_chart(struct worker_utilization *wu) {

2334 , (wu->per_job_type[i].units)?string2str(wu->per_job_type[i].units):"value"

2335 , "netdata"

2336 , "stats"

2337 - , wu->priority + 5

2337 + , wu->priority + 5 + i

2338 , localhost->rrd_update_every

2339 , RRDSET_TYPE_LINE

2340 );

@@ -2378,7 +2378,7 @@ static void workers_utilization_update_chart(struct worker_utilization *wu) {

2378 snprintf(context, RRD_ID_LENGTH_MAX, "netdata.workers.%s.rate.%s", wu->name_lowercase, job_name_sanitized);

2379

2380 char title[1000 + 1];

2381 - snprintf(title, 1000, "Netdata Workers %s Rate of %s", wu->name_lowercase, string2str(wu->per_job_type[i].name));

2381 + snprintf(title, 1000, "Netdata Workers %s rate of %s", wu->name_lowercase, string2str(wu->per_job_type[i].name));

2382

2383 wu->per_job_type[i].st = rrdset_create_localhost(

2384 "netdata"

@@ -2390,7 +2390,7 @@ static void workers_utilization_update_chart(struct worker_utilization *wu) {

2390 , (wu->per_job_type[i].units)?string2str(wu->per_job_type[i].units):"rate"

2391 , "netdata"

2392 , "stats"

2393 - , wu->priority + 5

2393 + , wu->priority + 5 + i

2394 , localhost->rrd_update_every

2395 , RRDSET_TYPE_LINE

2396 );

database/rrdcontext.c

-2

@@ -3253,8 +3253,6 @@ static void rrdcontext_garbage_collect_single_host(RRDHOST *host, bool worker_jo

3253 "RRDCONTEXT: context '%s' of host '%s', deleted from rrdmetrics dictionary.",

3254 string2str(rc->id),

3255 rrdhost_hostname(host));

3256 -

3257 - fprintf(stderr, "RRDCONTEXT: deleted context '%s'", string2str(rc->id));

3256 }

3257

3258 // the item is referenced in the dictionary

streaming/replication.c

+126 -87

@@ -4,7 +4,7 @@

4 #include "Judy.h"

6 #define STREAMING_START_MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED 50

7 -#define MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED 20

7 +#define MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED 50

8 #define MIN_SENDER_BUFFER_PERCENTAGE_ALLOWED 10

10 #define WORKER_JOB_FIND_NEXT 1

@@ -14,17 +14,17 @@

14 #define WORKER_JOB_CHECK_CONSISTENCY 5

15 #define WORKER_JOB_BUFFER_COMMIT 6

16 #define WORKER_JOB_CLEANUP 7

17 +#define WORKER_JOB_WAIT 8

19 // master thread worker jobs

19 -#define WORKER_JOB_STATISTICS 8

20 -#define WORKER_JOB_CUSTOM_METRIC_PENDING_REQUESTS 9

21 -#define WORKER_JOB_CUSTOM_METRIC_COMPLETION 10

22 -#define WORKER_JOB_CUSTOM_METRIC_ADDED 11

23 -#define WORKER_JOB_CUSTOM_METRIC_DONE 12

24 -#define WORKER_JOB_CUSTOM_METRIC_SKIPPED_NOT_CONNECTED 13

25 -#define WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM 14

26 -#define WORKER_JOB_CUSTOM_METRIC_WAITS 15

27 -#define WORKER_JOB_CUSTOM_METRIC_SENDER_RESETS 16

20 +#define WORKER_JOB_STATISTICS 9

21 +#define WORKER_JOB_CUSTOM_METRIC_PENDING_REQUESTS 10

22 +#define WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM 11

23 +#define WORKER_JOB_CUSTOM_METRIC_COMPLETION 12

24 +#define WORKER_JOB_CUSTOM_METRIC_ADDED 13

25 +#define WORKER_JOB_CUSTOM_METRIC_DONE 14

26 +#define WORKER_JOB_CUSTOM_METRIC_SENDER_RESETS 15

27 +#define WORKER_JOB_CUSTOM_METRIC_SENDER_FULL 16

29 #define ITERATIONS_IDLE_WITHOUT_PENDING_TO_RUN_SENDER_VERIFICATION 30

30 #define SECONDS_TO_RESET_POINT_IN_TIME 10

@@ -591,6 +591,7 @@ struct replication_request {

591 Word_t unique_id; // auto-increment, later requests have bigger

592 bool found; // used as a result boolean for the find call

593 bool indexed_in_judy; // true when the request is indexed in judy

594 + bool not_indexed_buffer_full; // true when the request is not indexed because the sender is full

595 };

596

597 // replication sort entry in JudyL array

@@ -605,7 +606,7 @@ struct replication_sort_entry {

606

607 // the global variables for the replication thread

608 static struct replication_thread {

608 - netdata_mutex_t mutex;

609 + SPINLOCK spinlock;

610

611 struct {

612 size_t pending; // number of requests pending in the queue

@@ -614,9 +615,8 @@ static struct replication_thread {

615 // statistics

616 size_t added; // number of requests added to the queue

617 size_t removed; // number of requests removed from the queue

617 - size_t skipped_not_connected; // number of requests skipped, because the sender is not connected to a parent

618 - size_t skipped_no_room; // number of requests skipped, because the sender has no room for responses

619 -// size_t skipped_no_room_since_last_reset;

618 + size_t pending_no_room; // number of requests skipped, because the sender has no room for responses

619 + size_t senders_full; // number of times a sender reset our last position in the queue

620 size_t sender_resets; // number of times a sender reset our last position in the queue

621 time_t first_time_t; // the minimum 'after' we encountered

622

@@ -634,7 +634,6 @@ static struct replication_thread {

634 } atomic; // access should be with atomic operations

635

636 struct {

637 - size_t waits;

637 size_t last_executed; // caching of the atomic.executed to report number of requests executed since last time

638

639 netdata_thread_t **threads_ptrs;

@@ -642,17 +641,16 @@ static struct replication_thread {

641 } main_thread; // access is allowed only by the main thread

642

643 } replication_globals = {

645 - .mutex = NETDATA_MUTEX_INITIALIZER,

644 + .spinlock = NETDATA_SPINLOCK_INITIALIZER,

645 .unsafe = {

646 .pending = 0,

647 .unique_id = 0,

648

649 .added = 0,

650 .removed = 0,

652 - .skipped_not_connected = 0,

653 - .skipped_no_room = 0,

654 -// .skipped_no_room_since_last_reset = 0,

651 + .pending_no_room = 0,

652 .sender_resets = 0,

653 + .senders_full = 0,

654

655 .first_time_t = 0,

656

@@ -667,7 +665,6 @@ static struct replication_thread {

665 .latest_first_time = 0,

666 },

667 .main_thread = {

670 - .waits = 0,

668 .last_executed = 0,

669 .threads = 0,

670 .threads_ptrs = NULL,

@@ -682,11 +679,11 @@ static inline bool replication_recursive_lock_mode(char mode) {

679

680 if(mode == 'L') { // (L)ock

681 if(++recursions == 1)

685 - netdata_mutex_lock(&replication_globals.mutex);

682 + netdata_spinlock_lock(&replication_globals.spinlock);

683 }

684 else if(mode == 'U') { // (U)nlock

685 if(--recursions == 0)

689 - netdata_mutex_unlock(&replication_globals.mutex);

686 + netdata_spinlock_unlock(&replication_globals.spinlock);

687 }

688 else if(mode == 'C') { // (C)heck

689 if(recursions > 0)

@@ -736,6 +733,7 @@ static struct replication_sort_entry *replication_sort_entry_create_unsafe(struc

733 // save the unique id into the request, to be able to delete it later

734 rq->unique_id = rse->unique_id;

735 rq->indexed_in_judy = false;

736 + rq->not_indexed_buffer_full = false;

737 return rse;

738 }

739

@@ -743,9 +741,20 @@ static void replication_sort_entry_destroy(struct replication_sort_entry *rse) {

741 freez(rse);

742 }

743

746 -static struct replication_sort_entry *replication_sort_entry_add(struct replication_request *rq) {

744 +static void replication_sort_entry_add(struct replication_request *rq) {

745 replication_recursive_lock();

746

747 + if(rrdpush_sender_replication_buffer_full_get(rq->sender)) {

748 + rq->indexed_in_judy = false;

749 + rq->not_indexed_buffer_full = true;

750 + replication_globals.unsafe.pending_no_room++;

751 + replication_recursive_unlock();

752 + return;

753 + }

754 +

755 + if(rq->not_indexed_buffer_full)

756 + replication_globals.unsafe.pending_no_room--;

757 +

758 struct replication_sort_entry *rse = replication_sort_entry_create_unsafe(rq);

759

760 // if(rq->after < (time_t)replication_globals.protected.queue.after &&

@@ -770,13 +779,12 @@ static struct replication_sort_entry *replication_sort_entry_add(struct replicat

779 Pvoid_t *item = JudyLIns(inner_judy_ptr, rq->unique_id, PJE0);

780 *item = rse;

781 rq->indexed_in_judy = true;

782 + rq->not_indexed_buffer_full = false;

783

784 if(!replication_globals.unsafe.first_time_t || rq->after < replication_globals.unsafe.first_time_t)

785 replication_globals.unsafe.first_time_t = rq->after;

786

787 replication_recursive_unlock();

778 -

779 - return rse;

788 }

789

790 static bool replication_sort_entry_unlink_and_free_unsafe(struct replication_sort_entry *rse, Pvoid_t **inner_judy_ppptr) {

@@ -806,7 +814,7 @@ static bool replication_sort_entry_unlink_and_free_unsafe(struct replication_sor

814 return inner_judy_deleted;

815 }

816

809 -static void replication_sort_entry_del(struct replication_request *rq) {

817 +static void replication_sort_entry_del(struct replication_request *rq, bool buffer_full) {

818 Pvoid_t *inner_judy_pptr;

819 struct replication_sort_entry *rse_to_delete = NULL;

820

@@ -819,6 +827,11 @@ static void replication_sort_entry_del(struct replication_request *rq) {

827 if (our_item_pptr) {

828 rse_to_delete = *our_item_pptr;

829 replication_sort_entry_unlink_and_free_unsafe(rse_to_delete, &inner_judy_pptr);

830 +

831 + if(buffer_full) {

832 + replication_globals.unsafe.pending_no_room++;

833 + rq->not_indexed_buffer_full = true;

834 + }

835 }

836 }

837

@@ -877,44 +890,17 @@ static struct replication_request replication_request_get_first_available() {

890 while (!rq_to_return.found && (our_item_pptr = JudyLNext(*inner_judy_pptr, &replication_globals.unsafe.queue.unique_id, PJE0))) {

891 struct replication_sort_entry *rse = *our_item_pptr;

892 struct replication_request *rq = rse->rq;

880 - struct sender_state *s = rq->sender;

881 -

882 - if (likely(rrdpush_sender_get_buffer_used_percent(s) <= MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED)) {

883 - // there is room for this request in the sender buffer

884 -

885 - bool sender_is_connected =

886 - rrdhost_flag_check(s->host, RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED);

887 -

888 - bool sender_has_been_flushed_since_this_request =

889 - rq->sender_last_flush_ut != rrdpush_sender_get_flush_time(s);

893

891 - if (unlikely(!sender_is_connected || sender_has_been_flushed_since_this_request)) {

892 - // skip this request, the sender is not connected, or it has reconnected

894 + // copy the request to return it

895 + rq_to_return = *rq;

896 + rq_to_return.chart_id = string_dup(rq_to_return.chart_id);

897

894 - replication_globals.unsafe.skipped_not_connected++;

895 - if (replication_sort_entry_unlink_and_free_unsafe(rse, &inner_judy_pptr))

896 - // we removed the item from the outer JudyL

897 - break;

898 - }

899 - else {

900 - // this request is good to execute

898 + // set the return result to found

899 + rq_to_return.found = true;

900

902 - // copy the request to return it

903 - rq_to_return = *rq;

904 - rq_to_return.chart_id = string_dup(rq_to_return.chart_id);

905 -

906 - // set the return result to found

907 - rq_to_return.found = true;

908 -

909 - if (replication_sort_entry_unlink_and_free_unsafe(rse, &inner_judy_pptr))

910 - // we removed the item from the outer JudyL

911 - break;

912 - }

913 - }

914 - else {

915 - replication_globals.unsafe.skipped_no_room++;

916 -// replication_globals.protected.skipped_no_room_since_last_reset++;

917 - }

901 + if (replication_sort_entry_unlink_and_free_unsafe(rse, &inner_judy_pptr))

902 + // we removed the item from the outer JudyL

903 + break;

904 }

905

906 // call JudyLNext from now on

@@ -959,7 +945,20 @@ static bool replication_request_conflict_callback(const DICTIONARY_ITEM *item __

945

946 replication_recursive_lock();

947

962 - if(!rq->indexed_in_judy) {

948 + if(!rq->indexed_in_judy && rq->not_indexed_buffer_full) {

949 + // we can replace this command

950 + internal_error(

951 + true,

952 + "STREAM %s [send to %s]: REPLAY: 'host:%s/chart:%s' replacing duplicate replication command received (existing from %llu to %llu [%s], new from %llu to %llu [%s])",

953 + rrdhost_hostname(s->host), s->connected_to, rrdhost_hostname(s->host), dictionary_acquired_item_name(item),

954 + (unsigned long long)rq->after, (unsigned long long)rq->before, rq->start_streaming ? "true" : "false",

955 + (unsigned long long)rq_new->after, (unsigned long long)rq_new->before, rq_new->start_streaming ? "true" : "false");

956 +

957 + rq->after = rq_new->after;

958 + rq->before = rq_new->before;

959 + rq->start_streaming = rq_new->start_streaming;

960 + }

961 + else if(!rq->indexed_in_judy) {

962 replication_sort_entry_add(rq);

963 internal_error(

964 true,

@@ -991,7 +990,13 @@ static void replication_request_delete_callback(const DICTIONARY_ITEM *item __ma

990 rrdpush_sender_replicating_charts_minus_one(rq->sender);

991

992 if(rq->indexed_in_judy)

994 - replication_sort_entry_del(rq);

993 + replication_sort_entry_del(rq, false);

994 +

995 + else if(rq->not_indexed_buffer_full) {

996 + replication_recursive_lock();

997 + replication_globals.unsafe.pending_no_room--;

998 + replication_recursive_unlock();

999 + }

1000

1001 string_freez(rq->chart_id);

1002 }

@@ -1046,6 +1051,7 @@ static bool replication_execute_request(struct replication_request *rq, bool wor

1051

1052 cleanup:

1053 string_freez(rq->chart_id);

1054 + worker_is_idle();

1055 return ret;

1056 }

1057

@@ -1060,6 +1066,8 @@ void replication_add_request(struct sender_state *sender, const char *chart_id,

1066 .before = before,

1067 .start_streaming = start_streaming,

1068 .sender_last_flush_ut = rrdpush_sender_get_flush_time(sender),

1069 + .indexed_in_judy = false,

1070 + .not_indexed_buffer_full = false,

1071 };

1072

1073 if(start_streaming && rrdpush_sender_get_buffer_used_percent(sender) <= STREAMING_START_MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED)

@@ -1094,15 +1102,36 @@ void replication_recalculate_buffer_used_ratio_unsafe(struct sender_state *s) {

1102 size_t available = cbuffer_available_size_unsafe(s->host->sender->buffer);

1103 size_t percentage = (s->buffer->max_size - available) * 100 / s->buffer->max_size;

1104

1097 - if(percentage > MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED)

1098 - s->replication.unsafe.reached_max = true;

1105 + if(unlikely(percentage > MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED && !rrdpush_sender_replication_buffer_full_get(s))) {

1106 + rrdpush_sender_replication_buffer_full_set(s, true);

1107 +

1108 + struct replication_request *rq;

1109 + dfe_start_read(s->replication.requests, rq) {

1110 + if(rq->indexed_in_judy && !rq->not_indexed_buffer_full) {

1111 + replication_sort_entry_del(rq, true);

1112 + }

1113 + }

1114 + dfe_done(rq);

1115

1100 - if(s->replication.unsafe.reached_max &&

1101 - percentage <= MIN_SENDER_BUFFER_PERCENTAGE_ALLOWED) {

1102 - s->replication.unsafe.reached_max = false;

1116 replication_recursive_lock();

1104 -// replication_set_next_point_in_time(0, 0);

1117 + replication_globals.unsafe.senders_full++;

1118 + replication_recursive_unlock();

1119 + }

1120 + else if(unlikely(percentage < MIN_SENDER_BUFFER_PERCENTAGE_ALLOWED && rrdpush_sender_replication_buffer_full_get(s))) {

1121 + rrdpush_sender_replication_buffer_full_set(s, false);

1122 +

1123 + struct replication_request *rq;

1124 + dfe_start_read(s->replication.requests, rq) {

1125 + if(!rq->indexed_in_judy && rq->not_indexed_buffer_full) {

1126 + replication_sort_entry_add(rq);

1127 + }

1128 + }

1129 + dfe_done(rq);

1130 +

1131 + replication_recursive_lock();

1132 + replication_globals.unsafe.senders_full--;

1133 replication_globals.unsafe.sender_resets++;

1134 + // replication_set_next_point_in_time(0, 0);

1135 replication_recursive_unlock();

1136 }

1137

@@ -1188,17 +1217,17 @@ static void replication_initialize_workers(bool master) {

1217 worker_register_job_name(WORKER_JOB_CHECK_CONSISTENCY, "check consistency");

1218 worker_register_job_name(WORKER_JOB_BUFFER_COMMIT, "commit");

1219 worker_register_job_name(WORKER_JOB_CLEANUP, "cleanup");

1220 + worker_register_job_name(WORKER_JOB_WAIT, "wait");

1221

1222 if(master) {

1223 worker_register_job_name(WORKER_JOB_STATISTICS, "statistics");

1224 worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_PENDING_REQUESTS, "pending requests", "requests", WORKER_METRIC_ABSOLUTE);

1225 + worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM, "no room requests", "requests", WORKER_METRIC_ABSOLUTE);

1226 worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_COMPLETION, "completion", "%", WORKER_METRIC_ABSOLUTE);

1227 worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_ADDED, "added requests", "requests/s", WORKER_METRIC_INCREMENTAL_TOTAL);

1228 worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_DONE, "finished requests", "requests/s", WORKER_METRIC_INCREMENTAL_TOTAL);

1198 - worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NOT_CONNECTED, "not connected requests", "requests/s", WORKER_METRIC_INCREMENTAL_TOTAL);

1199 - worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM, "no room requests", "requests/s", WORKER_METRIC_INCREMENTAL_TOTAL);

1229 worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_SENDER_RESETS, "sender resets", "resets/s", WORKER_METRIC_INCREMENTAL_TOTAL);

1201 - worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_WAITS, "waits", "waits/s", WORKER_METRIC_INCREMENTAL_TOTAL);

1230 + worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_SENDER_FULL, "senders full", "senders", WORKER_METRIC_ABSOLUTE);

1231 }

1232 }

1233

@@ -1210,8 +1239,10 @@ static int replication_execute_next_pending_request(void) {

1239 worker_is_busy(WORKER_JOB_FIND_NEXT);

1240 struct replication_request rq = replication_request_get_first_available();

1241

1213 - if(unlikely(!rq.found))

1242 + if(unlikely(!rq.found)) {

1243 + worker_is_idle();

1244 return REQUEST_QUEUE_EMPTY;

1245 + }

1246

1247 // delete the request from the dictionary

1248 worker_is_busy(WORKER_JOB_DELETE_ENTRY);

@@ -1221,9 +1252,12 @@ static int replication_execute_next_pending_request(void) {

1252

1253 replication_set_latest_first_time(rq.after);

1254

1224 - if(unlikely(!replication_execute_request(&rq, true)))

1255 + if(unlikely(!replication_execute_request(&rq, true))) {

1256 + worker_is_idle();

1257 return REQUEST_CHART_NOT_FOUND;

1258 + }

1259

1260 + worker_is_idle();

1261 return REQUEST_OK;

1262 }

1263

@@ -1238,6 +1272,7 @@ static void *replication_worker_thread(void *ptr) {

1272

1273 while(!netdata_exit) {

1274 if(unlikely(replication_execute_next_pending_request() == REQUEST_QUEUE_EMPTY)) {

1275 + worker_is_busy(WORKER_JOB_WAIT);

1276 worker_is_idle();

1277 sleep_usec(1 * USEC_PER_SEC);

1278 }

@@ -1305,6 +1340,7 @@ void *replication_thread_main(void *ptr __maybe_unused) {

1340 if(unlikely(now_mono_ut - last_now_mono_ut > default_rrd_update_every * USEC_PER_SEC)) {

1341 last_now_mono_ut = now_mono_ut;

1342

1343 + worker_is_busy(WORKER_JOB_STATISTICS);

1344 replication_recursive_lock();

1345

1346 size_t current_executed = __atomic_load_n(&replication_globals.atomic.executed, __ATOMIC_RELAXED);

@@ -1321,19 +1357,21 @@ void *replication_thread_main(void *ptr __maybe_unused) {

1357 replication_reset_next_point_in_time_countdown = SECONDS_TO_RESET_POINT_IN_TIME;

1358 }

1359

1324 - if(!replication_globals.unsafe.pending && --run_verification_countdown == 0) {

1325 - // reset the statistics about completion percentage

1326 - replication_globals.unsafe.first_time_t = 0;

1327 - replication_set_latest_first_time(0);

1360 + if(--run_verification_countdown == 0) {

1361 + if (!replication_globals.unsafe.pending && !replication_globals.unsafe.pending_no_room) {

1362 + // reset the statistics about completion percentage

1363 + replication_globals.unsafe.first_time_t = 0;

1364 + replication_set_latest_first_time(0);

1365

1329 - verify_all_hosts_charts_are_streaming_now();

1366 + verify_all_hosts_charts_are_streaming_now();

1367

1331 - run_verification_countdown = LONG_MAX;

1332 - slow = true;

1368 + run_verification_countdown = LONG_MAX;

1369 + slow = true;

1370 + }

1371 + else

1372 + run_verification_countdown = ITERATIONS_IDLE_WITHOUT_PENDING_TO_RUN_SENDER_VERIFICATION;

1373 }

1374

1335 - worker_is_busy(WORKER_JOB_STATISTICS);

1336 -

1375 time_t latest_first_time_t = replication_get_latest_first_time();

1376 if(latest_first_time_t && replication_globals.unsafe.pending) {

1377 // completion percentage statistics

@@ -1349,15 +1387,17 @@ void *replication_thread_main(void *ptr __maybe_unused) {

1387 worker_set_metric(WORKER_JOB_CUSTOM_METRIC_PENDING_REQUESTS, (NETDATA_DOUBLE)replication_globals.unsafe.pending);

1388 worker_set_metric(WORKER_JOB_CUSTOM_METRIC_ADDED, (NETDATA_DOUBLE)replication_globals.unsafe.added);

1389 worker_set_metric(WORKER_JOB_CUSTOM_METRIC_DONE, (NETDATA_DOUBLE)__atomic_load_n(&replication_globals.atomic.executed, __ATOMIC_RELAXED));

1352 - worker_set_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NOT_CONNECTED, (NETDATA_DOUBLE)replication_globals.unsafe.skipped_not_connected);

1353 - worker_set_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM, (NETDATA_DOUBLE)replication_globals.unsafe.skipped_no_room);

1390 + worker_set_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM, (NETDATA_DOUBLE)replication_globals.unsafe.pending_no_room);

1391 worker_set_metric(WORKER_JOB_CUSTOM_METRIC_SENDER_RESETS, (NETDATA_DOUBLE)replication_globals.unsafe.sender_resets);

1355 - worker_set_metric(WORKER_JOB_CUSTOM_METRIC_WAITS, (NETDATA_DOUBLE)replication_globals.main_thread.waits);

1392 + worker_set_metric(WORKER_JOB_CUSTOM_METRIC_SENDER_FULL, (NETDATA_DOUBLE)replication_globals.unsafe.senders_full);

1393

1394 replication_recursive_unlock();

1395 + worker_is_idle();

1396 }

1397

1398 if(unlikely(replication_execute_next_pending_request() == REQUEST_QUEUE_EMPTY)) {

1399 +

1400 + worker_is_busy(WORKER_JOB_WAIT);

1401 replication_recursive_lock();

1402

1403 // the timeout also defines now frequently we will traverse all the pending requests

@@ -1388,7 +1428,6 @@ void *replication_thread_main(void *ptr __maybe_unused) {

1428 last_sender_resets = replication_globals.unsafe.sender_resets;

1429 }

1430

1391 - replication_globals.main_thread.waits++;

1431 replication_recursive_unlock();

1432

1433 worker_is_idle();

streaming/rrdpush.h

+6 -6

@@ -168,12 +168,9 @@ struct sender_state {

168 struct {

169 size_t pending_requests; // the currently outstanding replication requests

170 size_t charts_replicating; // the number of unique charts having pending replication requests (on every request one is added and is removed when we finish it - it does not track completion of the replication for this chart)

171 + bool reached_max; // true when the sender buffer should not get more replication responses

172 } atomic;

173

173 - struct {

174 - bool reached_max; // used to avoid resetting the replication thread too frequently

175 - } unsafe; // protected by sender mutex

176 -

174 } replication;

175

176 struct {

@@ -182,10 +179,13 @@ struct sender_state {

179 } atomic;

180 };

181

185 -#define rrdpush_sender_set_buffer_used_percent(sender, value) __atomic_store_n(&((sender)->atomic.buffer_used_percentage), value, __ATOMIC_RELAXED);

182 +#define rrdpush_sender_replication_buffer_full_set(sender, value) __atomic_store_n(&((sender)->replication.atomic.reached_max), value, __ATOMIC_SEQ_CST)

183 +#define rrdpush_sender_replication_buffer_full_get(sender) __atomic_load_n(&((sender)->replication.atomic.reached_max), __ATOMIC_SEQ_CST)

184 +

185 +#define rrdpush_sender_set_buffer_used_percent(sender, value) __atomic_store_n(&((sender)->atomic.buffer_used_percentage), value, __ATOMIC_RELAXED)

186 #define rrdpush_sender_get_buffer_used_percent(sender) __atomic_load_n(&((sender)->atomic.buffer_used_percentage), __ATOMIC_RELAXED)

187

188 -#define rrdpush_sender_set_flush_time(sender) __atomic_store_n(&((sender)->atomic.last_flush_time_ut), now_realtime_usec(), __ATOMIC_RELAXED);

188 +#define rrdpush_sender_set_flush_time(sender) __atomic_store_n(&((sender)->atomic.last_flush_time_ut), now_realtime_usec(), __ATOMIC_RELAXED)

189 #define rrdpush_sender_get_flush_time(sender) __atomic_load_n(&((sender)->atomic.last_flush_time_ut), __ATOMIC_RELAXED)

190

191 #define rrdpush_sender_replicating_charts(sender) __atomic_load_n(&((sender)->replication.atomic.charts_replicating), __ATOMIC_RELAXED)