[WIP] Feature cherry-up

.github/workflows/ci.yml

@@ -2,9 +2,9 @@ name: CI
 
 on:
   push:
-    branches: [ xredis_2_ror ]
+    branches: [ main,deprecated/xredis_2_ror_v1 ]
   pull_request:
-    branches: [ xredis_2_ror ]
+    branches: [ main,deprecated/xredis_2_ror_v1 ]
 
 jobs:
    unit:

redis.conf

@@ -2422,9 +2422,9 @@ rocksdb.ratelimiter.rate_per_sec 1024mb
 # rocksdb.data.max_bytes_for_level_base 512mb
 # rocksdb.meta.max_bytes_for_level_base 256mb
 
-# Default: 10
+# Default: 
 #
-# rocksdb.data.max_bytes_for_level_multiplier 10
+# rocksdb.data.max_bytes_for_level_multiplier 8
 # rocksdb.meta.max_bytes_for_level_multiplier 10
 
 # If true, RocksDB will pick target size of each level dynamically.
@@ -2500,9 +2500,9 @@ rocksdb.ratelimiter.rate_per_sec 1024mb
 #
 # max_bytes_for_level_multiplier_additional is ignored with this flag on.
 #
-# Default: no
+# Default: 
 #
-# rocksdb.data.compaction_dynamic_level_bytes no
+# rocksdb.data.compaction_dynamic_level_bytes yes
 # rocksdb.meta.compaction_dynamic_level_bytes no
 
 # If suggest_compact_deletion_percentage > 0, using CompactOnDeletionCollector,

src/config.c

@@ -2612,6 +2612,17 @@ static int updateRocksdbMetaBlobGarbageCollectionForceThresholdPercentage(long l
     return updateRocksdbCFOptionPersent(META_CF, "blob_garbage_collection_force_threshold", val, err);
 }
 
+static int updateRocksdbDataLevel0FileNumCompactionTrigger(long long val, long long prev, const char **err) {
+    UNUSED(prev);
+    return updateRocksdbCFOptionNumber(DATA_CF, "level0_file_num_compaction_trigger", val, err);
+           updateRocksdbCFOptionNumber(SCORE_CF, "level0_file_num_compaction_trigger", val, err);
+}
+
+static int updateRocksdbMetaLevel0FileNumCompactionTrigger(long long val, long long prev, const char **err) {
+    UNUSED(prev);
+    return updateRocksdbCFOptionNumber(META_CF, "level0_file_num_compaction_trigger", val, err);
+}
+
 const char *rocksdbCompressionTypeName(int val) {
     const char *name = configEnumGetNameOrUnknown(rocksdb_compression_enum, val);
     if (!strcmp(name, "no")) {
@@ -2912,7 +2923,7 @@ standardConfig configs[] = {
     createBoolConfig("rocksdb.enable_pipelined_write", NULL, IMMUTABLE_CONFIG, server.rocksdb_enable_pipelined_write, 0, NULL, NULL),
     createBoolConfig("rocksdb.data.disable_auto_compactions", "rocksdb.disable_auto_compactions", MODIFIABLE_CONFIG, server.rocksdb_data_disable_auto_compactions, 0, NULL, updateRocksdbDataDisableAutoCompactions),
     createBoolConfig("rocksdb.meta.disable_auto_compactions", NULL, MODIFIABLE_CONFIG, server.rocksdb_meta_disable_auto_compactions, 0, NULL, updateRocksdbMetaDisableAutoCompactions),
-    createBoolConfig("rocksdb.data.compaction_dynamic_level_bytes", "rocksdb.compaction_dynamic_level_bytes", IMMUTABLE_CONFIG, server.rocksdb_data_compaction_dynamic_level_bytes, 0, NULL, NULL),
+    createBoolConfig("rocksdb.data.compaction_dynamic_level_bytes", "rocksdb.compaction_dynamic_level_bytes", IMMUTABLE_CONFIG, server.rocksdb_data_compaction_dynamic_level_bytes, 1, NULL, NULL),
     createBoolConfig("rocksdb.meta.compaction_dynamic_level_bytes", NULL, IMMUTABLE_CONFIG, server.rocksdb_meta_compaction_dynamic_level_bytes, 0, NULL, NULL),
     createBoolConfig("rocksdb.data.enable_blob_files", "rocksdb.enable_blob_files", MODIFIABLE_CONFIG, server.rocksdb_data_enable_blob_files, 0, NULL, updateRocksdbDataEnableBlobFiles),
     createBoolConfig("rocksdb.meta.enable_blob_files", NULL, MODIFIABLE_CONFIG, server.rocksdb_meta_enable_blob_files, 0, NULL, updateRocksdbMetaEnableBlobFiles),
@@ -3034,16 +3045,18 @@ standardConfig configs[] = {
     createIntConfig("rocksdb.meta.block_size", NULL, IMMUTABLE_CONFIG, 512, INT_MAX, server.rocksdb_meta_block_size, 8192, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("rocksdb.data.level0_slowdown_writes_trigger", "rocksdb.level0_slowdown_writes_trigger", MODIFIABLE_CONFIG, 1, INT_MAX, server.rocksdb_data_level0_slowdown_writes_trigger, 20, INTEGER_CONFIG, NULL, updateRocksdbDataLevel0SlowdownWritesTrigger),
     createIntConfig("rocksdb.meta.level0_slowdown_writes_trigger", NULL, MODIFIABLE_CONFIG, 1, INT_MAX, server.rocksdb_meta_level0_slowdown_writes_trigger, 20, INTEGER_CONFIG, NULL, updateRocksdbMetaLevel0SlowdownWritesTrigger),
-    createIntConfig("rocksdb.data.max_bytes_for_level_multiplier", "rocksdb.max_bytes_for_level_multiplier", MODIFIABLE_CONFIG, 1, INT_MAX, server.rocksdb_data_max_bytes_for_level_multiplier, 10, INTEGER_CONFIG, NULL, updateRocksdbDataMaxBytesForLevelMultiplier),
+    createIntConfig("rocksdb.data.max_bytes_for_level_multiplier", "rocksdb.max_bytes_for_level_multiplier", MODIFIABLE_CONFIG, 1, INT_MAX, server.rocksdb_data_max_bytes_for_level_multiplier, 8, INTEGER_CONFIG, NULL, updateRocksdbDataMaxBytesForLevelMultiplier),
     createIntConfig("rocksdb.meta.max_bytes_for_level_multiplier", NULL, MODIFIABLE_CONFIG, 1, INT_MAX, server.rocksdb_meta_max_bytes_for_level_multiplier, 10, INTEGER_CONFIG, NULL, updateRocksdbMetaMaxBytesForLevelMultiplier),
     createIntConfig("rocksdb.data.suggest_compact_deletion_percentage", "rocksdb.suggest_compact_deletion_percentage", IMMUTABLE_CONFIG, 0, 100, server.rocksdb_data_suggest_compact_deletion_percentage, 95, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("rocksdb.meta.suggest_compact_deletion_percentage", NULL, IMMUTABLE_CONFIG, 0, 100, server.rocksdb_meta_suggest_compact_deletion_percentage, 95, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("rocksdb.WAL_ttl_seconds", NULL, IMMUTABLE_CONFIG, 0, INT_MAX, server.rocksdb_WAL_ttl_seconds, 18000, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("rocksdb.WAL_size_limit_MB", NULL, IMMUTABLE_CONFIG, 0, INT_MAX, server.rocksdb_WAL_size_limit_MB, 16384, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("rocksdb.data.blob_garbage_collection_age_cutoff_percentage", "rocksdb.blob_garbage_collection_age_cutoff_percentage", MODIFIABLE_CONFIG, 0, INT_MAX, server.rocksdb_data_blob_garbage_collection_age_cutoff_percentage, 5, INTEGER_CONFIG, NULL, updateRocksdbDataBlobGarbageCollectionAgeCutoffPercentage),
     createIntConfig("rocksdb.meta.blob_garbage_collection_age_cutoff_percentage", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, server.rocksdb_meta_blob_garbage_collection_age_cutoff_percentage, 5, INTEGER_CONFIG, NULL, updateRocksdbMetaBlobGarbageCollectionAgeCutoffPercentage),
-    createIntConfig("rocksdb.data.blob_garbage_collection_force_threshold_percentage", "rocksdb.blob_garbage_collection_force_threshold_percentage", MODIFIABLE_CONFIG, 0, INT_MAX, server.rocksdb_data_blob_garbage_collection_force_threshold_percentage, 90, INTEGER_CONFIG, NULL, updateRocksdbDataBlobGarbageCollectionForceThresholdPercentage),
+    createIntConfig("rocksdb.data.blob_garbage_collection_force_threshold_percentage", "rocksdb.blob_garbage_collection_force_threshold_percentage", MODIFIABLE_CONFIG, 0, INT_MAX, server.rocksdb_data_blob_garbage_collection_force_threshold_percentage, 50, INTEGER_CONFIG, NULL, updateRocksdbDataBlobGarbageCollectionForceThresholdPercentage),
     createIntConfig("rocksdb.meta.blob_garbage_collection_force_threshold_percentage", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, server.rocksdb_meta_blob_garbage_collection_force_threshold_percentage, 90, INTEGER_CONFIG, NULL, updateRocksdbMetaBlobGarbageCollectionForceThresholdPercentage),
+    createIntConfig("rocksdb.data.level0_file_num_compaction_trigger", "rocksdb.level0_file_num_compaction_trigger", MODIFIABLE_CONFIG, 0, INT_MAX, server.rocksdb_data_level0_file_num_compaction_trigger, 4, INTEGER_CONFIG, NULL, updateRocksdbDataLevel0FileNumCompactionTrigger),
+    createIntConfig("rocksdb.meta.level0_file_num_compaction_trigger", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, server.rocksdb_meta_level0_file_num_compaction_trigger, 4, INTEGER_CONFIG, NULL, updateRocksdbMetaLevel0FileNumCompactionTrigger),
 #endif
 
     /* Unsigned int configs */

src/ctrip_swap.c

@@ -299,15 +299,6 @@ void keyRequestSwapFinished(swapData *data, void *pd, int errcode) {
     ctx->finished(ctx->c,ctx);
 }
 
-/* Expired key should delete only if server is master, check expireIfNeeded
- * for more details. */
-int keyExpiredAndShouldDelete(redisDb *db, robj *key) {
-    if (!keyIsExpired(db,key)) return 0;
-    if (server.masterhost != NULL) return 0;
-    if (checkClientPauseTimeoutAndReturnIfPaused()) return 0;
-    return 1;
-}
-
 #define NOSWAP_REASON_KEYNOTEXISTS 1
 #define NOSWAP_REASON_NOTKEYLEVEL 2
 #define NOSWAP_REASON_KEYNOTSUPPORT 3

src/ctrip_swap_cmd.c

@@ -1007,6 +1007,9 @@ struct redisCommand redisCommandTable[SWAP_CMD_COUNT] = {
      "admin no-script ok-loading fast may-replicate",
      0,NULL,NULL,SWAP_NOP,0,0,0,0,0,0,0},
 
+    {"import",importCommand,-2,
+     "admin use-memory ok-loading fast @dangerous",
+     0,NULL,NULL,SWAP_NOP,0,0,0,0,0,0,0},
 };
 
 struct SwapDataTypeItem {

src/ctrip_swap_data.c

@@ -61,6 +61,8 @@ int swapDataMarkedPropagateExpire(swapData *data) {
 }
 
 static int swapDataExpiredAndShouldDelete(swapData *data) {
+    if (isImportingExpireDisabled()) return 0;
+
     if (!timestampIsExpired(data->expire)) return 0;
     if (server.masterhost != NULL) return 0;
     if (checkClientPauseTimeoutAndReturnIfPaused()) return 0;

src/ctrip_swap_expire.c

@@ -61,11 +61,13 @@ int scanMetaExpireIfNeeded(redisDb *db, scanMeta *meta) {
     if (server.masterhost != NULL) return 1;
     if (checkClientPauseTimeoutAndReturnIfPaused()) return 1;
 
-    /* Delete the key */
-    c = server.swap_expire_clients[db->id];
-    key = createStringObject(meta->key,sdslen(meta->key));
-    submitExpireClientRequest(c, key, 0);
-    decrRefCount(key);
+    if (!isImportingExpireDisabled()) {
+        /* Delete the key */
+        c = server.swap_expire_clients[db->id];
+        key = createStringObject(meta->key,sdslen(meta->key));
+        submitExpireClientRequest(c, key, 0);
+        decrRefCount(key);
+    }
 
     return 1;
 }

src/ctrip_swap_rocks.c

@@ -96,7 +96,6 @@ static int rocksOpen(rocks *rocks) {
     rocksdb_options_optimize_for_point_lookup(rocks->db_opts, 1);
 
     rocksdb_options_set_min_write_buffer_number_to_merge(rocks->db_opts, 2);
-    rocksdb_options_set_level0_file_num_compaction_trigger(rocks->db_opts, 2);
     rocksdb_options_set_max_bytes_for_level_base(rocks->db_opts, 256*MB);
     rocksdb_options_compaction_readahead_size(rocks->db_opts, 2*1024*1024); /* default 0 */
 
@@ -152,6 +151,7 @@ static int rocksOpen(rocks *rocks) {
     rocksdb_options_set_max_bytes_for_level_base(rocks->cf_opts[DATA_CF],server.rocksdb_data_max_bytes_for_level_base);
     rocksdb_options_set_max_bytes_for_level_multiplier(rocks->cf_opts[DATA_CF], server.rocksdb_data_max_bytes_for_level_multiplier);
     rocksdb_options_set_level_compaction_dynamic_level_bytes(rocks->cf_opts[DATA_CF], server.rocksdb_data_compaction_dynamic_level_bytes);
+    rocksdb_options_set_level0_file_num_compaction_trigger(rocks->cf_opts[DATA_CF], server.rocksdb_data_level0_file_num_compaction_trigger);
     if (server.rocksdb_data_suggest_compact_deletion_percentage) {
         double deletion_ratio = (double)server.rocksdb_data_suggest_compact_deletion_percentage / 100;
         rocksdb_options_add_compact_on_deletion_collector_factory(rocks->cf_opts[DATA_CF],
@@ -189,6 +189,7 @@ static int rocksOpen(rocks *rocks) {
     rocksdb_options_set_max_bytes_for_level_base(rocks->cf_opts[SCORE_CF],server.rocksdb_data_max_bytes_for_level_base);
     rocksdb_options_set_max_bytes_for_level_multiplier(rocks->cf_opts[SCORE_CF], server.rocksdb_data_max_bytes_for_level_multiplier);
     rocksdb_options_set_level_compaction_dynamic_level_bytes(rocks->cf_opts[SCORE_CF], server.rocksdb_data_compaction_dynamic_level_bytes);
+    rocksdb_options_set_level0_file_num_compaction_trigger(rocks->cf_opts[SCORE_CF], server.rocksdb_data_level0_file_num_compaction_trigger);
     if (server.rocksdb_data_suggest_compact_deletion_percentage) {
         double deletion_ratio = (double)server.rocksdb_data_suggest_compact_deletion_percentage / 100;
         rocksdb_options_add_compact_on_deletion_collector_factory(rocks->cf_opts[SCORE_CF],
@@ -226,6 +227,7 @@ static int rocksOpen(rocks *rocks) {
     rocksdb_options_set_max_bytes_for_level_base(rocks->cf_opts[META_CF],server.rocksdb_meta_max_bytes_for_level_base);
     rocksdb_options_set_max_bytes_for_level_multiplier(rocks->cf_opts[META_CF], server.rocksdb_meta_max_bytes_for_level_multiplier);
     rocksdb_options_set_level_compaction_dynamic_level_bytes(rocks->cf_opts[META_CF], server.rocksdb_meta_compaction_dynamic_level_bytes);
+    rocksdb_options_set_level0_file_num_compaction_trigger(rocks->cf_opts[META_CF], server.rocksdb_meta_level0_file_num_compaction_trigger);
     if (server.rocksdb_meta_suggest_compact_deletion_percentage) {
         double deletion_ratio = (double)server.rocksdb_meta_suggest_compact_deletion_percentage / 100;
         rocksdb_options_add_compact_on_deletion_collector_factory(rocks->cf_opts[META_CF],

src/ctrip_swap_server.h

@@ -258,6 +258,8 @@ typedef struct swapBatchLimitsConfig {
     int rocksdb_meta_blob_garbage_collection_age_cutoff_percentage; \
     int rocksdb_data_blob_garbage_collection_force_threshold_percentage;  \
     int rocksdb_meta_blob_garbage_collection_force_threshold_percentage;  \
+    int rocksdb_data_level0_file_num_compaction_trigger;  \
+    int rocksdb_meta_level0_file_num_compaction_trigger;  \
     /* swap block*/ \
     struct swapUnblockCtx* swap_dependency_block_ctx; \
     /* absent cache */ \

src/ctrip_swap_thread.c

@@ -47,8 +47,8 @@ void *swapThreadMain (void *arg) {
         listRewind(thread->pending_reqs, &li);
         processing_reqs = listCreate();
         while ((ln = listNext(&li))) {
-            swapRequest *req = listNodeValue(ln);
-            listAddNodeHead(processing_reqs, req);
+            swapRequestBatch *reqs = listNodeValue(ln);
+            listAddNodeTail(processing_reqs, reqs);
             listDelNode(thread->pending_reqs, ln);
         }
         pthread_mutex_unlock(&thread->lock);

src/db.c

@@ -1769,10 +1769,13 @@ int expireIfNeeded(redisDb *db, robj *key) {
      * have failed over and the new primary will send us the expire. */
     if (checkClientPauseTimeoutAndReturnIfPaused()) return 1;
 
-    /* Delete the key */
+    if (!isImportingExpireDisabled()) {
+        /* Delete the key */
 #ifndef ENABLE_SWAP
-    deleteExpiredKeyAndPropagate(db,key);
+        deleteExpiredKeyAndPropagate(db,key);
 #endif
+    }
+
     return 1;
 }