Avoid used_memory contention when update from multiple threads.

MeirShpilraien · Jul 21, 2024 · 0aa7245 · 0aa7245
1 parent a331978
commit 0aa7245
Show file tree

Hide file tree

Showing 3 changed files with 44 additions and 13 deletions.
diff --git a/src/config.h b/src/config.h
@@ -32,6 +32,14 @@
 #define redis_stat stat
 #endif
 
+#ifndef CACHE_LINE_SIZE
+#if defined(__aarch64__) && defined(__APPLE__)
+#define CACHE_LINE_SIZE 128
+#else
+#define CACHE_LINE_SIZE 64
+#endif
+#endif
+
 /* Test for proc filesystem */
 #ifdef __linux__
 #define HAVE_PROC_STAT 1

diff --git a/src/networking.c b/src/networking.c
@@ -4213,13 +4213,6 @@ void processEventsWhileBlocked(void) {
  * ========================================================================== */
 
 #define IO_THREADS_MAX_NUM 128
-#ifndef CACHE_LINE_SIZE
-#if defined(__aarch64__) && defined(__APPLE__)
-#define CACHE_LINE_SIZE 128
-#else
-#define CACHE_LINE_SIZE 64
-#endif
-#endif
 
 typedef struct __attribute__((aligned(CACHE_LINE_SIZE))) threads_pending {
     redisAtomic unsigned long value;

diff --git a/src/zmalloc.c b/src/zmalloc.c
@@ -67,10 +67,31 @@ void zlibc_free(void *ptr) {
 #define dallocx(ptr,flags) je_dallocx(ptr,flags)
 #endif
 
-#define update_zmalloc_stat_alloc(__n) atomicIncr(used_memory,(__n))
-#define update_zmalloc_stat_free(__n) atomicDecr(used_memory,(__n))
+#define MAX_THREADS 16 /* Keep it a power of 2 so we can use '&' instead of '%'. */
+#define THREAD_MASK (MAX_THREADS - 1)
 
-static redisAtomic size_t used_memory = 0;
+typedef struct used_memory_entry {
+    redisAtomic long used_memory;
+    char padding[CACHE_LINE_SIZE - sizeof(long)];
+} used_memory_entry;
+
+static __attribute__((aligned(CACHE_LINE_SIZE))) used_memory_entry used_memory[MAX_THREADS];
+static redisAtomic size_t num_active_threads = 0;
+static __thread long my_thread_index = -1;
+
+void update_zmalloc_stat_alloc(long num) {
+    if (my_thread_index == -1) {
+        my_thread_index = atomicIncr(num_active_threads, 1) & THREAD_MASK;
+    }
+    atomicIncr(used_memory[my_thread_index].used_memory, num);
+}
+
+void update_zmalloc_stat_free(long num) {
+    if (my_thread_index == -1) {
+        my_thread_index = atomicIncr(num_active_threads, 1) & THREAD_MASK;
+    }
+    atomicDecr(used_memory[my_thread_index].used_memory, num);
+}
 
 static void zmalloc_default_oom(size_t size) {
     fprintf(stderr, "zmalloc: Out of memory trying to allocate %zu bytes\n",
@@ -436,9 +457,18 @@ char *zstrdup(const char *s) {
 }
 
 size_t zmalloc_used_memory(void) {
-    size_t um;
-    atomicGet(used_memory,um);
-    return um;
+    size_t local_num_active_threads;
+    size_t totol_mem = 0;
+    atomicGet(num_active_threads,local_num_active_threads);
+    if (local_num_active_threads > MAX_THREADS) {
+        local_num_active_threads = MAX_THREADS;
+    }
+    for (size_t i = 0 ; i < local_num_active_threads ; ++i) {
+        size_t thread_used_mem;
+        atomicGet(used_memory[i].used_memory, thread_used_mem);
+        totol_mem += thread_used_mem;
+    }
+    return totol_mem;
 }
 
 void zmalloc_set_oom_handler(void (*oom_handler)(size_t)) {