Make object refcounts atomic, technically it doesn't need to be currently but the semantics are complicated and error prone. This is much safer with little performance impact

Former-commit-id: e310d33c121550f69b1c06d101db0c3f944a7fb0

src/debug.cpp

@@ -436,7 +436,7 @@ NULL
             "Value at:%p refcount:%d "
             "encoding:%s serializedlength:%zu "
             "lru:%d lru_seconds_idle:%llu%s",
-            (void*)val, val->refcount,
+            (void*)val, static_cast<int>(val->refcount),
             strenc, rdbSavedObjectLen(val),
             val->lru, estimateObjectIdleTime(val)/1000, extra);
     } else if (!strcasecmp(szFromObj(c->argv[1]),"sdslen") && c->argc == 3) {
@@ -721,14 +721,14 @@ void _serverAssertPrintClientInfo(const client *c) {
             arg = buf;
         }
         serverLog(LL_WARNING,"client->argv[%d] = \"%s\" (refcount: %d)",
-            j, arg, c->argv[j]->refcount);
+            j, arg, static_cast<int>(c->argv[j]->refcount));
     }
 }
 
 void serverLogObjectDebugInfo(robj_roptr o) {
     serverLog(LL_WARNING,"Object type: %d", o->type);
     serverLog(LL_WARNING,"Object encoding: %d", o->encoding);
-    serverLog(LL_WARNING,"Object refcount: %d", o->refcount);
+    serverLog(LL_WARNING,"Object refcount: %d", static_cast<int>(o->refcount));
     if (o->type == OBJ_STRING && sdsEncodedObject(o)) {
         serverLog(LL_WARNING,"Object raw string len: %zu", sdslen(szFromObj(o)));
         if (sdslen(szFromObj(o)) < 4096) {

src/networking.cpp

@@ -2290,6 +2290,7 @@ sds getAllClientsInfoString(int type) {
     listRewind(g_pserver->clients,&li);
     while ((ln = listNext(&li)) != NULL) {
         client = reinterpret_cast<struct client*>(listNodeValue(ln));
+        std::unique_lock<decltype(client->lock)> lock(client->lock);
         if (type != -1 && getClientType(client) != type) continue;
         o = catClientInfoString(o,client);
         o = sdscatlen(o,"\n",1);

src/object.cpp

@@ -43,7 +43,7 @@ robj *createObject(int type, void *ptr) {
     o->type = type;
     o->encoding = OBJ_ENCODING_RAW;
     o->m_ptr = ptr;
-    o->refcount = 1;
+    o->refcount.store(1, std::memory_order_relaxed);
     o->mvcc_tstamp = OBJ_MVCC_INVALID;
 
     /* Set the LRU to the current lruclock (minutes resolution), or
@@ -69,7 +69,7 @@ robj *createObject(int type, void *ptr) {
  */
 robj *makeObjectShared(robj *o) {
     serverAssert(o->refcount == 1);
-    o->refcount = OBJ_SHARED_REFCOUNT;
+    o->refcount.store(OBJ_SHARED_REFCOUNT, std::memory_order_relaxed);
     return o;
 }
 
@@ -91,7 +91,7 @@ robj *createEmbeddedStringObject(const char *ptr, size_t len) {
 
     o->type = OBJ_STRING;
     o->encoding = OBJ_ENCODING_EMBSTR;
-    o->refcount = 1;
+    o->refcount.store(1, std::memory_order_relaxed);
     o->mvcc_tstamp = OBJ_MVCC_INVALID;
 
     if (g_pserver->maxmemory_policy & MAXMEMORY_FLAG_LFU) {
@@ -352,7 +352,7 @@ void freeStreamObject(robj_roptr o) {
 }
 
 void incrRefCount(robj_roptr o) {
-    if (o->refcount != OBJ_SHARED_REFCOUNT) o->refcount++;
+    if (o->refcount != OBJ_SHARED_REFCOUNT) o->refcount.fetch_add(1, std::memory_order_acquire);
 }
 
 void decrRefCount(robj_roptr o) {
@@ -370,7 +370,7 @@ void decrRefCount(robj_roptr o) {
         zfree(o.unsafe_robjcast());
     } else {
         if (o->refcount <= 0) serverPanic("decrRefCount against refcount <= 0");
-        if (o->refcount != OBJ_SHARED_REFCOUNT) o->refcount--;
+        if (o->refcount != OBJ_SHARED_REFCOUNT) o->refcount.fetch_sub(1, std::memory_order_acquire);
     }
 }
 

src/server.h

@@ -709,13 +709,14 @@ typedef struct RedisModuleDigest {
 
 #define OBJ_SHARED_REFCOUNT INT_MAX
 #define OBJ_MVCC_INVALID (0xFFFFFFFFFFFFFFFFULL)
+
 typedef struct redisObject {
     unsigned type:4;
     unsigned encoding:4;
     unsigned lru:LRU_BITS; /* LRU time (relative to global lru_clock) or
                             * LFU data (least significant 8 bits frequency
                             * and most significant 16 bits access time). */
-    mutable int refcount;
+    mutable std::atomic<int> refcount;
     uint64_t mvcc_tstamp;
     void *m_ptr;
 } robj;