Replication fixes #3 (#14035)

* cleanup and additional information about replication

* fix deadlock on sender mutex

* do not ignore start streaming empty requests; when there duplicate requests, merge them

* flipped the flag

* final touch

* added queued flag on the charts to prevent them from being obsoleted by the service thread

1 files changed, 23 insertions, 17 deletions

diff --git a/daemon/service.c b/daemon/service.c
index 28fdc84756..a7db7ceb79 100644
--- a/daemon/service.c
+++ b/daemon/service.c
@@ -138,6 +138,9 @@ static void svc_rrdhost_cleanup_obsolete_charts(RRDHOST *host) {
     time_t now = now_realtime_sec();
     RRDSET *st;
     rrdset_foreach_reentrant(st, host) {
+        if(rrdset_is_replicating(st))
+            continue;
+
         if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_OBSOLETE)
                      && st->last_accessed_time + rrdset_free_obsolete_time < now
                      && st->last_updated.tv_sec + rrdset_free_obsolete_time < now
@@ -160,7 +163,7 @@ static void svc_rrdset_check_obsoletion(RRDHOST *host) {
     time_t last_entry_t;
     RRDSET *st;
     rrdset_foreach_read(st, host) {
-        if(!rrdset_flag_check(st, RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED))
+        if(rrdset_is_replicating(st))
             continue;
 
         last_entry_t = rrdset_last_entry_t(st);
@@ -181,6 +184,8 @@ static void svc_rrd_cleanup_obsolete_charts_from_all_hosts() {
 
     RRDHOST *host;
     rrdhost_foreach_read(host) {
+        if(rrdhost_receiver_replicating_charts(host) || rrdhost_sender_replicating_charts(host))
+            continue;
 
         if(rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_OBSOLETE_CHARTS|RRDHOST_FLAG_PENDING_OBSOLETE_DIMENSIONS)) {
             rrdhost_flag_clear(host, RRDHOST_FLAG_PENDING_OBSOLETE_CHARTS|RRDHOST_FLAG_PENDING_OBSOLETE_DIMENSIONS);
@@ -215,25 +220,26 @@ static void svc_rrdhost_cleanup_orphan_hosts(RRDHOST *protected_host) {
 
 restart_after_removal:
     rrdhost_foreach_write(host) {
-        if(rrdhost_should_be_removed(host, protected_host, now)) {
-            info("Host '%s' with machine guid '%s' is obsolete - cleaning up.", rrdhost_hostname(host), host->machine_guid);
+        if(!rrdhost_should_be_removed(host, protected_host, now))
+            continue;
 
-            if (rrdhost_option_check(host, RRDHOST_OPTION_DELETE_ORPHAN_HOST)
-                /* don't delete multi-host DB host files */
-                && !(host->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE && is_storage_engine_shared(host->db[0].instance))
-            ) {
-                worker_is_busy(WORKER_JOB_DELETE_HOST_CHARTS);
-                rrdhost_delete_charts(host);
-            }
-            else {
-                worker_is_busy(WORKER_JOB_SAVE_HOST_CHARTS);
-                rrdhost_save_charts(host);
-            }
+        info("Host '%s' with machine guid '%s' is obsolete - cleaning up.", rrdhost_hostname(host), host->machine_guid);
 
-            worker_is_busy(WORKER_JOB_FREE_HOST);
-            rrdhost_free(host, 0);
-            goto restart_after_removal;
+        if (rrdhost_option_check(host, RRDHOST_OPTION_DELETE_ORPHAN_HOST)
+            /* don't delete multi-host DB host files */
+            && !(host->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE && is_storage_engine_shared(host->db[0].instance))
+        ) {
+            worker_is_busy(WORKER_JOB_DELETE_HOST_CHARTS);
+            rrdhost_delete_charts(host);
         }
+        else {
+            worker_is_busy(WORKER_JOB_SAVE_HOST_CHARTS);
+            rrdhost_save_charts(host);
+        }
+
+        worker_is_busy(WORKER_JOB_FREE_HOST);
+        rrdhost_free(host, 0);
+        goto restart_after_removal;
     }
 
     rrd_unlock();