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

1 files changed, 6 insertions, 6 deletions

diff --git a/streaming/rrdpush.h b/streaming/rrdpush.h
index c5f7618c13..a0c7e8de23 100644
--- a/streaming/rrdpush.h
+++ b/streaming/rrdpush.h
@@ -168,12 +168,9 @@ struct sender_state {
         struct {
             size_t pending_requests;            // the currently outstanding replication requests
             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)
+            bool reached_max;                   // true when the sender buffer should not get more replication responses
         } atomic;
 
-        struct {
-            bool reached_max;                   // used to avoid resetting the replication thread too frequently
-        } unsafe;                               // protected by sender mutex
-
     } replication;
 
     struct {
@@ -182,10 +179,13 @@ struct sender_state {
     } atomic;
 };
 
-#define rrdpush_sender_set_buffer_used_percent(sender, value) __atomic_store_n(&((sender)->atomic.buffer_used_percentage), value, __ATOMIC_RELAXED);
+#define rrdpush_sender_replication_buffer_full_set(sender, value) __atomic_store_n(&((sender)->replication.atomic.reached_max), value, __ATOMIC_SEQ_CST)
+#define rrdpush_sender_replication_buffer_full_get(sender) __atomic_load_n(&((sender)->replication.atomic.reached_max), __ATOMIC_SEQ_CST)
+
+#define rrdpush_sender_set_buffer_used_percent(sender, value) __atomic_store_n(&((sender)->atomic.buffer_used_percentage), value, __ATOMIC_RELAXED)
 #define rrdpush_sender_get_buffer_used_percent(sender) __atomic_load_n(&((sender)->atomic.buffer_used_percentage), __ATOMIC_RELAXED)
 
-#define rrdpush_sender_set_flush_time(sender) __atomic_store_n(&((sender)->atomic.last_flush_time_ut), now_realtime_usec(), __ATOMIC_RELAXED);
+#define rrdpush_sender_set_flush_time(sender) __atomic_store_n(&((sender)->atomic.last_flush_time_ut), now_realtime_usec(), __ATOMIC_RELAXED)
 #define rrdpush_sender_get_flush_time(sender) __atomic_load_n(&((sender)->atomic.last_flush_time_ut), __ATOMIC_RELAXED)
 
 #define rrdpush_sender_replicating_charts(sender) __atomic_load_n(&((sender)->replication.atomic.charts_replicating), __ATOMIC_RELAXED)