make refcount atomic

src/aof.c

@@ -2423,7 +2423,7 @@ int rewriteObject(rio *r, robj *key, robj *o, int dbid, long long expiretime) {
     }
 
     /* If modules metadata is available */
-    if ((getModuleMetaBits(o->metabits)) && (keyMetaOnAof(r, key, o, dbid) == 0))
+    if ((getModuleMetaBits(o->flags.metabits)) && (keyMetaOnAof(r, key, o, dbid) == 0))
         return C_ERR;
 
     return C_OK;

src/cluster.c

@@ -94,7 +94,7 @@ void createDumpPayload(rio *payload, robj *o, robj *key, int dbid, int skip_chec
     rioInitWithBuffer(payload,sdsempty());
 
     /* Save key metadata if present without (handles TTL separately via command args) */
-    if (getModuleMetaBits(o->metabits))
+    if (getModuleMetaBits(o->flags.metabits))
         serverAssert(rdbSaveKeyMetadata(payload, key, o, dbid) != -1);
     serverAssert(rdbSaveObjectType(payload,o));
     serverAssert(rdbSaveObject(payload,o,key,dbid));

src/cluster_asm.c

@@ -3439,7 +3439,7 @@ void asmActiveTrimDeleteKey(redisDb *db, robj *keyobj) {
         debugDelay(asmManager->debug_active_trim_delay);
 
     /* The key needs to be converted from static to heap before deletion. */
-    int static_key = keyobj->refcount == OBJ_STATIC_REFCOUNT;
+    int static_key = keyobj->flags.refcount == OBJ_STATIC_REFCOUNT;
     if (static_key) keyobj = createStringObject(keyobj->ptr, sdslen(keyobj->ptr));
 
     dbDelete(db, keyobj);

src/db.c

@@ -59,7 +59,7 @@ void updateLFU(robj *val) {
 
 /* Update LRM when an object is modified. */
 void updateLRM(robj *o) {
-    if (o->refcount == OBJ_SHARED_REFCOUNT)
+    if (o->flags.refcount == OBJ_SHARED_REFCOUNT)
         return;
     if (server.maxmemory_policy & MAXMEMORY_FLAG_LRM) {
         o->lru = LRU_CLOCK();
@@ -591,10 +591,10 @@ static void dbSetValue(redisDb *db, robj *key, robj **valref, dictEntryLink link
     long long oldExpire = getExpire(db, key->ptr, old);
 
     /* All metadata will be kept if not `overwrite` for the new object  */
-    uint32_t newKeyMetaBits = old->metabits;
+    uint32_t newKeyMetaBits = old->flags.metabits;
     /* clear expire if not keepTTL or no old expire */
     if ((!keepTTL) || (oldExpire == -1))
-        newKeyMetaBits &= ~KEY_META_MASK_EXPIRE; 
+        newKeyMetaBits &= ~KEY_META_MASK_EXPIRE;
 
     if (overwrite) {
         /* On overwrite, discard module metadata excluding expire if set */
@@ -604,7 +604,7 @@ static void dbSetValue(redisDb *db, robj *key, robj **valref, dictEntryLink link
         incrRefCount(old);
 
         /* Free related metadata. Ignore builtin metadata (currently only expire) */
-        if (getModuleMetaBits(old->metabits)) {
+        if (getModuleMetaBits(old->flags.metabits)) {
             keyMetaOnUnlink(db, key, old);
             freeModuleMeta = 1;
         }
@@ -622,8 +622,8 @@ static void dbSetValue(redisDb *db, robj *key, robj **valref, dictEntryLink link
     if (server.memory_tracking_enabled)
         oldsize = kvobjAllocSize(old);
 
-    if ((old->refcount == 1 && old->encoding != OBJ_ENCODING_EMBSTR) &&
-        (val->refcount == 1 && val->encoding != OBJ_ENCODING_EMBSTR) && (!freeModuleMeta))
+    if ((old->flags.refcount == 1 && old->encoding != OBJ_ENCODING_EMBSTR) &&
+        (val->flags.refcount == 1 && val->encoding != OBJ_ENCODING_EMBSTR) && (!freeModuleMeta))
     {
         /* Keep old object in the database. Just swap it's ptr, type and
          * encoding with the content of val. */
@@ -687,7 +687,7 @@ static void dbSetValue(redisDb *db, robj *key, robj **valref, dictEntryLink link
         }
     }
 
-    if (server.io_threads_num > 1 && old->encoding == OBJ_ENCODING_RAW && old->refcount == 1) {
+    if (server.io_threads_num > 1 && old->encoding == OBJ_ENCODING_RAW && old->flags.refcount == 1) {
         /* In multi-threaded mode, the OBJ_ENCODING_RAW string object usually is
          * allocated in the IO thread, so we defer the free to the IO thread.
          * Besides, we never free a string object in BIO threads, so, even with
@@ -853,7 +853,7 @@ int dbGenericDelete(redisDb *db, robj *key, int async, int flags) {
          * need to incr to retain kv */
         incrRefCount(kv); /* refcnt=1->2 */
         /* Metadata hook: notify unlink for key metadata cleanup. */
-        if (getModuleMetaBits(kv->metabits)) keyMetaOnUnlink(db, key, kv);
+        if (getModuleMetaBits(kv->flags.metabits)) keyMetaOnUnlink(db, key, kv);
         /* Tells the module that the key has been unlinked from the database. */
         moduleNotifyKeyUnlink(key, kv, db->id, flags);
         /* We want to try to unblock any module clients or clients using a blocking XREADGROUP */
@@ -950,7 +950,7 @@ kvobj *dbUnshareStringValue(redisDb *db, robj *key, kvobj *kv) {
  * which can be used if we already have one, thus saving the dbFind call. */
 kvobj *dbUnshareStringValueByLink(redisDb *db, robj *key, kvobj *o, dictEntryLink link) {
     serverAssert(o->type == OBJ_STRING);
-    if (o->refcount != 1 || o->encoding != OBJ_ENCODING_RAW) {
+    if (o->flags.refcount != 1 || o->encoding != OBJ_ENCODING_RAW) {
         robj *decoded = getDecodedObject(o);
         o = createRawStringObject(decoded->ptr, sdslen(decoded->ptr));
         decrRefCount(decoded);
@@ -2166,7 +2166,7 @@ void renameGenericCommand(client *c, int nx) {
     /* Prepare metadata for the renamed key */
     KeyMetaSpec keymeta;
     keyMetaSpecInit(&keymeta);
-    if (o->metabits) keyMetaOnRename(c->db, o, c->argv[1], c->argv[2], &keymeta);
+    if (o->flags.metabits) keyMetaOnRename(c->db, o, c->argv[1], c->argv[2], &keymeta);
 
     dbDelete(c->db,c->argv[1]);
 
@@ -2381,7 +2381,7 @@ void copyCommand(client *c) {
     /* Prepare metadata for the new key */
     KeyMetaSpec keymeta;
     keyMetaSpecInit(&keymeta);
-    if (o->metabits) keyMetaOnCopy(o, key, newkey, c->db->id, dst->id, &keymeta);
+    if (o->flags.metabits) keyMetaOnCopy(o, key, newkey, c->db->id, dst->id, &keymeta);
 
     kvobj *kvCopy = dbAddInternal(dst, newkey, &newobj, NULL, &keymeta);
 
@@ -2696,7 +2696,7 @@ static void deleteKeyAndPropagate(redisDb *db, robj *keyobj, int notify_type, lo
     char *notify_name = notify_type == NOTIFY_EXPIRED ? "expired" : "evicted";
 
     /* The key needs to be converted from static to heap before deleted */
-    int static_key = keyobj->refcount == OBJ_STATIC_REFCOUNT;
+    int static_key = keyobj->flags.refcount == OBJ_STATIC_REFCOUNT;
     if (static_key) {
         keyobj = createStringObject(keyobj->ptr, sdslen(keyobj->ptr));
     }

src/debug.c

@@ -685,7 +685,7 @@ NULL
             "Value at:%p refcount:%d "
             "encoding:%s serializedlength:%zu "
             "lru:%d lru_seconds_idle:%llu%s",
-            (void*)kv, kv->refcount,
+            (void*)kv, kv->flags.refcount,
             strenc, rdbSavedObjectLen(kv, c->argv[2], c->db->id),
             kv->lru, estimateObjectIdleTime(kv)/1000, extra);
     } else if (!strcasecmp(c->argv[1]->ptr,"sdslen") && c->argc == 3) {
@@ -1250,14 +1250,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, c->argv[j]->flags.refcount);
     }
 }
 
 void serverLogObjectDebugInfo(const robj *o) {
     serverLog(LL_WARNING,"Object type: %u", o->type);
     serverLog(LL_WARNING,"Object encoding: %u", o->encoding);
-    serverLog(LL_WARNING,"Object refcount: %d", o->refcount);
+    serverLog(LL_WARNING,"Object refcount: %d", o->flags.refcount);
 #if UNSAFE_CRASH_REPORT
     /* This code is now disabled. o->ptr may be unreliable to print. in some
      * cases a ziplist could have already been freed by realloc, but not yet

src/defrag.c

@@ -293,9 +293,9 @@ void *activeDefragHfieldAndUpdateRef(void *ptr, void *privdata) {
  * reference count is not 1, in these cases, the caller must explicitly pass
  * in the reference count, otherwise defragmentation will not be performed.
  * Note that the caller is responsible for updating any other references to the robj. */
-robj *activeDefragStringObEx(robj* ob, int expected_refcount) {
+robj *activeDefragStringObEx(robj* ob, unsigned int expected_refcount) {
     robj *ret = NULL;
-    if (ob->refcount!=expected_refcount)
+    if (ob->flags.refcount!=expected_refcount)
         return NULL;
 
     /* try to defrag robj (only if not an EMBSTR type (handled below). */
@@ -1059,7 +1059,7 @@ robj *activeDefragKvobj(kvobj* kv, int without_free) {
     long offsetEmbstr = LONG_MIN;
 
     /* Don't defrag kvobj's with multiple references (refcount > 1) */
-    if (kv->refcount != 1)
+    if (kv->flags.refcount != 1)
         return NULL;
 
     /* Calculate offset for EMBSTR strings */
@@ -1131,7 +1131,7 @@ void defragKey(defragKeysCtx *ctx, dictEntry *de, dictEntryLink link) {
         /* Only defrag strings with refcount==1 (String might be shared as dict 
          * keys, e.g. pub/sub channels, and may be accessed by IO threads. Other 
          * types are never used as dict keys) */
-        if ((ob->refcount==1) && (ob->encoding == OBJ_ENCODING_RAW)) {
+        if ((ob->flags.refcount==1) && (ob->encoding == OBJ_ENCODING_RAW)) {
             /* For RAW strings, defrag the separate SDS allocation */
             sds newsds = activeDefragSds((sds)ob->ptr);
             if (newsds) ob->ptr = newsds;
@@ -1254,7 +1254,7 @@ void defragPubsubScanCallback(void *privdata, const dictEntry *de, dictEntryLink
     dict *newclients, *clients = dictGetVal(de);
 
     /* Try to defrag the channel name. */
-    serverAssert(channel->refcount == (int)dictSize(clients) + 1);
+    serverAssert(channel->flags.refcount == dictSize(clients) + 1);
     newchannel = activeDefragStringObEx(channel, dictSize(clients) + 1);
     if (newchannel) {
         kvstoreDictSetKey(pubsub_channels, ctx->kvstate.slot, (dictEntry*)de, newchannel);

src/iothread.c

@@ -183,8 +183,6 @@ void keepClientInMainThread(client *c) {
     c->io_flags |= CLIENT_IO_READ_ENABLED | CLIENT_IO_WRITE_ENABLED;
     c->tid = IOTHREAD_MAIN_THREAD_ID;
     freeClientDeferredObjects(c, 1); /* Free deferred objects. */
-    freeClientIODeferredObjects(c, 1); /* Free IO deferred objects. */
-    tryUnlinkClientFromPendingRefReply(c, 0);
     /* Main thread starts to manage it. */
     server.io_threads_clients_num[c->tid]++;
 }
@@ -586,10 +584,6 @@ int processClientsFromIOThread(IOThread *t) {
         /* Let main thread to run it, set running thread id first. */
         c->running_tid = IOTHREAD_MAIN_THREAD_ID;
 
-        /* Free objects queued by IO thread for deferred freeing. */
-        freeClientIODeferredObjects(c, 0);
-        tryUnlinkClientFromPendingRefReply(c, 0);
-
         /* If a read error occurs, handle it in the main thread first, since we
          * want to print logs about client information before freeing. */
         if (isClientReadErrorFatal(c)) handleClientReadError(c);