Adding upstream version 1.36.0.upstream/1.36.0

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 1461 insertions, 334 deletions

diff --git a/libnetdata/dictionary/dictionary.c b/libnetdata/dictionary/dictionary.c
index 42285037d..c1325ecb5 100644
--- a/libnetdata/dictionary/dictionary.c
+++ b/libnetdata/dictionary/dictionary.c
@@ -1,7 +1,12 @@
 // SPDX-License-Identifier: GPL-3.0-or-later
 
-// NOT TO BE USED BY USERS YET
-#define DICTIONARY_FLAG_REFERENCE_COUNTERS (1 << 6) // maintain reference counter in walkthrough and foreach
+// NOT TO BE USED BY USERS
+#define DICTIONARY_FLAG_EXCLUSIVE_ACCESS    (1 << 29) // there is only one thread accessing the dictionary
+#define DICTIONARY_FLAG_DESTROYED           (1 << 30) // this dictionary has been destroyed
+#define DICTIONARY_FLAG_DEFER_ALL_DELETIONS (1 << 31) // defer all deletions of items in the dictionary
+
+// our reserved flags that cannot be set by users
+#define DICTIONARY_FLAGS_RESERVED (DICTIONARY_FLAG_EXCLUSIVE_ACCESS|DICTIONARY_FLAG_DESTROYED|DICTIONARY_FLAG_DEFER_ALL_DELETIONS)
 
 typedef struct dictionary DICTIONARY;
 #define DICTIONARY_INTERNALS
@@ -19,99 +24,49 @@ typedef struct dictionary DICTIONARY;
 #include <Judy.h>
 #endif
 
-/*
- * This version uses JudyHS arrays to index the dictionary
- *
- * The following output is from the unit test, at the end of this file:
- *
- * This is the JudyHS version:
- *
- * 1000000 x dictionary_set() (dictionary size 0 entries, 0 KB)...
- * 1000000 x dictionary_get(existing) (dictionary size 1000000 entries, 74001 KB)...
- * 1000000 x dictionary_get(non-existing) (dictionary size 1000000 entries, 74001 KB)...
- * Walking through the dictionary (dictionary size 1000000 entries, 74001 KB)...
- * 1000000 x dictionary_del(existing) (dictionary size 1000000 entries, 74001 KB)...
- * 1000000 x dictionary_set() (dictionary size 0 entries, 0 KB)...
- * Destroying dictionary (dictionary size 1000000 entries, 74001 KB)...
- *
- * TIMINGS:
- * adding 316027 usec, positive search 156740 usec, negative search 84524, walk through 15036 usec, deleting 361444, destroy 107394 usec
- *
- * This is from the JudySL version:
- *
- * Creating dictionary of 1000000 entries...
- * Checking index of 1000000 entries...
- * Walking 1000000 entries and checking name-value pairs...
- * Created and checked 1000000 entries, found 0 errors - used 58376 KB of memory
- * Destroying dictionary of 1000000 entries...
- * Deleted 1000000 entries
- * create 338975 usec, check 156080 usec, walk 80764 usec, destroy 444569 usec
- *
- * This is the AVL version:
- *
- * Creating dictionary of 1000000 entries...
- * Checking index of 1000000 entries...
- * Walking 1000000 entries and checking name-value pairs...
- * Created and checked 1000000 entries, found 0 errors - used 89626 KB of memory
- * Destroying dictionary of 1000000 entries...
- * create 413892 usec, check 220006 usec, walk 34247 usec, destroy 98062 usec
- *
- * So, the JudySL is a lot slower to WALK and DESTROY (DESTROY does a WALK)
- * It is slower, because for every item, JudySL copies the KEY/NAME to a
- * caller supplied buffer (Index). So, by just walking over 1 million items,
- * JudySL does 1 million strcpy() !!!
- *
- * It also seems that somehow JudySLDel() is unbelievably slow too!
- *
- */
+typedef enum name_value_flags {
+    NAME_VALUE_FLAG_NONE                   = 0,
+    NAME_VALUE_FLAG_NAME_IS_ALLOCATED      = (1 << 0), // the name pointer is a STRING
+    NAME_VALUE_FLAG_DELETED                = (1 << 1), // this item is deleted, so it is not available for traversal
+    NAME_VALUE_FLAG_NEW_OR_UPDATED         = (1 << 2), // this item is new or just updated (used by the react callback)
 
+    // IMPORTANT: IF YOU ADD ANOTHER FLAG, YOU NEED TO ALLOCATE ANOTHER BIT TO FLAGS IN NAME_VALUE !!!
+} NAME_VALUE_FLAGS;
 
 /*
  * Every item in the dictionary has the following structure.
  */
+
 typedef struct name_value {
 #ifdef DICTIONARY_WITH_AVL
     avl_t avl_node;
 #endif
 
+#ifdef NETDATA_INTERNAL_CHECKS
+    DICTIONARY *dict;
+#endif
+
     struct name_value *next;    // a double linked list to allow fast insertions and deletions
     struct name_value *prev;
 
-    char *name;                 // the name of the dictionary item
+    uint32_t refcount;          // the reference counter
+    uint32_t value_len:29;      // the size of the value (assumed binary)
+    uint8_t flags:3;            // the flags for this item
+
     void *value;                // the value of the dictionary item
+    union {
+        STRING *string_name;    // the name of the dictionary item
+        char *caller_name;      // the user supplied string pointer
+    };
 } NAME_VALUE;
 
-/*
- * When DICTIONARY_FLAG_WITH_STATISTICS is set, we need to keep track of all the memory
- * we allocate and free. So, we need to keep track of the sizes of all names and values.
- * We do this by overloading NAME_VALUE with the following additional fields.
- */
-
-typedef enum name_value_flags {
-    NAME_VALUE_FLAG_NONE                   = 0,
-    NAME_VALUE_FLAG_DELETED                = (1 << 0), // this item is deleted
-} NAME_VALUE_FLAGS;
-
-typedef struct name_value_with_stats {
-    NAME_VALUE name_value_data_here;    // never used - just to put the lengths at the right position
-
-    size_t name_len;                    // the size of the name, including the terminating zero
-    size_t value_len;                   // the size of the value (assumed binary)
-
-    size_t refcount;                    // the reference counter
-    NAME_VALUE_FLAGS flags;             // the flags for this item
-} NAME_VALUE_WITH_STATS;
-
-struct dictionary_stats {
-    size_t inserts;
-    size_t deletes;
-    size_t searches;
-    size_t resets;
-    size_t entries;
-    size_t memory;
-};
-
 struct dictionary {
+#ifdef NETDATA_INTERNAL_CHECKS
+    const char *creation_function;
+    const char *creation_file;
+    size_t creation_line;
+#endif
+
     DICTIONARY_FLAGS flags;             // the flags of the dictionary
 
     NAME_VALUE *first_item;             // the double linked list base pointers
@@ -120,26 +75,51 @@ struct dictionary {
 #ifdef DICTIONARY_WITH_AVL
     avl_tree_type values_index;
     NAME_VALUE *hash_base;
+    void *(*get_thread_static_name_value)(const char *name);
 #endif
 
 #ifdef DICTIONARY_WITH_JUDYHS
     Pvoid_t JudyHSArray;                // the hash table
 #endif
 
-    netdata_rwlock_t *rwlock;           // the r/w lock when DICTIONARY_FLAG_SINGLE_THREADED is not set
+    netdata_rwlock_t rwlock;            // the r/w lock when DICTIONARY_FLAG_SINGLE_THREADED is not set
 
     void (*ins_callback)(const char *name, void *value, void *data);
     void *ins_callback_data;
 
+    void (*react_callback)(const char *name, void *value, void *data);
+    void *react_callback_data;
+
     void (*del_callback)(const char *name, void *value, void *data);
     void *del_callback_data;
 
     void (*conflict_callback)(const char *name, void *old_value, void *new_value, void *data);
     void *conflict_callback_data;
 
-    struct dictionary_stats *stats;     // the statistics when DICTIONARY_FLAG_WITH_STATISTICS is set
+    size_t version;                   // the current version of the dictionary
+    size_t inserts;                   // how many index insertions have been performed
+    size_t deletes;                   // how many index deletions have been performed
+    size_t searches;                  // how many index searches have been performed
+    size_t resets;                    // how many times items have reset their values
+    size_t walkthroughs;              // how many walkthroughs have been done
+    long int memory;                  // how much memory the dictionary has currently allocated
+    long int entries;                 // how many items are currently in the index (the linked list may have more)
+    long int referenced_items;        // how many items of the dictionary are currently being used by 3rd parties
+    long int pending_deletion_items;  // how many items of the dictionary have been deleted, but have not been removed yet
+    int readers;                      // how many readers are currently using the dictionary
+    int writers;                      // how many writers are currently using the dictionary
+
+    size_t scratchpad_size;           // the size of the scratchpad in bytes
+    uint8_t scratchpad[];             // variable size scratchpad requested by the caller
 };
 
+static inline void linkedlist_namevalue_unlink_unsafe(DICTIONARY *dict, NAME_VALUE *nv);
+static size_t namevalue_destroy_unsafe(DICTIONARY *dict, NAME_VALUE *nv);
+static inline const char *namevalue_get_name(NAME_VALUE *nv);
+
+// ----------------------------------------------------------------------------
+// callbacks registration
+
 void dictionary_register_insert_callback(DICTIONARY *dict, void (*ins_callback)(const char *name, void *value, void *data), void *data) {
     dict->ins_callback = ins_callback;
     dict->ins_callback_data = data;
@@ -155,63 +135,156 @@ void dictionary_register_conflict_callback(DICTIONARY *dict, void (*conflict_cal
     dict->conflict_callback_data = data;
 }
 
+void dictionary_register_react_callback(DICTIONARY *dict, void (*react_callback)(const char *name, void *value, void *data), void *data) {
+    dict->react_callback = react_callback;
+    dict->react_callback_data = data;
+}
+
 // ----------------------------------------------------------------------------
 // dictionary statistics maintenance
 
-size_t dictionary_stats_allocated_memory(DICTIONARY *dict) {
-    if(unlikely(dict->flags & DICTIONARY_FLAG_WITH_STATISTICS))
-        return dict->stats->memory;
-    return 0;
+long int dictionary_stats_allocated_memory(DICTIONARY *dict) {
+    return dict->memory;
 }
-size_t dictionary_stats_entries(DICTIONARY *dict) {
-    if(unlikely(dict->flags & DICTIONARY_FLAG_WITH_STATISTICS))
-        return dict->stats->entries;
-    return 0;
+long int dictionary_stats_entries(DICTIONARY *dict) {
+    return dict->entries;
+}
+size_t dictionary_stats_version(DICTIONARY *dict) {
+    return dict->version;
 }
 size_t dictionary_stats_searches(DICTIONARY *dict) {
-    if(unlikely(dict->flags & DICTIONARY_FLAG_WITH_STATISTICS))
-        return dict->stats->searches;
-    return 0;
+    return dict->searches;
 }
 size_t dictionary_stats_inserts(DICTIONARY *dict) {
-    if(unlikely(dict->flags & DICTIONARY_FLAG_WITH_STATISTICS))
-        return dict->stats->inserts;
-    return 0;
+    return dict->inserts;
 }
 size_t dictionary_stats_deletes(DICTIONARY *dict) {
-    if(unlikely(dict->flags & DICTIONARY_FLAG_WITH_STATISTICS))
-        return dict->stats->deletes;
-    return 0;
+    return dict->deletes;
 }
 size_t dictionary_stats_resets(DICTIONARY *dict) {
-    if(unlikely(dict->flags & DICTIONARY_FLAG_WITH_STATISTICS))
-        return dict->stats->resets;
-    return 0;
+    return dict->resets;
+}
+size_t dictionary_stats_walkthroughs(DICTIONARY *dict) {
+    return dict->walkthroughs;
+}
+size_t dictionary_stats_referenced_items(DICTIONARY *dict) {
+    return __atomic_load_n(&dict->referenced_items, __ATOMIC_SEQ_CST);
 }
 
 static inline void DICTIONARY_STATS_SEARCHES_PLUS1(DICTIONARY *dict) {
-    if(unlikely(dict->flags & DICTIONARY_FLAG_WITH_STATISTICS))
-        dict->stats->searches++;
+    if(dict->flags & DICTIONARY_FLAG_EXCLUSIVE_ACCESS) {
+        dict->searches++;
+    }
+    else {
+        __atomic_fetch_add(&dict->searches, 1, __ATOMIC_RELAXED);
+    }
 }
 static inline void DICTIONARY_STATS_ENTRIES_PLUS1(DICTIONARY *dict, size_t size) {
-    if(unlikely(dict->flags & DICTIONARY_FLAG_WITH_STATISTICS)) {
-        dict->stats->inserts++;
-        dict->stats->entries++;
-        dict->stats->memory += size;
+    if(dict->flags & DICTIONARY_FLAG_EXCLUSIVE_ACCESS) {
+        dict->version++;
+        dict->inserts++;
+        dict->entries++;
+        dict->memory += (long)size;
+    }
+    else {
+        __atomic_fetch_add(&dict->version, 1, __ATOMIC_SEQ_CST);
+        __atomic_fetch_add(&dict->inserts, 1, __ATOMIC_RELAXED);
+        __atomic_fetch_add(&dict->entries, 1, __ATOMIC_RELAXED);
+        __atomic_fetch_add(&dict->memory, (long)size, __ATOMIC_RELAXED);
     }
 }
-static inline void DICTIONARY_STATS_ENTRIES_MINUS1(DICTIONARY *dict, size_t size) {
-    if(unlikely(dict->flags & DICTIONARY_FLAG_WITH_STATISTICS)) {
-        dict->stats->deletes++;
-        dict->stats->entries--;
-        dict->stats->memory -= size;
+static inline void DICTIONARY_STATS_ENTRIES_MINUS1(DICTIONARY *dict) {
+    if(dict->flags & DICTIONARY_FLAG_EXCLUSIVE_ACCESS) {
+        dict->version++;
+        dict->deletes++;
+        dict->entries--;
+    }
+    else {
+        __atomic_fetch_add(&dict->version, 1, __ATOMIC_SEQ_CST);
+        __atomic_fetch_add(&dict->deletes, 1, __ATOMIC_RELAXED);
+        __atomic_fetch_sub(&dict->entries, 1, __ATOMIC_RELAXED);
+    }
+}
+static inline void DICTIONARY_STATS_ENTRIES_MINUS_MEMORY(DICTIONARY *dict, size_t size) {
+    if(dict->flags & DICTIONARY_FLAG_EXCLUSIVE_ACCESS) {
+        dict->memory -= (long)size;
+    }
+    else {
+        __atomic_fetch_sub(&dict->memory, (long)size, __ATOMIC_RELAXED);
     }
 }
 static inline void DICTIONARY_STATS_VALUE_RESETS_PLUS1(DICTIONARY *dict, size_t oldsize, size_t newsize) {
-    if(unlikely(dict->flags & DICTIONARY_FLAG_WITH_STATISTICS)) {
-        dict->stats->resets++;
-        dict->stats->memory += newsize;
-        dict->stats->memory -= oldsize;
+    if(dict->flags & DICTIONARY_FLAG_EXCLUSIVE_ACCESS) {
+        dict->version++;
+        dict->resets++;
+        dict->memory += (long)newsize;
+        dict->memory -= (long)oldsize;
+    }
+    else {
+        __atomic_fetch_add(&dict->version, 1, __ATOMIC_SEQ_CST);
+        __atomic_fetch_add(&dict->resets, 1, __ATOMIC_RELAXED);
+        __atomic_fetch_add(&dict->memory, (long)newsize, __ATOMIC_RELAXED);
+        __atomic_fetch_sub(&dict->memory, (long)oldsize, __ATOMIC_RELAXED);
+    }
+}
+
+static inline void DICTIONARY_STATS_WALKTHROUGHS_PLUS1(DICTIONARY *dict) {
+    if(dict->flags & DICTIONARY_FLAG_EXCLUSIVE_ACCESS) {
+        dict->walkthroughs++;
+    }
+    else {
+        __atomic_fetch_add(&dict->walkthroughs, 1, __ATOMIC_RELAXED);
+    }
+}
+
+static inline size_t DICTIONARY_STATS_REFERENCED_ITEMS_PLUS1(DICTIONARY *dict) {
+    return __atomic_add_fetch(&dict->referenced_items, 1, __ATOMIC_SEQ_CST);
+}
+
+static inline size_t DICTIONARY_STATS_REFERENCED_ITEMS_MINUS1(DICTIONARY *dict) {
+    return __atomic_sub_fetch(&dict->referenced_items, 1, __ATOMIC_SEQ_CST);
+}
+
+static inline size_t DICTIONARY_STATS_PENDING_DELETES_PLUS1(DICTIONARY *dict) {
+    return __atomic_add_fetch(&dict->pending_deletion_items, 1, __ATOMIC_SEQ_CST);
+}
+
+static inline size_t DICTIONARY_STATS_PENDING_DELETES_MINUS1(DICTIONARY *dict) {
+    return __atomic_sub_fetch(&dict->pending_deletion_items, 1, __ATOMIC_SEQ_CST);
+}
+
+static inline size_t DICTIONARY_STATS_PENDING_DELETES_GET(DICTIONARY *dict) {
+    return __atomic_load_n(&dict->pending_deletion_items, __ATOMIC_SEQ_CST);
+}
+
+static inline int DICTIONARY_NAME_VALUE_REFCOUNT_GET(NAME_VALUE *nv) {
+    return __atomic_load_n(&nv->refcount, __ATOMIC_SEQ_CST);
+}
+
+// ----------------------------------------------------------------------------
+// garbage collector
+// it is called every time someone gets a write lock to the dictionary
+
+static void garbage_collect_pending_deletes_unsafe(DICTIONARY *dict) {
+    if(!(dict->flags & DICTIONARY_FLAG_EXCLUSIVE_ACCESS)) return;
+
+    if(likely(!DICTIONARY_STATS_PENDING_DELETES_GET(dict))) return;
+
+    NAME_VALUE *nv = dict->first_item;
+    while(nv) {
+        if((nv->flags & NAME_VALUE_FLAG_DELETED) && DICTIONARY_NAME_VALUE_REFCOUNT_GET(nv) == 0) {
+            NAME_VALUE *nv_next = nv->next;
+
+            linkedlist_namevalue_unlink_unsafe(dict, nv);
+            namevalue_destroy_unsafe(dict, nv);
+
+            size_t pending = DICTIONARY_STATS_PENDING_DELETES_MINUS1(dict);
+            if(!pending) break;
+
+            nv = nv_next;
+        }
+        else
+            nv = nv->next;
     }
 }
 
@@ -220,41 +293,104 @@ static inline void DICTIONARY_STATS_VALUE_RESETS_PLUS1(DICTIONARY *dict, size_t
 
 static inline size_t dictionary_lock_init(DICTIONARY *dict) {
     if(likely(!(dict->flags & DICTIONARY_FLAG_SINGLE_THREADED))) {
-            dict->rwlock = mallocz(sizeof(netdata_rwlock_t));
-            netdata_rwlock_init(dict->rwlock);
-            return sizeof(netdata_rwlock_t);
+        netdata_rwlock_init(&dict->rwlock);
+
+        if(dict->flags & DICTIONARY_FLAG_EXCLUSIVE_ACCESS)
+            dict->flags &= ~DICTIONARY_FLAG_EXCLUSIVE_ACCESS;
+
+        return 0;
     }
-    dict->rwlock = NULL;
+
+    // we are single threaded
+    dict->flags |= DICTIONARY_FLAG_EXCLUSIVE_ACCESS;
     return 0;
 }
 
 static inline size_t dictionary_lock_free(DICTIONARY *dict) {
     if(likely(!(dict->flags & DICTIONARY_FLAG_SINGLE_THREADED))) {
-        netdata_rwlock_destroy(dict->rwlock);
-        freez(dict->rwlock);
-        return sizeof(netdata_rwlock_t);
+        netdata_rwlock_destroy(&dict->rwlock);
+        return 0;
     }
     return 0;
 }
 
-static inline void dictionary_lock_rlock(DICTIONARY *dict) {
-    if(likely(!(dict->flags & DICTIONARY_FLAG_SINGLE_THREADED))) {
-        // debug(D_DICTIONARY, "Dictionary READ lock");
-        netdata_rwlock_rdlock(dict->rwlock);
+static void dictionary_lock(DICTIONARY *dict, char rw) {
+    if(rw == 'u' || rw == 'U') return;
+
+    if(rw == 'r' || rw == 'R') {
+        // read lock
+        __atomic_add_fetch(&dict->readers, 1, __ATOMIC_RELAXED);
+    }
+    else {
+        // write lock
+        __atomic_add_fetch(&dict->writers, 1, __ATOMIC_RELAXED);
+    }
+
+    if(likely(dict->flags & DICTIONARY_FLAG_SINGLE_THREADED))
+        return;
+
+    if(rw == 'r' || rw == 'R') {
+        // read lock
+        netdata_rwlock_rdlock(&dict->rwlock);
+
+        if(dict->flags & DICTIONARY_FLAG_EXCLUSIVE_ACCESS) {
+            internal_error(true, "DICTIONARY: left-over exclusive access to dictionary created by %s (%zu@%s) found", dict->creation_function, dict->creation_line, dict->creation_file);
+            dict->flags &= ~DICTIONARY_FLAG_EXCLUSIVE_ACCESS;
+        }
+    }
+    else {
+        // write lock
+        netdata_rwlock_wrlock(&dict->rwlock);
+
+        dict->flags |= DICTIONARY_FLAG_EXCLUSIVE_ACCESS;
     }
 }
 
-static inline void dictionary_lock_wrlock(DICTIONARY *dict) {
-    if(likely(!(dict->flags & DICTIONARY_FLAG_SINGLE_THREADED))) {
-        // debug(D_DICTIONARY, "Dictionary WRITE lock");
-        netdata_rwlock_wrlock(dict->rwlock);
+static void dictionary_unlock(DICTIONARY *dict, char rw) {
+    if(rw == 'u' || rw == 'U') return;
+
+    if(rw == 'r' || rw == 'R') {
+        // read unlock
+        __atomic_sub_fetch(&dict->readers, 1, __ATOMIC_RELAXED);
+    }
+    else {
+        // write unlock
+        garbage_collect_pending_deletes_unsafe(dict);
+        __atomic_sub_fetch(&dict->writers, 1, __ATOMIC_RELAXED);
     }
+
+    if(likely(dict->flags & DICTIONARY_FLAG_SINGLE_THREADED))
+        return;
+
+    if(dict->flags & DICTIONARY_FLAG_EXCLUSIVE_ACCESS)
+        dict->flags &= ~DICTIONARY_FLAG_EXCLUSIVE_ACCESS;
+
+    netdata_rwlock_unlock(&dict->rwlock);
 }
 
-static inline void dictionary_unlock(DICTIONARY *dict) {
-    if(likely(!(dict->flags & DICTIONARY_FLAG_SINGLE_THREADED))) {
-        // debug(D_DICTIONARY, "Dictionary UNLOCK lock");
-        netdata_rwlock_unlock(dict->rwlock);
+// ----------------------------------------------------------------------------
+// deferred deletions
+
+void dictionary_defer_all_deletions_unsafe(DICTIONARY *dict, char rw) {
+    if(rw == 'r' || rw == 'R') {
+        // read locked - no need to defer deletions
+        ;
+    }
+    else {
+        // write locked - defer deletions
+        dict->flags |= DICTIONARY_FLAG_DEFER_ALL_DELETIONS;
+    }
+}
+
+void dictionary_restore_all_deletions_unsafe(DICTIONARY *dict, char rw) {
+    if(rw == 'r' || rw == 'R') {
+        // read locked - no need to defer deletions
+        internal_error(dict->flags & DICTIONARY_FLAG_DEFER_ALL_DELETIONS, "DICTIONARY: deletions are deferred on a read lock");
+    }
+    else {
+        // write locked - defer deletions
+        if(dict->flags & DICTIONARY_FLAG_DEFER_ALL_DELETIONS)
+            dict->flags &= ~DICTIONARY_FLAG_DEFER_ALL_DELETIONS;
     }
 }
 
@@ -277,39 +413,90 @@ static inline size_t reference_counter_free(DICTIONARY *dict) {
     return 0;
 }
 
-static void reference_counter_acquire(DICTIONARY *dict, NAME_VALUE *nv) {
-    if(unlikely(dict->flags & DICTIONARY_FLAG_REFERENCE_COUNTERS)) {
-        NAME_VALUE_WITH_STATS *nvs = (NAME_VALUE_WITH_STATS *)nv;
-        __atomic_fetch_add(&nvs->refcount, 1, __ATOMIC_SEQ_CST);
-    }
+static int reference_counter_increase(NAME_VALUE *nv) {
+    int refcount = __atomic_add_fetch(&nv->refcount, 1, __ATOMIC_SEQ_CST);
+    if(refcount == 1)
+        fatal("DICTIONARY: request to dup item '%s' but its reference counter was zero", namevalue_get_name(nv));
+    return refcount;
 }
 
-static void reference_counter_release(DICTIONARY *dict, NAME_VALUE *nv) {
-    if(unlikely(dict->flags & DICTIONARY_FLAG_REFERENCE_COUNTERS)) {
-        NAME_VALUE_WITH_STATS *nvs = (NAME_VALUE_WITH_STATS *)nv;
-        __atomic_fetch_sub(&nvs->refcount, 1, __ATOMIC_SEQ_CST);
+static int reference_counter_acquire(DICTIONARY *dict, NAME_VALUE *nv) {
+    int refcount;
+    if(likely(dict->flags & DICTIONARY_FLAG_SINGLE_THREADED))
+        refcount = ++nv->refcount;
+    else
+        refcount = __atomic_add_fetch(&nv->refcount, 1, __ATOMIC_SEQ_CST);
+
+    if(refcount == 1) {
+        // referenced items counts number of unique items referenced
+        // so, we increase it only when refcount == 1
+        DICTIONARY_STATS_REFERENCED_ITEMS_PLUS1(dict);
+
+        // if this is a deleted item, but the counter increased to 1
+        // we need to remove it from the pending items to delete
+        if (nv->flags & NAME_VALUE_FLAG_DELETED)
+            DICTIONARY_STATS_PENDING_DELETES_MINUS1(dict);
     }
+
+    return refcount;
 }
 
-static int reference_counter_mark_deleted(DICTIONARY *dict, NAME_VALUE *nv) {
-    if(unlikely(dict->flags & DICTIONARY_FLAG_REFERENCE_COUNTERS)) {
-        NAME_VALUE_WITH_STATS *nvs = (NAME_VALUE_WITH_STATS *)nv;
-        nvs->flags |= NAME_VALUE_FLAG_DELETED;
-        return 1;
+static uint32_t reference_counter_release(DICTIONARY *dict, NAME_VALUE *nv, bool can_get_write_lock) {
+    // this function may be called without any lock on the dictionary
+    // or even when someone else has a write lock on the dictionary
+    // so, we cannot check for EXCLUSIVE ACCESS
+
+    uint32_t refcount;
+    if(likely(dict->flags & DICTIONARY_FLAG_SINGLE_THREADED))
+        refcount = nv->refcount--;
+    else
+        refcount = __atomic_fetch_sub(&nv->refcount, 1, __ATOMIC_SEQ_CST);
+
+    if(refcount == 0) {
+        internal_error(true, "DICTIONARY: attempted to release item without references: '%s' on dictionary created by %s() (%zu@%s)", namevalue_get_name(nv), dict->creation_function, dict->creation_line, dict->creation_file);
+        fatal("DICTIONARY: attempted to release item without references: '%s'", namevalue_get_name(nv));
     }
-    return 0;
+
+    if(refcount == 1) {
+        if((nv->flags & NAME_VALUE_FLAG_DELETED))
+            DICTIONARY_STATS_PENDING_DELETES_PLUS1(dict);
+
+        // referenced items counts number of unique items referenced
+        // so, we decrease it only when refcount == 0
+        DICTIONARY_STATS_REFERENCED_ITEMS_MINUS1(dict);
+    }
+
+    if(can_get_write_lock && DICTIONARY_STATS_PENDING_DELETES_GET(dict)) {
+        // we can garbage collect now
+
+        dictionary_lock(dict, DICTIONARY_LOCK_WRITE);
+        garbage_collect_pending_deletes_unsafe(dict);
+        dictionary_unlock(dict, DICTIONARY_LOCK_WRITE);
+    }
+
+    return refcount;
 }
 
 // ----------------------------------------------------------------------------
 // hash table
 
 #ifdef DICTIONARY_WITH_AVL
+static inline const char *namevalue_get_name(NAME_VALUE *nv);
+
 static int name_value_compare(void* a, void* b) {
-    return strcmp(((NAME_VALUE *)a)->name, ((NAME_VALUE *)b)->name);
+    return strcmp(namevalue_get_name((NAME_VALUE *)a), namevalue_get_name((NAME_VALUE *)b));
+}
+
+static void *get_thread_static_name_value(const char *name) {
+    static __thread NAME_VALUE tmp = { 0 };
+    tmp.flags = NAME_VALUE_FLAG_NONE;
+    tmp.caller_name = (char *)name;
+    return &tmp;
 }
 
 static void hashtable_init_unsafe(DICTIONARY *dict) {
     avl_init(&dict->values_index, name_value_compare);
+    dict->get_thread_static_name_value = get_thread_static_name_value;
 }
 
 static size_t hashtable_destroy_unsafe(DICTIONARY *dict) {
@@ -317,7 +504,7 @@ static size_t hashtable_destroy_unsafe(DICTIONARY *dict) {
     return 0;
 }
 
-static inline int hashtable_delete_unsafe(DICTIONARY *dict, const char *name, size_t name_len, NAME_VALUE *nv) {
+static inline int hashtable_delete_unsafe(DICTIONARY *dict, const char *name, size_t name_len, void *nv) {
     (void)name;
     (void)name_len;
 
@@ -330,9 +517,8 @@ static inline int hashtable_delete_unsafe(DICTIONARY *dict, const char *name, si
 static inline NAME_VALUE *hashtable_get_unsafe(DICTIONARY *dict, const char *name, size_t name_len) {
     (void)name_len;
 
-    NAME_VALUE tmp;
-    tmp.name = (char *)name;
-    return (NAME_VALUE *)avl_search(&(dict->values_index), (avl_t *) &tmp);
+    void *tmp = dict->get_thread_static_name_value(name);
+    return (NAME_VALUE *)avl_search(&(dict->values_index), (avl_t *)tmp);
 }
 
 static inline NAME_VALUE **hashtable_insert_unsafe(DICTIONARY *dict, const char *name, size_t name_len) {
@@ -346,13 +532,10 @@ static inline NAME_VALUE **hashtable_insert_unsafe(DICTIONARY *dict, const char
     return &dict->hash_base;
 }
 
-static inline void hashtable_inserted_name_value_unsafe(DICTIONARY *dict, const char *name, size_t name_len, NAME_VALUE *nv) {
+static inline void hashtable_inserted_name_value_unsafe(DICTIONARY *dict, void *nv) {
     // we have our new NAME_VALUE object.
     // Let's index it.
 
-    (void)name;
-    (void)name_len;
-
     if(unlikely(avl_insert(&((dict)->values_index), (avl_t *)(nv)) != (avl_t *)nv))
         error("dictionary: INTERNAL ERROR: duplicate insertion to dictionary.");
 }
@@ -380,6 +563,8 @@ static size_t hashtable_destroy_unsafe(DICTIONARY *dict) {
 }
 
 static inline NAME_VALUE **hashtable_insert_unsafe(DICTIONARY *dict, const char *name, size_t name_len) {
+    internal_error(!(dict->flags & DICTIONARY_FLAG_EXCLUSIVE_ACCESS), "DICTIONARY: inserting item from the index without exclusive access to the dictionary created by %s() (%zu@%s)", dict->creation_function, dict->creation_line, dict->creation_file);
+
     JError_t J_Error;
     Pvoid_t *Rc = JudyHSIns(&dict->JudyHSArray, (void *)name, name_len, &J_Error);
     if (unlikely(Rc == PJERR)) {
@@ -397,9 +582,10 @@ static inline NAME_VALUE **hashtable_insert_unsafe(DICTIONARY *dict, const char
     return (NAME_VALUE **)Rc;
 }
 
-static inline int hashtable_delete_unsafe(DICTIONARY *dict, const char *name, size_t name_len, NAME_VALUE *nv) {
-    (void)nv;
+static inline int hashtable_delete_unsafe(DICTIONARY *dict, const char *name, size_t name_len, void *nv) {
+    internal_error(!(dict->flags & DICTIONARY_FLAG_EXCLUSIVE_ACCESS), "DICTIONARY: deleting item from the index without exclusive access to the dictionary created by %s() (%zu@%s)", dict->creation_function, dict->creation_line, dict->creation_file);
 
+    (void)nv;
     if(unlikely(!dict->JudyHSArray)) return 0;
 
     JError_t J_Error;
@@ -440,10 +626,8 @@ static inline NAME_VALUE *hashtable_get_unsafe(DICTIONARY *dict, const char *nam
     }
 }
 
-static inline void hashtable_inserted_name_value_unsafe(DICTIONARY *dict, const char *name, size_t name_len, NAME_VALUE *nv) {
+static inline void hashtable_inserted_name_value_unsafe(DICTIONARY *dict, void *nv) {
     (void)dict;
-    (void)name;
-    (void)name_len;
     (void)nv;
     ;
 }
@@ -454,6 +638,8 @@ static inline void hashtable_inserted_name_value_unsafe(DICTIONARY *dict, const
 // linked list management
 
 static inline void linkedlist_namevalue_link_unsafe(DICTIONARY *dict, NAME_VALUE *nv) {
+    internal_error(!(dict->flags & DICTIONARY_FLAG_EXCLUSIVE_ACCESS), "DICTIONARY: adding item to the linked-list without exclusive access to the dictionary created by %s() (%zu@%s)", dict->creation_function, dict->creation_line, dict->creation_file);
+
     if (unlikely(!dict->first_item)) {
         // we are the only ones here
         nv->next = NULL;
@@ -481,6 +667,8 @@ static inline void linkedlist_namevalue_link_unsafe(DICTIONARY *dict, NAME_VALUE
 }
 
 static inline void linkedlist_namevalue_unlink_unsafe(DICTIONARY *dict, NAME_VALUE *nv) {
+    internal_error(!(dict->flags & DICTIONARY_FLAG_EXCLUSIVE_ACCESS), "DICTIONARY: removing item from the linked-list without exclusive access to the dictionary created by %s() (%zu@%s)", dict->creation_function, dict->creation_line, dict->creation_file);
+
     if(nv->next) nv->next->prev = nv->prev;
     if(nv->prev) nv->prev->next = nv->next;
     if(dict->first_item == nv) dict->first_item = nv->next;
@@ -490,54 +678,47 @@ static inline void linkedlist_namevalue_unlink_unsafe(DICTIONARY *dict, NAME_VAL
 // ----------------------------------------------------------------------------
 // NAME_VALUE methods
 
-static inline size_t namevalue_alloc_size(DICTIONARY *dict) {
-    return (dict->flags & DICTIONARY_FLAG_WITH_STATISTICS) ? sizeof(NAME_VALUE_WITH_STATS) : sizeof(NAME_VALUE);
-}
-
-static inline size_t namevalue_get_namelen(DICTIONARY *dict, NAME_VALUE *nv) {
-    if(unlikely(dict->flags & DICTIONARY_FLAG_WITH_STATISTICS)) {
-        NAME_VALUE_WITH_STATS *nvs = (NAME_VALUE_WITH_STATS *)nv;
-        return nvs->name_len;
+static inline size_t namevalue_set_name(DICTIONARY *dict, NAME_VALUE *nv, const char *name, size_t name_len) {
+    if(likely(dict->flags & DICTIONARY_FLAG_NAME_LINK_DONT_CLONE)) {
+        nv->caller_name = (char *)name;
+        return 0;
     }
-    return 0;
+
+    nv->string_name = string_strdupz(name);
+    nv->flags |= NAME_VALUE_FLAG_NAME_IS_ALLOCATED;
+    return name_len;
 }
-static inline size_t namevalue_get_valuelen(DICTIONARY *dict, NAME_VALUE *nv) {
-    if(unlikely(dict->flags & DICTIONARY_FLAG_WITH_STATISTICS)) {
-        NAME_VALUE_WITH_STATS *nvs = (NAME_VALUE_WITH_STATS *)nv;
-        return nvs->value_len;
-    }
+
+static inline size_t namevalue_free_name(DICTIONARY *dict, NAME_VALUE *nv) {
+    if(unlikely(!(dict->flags & DICTIONARY_FLAG_NAME_LINK_DONT_CLONE)))
+        string_freez(nv->string_name);
+
     return 0;
 }
-static inline void namevalue_set_valuelen(DICTIONARY *dict, NAME_VALUE *nv, size_t value_len) {
-    if(unlikely(dict->flags & DICTIONARY_FLAG_WITH_STATISTICS)) {
-        NAME_VALUE_WITH_STATS *nvs = (NAME_VALUE_WITH_STATS *)nv;
-        nvs->value_len = value_len;
-    }
-}
-static inline void namevalue_set_namevaluelen(DICTIONARY *dict, NAME_VALUE *nv, size_t name_len, size_t value_len) {
-    if(unlikely(dict->flags & DICTIONARY_FLAG_WITH_STATISTICS)) {
-        NAME_VALUE_WITH_STATS *nvs = (NAME_VALUE_WITH_STATS *)nv;
-        nvs->name_len = name_len;
-        nvs->value_len = value_len;
-    }
+
+static inline const char *namevalue_get_name(NAME_VALUE *nv) {
+    if(nv->flags & NAME_VALUE_FLAG_NAME_IS_ALLOCATED)
+        return string2str(nv->string_name);
+    else
+        return nv->caller_name;
 }
 
 static NAME_VALUE *namevalue_create_unsafe(DICTIONARY *dict, const char *name, size_t name_len, void *value, size_t value_len) {
     debug(D_DICTIONARY, "Creating name value entry for name '%s'.", name);
 
-    size_t size = namevalue_alloc_size(dict);
+    size_t size = sizeof(NAME_VALUE);
     NAME_VALUE *nv = mallocz(size);
     size_t allocated = size;
 
-    namevalue_set_namevaluelen(dict, nv, name_len, value_len);
+#ifdef NETDATA_INTERNAL_CHECKS
+    nv->dict = dict;
+#endif
 
-    if(likely(dict->flags & DICTIONARY_FLAG_NAME_LINK_DONT_CLONE))
-        nv->name = (char *)name;
-    else {
-        nv->name = mallocz(name_len);
-        memcpy(nv->name, name, name_len);
-        allocated += name_len;
-    }
+    nv->refcount = 0;
+    nv->flags = NAME_VALUE_FLAG_NONE;
+    nv->value_len = value_len;
+
+    allocated += namevalue_set_name(dict, nv, name, name_len);
 
     if(likely(dict->flags & DICTIONARY_FLAG_VALUE_LINK_DONT_CLONE))
         nv->value = value;
@@ -556,7 +737,7 @@ static NAME_VALUE *namevalue_create_unsafe(DICTIONARY *dict, const char *name, s
             }
         }
         else {
-            // the caller want an item without any value
+            // the caller wants an item without any value
             nv->value = NULL;
         }
 
@@ -565,107 +746,189 @@ static NAME_VALUE *namevalue_create_unsafe(DICTIONARY *dict, const char *name, s
 
     DICTIONARY_STATS_ENTRIES_PLUS1(dict, allocated);
 
+    if(dict->ins_callback)
+        dict->ins_callback(namevalue_get_name(nv), nv->value, dict->ins_callback_data);
+
     return nv;
 }
 
 static void namevalue_reset_unsafe(DICTIONARY *dict, NAME_VALUE *nv, void *value, size_t value_len) {
-    debug(D_DICTIONARY, "Dictionary entry with name '%s' found. Changing its value.", nv->name);
+    debug(D_DICTIONARY, "Dictionary entry with name '%s' found. Changing its value.", namevalue_get_name(nv));
+
+    DICTIONARY_STATS_VALUE_RESETS_PLUS1(dict, nv->value_len, value_len);
+
+    if(dict->del_callback)
+        dict->del_callback(namevalue_get_name(nv), nv->value, dict->del_callback_data);
 
     if(likely(dict->flags & DICTIONARY_FLAG_VALUE_LINK_DONT_CLONE)) {
-        debug(D_DICTIONARY, "Dictionary: linking value to '%s'", nv->name);
+        debug(D_DICTIONARY, "Dictionary: linking value to '%s'", namevalue_get_name(nv));
         nv->value = value;
-        namevalue_set_valuelen(dict, nv, value_len);
+        nv->value_len = value_len;
     }
     else {
-        debug(D_DICTIONARY, "Dictionary: cloning value to '%s'", nv->name);
-        DICTIONARY_STATS_VALUE_RESETS_PLUS1(dict, namevalue_get_valuelen(dict, nv), value_len);
-
-        void *old = nv->value;
-        void *new = mallocz(value_len);
-        memcpy(new, value, value_len);
-        nv->value = new;
-        namevalue_set_valuelen(dict, nv, value_len);
+        debug(D_DICTIONARY, "Dictionary: cloning value to '%s'", namevalue_get_name(nv));
+
+        void *oldvalue = nv->value;
+        void *newvalue = NULL;
+        if(value_len) {
+            newvalue = mallocz(value_len);
+            if(value) memcpy(newvalue, value, value_len);
+            else memset(newvalue, 0, value_len);
+        }
+        nv->value = newvalue;
+        nv->value_len = value_len;
 
-        debug(D_DICTIONARY, "Dictionary: freeing old value of '%s'", nv->name);
-        freez(old);
+        debug(D_DICTIONARY, "Dictionary: freeing old value of '%s'", namevalue_get_name(nv));
+        freez(oldvalue);
     }
+
+    if(dict->ins_callback)
+        dict->ins_callback(namevalue_get_name(nv), nv->value, dict->ins_callback_data);
 }
 
 static size_t namevalue_destroy_unsafe(DICTIONARY *dict, NAME_VALUE *nv) {
-    debug(D_DICTIONARY, "Destroying name value entry for name '%s'.", nv->name);
+    debug(D_DICTIONARY, "Destroying name value entry for name '%s'.", namevalue_get_name(nv));
+
+    if(dict->del_callback)
+        dict->del_callback(namevalue_get_name(nv), nv->value, dict->del_callback_data);
 
     size_t freed = 0;
 
     if(unlikely(!(dict->flags & DICTIONARY_FLAG_VALUE_LINK_DONT_CLONE))) {
-        debug(D_DICTIONARY, "Dictionary freeing value of '%s'", nv->name);
+        debug(D_DICTIONARY, "Dictionary freeing value of '%s'", namevalue_get_name(nv));
         freez(nv->value);
-        freed += namevalue_get_valuelen(dict, nv);
+        freed += nv->value_len;
     }
 
     if(unlikely(!(dict->flags & DICTIONARY_FLAG_NAME_LINK_DONT_CLONE))) {
-        debug(D_DICTIONARY, "Dictionary freeing name '%s'", nv->name);
-        freez(nv->name);
-        freed += namevalue_get_namelen(dict, nv);
+        debug(D_DICTIONARY, "Dictionary freeing name '%s'", namevalue_get_name(nv));
+        freed += namevalue_free_name(dict, nv);
     }
 
     freez(nv);
-    freed += namevalue_alloc_size(dict);
+    freed += sizeof(NAME_VALUE);
 
-    DICTIONARY_STATS_ENTRIES_MINUS1(dict, freed);
+    DICTIONARY_STATS_ENTRIES_MINUS_MEMORY(dict, freed);
 
     return freed;
 }
 
+// if a dictionary item can be deleted, return true, otherwise return false
+static bool name_value_can_be_deleted(DICTIONARY *dict, NAME_VALUE *nv) {
+    if(unlikely(dict->flags & DICTIONARY_FLAG_DEFER_ALL_DELETIONS))
+        return false;
+
+    if(unlikely(DICTIONARY_NAME_VALUE_REFCOUNT_GET(nv) > 0))
+        return false;
+
+    return true;
+}
+
 // ----------------------------------------------------------------------------
 // API - dictionary management
-
-DICTIONARY *dictionary_create(DICTIONARY_FLAGS flags) {
+#ifdef NETDATA_INTERNAL_CHECKS
+DICTIONARY *dictionary_create_advanced_with_trace(DICTIONARY_FLAGS flags, size_t scratchpad_size, const char *function, size_t line, const char *file) {
+#else
+DICTIONARY *dictionary_create_advanced(DICTIONARY_FLAGS flags, size_t scratchpad_size) {
+#endif
     debug(D_DICTIONARY, "Creating dictionary.");
 
-    if((flags & DICTIONARY_FLAG_REFERENCE_COUNTERS) && (flags & DICTIONARY_FLAG_SINGLE_THREADED)) {
-        error("DICTIONARY: requested reference counters on single threaded dictionary. Not adding reference counters.");
-        flags &= ~DICTIONARY_FLAG_REFERENCE_COUNTERS;
-    }
+    if(unlikely(flags & DICTIONARY_FLAGS_RESERVED))
+        flags &= ~DICTIONARY_FLAGS_RESERVED;
 
-    if(flags & DICTIONARY_FLAG_REFERENCE_COUNTERS) {
-        // we need statistics to allocate the extra NAME_VALUE attributes
-        flags |= DICTIONARY_FLAG_WITH_STATISTICS;
-    }
-
-    DICTIONARY *dict = callocz(1, sizeof(DICTIONARY));
-    size_t allocated = sizeof(DICTIONARY);
+    DICTIONARY *dict = callocz(1, sizeof(DICTIONARY) + scratchpad_size);
+    size_t allocated = sizeof(DICTIONARY) + scratchpad_size;
 
+    dict->scratchpad_size = scratchpad_size;
     dict->flags = flags;
     dict->first_item = dict->last_item = NULL;
 
     allocated += dictionary_lock_init(dict);
     allocated += reference_counter_init(dict);
-
-    if(flags & DICTIONARY_FLAG_WITH_STATISTICS) {
-        dict->stats = callocz(1, sizeof(struct dictionary_stats));
-        allocated += sizeof(struct dictionary_stats);
-        dict->stats->memory = allocated;
-    }
-    else
-        dict->stats = NULL;
+    dict->memory = (long)allocated;
 
     hashtable_init_unsafe(dict);
+
+#ifdef NETDATA_INTERNAL_CHECKS
+    dict->creation_function = function;
+    dict->creation_file = file;
+    dict->creation_line = line;
+#endif
+
     return (DICTIONARY *)dict;
 }
 
+void *dictionary_scratchpad(DICTIONARY *dict) {
+    return &dict->scratchpad;
+}
+
 size_t dictionary_destroy(DICTIONARY *dict) {
+    if(!dict) return 0;
+
+    NAME_VALUE *nv;
+
     debug(D_DICTIONARY, "Destroying dictionary.");
 
-    dictionary_lock_wrlock(dict);
+    long referenced_items = 0;
+    size_t retries = 0;
+    do {
+        referenced_items = __atomic_load_n(&dict->referenced_items, __ATOMIC_SEQ_CST);
+        if (referenced_items) {
+            dictionary_lock(dict, DICTIONARY_LOCK_WRITE);
+
+            // there are referenced items
+            // delete all items individually, so that only the referenced will remain
+            NAME_VALUE *nv_next;
+            for (nv = dict->first_item; nv; nv = nv_next) {
+                nv_next = nv->next;
+                size_t refcount = DICTIONARY_NAME_VALUE_REFCOUNT_GET(nv);
+                if (!refcount && !(nv->flags & NAME_VALUE_FLAG_DELETED))
+                    dictionary_del_unsafe(dict, namevalue_get_name(nv));
+            }
+
+            internal_error(
+                retries == 0,
+                "DICTIONARY: waiting (try %zu) for destruction of dictionary created from %s() %zu@%s, because it has %ld referenced items in it (%ld total).",
+                retries + 1,
+                dict->creation_function,
+                dict->creation_line,
+                dict->creation_file,
+                referenced_items,
+                dict->entries);
+
+            dictionary_unlock(dict, DICTIONARY_LOCK_WRITE);
+            sleep_usec(10000);
+        }
+    } while(referenced_items > 0 && ++retries < 10);
+
+    if(referenced_items) {
+        dictionary_lock(dict, DICTIONARY_LOCK_WRITE);
+
+        dict->flags |= DICTIONARY_FLAG_DESTROYED;
+        internal_error(
+            true,
+            "DICTIONARY: delaying destruction of dictionary created from %s() %zu@%s after %zu retries, because it has %ld referenced items in it (%ld total).",
+            dict->creation_function,
+            dict->creation_line,
+            dict->creation_file,
+            retries,
+            referenced_items,
+            dict->entries);
+
+        dictionary_unlock(dict, DICTIONARY_LOCK_WRITE);
+        return 0;
+    }
+
+    dictionary_lock(dict, DICTIONARY_LOCK_WRITE);
 
     size_t freed = 0;
-    NAME_VALUE *nv = dict->first_item;
+    nv = dict->first_item;
     while (nv) {
         // cache nv->next
         // because we are going to free nv
-        NAME_VALUE *nvnext = nv->next;
+        NAME_VALUE *nv_next = nv->next;
         freed += namevalue_destroy_unsafe(dict, nv);
-        nv = nvnext;
+        nv = nv_next;
         // to speed up destruction, we don't
         // unlink nv from the linked-list here
     }
@@ -676,31 +939,31 @@ size_t dictionary_destroy(DICTIONARY *dict) {
     // destroy the dictionary
     freed += hashtable_destroy_unsafe(dict);
 
-    dictionary_unlock(dict);
+    dictionary_unlock(dict, DICTIONARY_LOCK_WRITE);
     freed += dictionary_lock_free(dict);
     freed += reference_counter_free(dict);
-
-    if(unlikely(dict->flags & DICTIONARY_FLAG_WITH_STATISTICS)) {
-        freez(dict->stats);
-        dict->stats = NULL;
-        freed += sizeof(struct dictionary_stats);
-    }
-
+    freed += sizeof(DICTIONARY) + dict->scratchpad_size;
     freez(dict);
-    freed += sizeof(DICTIONARY);
 
     return freed;
 }
 
 // ----------------------------------------------------------------------------
-// API - items management
+// helpers
 
-void *dictionary_set_unsafe(DICTIONARY *dict, const char *name, void *value, size_t value_len) {
-    if(unlikely(!name || !*name)) {
-        error("Attempted to dictionary_set() a dictionary item without a name");
+static NAME_VALUE *dictionary_set_name_value_unsafe(DICTIONARY *dict, const char *name, void *value, size_t value_len) {
+    if(unlikely(!name)) {
+        internal_error(true, "DICTIONARY: attempted to dictionary_set() a dictionary item without a name");
         return NULL;
     }
 
+    if(unlikely(dict->flags & DICTIONARY_FLAG_DESTROYED)) {
+        internal_error(true, "DICTIONARY: attempted to dictionary_set() on a destroyed dictionary");
+        return NULL;
+    }
+
+    internal_error(!(dict->flags & DICTIONARY_FLAG_EXCLUSIVE_ACCESS), "DICTIONARY: inserting dictionary item '%s' without exclusive access to dictionary", name);
+
     size_t name_len = strlen(name) + 1; // we need the terminating null too
 
     debug(D_DICTIONARY, "SET dictionary entry with name '%s'.", name);
@@ -720,11 +983,9 @@ void *dictionary_set_unsafe(DICTIONARY *dict, const char *name, void *value, siz
     if(likely(*pnv == 0)) {
         // a new item added to the index
         nv = *pnv = namevalue_create_unsafe(dict, name, name_len, value, value_len);
-        hashtable_inserted_name_value_unsafe(dict, name, name_len, nv);
+        hashtable_inserted_name_value_unsafe(dict, nv);
         linkedlist_namevalue_link_unsafe(dict, nv);
-
-        if(dict->ins_callback)
-            dict->ins_callback(nv->name, nv->value, dict->ins_callback_data);
+        nv->flags |= NAME_VALUE_FLAG_NEW_OR_UPDATED;
     }
     else {
         // the item is already in the index
@@ -732,33 +993,34 @@ void *dictionary_set_unsafe(DICTIONARY *dict, const char *name, void *value, siz
         // or overwrite the value, depending on dictionary flags
 
         nv = *pnv;
-        if(!(dict->flags & DICTIONARY_FLAG_DONT_OVERWRITE_VALUE)) {
-
-            if(dict->del_callback)
-                dict->del_callback(nv->name, nv->value, dict->del_callback_data);
 
+        if(!(dict->flags & DICTIONARY_FLAG_DONT_OVERWRITE_VALUE)) {
             namevalue_reset_unsafe(dict, nv, value, value_len);
+            nv->flags |= NAME_VALUE_FLAG_NEW_OR_UPDATED;
+        }
 
-            if(dict->ins_callback)
-                dict->ins_callback(nv->name, nv->value, dict->ins_callback_data);
+        else if(dict->conflict_callback) {
+            dict->conflict_callback(namevalue_get_name(nv), nv->value, value, dict->conflict_callback_data);
+            nv->flags |= NAME_VALUE_FLAG_NEW_OR_UPDATED;
+        }
+
+        else {
+            // make sure this flag is not set
+            nv->flags &= ~NAME_VALUE_FLAG_NEW_OR_UPDATED;
         }
-        else if(dict->conflict_callback)
-            dict->conflict_callback(nv->name, nv->value, value, dict->conflict_callback_data);
     }
 
-    return nv->value;
+    return nv;
 }
 
-void *dictionary_set(DICTIONARY *dict, const char *name, void *value, size_t value_len) {
-    dictionary_lock_wrlock(dict);
-    void *ret = dictionary_set_unsafe(dict, name, value, value_len);
-    dictionary_unlock(dict);
-    return ret;
-}
+static NAME_VALUE *dictionary_get_name_value_unsafe(DICTIONARY *dict, const char *name) {
+    if(unlikely(!name)) {
+        internal_error(true, "attempted to dictionary_get() without a name");
+        return NULL;
+    }
 
-void *dictionary_get_unsafe(DICTIONARY *dict, const char *name) {
-    if(unlikely(!name || !*name)) {
-        error("Attempted to dictionary_get() without a name");
+    if(unlikely(dict->flags & DICTIONARY_FLAG_DESTROYED)) {
+        internal_error(true, "DICTIONARY: attempted to dictionary_get() on a destroyed dictionary");
         return NULL;
     }
 
@@ -773,22 +1035,168 @@ void *dictionary_get_unsafe(DICTIONARY *dict, const char *name) {
     }
 
     debug(D_DICTIONARY, "Found dictionary entry with name '%s'.", name);
+    return nv;
+}
+
+// ----------------------------------------------------------------------------
+// API - items management
+
+void *dictionary_set_unsafe(DICTIONARY *dict, const char *name, void *value, size_t value_len) {
+    NAME_VALUE *nv = dictionary_set_name_value_unsafe(dict, name, value, value_len);
+
+    if(unlikely(dict->react_callback && nv && (nv->flags & NAME_VALUE_FLAG_NEW_OR_UPDATED))) {
+        // we need to call the react callback with a reference counter on nv
+        reference_counter_acquire(dict, nv);
+        dict->react_callback(namevalue_get_name(nv), nv->value, dict->react_callback_data);
+        reference_counter_release(dict, nv, false);
+    }
+
+    return nv ? nv->value : NULL;
+}
+
+void *dictionary_set(DICTIONARY *dict, const char *name, void *value, size_t value_len) {
+    dictionary_lock(dict, DICTIONARY_LOCK_WRITE);
+    NAME_VALUE *nv = dictionary_set_name_value_unsafe(dict, name, value, value_len);
+
+    // we need to get a reference counter for the react callback
+    // before we unlock the dictionary
+    if(unlikely(dict->react_callback && nv && (nv->flags & NAME_VALUE_FLAG_NEW_OR_UPDATED)))
+        reference_counter_acquire(dict, nv);
+
+    dictionary_unlock(dict, DICTIONARY_LOCK_WRITE);
+
+    if(unlikely(dict->react_callback && nv && (nv->flags & NAME_VALUE_FLAG_NEW_OR_UPDATED))) {
+        // we got the reference counter we need, above
+        dict->react_callback(namevalue_get_name(nv), nv->value, dict->react_callback_data);
+        reference_counter_release(dict, nv, false);
+    }
+
+    return nv ? nv->value : NULL;
+}
+
+DICTIONARY_ITEM *dictionary_set_and_acquire_item_unsafe(DICTIONARY *dict, const char *name, void *value, size_t value_len) {
+    NAME_VALUE *nv = dictionary_set_name_value_unsafe(dict, name, value, value_len);
+
+    if(unlikely(!nv))
+        return NULL;
+
+    reference_counter_acquire(dict, nv);
+
+    if(unlikely(dict->react_callback && (nv->flags & NAME_VALUE_FLAG_NEW_OR_UPDATED))) {
+        dict->react_callback(namevalue_get_name(nv), nv->value, dict->react_callback_data);
+    }
+
+    return (DICTIONARY_ITEM *)nv;
+}
+
+DICTIONARY_ITEM *dictionary_set_and_acquire_item(DICTIONARY *dict, const char *name, void *value, size_t value_len) {
+    dictionary_lock(dict, DICTIONARY_LOCK_WRITE);
+    NAME_VALUE *nv = dictionary_set_name_value_unsafe(dict, name, value, value_len);
+
+    // we need to get the reference counter before we unlock
+    if(nv) reference_counter_acquire(dict, nv);
+
+    dictionary_unlock(dict, DICTIONARY_LOCK_WRITE);
+
+    if(unlikely(dict->react_callback && nv && (nv->flags & NAME_VALUE_FLAG_NEW_OR_UPDATED))) {
+        // we already have a reference counter, for the caller, no need for another one
+        dict->react_callback(namevalue_get_name(nv), nv->value, dict->react_callback_data);
+    }
+
+    return (DICTIONARY_ITEM *)nv;
+}
+
+void *dictionary_get_unsafe(DICTIONARY *dict, const char *name) {
+    NAME_VALUE *nv = dictionary_get_name_value_unsafe(dict, name);
+
+    if(unlikely(!nv))
+        return NULL;
+
     return nv->value;
 }
 
 void *dictionary_get(DICTIONARY *dict, const char *name) {
-    dictionary_lock_rlock(dict);
+    dictionary_lock(dict, DICTIONARY_LOCK_READ);
     void *ret = dictionary_get_unsafe(dict, name);
-    dictionary_unlock(dict);
+    dictionary_unlock(dict, DICTIONARY_LOCK_READ);
     return ret;
 }
 
+DICTIONARY_ITEM *dictionary_get_and_acquire_item_unsafe(DICTIONARY *dict, const char *name) {
+    NAME_VALUE *nv = dictionary_get_name_value_unsafe(dict, name);
+
+    if(unlikely(!nv))
+        return NULL;
+
+    reference_counter_acquire(dict, nv);
+    return (DICTIONARY_ITEM *)nv;
+}
+
+DICTIONARY_ITEM *dictionary_get_and_acquire_item(DICTIONARY *dict, const char *name) {
+    dictionary_lock(dict, DICTIONARY_LOCK_READ);
+    void *ret = dictionary_get_and_acquire_item_unsafe(dict, name);
+    dictionary_unlock(dict, DICTIONARY_LOCK_READ);
+    return ret;
+}
+
+DICTIONARY_ITEM *dictionary_acquired_item_dup(DICTIONARY_ITEM *item) {
+    if(unlikely(!item)) return NULL;
+    reference_counter_increase((NAME_VALUE *)item);
+    return item;
+}
+
+const char *dictionary_acquired_item_name(DICTIONARY_ITEM *item) {
+    if(unlikely(!item)) return NULL;
+    return namevalue_get_name((NAME_VALUE *)item);
+}
+
+void *dictionary_acquired_item_value(DICTIONARY_ITEM *item) {
+    if(unlikely(!item)) return NULL;
+    return ((NAME_VALUE *)item)->value;
+}
+
+void dictionary_acquired_item_release_unsafe(DICTIONARY *dict, DICTIONARY_ITEM *item) {
+    if(unlikely(!item)) return;
+
+#ifdef NETDATA_INTERNAL_CHECKS
+    if(((NAME_VALUE *)item)->dict != dict)
+        fatal("DICTIONARY: %s(): name_value item with name '%s' does not belong to this dictionary", __FUNCTION__, namevalue_get_name((NAME_VALUE *)item));
+#endif
+
+    reference_counter_release(dict, (NAME_VALUE *)item, false);
+}
+
+void dictionary_acquired_item_release(DICTIONARY *dict, DICTIONARY_ITEM *item) {
+    if(unlikely(!item)) return;
+
+#ifdef NETDATA_INTERNAL_CHECKS
+    if(((NAME_VALUE *)item)->dict != dict)
+        fatal("DICTIONARY: %s(): name_value item with name '%s' does not belong to this dictionary", __FUNCTION__, namevalue_get_name((NAME_VALUE *)item));
+#endif
+
+    // no need to get a lock here
+    // we pass the last parameter to reference_counter_release() as true
+    // so that the release may get a write-lock if required to clean up
+
+    reference_counter_release(dict, (NAME_VALUE *)item, true);
+
+    if(unlikely(dict->flags & DICTIONARY_FLAG_DESTROYED))
+        dictionary_destroy(dict);
+}
+
 int dictionary_del_unsafe(DICTIONARY *dict, const char *name) {
+    if(unlikely(dict->flags & DICTIONARY_FLAG_DESTROYED)) {
+        internal_error(true, "DICTIONARY: attempted to dictionary_del() on a destroyed dictionary");
+        return -1;
+    }
+
     if(unlikely(!name || !*name)) {
-        error("Attempted to dictionary_det() without a name");
+        internal_error(true, "DICTIONARY: attempted to dictionary_del() without a name");
         return -1;
     }
 
+    internal_error(!(dict->flags & DICTIONARY_FLAG_EXCLUSIVE_ACCESS), "DICTIONARY: INTERNAL ERROR: deleting dictionary item '%s' without exclusive access to dictionary", name);
+
     size_t name_len = strlen(name) + 1; // we need the terminating null too
 
     debug(D_DICTIONARY, "DEL dictionary entry with name '%s'.", name);
@@ -809,23 +1217,25 @@ int dictionary_del_unsafe(DICTIONARY *dict, const char *name) {
         if(hashtable_delete_unsafe(dict, name, name_len, nv) == 0)
             error("DICTIONARY: INTERNAL ERROR: tried to delete item with name '%s' that is not in the index", name);
 
-        if(!reference_counter_mark_deleted(dict, nv)) {
+        if(name_value_can_be_deleted(dict, nv)) {
             linkedlist_namevalue_unlink_unsafe(dict, nv);
-
-            if(dict->del_callback)
-                dict->del_callback(nv->name, nv->value, dict->del_callback_data);
-
             namevalue_destroy_unsafe(dict, nv);
         }
+        else
+            nv->flags |= NAME_VALUE_FLAG_DELETED;
+
         ret = 0;
+
+        DICTIONARY_STATS_ENTRIES_MINUS1(dict);
+
     }
     return ret;
 }
 
 int dictionary_del(DICTIONARY *dict, const char *name) {
-    dictionary_lock_wrlock(dict);
+    dictionary_lock(dict, DICTIONARY_LOCK_WRITE);
     int ret = dictionary_del_unsafe(dict, name);
-    dictionary_unlock(dict);
+    dictionary_unlock(dict, DICTIONARY_LOCK_WRITE);
     return ret;
 }
 
@@ -835,25 +1245,37 @@ int dictionary_del(DICTIONARY *dict, const char *name) {
 void *dictionary_foreach_start_rw(DICTFE *dfe, DICTIONARY *dict, char rw) {
     if(unlikely(!dfe || !dict)) return NULL;
 
+    if(unlikely(dict->flags & DICTIONARY_FLAG_DESTROYED)) {
+        internal_error(true, "DICTIONARY: attempted to dictionary_foreach_start_rw() on a destroyed dictionary");
+        dfe->last_item = NULL;
+        dfe->name = NULL;
+        dfe->value = NULL;
+        return NULL;
+    }
+
     dfe->dict = dict;
+    dfe->rw = rw;
     dfe->started_ut = now_realtime_usec();
 
-    if(rw == 'r' || rw == 'R')
-        dictionary_lock_rlock(dict);
-    else
-        dictionary_lock_wrlock(dict);
+    dictionary_lock(dict, dfe->rw);
+
+    DICTIONARY_STATS_WALKTHROUGHS_PLUS1(dict);
 
+    // get the first item from the list
     NAME_VALUE *nv = dict->first_item;
-    dfe->last_position_index = (void *)nv;
+
+    // skip all the deleted items
+    while(nv && (nv->flags & NAME_VALUE_FLAG_DELETED))
+        nv = nv->next;
 
     if(likely(nv)) {
-        dfe->next_position_index = (void *)nv->next;
-        dfe->name = nv->name;
-        dfe->value = (void *)nv->value;
+        dfe->last_item = nv;
+        dfe->name = (char *)namevalue_get_name(nv);
+        dfe->value = nv->value;
         reference_counter_acquire(dict, nv);
     }
     else {
-        dfe->next_position_index = NULL;
+        dfe->last_item = NULL;
         dfe->name = NULL;
         dfe->value = NULL;
     }
@@ -864,21 +1286,36 @@ void *dictionary_foreach_start_rw(DICTFE *dfe, DICTIONARY *dict, char rw) {
 void *dictionary_foreach_next(DICTFE *dfe) {
     if(unlikely(!dfe || !dfe->dict)) return NULL;
 
-    NAME_VALUE *nv = (NAME_VALUE *)dfe->last_position_index;
-    if(likely(nv))
-        reference_counter_release(dfe->dict, nv);
+    if(unlikely(dfe->dict->flags & DICTIONARY_FLAG_DESTROYED)) {
+        internal_error(true, "DICTIONARY: attempted to dictionary_foreach_next() on a destroyed dictionary");
+        dfe->last_item = NULL;
+        dfe->name = NULL;
+        dfe->value = NULL;
+        return NULL;
+    }
 
-    nv = dfe->last_position_index = dfe->next_position_index;
+    // the item we just did
+    NAME_VALUE *nv = (NAME_VALUE *)dfe->last_item;
 
-    if(likely(nv)) {
-        dfe->next_position_index = (void *)nv->next;
-        dfe->name = nv->name;
-        dfe->value = (void *)nv->value;
+    // get the next item from the list
+    NAME_VALUE *nv_next = (nv) ? nv->next : NULL;
+
+    // skip all the deleted items
+    while(nv_next && (nv_next->flags & NAME_VALUE_FLAG_DELETED))
+        nv_next = nv_next->next;
 
+    // release the old, so that it can possibly be deleted
+    if(likely(nv))
+        reference_counter_release(dfe->dict, nv, false);
+
+    if(likely(nv = nv_next)) {
+        dfe->last_item = nv;
+        dfe->name = (char *)namevalue_get_name(nv);
+        dfe->value = nv->value;
         reference_counter_acquire(dfe->dict, nv);
     }
     else {
-        dfe->next_position_index = NULL;
+        dfe->last_item = NULL;
         dfe->name = NULL;
         dfe->value = NULL;
     }
@@ -889,14 +1326,21 @@ void *dictionary_foreach_next(DICTFE *dfe) {
 usec_t dictionary_foreach_done(DICTFE *dfe) {
     if(unlikely(!dfe || !dfe->dict)) return 0;
 
-    NAME_VALUE *nv = (NAME_VALUE *)dfe->last_position_index;
-    if(nv)
-        reference_counter_release(dfe->dict, nv);
+    if(unlikely(dfe->dict->flags & DICTIONARY_FLAG_DESTROYED)) {
+        internal_error(true, "DICTIONARY: attempted to dictionary_foreach_next() on a destroyed dictionary");
+        return 0;
+    }
+
+    // the item we just did
+    NAME_VALUE *nv = (NAME_VALUE *)dfe->last_item;
+
+    // release it, so that it can possibly be deleted
+    if(likely(nv))
+        reference_counter_release(dfe->dict, nv, false);
 
-    dictionary_unlock((DICTIONARY *)dfe->dict);
+    dictionary_unlock(dfe->dict, dfe->rw);
     dfe->dict = NULL;
-    dfe->last_position_index = NULL;
-    dfe->next_position_index = NULL;
+    dfe->last_item = NULL;
     dfe->name = NULL;
     dfe->value = NULL;
 
@@ -912,21 +1356,40 @@ usec_t dictionary_foreach_done(DICTFE *dfe) {
 // do not use other dictionary calls while walking the dictionary - deadlock!
 
 int dictionary_walkthrough_rw(DICTIONARY *dict, char rw, int (*callback)(const char *name, void *entry, void *data), void *data) {
-    if(rw == 'r' || rw == 'R')
-        dictionary_lock_rlock(dict);
-    else
-        dictionary_lock_wrlock(dict);
+    if(unlikely(!dict)) return 0;
+
+    if(unlikely(dict->flags & DICTIONARY_FLAG_DESTROYED)) {
+        internal_error(true, "DICTIONARY: attempted to dictionary_walkthrough_rw() on a destroyed dictionary");
+        return 0;
+    }
+
+    dictionary_lock(dict, rw);
+
+    DICTIONARY_STATS_WALKTHROUGHS_PLUS1(dict);
 
     // written in such a way, that the callback can delete the active element
 
     int ret = 0;
     NAME_VALUE *nv = dict->first_item, *nv_next;
     while(nv) {
-        nv_next = nv->next;
 
+        // skip the deleted items
+        if(unlikely(nv->flags & NAME_VALUE_FLAG_DELETED)) {
+            nv = nv->next;
+            continue;
+        }
+
+        // get a reference counter, so that our item will not be deleted
+        // while we are using it
         reference_counter_acquire(dict, nv);
-        int r = callback(nv->name, nv->value, data);
-        reference_counter_release(dict, nv);
+
+        int r = callback(namevalue_get_name(nv), nv->value, data);
+
+        // since we have a reference counter, this item cannot be deleted
+        // until we release the reference counter, so the pointers are there
+        nv_next = nv->next;
+        reference_counter_release(dict, nv, false);
+
         if(unlikely(r < 0)) {
             ret = r;
             break;
@@ -937,12 +1400,249 @@ int dictionary_walkthrough_rw(DICTIONARY *dict, char rw, int (*callback)(const c
         nv = nv_next;
     }
 
-    dictionary_unlock(dict);
+    dictionary_unlock(dict, rw);
 
     return ret;
 }
 
 // ----------------------------------------------------------------------------
+// sorted walkthrough
+
+static int dictionary_sort_compar(const void *nv1, const void *nv2) {
+    return strcmp(namevalue_get_name((*(NAME_VALUE **)nv1)), namevalue_get_name((*(NAME_VALUE **)nv2)));
+}
+
+int dictionary_sorted_walkthrough_rw(DICTIONARY *dict, char rw, int (*callback)(const char *name, void *entry, void *data), void *data) {
+    if(unlikely(!dict || !dict->entries)) return 0;
+
+    if(unlikely(dict->flags & DICTIONARY_FLAG_DESTROYED)) {
+        internal_error(true, "DICTIONARY: attempted to dictionary_sorted_walkthrough_rw() on a destroyed dictionary");
+        return 0;
+    }
+
+    dictionary_lock(dict, rw);
+    dictionary_defer_all_deletions_unsafe(dict, rw);
+
+    DICTIONARY_STATS_WALKTHROUGHS_PLUS1(dict);
+
+    size_t count = dict->entries;
+    NAME_VALUE **array = mallocz(sizeof(NAME_VALUE *) * count);
+
+    size_t i;
+    NAME_VALUE *nv;
+    for(nv = dict->first_item, i = 0; nv && i < count ;nv = nv->next) {
+        if(likely(!(nv->flags & NAME_VALUE_FLAG_DELETED)))
+            array[i++] = nv;
+    }
+
+    internal_error(nv != NULL, "DICTIONARY: during sorting expected to have %zu items in dictionary, but there are more. Sorted results may be incomplete. Dictionary fails to maintain an accurate number of the number of entries it has.", count);
+
+    if(unlikely(i != count)) {
+        internal_error(true, "DICTIONARY: during sorting expected to have %zu items in dictionary, but there are %zu. Sorted results may be incomplete. Dictionary fails to maintain an accurate number of the number of entries it has.", count, i);
+        count = i;
+    }
+
+    qsort(array, count, sizeof(NAME_VALUE *), dictionary_sort_compar);
+
+    int ret = 0;
+    for(i = 0; i < count ;i++) {
+        nv = array[i];
+        if(likely(!(nv->flags & NAME_VALUE_FLAG_DELETED))) {
+            reference_counter_acquire(dict, nv);
+            int r = callback(namevalue_get_name(nv), nv->value, data);
+            reference_counter_release(dict, nv, false);
+            if (r < 0) {
+                ret = r;
+                break;
+            }
+            ret += r;
+        }
+    }
+
+    dictionary_restore_all_deletions_unsafe(dict, rw);
+    dictionary_unlock(dict, rw);
+    freez(array);
+
+    return ret;
+}
+
+// ----------------------------------------------------------------------------
+// STRING implementation - dedup all STRINGs
+
+typedef struct string_entry {
+#ifdef DICTIONARY_WITH_AVL
+    avl_t avl_node;
+#endif
+    uint32_t length;    // the string length with the terminating '\0'
+    uint32_t refcount;  // how many times this string is used
+    const char str[];   // the string itself
+} STRING_ENTRY;
+
+#ifdef DICTIONARY_WITH_AVL
+static int string_entry_compare(void* a, void* b) {
+    return strcmp(((STRING_ENTRY *)a)->str, ((STRING_ENTRY *)b)->str);
+}
+
+static void *get_thread_static_string_entry(const char *name) {
+    static __thread size_t _length = 0;
+    static __thread STRING_ENTRY *_tmp = NULL;
+
+    size_t size = sizeof(STRING_ENTRY) + strlen(name) + 1;
+    if(likely(_tmp && _length < size)) {
+        freez(_tmp);
+        _tmp = NULL;
+        _length = 0;
+    }
+
+    if(unlikely(!_tmp)) {
+        _tmp = callocz(1, size);
+        _length = size;
+    }
+
+    strcpy((char *)&_tmp->str[0], name);
+    return _tmp;
+}
+#endif
+
+DICTIONARY string_dictionary = {
+#ifdef DICTIONARY_WITH_AVL
+    .values_index = {
+        .root = NULL,
+        .compar = string_entry_compare
+    },
+    .get_thread_static_name_value = get_thread_static_string_entry,
+#endif
+
+    .flags = DICTIONARY_FLAG_EXCLUSIVE_ACCESS,
+    .rwlock = NETDATA_RWLOCK_INITIALIZER
+};
+
+static netdata_mutex_t string_mutex = NETDATA_MUTEX_INITIALIZER;
+
+STRING *string_dup(STRING *string) {
+    if(unlikely(!string)) return NULL;
+
+    STRING_ENTRY *se = (STRING_ENTRY *)string;
+    netdata_mutex_lock(&string_mutex);
+    se->refcount++;
+    netdata_mutex_unlock(&string_mutex);
+    return string;
+}
+
+STRING *string_strdupz(const char *str) {
+    if(unlikely(!str || !*str)) return NULL;
+
+    netdata_mutex_lock(&string_mutex);
+
+    size_t length = strlen(str) + 1;
+    STRING_ENTRY *se;
+    STRING_ENTRY **ptr = (STRING_ENTRY **)hashtable_insert_unsafe(&string_dictionary, str, length);
+    if(unlikely(*ptr == 0)) {
+        // a new item added to the index
+        size_t mem_size = sizeof(STRING_ENTRY) + length;
+        se = mallocz(mem_size);
+        strcpy((char *)se->str, str);
+        se->length = length;
+        se->refcount = 1;
+        *ptr = se;
+        hashtable_inserted_name_value_unsafe(&string_dictionary, se);
+        string_dictionary.version++;
+        string_dictionary.inserts++;
+        string_dictionary.entries++;
+        string_dictionary.memory += (long)mem_size;
+    }
+    else {
+        // the item is already in the index
+        se = *ptr;
+        se->refcount++;
+        string_dictionary.searches++;
+    }
+
+    netdata_mutex_unlock(&string_mutex);
+    return (STRING *)se;
+}
+
+void string_freez(STRING *string) {
+    if(unlikely(!string)) return;
+    netdata_mutex_lock(&string_mutex);
+
+    STRING_ENTRY *se = (STRING_ENTRY *)string;
+
+    if(se->refcount == 0)
+        fatal("STRING: tried to free string that has zero references.");
+
+    se->refcount--;
+    if(unlikely(se->refcount == 0)) {
+        if(hashtable_delete_unsafe(&string_dictionary, se->str, se->length, se) == 0)
+            error("STRING: INTERNAL ERROR: tried to delete '%s' that is not in the index", se->str);
+
+        size_t mem_size = sizeof(STRING_ENTRY) + se->length;
+        freez(se);
+        string_dictionary.version++;
+        string_dictionary.deletes++;
+        string_dictionary.entries--;
+        string_dictionary.memory -= (long)mem_size;
+    }
+
+    netdata_mutex_unlock(&string_mutex);
+}
+
+size_t string_length(STRING *string) {
+    if(unlikely(!string)) return 0;
+    return ((STRING_ENTRY *)string)->length - 1;
+}
+
+const char *string2str(STRING *string) {
+    if(unlikely(!string)) return "";
+    return ((STRING_ENTRY *)string)->str;
+}
+
+STRING *string_2way_merge(STRING *a, STRING *b) {
+    static STRING *X = NULL;
+
+    if(unlikely(!X)) {
+        X = string_strdupz("[x]");
+    }
+
+    if(unlikely(a == b)) return string_dup(a);
+    if(unlikely(a == X)) return string_dup(a);
+    if(unlikely(b == X)) return string_dup(b);
+    if(unlikely(!a)) return string_dup(X);
+    if(unlikely(!b)) return string_dup(X);
+
+    size_t alen = string_length(a);
+    size_t blen = string_length(b);
+    size_t length = alen + blen + string_length(X) + 1;
+    char buf1[length + 1], buf2[length + 1], *dst1;
+    const char *s1, *s2;
+
+    s1 = string2str(a);
+    s2 = string2str(b);
+    dst1 = buf1;
+    for( ; *s1 && *s2 && *s1 == *s2 ;s1++, s2++)
+        *dst1++ = *s1;
+
+    *dst1 = '\0';
+
+    if(*s1 != '\0' || *s2 != '\0') {
+        *dst1++ = '[';
+        *dst1++ = 'x';
+        *dst1++ = ']';
+
+        s1 = &(string2str(a))[alen - 1];
+        s2 = &(string2str(b))[blen - 1];
+        char *dst2 = &buf2[length];
+        *dst2 = '\0';
+        for (; *s1 && *s2 && *s1 == *s2; s1--, s2--)
+            *(--dst2) = *s1;
+
+        strcpy(dst1, dst2);
+    }
+
+    return string_strdupz(buf1);
+}
+
+// ----------------------------------------------------------------------------
 // unit test
 
 static void dictionary_unittest_free_char_pp(char **pp, size_t entries) {
@@ -983,6 +1683,22 @@ static size_t dictionary_unittest_set_clone(DICTIONARY *dict, char **names, char
     return errors;
 }
 
+static size_t dictionary_unittest_set_null(DICTIONARY *dict, char **names, char **values, size_t entries) {
+    (void)values;
+    size_t errors = 0;
+    long i = 0;
+    for(; i < (long)entries ;i++) {
+        void *val = dictionary_set(dict, names[i], NULL, 0);
+        if(val != NULL) { fprintf(stderr, ">>> %s() returns a non NULL value\n", __FUNCTION__); errors++; }
+    }
+    if(dictionary_stats_entries(dict) != i) {
+        fprintf(stderr, ">>> %s() dictionary items do not match\n", __FUNCTION__);
+        errors++;
+    }
+    return errors;
+}
+
+
 static size_t dictionary_unittest_set_nonclone(DICTIONARY *dict, char **names, char **values, size_t entries) {
     size_t errors = 0;
     for(size_t i = 0; i < entries ;i++) {
@@ -1182,7 +1898,7 @@ static size_t dictionary_unittest_destroy(DICTIONARY *dict, char **names, char *
 }
 
 static usec_t dictionary_unittest_run_and_measure_time(DICTIONARY *dict, char *message, char **names, char **values, size_t entries, size_t *errors, size_t (*callback)(DICTIONARY *dict, char **names, char **values, size_t entries)) {
-    fprintf(stderr, "%-40s... ", message);
+    fprintf(stderr, "%40s ... ", message);
 
     usec_t started = now_realtime_usec();
     size_t errs = callback(dict, names, values, entries);
@@ -1191,12 +1907,12 @@ static usec_t dictionary_unittest_run_and_measure_time(DICTIONARY *dict, char *m
 
     if(callback == dictionary_unittest_destroy) dict = NULL;
 
-    fprintf(stderr, " %zu errors, %zu items in dictionary, %llu usec \n", errs, dict? dictionary_stats_entries(dict):0, dt);
+    fprintf(stderr, " %zu errors, %ld items in dictionary, %llu usec \n", errs, dict? dictionary_stats_entries(dict):0, dt);
     *errors += errs;
     return dt;
 }
 
-void dictionary_unittest_clone(DICTIONARY *dict, char **names, char **values, size_t entries, size_t *errors) {
+static void dictionary_unittest_clone(DICTIONARY *dict, char **names, char **values, size_t entries, size_t *errors) {
     dictionary_unittest_run_and_measure_time(dict, "adding entries", names, values, entries, errors, dictionary_unittest_set_clone);
     dictionary_unittest_run_and_measure_time(dict, "getting entries", names, values, entries, errors, dictionary_unittest_get_clone);
     dictionary_unittest_run_and_measure_time(dict, "getting non-existing entries", names, values, entries, errors, dictionary_unittest_get_nonexisting);
@@ -1211,7 +1927,7 @@ void dictionary_unittest_clone(DICTIONARY *dict, char **names, char **values, si
     dictionary_unittest_run_and_measure_time(dict, "destroying empty dictionary", names, values, entries, errors, dictionary_unittest_destroy);
 }
 
-void dictionary_unittest_nonclone(DICTIONARY *dict, char **names, char **values, size_t entries, size_t *errors) {
+static void dictionary_unittest_nonclone(DICTIONARY *dict, char **names, char **values, size_t entries, size_t *errors) {
     dictionary_unittest_run_and_measure_time(dict, "adding entries", names, values, entries, errors, dictionary_unittest_set_nonclone);
     dictionary_unittest_run_and_measure_time(dict, "getting entries", names, values, entries, errors, dictionary_unittest_get_nonclone);
     dictionary_unittest_run_and_measure_time(dict, "getting non-existing entries", names, values, entries, errors, dictionary_unittest_get_nonexisting);
@@ -1226,6 +1942,217 @@ void dictionary_unittest_nonclone(DICTIONARY *dict, char **names, char **values,
     dictionary_unittest_run_and_measure_time(dict, "destroying empty dictionary", names, values, entries, errors, dictionary_unittest_destroy);
 }
 
+struct dictionary_unittest_sorting {
+    const char *oldname;
+    const char *oldvalue;
+    size_t count;
+};
+
+static int dictionary_unittest_sorting_callback(const char *name, void *value, void *data) {
+    struct dictionary_unittest_sorting *t = (struct dictionary_unittest_sorting *)data;
+    const char *v = (const char *)value;
+
+    int ret = 0;
+    if(t->oldname && strcmp(t->oldname, name) > 0) {
+        fprintf(stderr, "name '%s' should be after '%s'\n", t->oldname, name);
+        ret = 1;
+    }
+    t->count++;
+    t->oldname = name;
+    t->oldvalue = v;
+
+    return ret;
+}
+
+static size_t dictionary_unittest_sorted_walkthrough(DICTIONARY *dict, char **names, char **values, size_t entries) {
+    (void)names;
+    (void)values;
+    struct dictionary_unittest_sorting tmp = { .oldname = NULL, .oldvalue = NULL, .count = 0 };
+    size_t errors;
+    errors = dictionary_sorted_walkthrough_read(dict, dictionary_unittest_sorting_callback, &tmp);
+
+    if(tmp.count != entries) {
+        fprintf(stderr, "Expected %zu entries, counted %zu\n", entries, tmp.count);
+        errors++;
+    }
+    return errors;
+}
+
+static void dictionary_unittest_sorting(DICTIONARY *dict, char **names, char **values, size_t entries, size_t *errors) {
+    dictionary_unittest_run_and_measure_time(dict, "adding entries", names, values, entries, errors, dictionary_unittest_set_clone);
+    dictionary_unittest_run_and_measure_time(dict, "sorted walkthrough", names, values, entries, errors, dictionary_unittest_sorted_walkthrough);
+}
+
+static void dictionary_unittest_null_dfe(DICTIONARY *dict, char **names, char **values, size_t entries, size_t *errors) {
+    dictionary_unittest_run_and_measure_time(dict, "adding null value entries", names, values, entries, errors, dictionary_unittest_set_null);
+    dictionary_unittest_run_and_measure_time(dict, "traverse foreach read loop", names, values, entries, errors, dictionary_unittest_foreach);
+}
+
+
+static int check_dictionary_callback(const char *name, void *value, void *data) {
+    (void)name;
+    (void)value;
+    (void)data;
+    return 1;
+}
+
+static size_t check_dictionary(DICTIONARY *dict, size_t entries, size_t linked_list_members) {
+    size_t errors = 0;
+
+    fprintf(stderr, "dictionary entries %ld, expected %zu...\t\t\t\t\t", dictionary_stats_entries(dict), entries);
+    if (dictionary_stats_entries(dict) != (long)entries) {
+        fprintf(stderr, "FAILED\n");
+        errors++;
+    }
+    else
+        fprintf(stderr, "OK\n");
+
+    size_t ll = 0;
+    void *t;
+    dfe_start_read(dict, t)
+        ll++;
+    dfe_done(t);
+
+    fprintf(stderr, "dictionary foreach entries %zu, expected %zu...\t\t\t\t", ll, entries);
+    if(ll != entries) {
+        fprintf(stderr, "FAILED\n");
+        errors++;
+    }
+    else
+        fprintf(stderr, "OK\n");
+
+    ll = dictionary_walkthrough_read(dict, check_dictionary_callback, NULL);
+    fprintf(stderr, "dictionary walkthrough entries %zu, expected %zu...\t\t\t\t", ll, entries);
+    if(ll != entries) {
+        fprintf(stderr, "FAILED\n");
+        errors++;
+    }
+    else
+        fprintf(stderr, "OK\n");
+
+    ll = dictionary_sorted_walkthrough_read(dict, check_dictionary_callback, NULL);
+    fprintf(stderr, "dictionary sorted walkthrough entries %zu, expected %zu...\t\t\t", ll, entries);
+    if(ll != entries) {
+        fprintf(stderr, "FAILED\n");
+        errors++;
+    }
+    else
+        fprintf(stderr, "OK\n");
+
+    NAME_VALUE *nv;
+    for(ll = 0, nv = dict->first_item; nv ;nv = nv->next)
+        ll++;
+
+    fprintf(stderr, "dictionary linked list entries %zu, expected %zu...\t\t\t\t", ll, linked_list_members);
+    if(ll != linked_list_members) {
+        fprintf(stderr, "FAILED\n");
+        errors++;
+    }
+    else
+        fprintf(stderr, "OK\n");
+
+    return errors;
+}
+
+static int check_name_value_callback(const char *name, void *value, void *data) {
+    (void)name;
+    return value == data;
+}
+
+static size_t check_name_value_deleted_flag(DICTIONARY *dict, NAME_VALUE *nv, const char *name, const char *value, unsigned refcount, NAME_VALUE_FLAGS deleted_flags, bool searchable, bool browsable, bool linked) {
+    size_t errors = 0;
+
+    fprintf(stderr, "NAME_VALUE name is '%s', expected '%s'...\t\t\t\t", namevalue_get_name(nv), name);
+    if(strcmp(namevalue_get_name(nv), name) != 0) {
+        fprintf(stderr, "FAILED\n");
+        errors++;
+    }
+    else
+        fprintf(stderr, "OK\n");
+
+    fprintf(stderr, "NAME_VALUE value is '%s', expected '%s'...\t\t\t", (const char *)nv->value, value);
+    if(strcmp((const char *)nv->value, value) != 0) {
+        fprintf(stderr, "FAILED\n");
+        errors++;
+    }
+    else
+        fprintf(stderr, "OK\n");
+
+    fprintf(stderr, "NAME_VALUE refcount is %u, expected %u...\t\t\t\t\t", nv->refcount, refcount);
+    if (nv->refcount != refcount) {
+        fprintf(stderr, "FAILED\n");
+        errors++;
+    }
+    else
+        fprintf(stderr, "OK\n");
+
+    fprintf(stderr, "NAME_VALUE deleted flag is %s, expected %s...\t\t\t", (nv->flags & NAME_VALUE_FLAG_DELETED)?"TRUE":"FALSE", (deleted_flags & NAME_VALUE_FLAG_DELETED)?"TRUE":"FALSE");
+    if ((nv->flags & NAME_VALUE_FLAG_DELETED) != (deleted_flags & NAME_VALUE_FLAG_DELETED)) {
+        fprintf(stderr, "FAILED\n");
+        errors++;
+    }
+    else
+        fprintf(stderr, "OK\n");
+
+    void *v = dictionary_get(dict, name);
+    bool found = v == nv->value;
+    fprintf(stderr, "NAME_VALUE searchable %5s, expected %5s...\t\t\t\t", found?"true":"false", searchable?"true":"false");
+    if(found != searchable) {
+        fprintf(stderr, "FAILED\n");
+        errors++;
+    }
+    else
+        fprintf(stderr, "OK\n");
+
+    found = false;
+    void *t;
+    dfe_start_read(dict, t) {
+        if(t == nv->value) found = true;
+    }
+    dfe_done(t);
+
+    fprintf(stderr, "NAME_VALUE dfe browsable %5s, expected %5s...\t\t\t", found?"true":"false", browsable?"true":"false");
+    if(found != browsable) {
+        fprintf(stderr, "FAILED\n");
+        errors++;
+    }
+    else
+        fprintf(stderr, "OK\n");
+
+    found = dictionary_walkthrough_read(dict, check_name_value_callback, nv->value);
+    fprintf(stderr, "NAME_VALUE walkthrough browsable %5s, expected %5s...\t\t", found?"true":"false", browsable?"true":"false");
+    if(found != browsable) {
+        fprintf(stderr, "FAILED\n");
+        errors++;
+    }
+    else
+        fprintf(stderr, "OK\n");
+
+    found = dictionary_sorted_walkthrough_read(dict, check_name_value_callback, nv->value);
+    fprintf(stderr, "NAME_VALUE sorted walkthrough browsable %5s, expected %5s...\t", found?"true":"false", browsable?"true":"false");
+    if(found != browsable) {
+        fprintf(stderr, "FAILED\n");
+        errors++;
+    }
+    else
+        fprintf(stderr, "OK\n");
+
+    found = false;
+    NAME_VALUE *n;
+    for(n = dict->first_item; n ;n = n->next)
+        if(n == nv) found = true;
+
+    fprintf(stderr, "NAME_VALUE linked %5s, expected %5s...\t\t\t\t", found?"true":"false", linked?"true":"false");
+    if(found != linked) {
+        fprintf(stderr, "FAILED\n");
+        errors++;
+    }
+    else
+        fprintf(stderr, "OK\n");
+
+    return errors;
+}
+
 int dictionary_unittest(size_t entries) {
     if(entries < 10) entries = 10;
 
@@ -1237,23 +2164,23 @@ int dictionary_unittest(size_t entries) {
     char **values = dictionary_unittest_generate_values(entries);
 
     fprintf(stderr, "\nCreating dictionary single threaded, clone, %zu items\n", entries);
-    dict = dictionary_create(DICTIONARY_FLAG_SINGLE_THREADED|DICTIONARY_FLAG_WITH_STATISTICS);
+    dict = dictionary_create(DICTIONARY_FLAG_SINGLE_THREADED);
     dictionary_unittest_clone(dict, names, values, entries, &errors);
 
     fprintf(stderr, "\nCreating dictionary multi threaded, clone, %zu items\n", entries);
-    dict = dictionary_create(DICTIONARY_FLAG_WITH_STATISTICS);
+    dict = dictionary_create(DICTIONARY_FLAG_NONE);
     dictionary_unittest_clone(dict, names, values, entries, &errors);
 
     fprintf(stderr, "\nCreating dictionary single threaded, non-clone, add-in-front options, %zu items\n", entries);
-    dict = dictionary_create(DICTIONARY_FLAG_SINGLE_THREADED|DICTIONARY_FLAG_WITH_STATISTICS|DICTIONARY_FLAG_NAME_LINK_DONT_CLONE|DICTIONARY_FLAG_VALUE_LINK_DONT_CLONE|DICTIONARY_FLAG_ADD_IN_FRONT);
+    dict = dictionary_create(DICTIONARY_FLAG_SINGLE_THREADED|DICTIONARY_FLAG_NAME_LINK_DONT_CLONE|DICTIONARY_FLAG_VALUE_LINK_DONT_CLONE|DICTIONARY_FLAG_ADD_IN_FRONT);
     dictionary_unittest_nonclone(dict, names, values, entries, &errors);
 
     fprintf(stderr, "\nCreating dictionary multi threaded, non-clone, add-in-front options, %zu items\n", entries);
-    dict = dictionary_create(DICTIONARY_FLAG_WITH_STATISTICS|DICTIONARY_FLAG_NAME_LINK_DONT_CLONE|DICTIONARY_FLAG_VALUE_LINK_DONT_CLONE|DICTIONARY_FLAG_ADD_IN_FRONT);
+    dict = dictionary_create(DICTIONARY_FLAG_NAME_LINK_DONT_CLONE|DICTIONARY_FLAG_VALUE_LINK_DONT_CLONE|DICTIONARY_FLAG_ADD_IN_FRONT);
     dictionary_unittest_nonclone(dict, names, values, entries, &errors);
 
     fprintf(stderr, "\nCreating dictionary single-threaded, non-clone, don't overwrite options, %zu items\n", entries);
-    dict = dictionary_create(DICTIONARY_FLAG_SINGLE_THREADED|DICTIONARY_FLAG_WITH_STATISTICS|DICTIONARY_FLAG_NAME_LINK_DONT_CLONE|DICTIONARY_FLAG_VALUE_LINK_DONT_CLONE|DICTIONARY_FLAG_DONT_OVERWRITE_VALUE);
+    dict = dictionary_create(DICTIONARY_FLAG_SINGLE_THREADED|DICTIONARY_FLAG_NAME_LINK_DONT_CLONE|DICTIONARY_FLAG_VALUE_LINK_DONT_CLONE|DICTIONARY_FLAG_DONT_OVERWRITE_VALUE);
     dictionary_unittest_run_and_measure_time(dict, "adding entries", names, values, entries, &errors, dictionary_unittest_set_nonclone);
     dictionary_unittest_run_and_measure_time(dict, "resetting non-overwrite entries", names, values, entries, &errors, dictionary_unittest_reset_dont_overwrite_nonclone);
     dictionary_unittest_run_and_measure_time(dict, "traverse foreach read loop", names, values, entries, &errors, dictionary_unittest_foreach);
@@ -1262,22 +2189,222 @@ int dictionary_unittest(size_t entries) {
     dictionary_unittest_run_and_measure_time(dict, "destroying full dictionary", names, values, entries, &errors, dictionary_unittest_destroy);
 
     fprintf(stderr, "\nCreating dictionary multi-threaded, non-clone, don't overwrite options, %zu items\n", entries);
-    dict = dictionary_create(DICTIONARY_FLAG_WITH_STATISTICS|DICTIONARY_FLAG_NAME_LINK_DONT_CLONE|DICTIONARY_FLAG_VALUE_LINK_DONT_CLONE|DICTIONARY_FLAG_DONT_OVERWRITE_VALUE);
+    dict = dictionary_create(DICTIONARY_FLAG_NAME_LINK_DONT_CLONE|DICTIONARY_FLAG_VALUE_LINK_DONT_CLONE|DICTIONARY_FLAG_DONT_OVERWRITE_VALUE);
     dictionary_unittest_run_and_measure_time(dict, "adding entries", names, values, entries, &errors, dictionary_unittest_set_nonclone);
     dictionary_unittest_run_and_measure_time(dict, "walkthrough write delete this", names, values, entries, &errors, dictionary_unittest_walkthrough_delete_this);
     dictionary_unittest_run_and_measure_time(dict, "destroying empty dictionary", names, values, entries, &errors, dictionary_unittest_destroy);
 
     fprintf(stderr, "\nCreating dictionary multi-threaded, non-clone, don't overwrite options, %zu items\n", entries);
-    dict = dictionary_create(DICTIONARY_FLAG_WITH_STATISTICS|DICTIONARY_FLAG_NAME_LINK_DONT_CLONE|DICTIONARY_FLAG_VALUE_LINK_DONT_CLONE|DICTIONARY_FLAG_DONT_OVERWRITE_VALUE);
+    dict = dictionary_create(DICTIONARY_FLAG_NAME_LINK_DONT_CLONE|DICTIONARY_FLAG_VALUE_LINK_DONT_CLONE|DICTIONARY_FLAG_DONT_OVERWRITE_VALUE);
     dictionary_unittest_run_and_measure_time(dict, "adding entries", names, values, entries, &errors, dictionary_unittest_set_nonclone);
     dictionary_unittest_run_and_measure_time(dict, "foreach write delete this", names, values, entries, &errors, dictionary_unittest_foreach_delete_this);
     dictionary_unittest_run_and_measure_time(dict, "traverse foreach read loop empty", names, values, 0, &errors, dictionary_unittest_foreach);
     dictionary_unittest_run_and_measure_time(dict, "walkthrough read callback empty", names, values, 0, &errors, dictionary_unittest_walkthrough);
     dictionary_unittest_run_and_measure_time(dict, "destroying empty dictionary", names, values, entries, &errors, dictionary_unittest_destroy);
 
+    fprintf(stderr, "\nCreating dictionary single threaded, clone, %zu items\n", entries);
+    dict = dictionary_create(DICTIONARY_FLAG_SINGLE_THREADED);
+    dictionary_unittest_sorting(dict, names, values, entries, &errors);
+    dictionary_unittest_run_and_measure_time(dict, "destroying full dictionary", names, values, entries, &errors, dictionary_unittest_destroy);
+
+    fprintf(stderr, "\nCreating dictionary single threaded, clone, %zu items\n", entries);
+    dict = dictionary_create(DICTIONARY_FLAG_SINGLE_THREADED);
+    dictionary_unittest_null_dfe(dict, names, values, entries, &errors);
+    dictionary_unittest_run_and_measure_time(dict, "destroying full dictionary", names, values, entries, &errors, dictionary_unittest_destroy);
+
+    fprintf(stderr, "\nCreating dictionary single threaded, noclone, %zu items\n", entries);
+    dict = dictionary_create(DICTIONARY_FLAG_SINGLE_THREADED|DICTIONARY_FLAG_VALUE_LINK_DONT_CLONE);
+    dictionary_unittest_null_dfe(dict, names, values, entries, &errors);
+    dictionary_unittest_run_and_measure_time(dict, "destroying full dictionary", names, values, entries, &errors, dictionary_unittest_destroy);
+
+    // check reference counters
+    {
+        fprintf(stderr, "\nTesting reference counters:\n");
+        dict = dictionary_create(DICTIONARY_FLAG_NONE|DICTIONARY_FLAG_NAME_LINK_DONT_CLONE);
+        errors += check_dictionary(dict, 0, 0);
+
+        fprintf(stderr, "\nAdding test item to dictionary and acquiring it\n");
+        dictionary_set(dict, "test", "ITEM1", 6);
+        NAME_VALUE *nv = (NAME_VALUE *)dictionary_get_and_acquire_item(dict, "test");
+
+        errors += check_dictionary(dict, 1, 1);
+        errors += check_name_value_deleted_flag(dict, nv, "test", "ITEM1", 1, NAME_VALUE_FLAG_NONE, true, true, true);
+
+        fprintf(stderr, "\nChecking that reference counters are increased:\n");
+        void *t;
+        dfe_start_read(dict, t) {
+            errors += check_dictionary(dict, 1, 1);
+            errors +=
+                check_name_value_deleted_flag(dict, nv, "test", "ITEM1", 2, NAME_VALUE_FLAG_NONE, true, true, true);
+        }
+        dfe_done(t);
+
+        fprintf(stderr, "\nChecking that reference counters are decreased:\n");
+        errors += check_dictionary(dict, 1, 1);
+        errors += check_name_value_deleted_flag(dict, nv, "test", "ITEM1", 1, NAME_VALUE_FLAG_NONE, true, true, true);
+
+        fprintf(stderr, "\nDeleting the item we have acquired:\n");
+        dictionary_del(dict, "test");
+
+        errors += check_dictionary(dict, 0, 1);
+        errors += check_name_value_deleted_flag(dict, nv, "test", "ITEM1", 1, NAME_VALUE_FLAG_DELETED, false, false, true);
+
+        fprintf(stderr, "\nAdding another item with the same name of the item we deleted, while being acquired:\n");
+        dictionary_set(dict, "test", "ITEM2", 6);
+        errors += check_dictionary(dict, 1, 2);
+
+        fprintf(stderr, "\nAcquiring the second item:\n");
+        NAME_VALUE *nv2 = (NAME_VALUE *)dictionary_get_and_acquire_item(dict, "test");
+        errors += check_name_value_deleted_flag(dict, nv, "test", "ITEM1", 1, NAME_VALUE_FLAG_DELETED, false, false, true);
+        errors += check_name_value_deleted_flag(dict, nv2, "test", "ITEM2", 1, NAME_VALUE_FLAG_NONE, true, true, true);
+
+        fprintf(stderr, "\nReleasing the second item (the first is still acquired):\n");
+        dictionary_acquired_item_release(dict, (DICTIONARY_ITEM *)nv2);
+        errors += check_dictionary(dict, 1, 2);
+        errors += check_name_value_deleted_flag(dict, nv, "test", "ITEM1", 1, NAME_VALUE_FLAG_DELETED, false, false, true);
+        errors += check_name_value_deleted_flag(dict, nv2, "test", "ITEM2", 0, NAME_VALUE_FLAG_NONE, true, true, true);
+
+        fprintf(stderr, "\nDeleting the second item (the first is still acquired):\n");
+        dictionary_del(dict, "test");
+        errors += check_dictionary(dict, 0, 1);
+        errors += check_name_value_deleted_flag(dict, nv, "test", "ITEM1", 1, NAME_VALUE_FLAG_DELETED, false, false, true);
+
+        fprintf(stderr, "\nReleasing the first item (which we have already deleted):\n");
+        dictionary_acquired_item_release(dict, (DICTIONARY_ITEM *)nv);
+        errors += check_dictionary(dict, 0, 0);
+
+        fprintf(stderr, "\nAdding again the test item to dictionary and acquiring it\n");
+        dictionary_set(dict, "test", "ITEM1", 6);
+        nv = (NAME_VALUE *)dictionary_get_and_acquire_item(dict, "test");
+
+        errors += check_dictionary(dict, 1, 1);
+        errors += check_name_value_deleted_flag(dict, nv, "test", "ITEM1", 1, NAME_VALUE_FLAG_NONE, true, true, true);
+
+        fprintf(stderr, "\nDestroying the dictionary while we have acquired an item\n");
+        dictionary_destroy(dict);
+
+        fprintf(stderr, "Releasing the item (on a destroyed dictionary)\n");
+        dictionary_acquired_item_release(dict, (DICTIONARY_ITEM *)nv);
+        nv = NULL;
+        dict = NULL;
+    }
+
+    // check string
+    {
+        long string_entries_starting = dictionary_stats_entries(&string_dictionary);
+
+        fprintf(stderr, "\nChecking strings...\n");
+
+        STRING *s1 = string_strdupz("hello unittest");
+        STRING *s2 = string_strdupz("hello unittest");
+        if(s1 != s2) {
+            errors++;
+            fprintf(stderr, "ERROR: duplicating strings are not deduplicated\n");
+        }
+        else
+            fprintf(stderr, "OK: duplicating string are deduplicated\n");
+
+        STRING *s3 = string_dup(s1);
+        if(s3 != s1) {
+            errors++;
+            fprintf(stderr, "ERROR: cloning strings are not deduplicated\n");
+        }
+        else
+            fprintf(stderr, "OK: cloning string are deduplicated\n");
+
+        STRING_ENTRY *se = (STRING_ENTRY *)s1;
+        if(se->refcount != 3) {
+            errors++;
+            fprintf(stderr, "ERROR: string refcount is not 3\n");
+        }
+        else
+            fprintf(stderr, "OK: string refcount is 3\n");
+
+        STRING *s4 = string_strdupz("world unittest");
+        if(s4 == s1) {
+            errors++;
+            fprintf(stderr, "ERROR: string is sharing pointers on different strings\n");
+        }
+        else
+            fprintf(stderr, "OK: string is properly handling different strings\n");
+
+        usec_t start_ut, end_ut;
+        STRING **strings = mallocz(entries * sizeof(STRING *));
+
+        start_ut = now_realtime_usec();
+        for(size_t i = 0; i < entries ;i++) {
+            strings[i] = string_strdupz(names[i]);
+        }
+        end_ut = now_realtime_usec();
+        fprintf(stderr, "Created %zu strings in %llu usecs\n", entries, end_ut - start_ut);
+
+        start_ut = now_realtime_usec();
+        for(size_t i = 0; i < entries ;i++) {
+            strings[i] = string_dup(strings[i]);
+        }
+        end_ut = now_realtime_usec();
+        fprintf(stderr, "Cloned %zu strings in %llu usecs\n", entries, end_ut - start_ut);
+
+        start_ut = now_realtime_usec();
+        for(size_t i = 0; i < entries ;i++) {
+            string_freez(strings[i]);
+            string_freez(strings[i]);
+        }
+        end_ut = now_realtime_usec();
+        fprintf(stderr, "Freed %zu strings in %llu usecs\n", entries, end_ut - start_ut);
+
+        freez(strings);
+
+        if(dictionary_stats_entries(&string_dictionary) != string_entries_starting + 2) {
+            errors++;
+            fprintf(stderr, "ERROR: strings dictionary should have %ld items but it has %ld\n", string_entries_starting + 2, dictionary_stats_entries(&string_dictionary));
+        }
+        else
+            fprintf(stderr, "OK: strings dictionary has 2 items\n");
+    }
+
+    // check 2-way merge
+    {
+        struct testcase {
+            char *src1; char *src2; char *expected;
+        } tests[] = {
+            { "", "", ""},
+            { "a", "", "[x]"},
+            { "", "a", "[x]"},
+            { "a", "a", "a"},
+            { "abcd", "abcd", "abcd"},
+            { "foo_cs", "bar_cs", "[x]_cs"},
+            { "cp_UNIQUE_INFIX_cs", "cp_unique_infix_cs", "cp_[x]_cs"},
+            { "cp_UNIQUE_INFIX_ci_unique_infix_cs", "cp_unique_infix_ci_UNIQUE_INFIX_cs", "cp_[x]_cs"},
+            { "foo[1234]", "foo[4321]", "foo[[x]]"},
+            { NULL, NULL, NULL },
+        };
+
+        for (struct testcase *tc = &tests[0]; tc->expected != NULL; tc++) {
+            STRING *src1 = string_strdupz(tc->src1);
+            STRING *src2 = string_strdupz(tc->src2);
+            STRING *expected = string_strdupz(tc->expected);
+
+            STRING *result = string_2way_merge(src1, src2);
+            if (string_cmp(result, expected) != 0) {
+                fprintf(stderr, "string_2way_merge(\"%s\", \"%s\") -> \"%s\" (expected=\"%s\")\n",
+                        string2str(src1),
+                        string2str(src2),
+                        string2str(result),
+                        string2str(expected));
+                errors++;
+            }
+
+            string_freez(src1);
+            string_freez(src2);
+            string_freez(expected);
+            string_freez(result);
+        }
+    }
+
     dictionary_unittest_free_char_pp(names, entries);
     dictionary_unittest_free_char_pp(values, entries);
 
     fprintf(stderr, "\n%zu errors found\n", errors);
-    return (int)errors;
+    return  errors ? 1 : 0;
 }