memory mode: prefix swap related struct members, function names with swap_.

Author: patpatbear

src/blocked.c

@@ -65,11 +65,11 @@
 #include "latency.h"
 #include "monotonic.h"
 
+int serveClientBlockedOnList(client *receiver, robj *key, robj *dstkey, redisDb *db, robj *value, int wherefrom, int whereto
 #ifdef ENABLE_SWAP
-int serveClientBlockedOnList(client *receiver, robj *key, robj *dstkey, redisDb *db, robj *value, int wherefrom, int whereto, list* swap_wrong_type_error_keys);
-#else
-int serveClientBlockedOnList(client *receiver, robj *key, robj *dstkey, redisDb *db, robj *value, int wherefrom, int whereto);
+        , list* swap_wrong_type_error_keys
 #endif
+        );
 int getListPositionFromObjectOrReply(client *c, robj *arg, int *position);
 
 /* This structure represents the blocked key information that we store
@@ -272,11 +272,11 @@ void disconnectAllBlockedClients(void) {
 /* Helper function for handleClientsBlockedOnKeys(). This function is called
  * when there may be clients blocked on a list key, and there may be new
  * data to fetch (the key is ready). */
+void serveClientsBlockedOnListKey(robj *o, readyList *rl
 #ifdef ENABLE_SWAP
-void serveClientsBlockedOnListKey(robj *o, readyList *rl, list* swap_wrong_type_error_keys) {
-#else
-void serveClientsBlockedOnListKey(robj *o, readyList *rl) {
+, list* swap_wrong_type_error_keys
 #endif
+        ) {
     /* We serve clients in the same order they blocked for
      * this key, from the first blocked to the last. */
     dictEntry *de = dictFind(rl->db->blocking_keys,rl->key);
@@ -299,7 +299,7 @@ void serveClientsBlockedOnListKey(robj *o, readyList *rl) {
             int wherefrom = receiver->bpop.listpos.wherefrom;
             int whereto = receiver->bpop.listpos.whereto;
 #ifdef ENABLE_SWAP
-            robj *value = ctripListTypePop(o, wherefrom, rl->db, rl->key);
+            robj *value = swapListTypePop(o, wherefrom, rl->db, rl->key);
 #else
             robj *value = listTypePop(o, wherefrom);
 #endif
@@ -314,16 +314,16 @@ void serveClientsBlockedOnListKey(robj *o, readyList *rl) {
                 elapsedStart(&replyTimer);
                 if (serveClientBlockedOnList(receiver,
                     rl->key,dstkey,rl->db,value,
+                    wherefrom, whereto
 #ifdef ENABLE_SWAP
-                    wherefrom, whereto, swap_wrong_type_error_keys) == C_ERR)
-#else
-                    wherefrom, whereto) == C_ERR)
+                    , swap_wrong_type_error_keys
 #endif
+                    ) == C_ERR)
                 {
                     /* If we failed serving the client we need
                      * to also undo the POP operation. */
 #ifdef ENABLE_SWAP
-                    ctripListTypePush(o,value,wherefrom, rl->db, rl->key);
+                    swapListTypePush(o,value,wherefrom, rl->db, rl->key);
 #else
                     listTypePush(o,value,wherefrom);
 #endif
@@ -340,8 +340,7 @@ void serveClientsBlockedOnListKey(robj *o, readyList *rl) {
     }
 
 #ifdef ENABLE_SWAP
-    objectMeta *om = lookupMeta(rl->db, rl->key);
-    if (ctripListTypeLength(o, om) == 0) {
+    if (swapListTypeLength(o, lookupMeta(rl->db, rl->key)) == 0) {
 #else
     if (listTypeLength(o) == 0) {
 #endif

src/config.c

@@ -2998,7 +2998,7 @@ standardConfig configs[] = {
     createIntConfig("ctrip-monitor-port", NULL, IMMUTABLE_CONFIG, 0, 65535, server.ctrip_monitor_port, 6380, INTEGER_CONFIG, NULL, NULL), /* Monitor TCP port. */
     createIntConfig("swap-slow-expire-effort", NULL, MODIFIABLE_CONFIG, -10, 10, server.swap_slow_expire_effort, -5, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("swap-debug-evict-keys", NULL, MODIFIABLE_CONFIG, -1, INT_MAX, server.swap_debug_evict_keys, 0, INTEGER_CONFIG, NULL, NULL),
-    createIntConfig("ps-parallism-rdb", NULL, MODIFIABLE_CONFIG, 4, 16384, server.ps_parallism_rdb, 32, INTEGER_CONFIG, NULL, NULL),
+    createIntConfig("swap-ps-parallism-rdb", "ps-parallism-rdb", MODIFIABLE_CONFIG, 4, 16384, server.swap_ps_parallism_rdb, 32, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("swap-evict-step-max-subkeys", NULL, MODIFIABLE_CONFIG, 0, 65536, server.swap_evict_step_max_subkeys, 1024, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("swap-debug-rio-delay-micro", NULL, MODIFIABLE_CONFIG, -1, INT_MAX, server.swap_debug_rio_delay_micro, 0, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("swap-threads", NULL, IMMUTABLE_CONFIG, 4, 64, server.swap_threads_num, 4, INTEGER_CONFIG, NULL, NULL),

src/ctrip_swap.c

@@ -545,14 +545,14 @@ void swapMutexopCommand(client *c) {
 }
 
 int lockGlobalAndExec(clientKeyRequestFinished locked_op, uint64_t exclude_mark) {
-    if (exclude_mark && server.req_submitted&exclude_mark) {
+    if (exclude_mark && server.swap_req_submitted&exclude_mark) {
         return 0;
     }
     /* add flag before submit request otherwise when
      * global lock no block, flag may be del just after submit */
-    server.req_submitted |= exclude_mark;
+    server.swap_req_submitted |= exclude_mark;
 
-    client *c = server.mutex_client;
+    client *c = server.swap_mutex_client;
     getKeyRequestsResult result = GET_KEYREQUESTS_RESULT_INIT;
     getKeyRequestsPrepareResult(&result,1);
     getKeyRequestsAppendSubkeyResult(&result,REQUEST_LEVEL_SVR,NULL,0,NULL,

src/ctrip_swap.h

@@ -409,7 +409,7 @@ void swapPersistCtxPersistKeys(swapPersistCtx *ctx);
 sds genSwapPersistInfoString(sds info);
 void swapPersistKeyRequestFinished(swapPersistCtx *ctx, int dbid, robj *key, uint64_t persist_version);
 void loadDataFromDisk(void);
-void ctripLoadDataFromDisk(void);
+void swap_loadDataFromDisk(void);
 int submitEvictClientRequest(client *c, robj *key, int persist_keep, uint64_t persist_version);
 
 #define setObjectPersistKeep(o) do { \
@@ -864,7 +864,7 @@ void startSwapRewind(swap_rewind_type rewind_type);
 void endSwapRewind(void);
 void freeClientSwapCmdTrace(client *c);
 
-/* see server.req_submitted */
+/* see server.swap_req_submitted */
 #define REQ_SUBMITTED_NONE 0
 #define REQ_SUBMITTED_BGSAVE (1ULL<<0)
 #define REQ_SUBMITTED_REPL_START (1ULL<<1)
@@ -927,7 +927,7 @@ int swapDataSetupSet(swapData *d, OUT void **datactx);
 
 unsigned long swap_setTypeSize(const objectMeta *meta, const robj *o);
 unsigned long swap_setTypeSizeLookup(redisDb *db, robj *key, const robj *o);
-void ctrip_scardCommand(client *c);
+void swap_scardCommand(client *c);
 
 /* List */
 typedef struct listSwapData {
@@ -964,10 +964,10 @@ void clientArgRewritesRestore(client *c);
 void clientArgRewrite(client *c, argRewriteRequest arg_req, MOVE robj *new_arg);
 
 
-long ctripListTypeLength(robj *list, objectMeta *object_meta);
-void ctripListTypePush(robj *subject, robj *value, int where, redisDb *db, robj *key);
-robj *ctripListTypePop(robj *subject, int where, redisDb *db, robj *key);
-void ctripListMetaDelRange(redisDb *db, robj *key, long ltrim, long rtrim);
+long swapListTypeLength(robj *list, objectMeta *object_meta);
+void swapListTypePush(robj *subject, robj *value, int where, redisDb *db, robj *key);
+robj *swapListTypePop(robj *subject, int where, redisDb *db, robj *key);
+void swapListMetaDelRange(redisDb *db, robj *key, long ltrim, long rtrim);
 /* zset */
 typedef struct zsetSwapData {
   swapData sd;
@@ -1712,25 +1712,25 @@ typedef struct swapEvictKeysCtx {
     int ended;
 } swapEvictKeysCtx;
 
-void ctrip_startEvictionTimeProc(void);
-size_t ctrip_getMemoryToFree(size_t mem_used);
-void ctrip_performEvictionStart(swapEvictKeysCtx *sectx);
-int ctrip_performEvictionLoopStartShouldBreak(swapEvictKeysCtx *sectx);
+void swap_startEvictionTimeProc(void);
+size_t swap_getMemoryToFree(size_t mem_used);
+void swap_performEvictionStart(swapEvictKeysCtx *sectx);
+int swap_performEvictionLoopStartShouldBreak(swapEvictKeysCtx *sectx);
 size_t performEvictionSwapSelectedKey(swapEvictKeysCtx *sectx, redisDb *db, robj *keyobj);
-int ctrip_performEvictionLoopCheckShouldBreak(swapEvictKeysCtx *sectx);
-void ctrip_performEvictionEnd(swapEvictKeysCtx *sectx);
-static inline int ctrip_performEvictionLoopCheckInterval(int keys_freed) {
+int swap_performEvictionLoopCheckShouldBreak(swapEvictKeysCtx *sectx);
+void swap_performEvictionEnd(swapEvictKeysCtx *sectx);
+static inline int swap_performEvictionLoopCheckInterval(int keys_freed) {
     return keys_freed % server.swap_evict_loop_check_interval == 0;
 }
 /* used memory in disk swap mode */
 size_t coldFiltersUsedMemory(void); /* cuckoo filter not counted in maxmemory */
-static inline size_t ctrip_getUsedMemory(void) {
+static inline size_t swap_getUsedMemory(void) {
   int swap_inprogress_memory;
   atomicGet(server.swap_inprogress_memory, swap_inprogress_memory);
   return zmalloc_used_memory() - swap_inprogress_memory -
       coldFiltersUsedMemory() - swapPersistCtxUsedMemory(server.swap_persist_ctx);
 }
-static inline int ctrip_evictionTimeProcGetDelayMillis(void) {
+static inline int swap_evictionTimeProcGetDelayMillis(void) {
   if (swapEvictionReachedInprogressLimit()) return 1;
   else return 0;
 }
@@ -2302,7 +2302,7 @@ typedef struct rdbSaveRocksStats {
 } rdbSaveRocksStats;
 
 /* rdb save */
-int rdbSaveKeyHeader(rio *rdb, robj *key, robj *evict, unsigned char rdbtype, long long expiretime);
+int rdbSaveKeyHeader(rio *rdb, robj *key, robj *o, unsigned char rdbtype, long long expiretime);
 int rdbKeySaveHotExtensionInit(rdbKeySaveData *keydata, redisDb *db, sds keystr);
 int rdbKeySaveWarmColdInit(rdbKeySaveData *keydata, redisDb *db, decodedResult *dr);
 void rdbKeySaveDataDeinit(rdbKeySaveData *keydata);
@@ -2324,7 +2324,7 @@ int bitmapSaveInit(rdbKeySaveData *save, uint64_t version, const char *extend, s
 #define RDB_LOAD_BATCH_COUNT 50
 #define RDB_LOAD_BATCH_CAPACITY  (4*1024*1024)
 
-typedef struct ctripRdbLoadCtx {
+typedef struct swapRdbLoadObjectCtx {
   size_t errors;
   size_t idx;
   struct {
@@ -2336,10 +2336,10 @@ typedef struct ctripRdbLoadCtx {
     sds *rawkeys;
     sds *rawvals;
   } batch;
-} ctripRdbLoadCtx;
+} swapRdbLoadObjectCtx;
 
-void evictStartLoading(void);
-void evictStopLoading(int success);
+void swapStartLoading(void);
+void swapStopLoading(int success);
 
 struct rdbKeyLoadData;
 
@@ -2374,7 +2374,7 @@ static inline sds rdbVerbatimNew(unsigned char rdbtype) {
 
 int rdbLoadStringVerbatim(rio *rdb, sds *verbatim);
 int rdbLoadHashFieldsVerbatim(rio *rdb, unsigned long long len, sds *verbatim);
-int ctripRdbLoadObject(int rdbtype, rio *rdb, redisDb *db, sds key, long long expire, long long now, rdbKeyLoadData *keydata);
+int swapRdbLoadObject(int rdbtype, rio *rdb, redisDb *db, sds key, long long expire, long long now, rdbKeyLoadData *keydata);
 robj *rdbKeyLoadGetObject(rdbKeyLoadData *keydata);
 int rdbKeyLoadDataInit(rdbKeyLoadData *keydata, int rdbtype, redisDb *db, sds key, long long expire, long long now);
 void rdbKeyLoadDataDeinit(rdbKeyLoadData *keydata);
@@ -2610,11 +2610,22 @@ void swapDebugCommand(client *c);
 void swapExpiredCommand(client *c);
 const char *strObjectType(int type);
 int timestampIsExpired(mstime_t expire);
-size_t ctripDbSize(redisDb *db);
+size_t swap_dbSize(redisDb *db);
 long get_dir_size(char *dirname);
 robj *swapRandomKey(redisDb *db, metaScanResult *metas);
 void dbPauseRehash(redisDb *db);
 void dbResumeRehash(redisDb *db);
+int debugGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
+void commandProcessed(client *c);
+ssize_t rdbWriteRaw(rio *rdb, void *p, size_t len);
+void trackInstantaneousMetric(int metric, long long current_reading);
+long long getInstantaneousMetric(int metric);
+void swapInitServerConfig(void);
+void swapInitServer(void);
+void freeClientsInDeferedQueue(void);
+void swap_replicationStartPendingFork(void);
+void debugSwapOutCommand(client *c);
+size_t swapobjectComputeSize(robj *val, int samples, objectMeta *object_meta);
 
 void notifyKeyspaceEventDirty(int type, char *event, robj *key, int dbid, ...);
 void notifyKeyspaceEventDirtyKey(int type, char *event, robj *key, int dbid);

src/ctrip_swap_async.c

@@ -124,7 +124,7 @@ int asyncCompleteQueueInit() {
         return -1;
     }
 
-    server.CQ = cq;
+    server.swap_CQ = cq;
     return 0;
 }
 
@@ -137,7 +137,7 @@ void asyncCompleteQueueDeinit(asyncCompleteQueue *cq) {
 
 void asyncSwapRequestNotifyCallback(swapRequestBatch *reqs, void *pd) {
     UNUSED(pd);
-    asyncCompleteQueueAppend(server.CQ, reqs);
+    asyncCompleteQueueAppend(server.swap_CQ, reqs);
 }
 
 void asyncCompleteQueueAppend(asyncCompleteQueue *cq, swapRequestBatch *reqs) {
@@ -164,9 +164,9 @@ static int asyncCompleteQueueDrained() {
     int drained = 1;
 
     if (!swapThreadsDrained()) return 0;
-    pthread_mutex_lock(&server.CQ->lock);
-    if (listLength(server.CQ->complete_queue)) drained = 0;
-    pthread_mutex_unlock(&server.CQ->lock);
+    pthread_mutex_lock(&server.swap_CQ->lock);
+    if (listLength(server.swap_CQ->complete_queue)) drained = 0;
+    pthread_mutex_unlock(&server.swap_CQ->lock);
 
     return drained;
 }
@@ -176,7 +176,7 @@ int asyncCompleteQueueDrain(mstime_t time_limit) {
     mstime_t start = mstime();
 
     while (!asyncCompleteQueueDrained()) {
-        asyncCompleteQueueProcess(server.CQ);
+        asyncCompleteQueueProcess(server.swap_CQ);
 
         if (time_limit >= 0 && mstime() - start > time_limit) {
             result = -1;

src/ctrip_swap_blocked.c

@@ -210,7 +210,7 @@ int serveClientsBlockedOnListKeyWithoutTargetKey(robj *o, readyList *rl) {
             }
             int wherefrom = receiver->bpop.listpos.wherefrom;
             int whereto = receiver->bpop.listpos.whereto;
-            robj *value = ctripListTypePop(o, wherefrom, rl->db, rl->key);
+            robj *value = swapListTypePop(o, wherefrom, rl->db, rl->key);
             if (value) {
                 /* Protect receiver->bpop.target, that will be
                  * freed by the next unblockClient()
@@ -225,7 +225,7 @@ int serveClientsBlockedOnListKeyWithoutTargetKey(robj *o, readyList *rl) {
                 {
                     /* If we failed serving the client we need
                      * to also undo the POP operation. */
-                    ctripListTypePush(o,value,wherefrom, rl->db, rl->key);
+                    swapListTypePush(o,value,wherefrom, rl->db, rl->key);
                 }
                 updateStatsOnUnblock(receiver, 0, elapsedUs(replyTimer));
                 unblockClient(receiver);
@@ -240,7 +240,7 @@ int serveClientsBlockedOnListKeyWithoutTargetKey(robj *o, readyList *rl) {
     }
 end:
     om = lookupMeta(rl->db, rl->key);
-    if (ctripListTypeLength(o, om) == 0) {
+    if (swapListTypeLength(o, om) == 0) {
         dbDelete(rl->db,rl->key);
         notifyKeyspaceEvent(NOTIFY_GENERIC,"del",rl->key,rl->db->id);
         exists_list_blocked_with_target_key = 0;

src/ctrip_swap_cmd.c

@@ -226,7 +226,7 @@ struct redisCommand redisCommandTable[SWAP_CMD_COUNT] = {
      "read-only fast @set @swap_set",
      0,NULL,getKeyRequestSmembers,SWAP_IN,0,1,1,1,0,0,0},
 
-    {"scard",ctrip_scardCommand,2,
+    {"scard",swap_scardCommand,2,
      "read-only fast @set @swap_set",
      0,NULL,NULL,SWAP_IN,SWAP_IN_META,1,1,1,0,0,0},
 

src/ctrip_swap_compact.c

@@ -112,7 +112,7 @@ static unsigned char metaVersionFilterFilt(void* mvfilter_, int level, int cf, c
     if (state == FILTER_STATE_CLOSE) return 0;
     /* Since release 6.0, with compaction filter enabled, RocksDB always invoke filtering for any key,
      * even if it knows it will make a snapshot not repeatable. */
-    atomicGet(server.inflight_snapshot, inflight_snapshot);
+    atomicGet(server.rocksdb_inflight_snapshot, inflight_snapshot);
     if (inflight_snapshot > 0) return 0;
 
     updateCompactionFiltScanCount(cf);

src/ctrip_swap_debug.c

@@ -28,6 +28,24 @@
 
 #include "ctrip_swap.h"
 
+int debugGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    int *keys = NULL;
+    UNUSED(cmd);
+    if (argc == 3 && !strcasecmp(argv[1]->ptr,"object")) {
+        keys = getKeysPrepareResult(result,1);
+        result->numkeys = 1;
+        keys[0] = 2;
+    } else if (argc >= 3 && !strcasecmp(argv[1]->ptr,"digest-value")) {
+        keys = getKeysPrepareResult(result,argc-2);
+        result->numkeys = argc-2;
+        for (int i = 2; i < argc; i++) keys[i-2] = i;
+    } else {
+        keys = getKeysPrepareResult(result,1);
+        result->numkeys = 0;
+    }
+    return result->numkeys;
+}
+
 static sds debugRioGet(int cf, sds rawkey) {
     sds rawval;
     RIO _rio, *rio = &_rio;