streaming compression, query planner and replication fixes (#14023)

* streaming compression, query planner and replication fixes

* remove journal v2 stats from global statistics

* disable sql for checking past sql UUIDs

* single threaded replication

* final replication thread using dictionaries and JudyL for sorting the pending requests

* do not timeout the sending socket when there are pending replication requests

* streaming receiver using read() instead of fread()

* remove FILE * from streaming - now using posix read() and write()

* increase timeouts to 10 minutes

* apply sender timeout only when there are metrics that are supposed to be streamed

* error handling in replication

* remove retries on socket read timeout; better error messages

* take into account inbound traffic too to detect that a connection is stale

* remove race conditions from replication thread

* make sure deleted entries are marked as executed, so that even if deletion fails, they will not be executed

* 2 minutes timeout to retry streaming to a parent that already has this node

* remove unecessary condition check

* fix compilation warnings

* include judy in replication

* wrappers to handle retries for SSL_read and SSL_write

* compressed bytes read monitoring

* recursive locks on replication to make it faster during flush or cleanup

* replication completion chart at the receiver side

* simplified recursive mutex

* simplified recursive mutex again

2 files changed, 12 insertions, 5 deletions

diff --git a/libnetdata/dictionary/dictionary.c b/libnetdata/dictionary/dictionary.c
index e362acdd8b..29456200a9 100644
--- a/libnetdata/dictionary/dictionary.c
+++ b/libnetdata/dictionary/dictionary.c
@@ -2171,11 +2171,13 @@ int dictionary_walkthrough_rw(DICTIONARY *dict, char rw, int (*callback)(const D
 // ----------------------------------------------------------------------------
 // sorted walkthrough
 
+typedef int (*qsort_compar)(const void *item1, const void *item2);
+
 static int dictionary_sort_compar(const void *item1, const void *item2) {
     return strcmp(item_get_name((*(DICTIONARY_ITEM **)item1)), item_get_name((*(DICTIONARY_ITEM **)item2)));
 }
 
-int dictionary_sorted_walkthrough_rw(DICTIONARY *dict, char rw, int (*callback)(const DICTIONARY_ITEM *item, void *entry, void *data), void *data) {
+int dictionary_sorted_walkthrough_rw(DICTIONARY *dict, char rw, int (*callback)(const DICTIONARY_ITEM *item, void *entry, void *data), void *data, dictionary_sorted_compar compar) {
     if(unlikely(!dict || !callback)) return 0;
 
     if(unlikely(is_dictionary_destroyed(dict))) {
@@ -2200,7 +2202,10 @@ int dictionary_sorted_walkthrough_rw(DICTIONARY *dict, char rw, int (*callback)(
     if(unlikely(i != entries))
         entries = i;
 
-    qsort(array, entries, sizeof(DICTIONARY_ITEM *), dictionary_sort_compar);
+    if(compar)
+        qsort(array, entries, sizeof(DICTIONARY_ITEM *), (qsort_compar)compar);
+    else
+        qsort(array, entries, sizeof(DICTIONARY_ITEM *), dictionary_sort_compar);
 
     bool callit = true;
     int ret = 0, r;
diff --git a/libnetdata/dictionary/dictionary.h b/libnetdata/dictionary/dictionary.h
index d002385cb3..4e2ba751fa 100644
--- a/libnetdata/dictionary/dictionary.h
+++ b/libnetdata/dictionary/dictionary.h
@@ -230,9 +230,11 @@ size_t dictionary_acquired_item_references(DICT_ITEM_CONST DICTIONARY_ITEM *item
 #define dictionary_walkthrough_write(dict, callback, data) dictionary_walkthrough_rw(dict, 'w', callback, data)
 int dictionary_walkthrough_rw(DICTIONARY *dict, char rw, int (*callback)(const DICTIONARY_ITEM *item, void *value, void *data), void *data);
 
-#define dictionary_sorted_walkthrough_read(dict, callback, data) dictionary_sorted_walkthrough_rw(dict, 'r', callback, data)
-#define dictionary_sorted_walkthrough_write(dict, callback, data) dictionary_sorted_walkthrough_rw(dict, 'w', callback, data)
-int dictionary_sorted_walkthrough_rw(DICTIONARY *dict, char rw, int (*callback)(const DICTIONARY_ITEM *item, void *entry, void *data), void *data);
+typedef int (*dictionary_sorted_compar)(const DICTIONARY_ITEM **item1, const DICTIONARY_ITEM **item2);
+
+#define dictionary_sorted_walkthrough_read(dict, callback, data) dictionary_sorted_walkthrough_rw(dict, 'r', callback, data, NULL)
+#define dictionary_sorted_walkthrough_write(dict, callback, data) dictionary_sorted_walkthrough_rw(dict, 'w', callback, data, NULL)
+int dictionary_sorted_walkthrough_rw(DICTIONARY *dict, char rw, int (*callback)(const DICTIONARY_ITEM *item, void *entry, void *data), void *data, dictionary_sorted_compar compar);
 
 // ----------------------------------------------------------------------------
 // Traverse with foreach