[fix] self-describing subkey_size to keep consistent during persisted…

… data loading

src/bitops.c

@@ -495,7 +495,6 @@ robj *lookupStringForBitCommand(client *c, uint64_t maxbit) {
             serverAssert(om->swap_type == SWAP_TYPE_BITMAP);
             metaBitmapInit(&meta_bitmap, objectMetaGetPtr(om), o);
         } else {
-            //TODO confirm fixed how?
             /* maybe it is a empty string object. */
             /* it is never processed as bitmap in ror. */
             metaBitmapInit(&meta_bitmap, NULL, o);

src/ctrip_swap.h

@@ -543,7 +543,7 @@ int objectMetaDecode(struct objectMeta *object_meta, const char *extend, size_t
 int keyIsHot(objectMeta *object_meta, robj *value);
 sds dumpObjectMeta(objectMeta *object_meta);
 int objectMetaEqual(struct objectMeta *oma, struct objectMeta *omb);
-int objectMetaRebuildFeed(struct objectMeta *object_meta, uint64_t version, const char *subkey, size_t sublen, robj *subval);
+int objectMetaRebuildFeed(struct objectMeta *rebuild_meta, uint64_t version, const char *subkey, size_t sublen, robj *subval);
 
 static inline void *objectMetaGetPtr(objectMeta *object_meta) {
   return (void*)(long)object_meta->ptr;
@@ -988,6 +988,8 @@ void bitmapMetaFree(struct bitmapMeta *bitmap_meta);
 
 size_t bitmapMetaGetSize(struct bitmapMeta *bitmap_meta);
 
+size_t bitmapMetaGetSubkeySize(struct bitmapMeta *bitmap_meta);
+
 objectMeta *createBitmapObjectMeta(uint64_t version, MOVE struct bitmapMeta *bitmap_meta);
 
 int swapDataSetupBitmap(swapData *d, void **pdatactx);

src/ctrip_swap_list.c

@@ -1278,7 +1278,7 @@ void listObjectMetaDup(struct objectMeta *dup_meta, struct objectMeta *object_me
 
 static inline long listDecodeRidx(const char *str, size_t len);
 
-int listObjectMetaRebuildFeed(struct objectMeta *rebuild_meta,
+int listObjectMetaRebuildFeed(struct objectMeta *rebuild_meta, 
         uint64_t version, const char *subkey, size_t sublen, robj *subval) {
     UNUSED(subval);
     long ridx;

src/ctrip_swap_object.c

@@ -216,7 +216,7 @@ int lenObjectMetaIsHot(objectMeta *object_meta, robj *value) {
     return object_meta->len == 0;
 }
 
-static inline int lenObjectMetaRebuildFeed(struct objectMeta *rebuild_meta,
+static inline int lenObjectMetaRebuildFeed(struct objectMeta *rebuild_meta, 
         uint64_t version, const char *subkey, size_t sublen, robj *subval) {
     UNUSED(sublen), UNUSED(version), UNUSED(subval);
 

src/ctrip_swap_persist.c

@@ -437,7 +437,7 @@ sds genSwapPersistInfoString(sds info) {
 #define INIT_FIX_SKIP -2
 
 struct listMeta *listMetaCreate();
-struct bitmapMeta *bitmapMetaCreate();
+struct bitmapMeta *bitmapMetaCreate(size_t subkey_size);
 
 int keyLoadFixDataInit(keyLoadFixData *fix, redisDb *db, decodedResult *dr) {
     uint64_t version;
@@ -480,7 +480,8 @@ int keyLoadFixDataInit(keyLoadFixData *fix, redisDb *db, decodedResult *dr) {
         rebuild_meta = createListObjectMeta(dm->version, listMetaCreate());
         break;
     case SWAP_TYPE_BITMAP:
-        rebuild_meta = createBitmapObjectMeta(dm->version, bitmapMetaCreate());
+        rebuild_meta = createBitmapObjectMeta(dm->version, 
+            bitmapMetaCreate(bitmapMetaGetSubkeySize(objectMetaGetPtr(cold_meta))));
         break;
     default:
         rebuild_meta = NULL;

tests/swap/integration/persist.tcl

@@ -354,6 +354,59 @@ start_server {tags {persist} overrides {swap-persist-enabled yes swap-dirty-subk
         set dbsize_after [r dbsize]
         assert {[expr $dbsize_before - $dbsize_after] < 1000}
     }
+
+    test {persist and restart check bitmap subkey size} {
+
+        set bak_bitmap_subkey_size [lindex [r config get swap-bitmap-subkey-size] 1]
+
+        r CONFIG SET swap-bitmap-subkey-size 4096
+
+        r setbit mybitmap7 32767 1
+        r setbit mybitmap7 65535 1
+        r setbit mybitmap7 98303 1
+        r setbit mybitmap7 131071 1
+        r setbit mybitmap7 163839 1
+        r setbit mybitmap7 196607 1
+        r setbit mybitmap7 229375 1
+        r setbit mybitmap7 262143 1
+        r setbit mybitmap7 294911 1
+        r setbit mybitmap7 327679 1
+        r setbit mybitmap7 335871 1
+
+        # only one subkey in mybitmap8 
+        r setbit mybitmap8 300 1
+
+        # mybitmap9 is empty string
+        r set mybitmap9 ""
+        assert_equal {0} [r bitcount mybitmap9]
+
+        wait_key_clean r mybitmap7
+        assert [object_is_hot r mybitmap7]
+
+        wait_key_clean r mybitmap8
+        assert [object_is_hot r mybitmap8]
+
+        wait_key_clean r mybitmap9
+        assert [object_is_hot r mybitmap9]
+
+        assert_equal [object_meta_subkey_size r mybitmap7] 4096
+        assert_equal [object_meta_subkey_size r mybitmap8] 4096
+        assert_equal [object_meta_subkey_size r mybitmap9] 4096
+
+        r CONFIG SET swap-bitmap-subkey-size 2048
+        restart_server 0 true false
+
+        assert_equal [object_meta_subkey_size r mybitmap7] 4096
+        assert_equal [object_meta_subkey_size r mybitmap8] 4096
+        assert_equal [object_meta_subkey_size r mybitmap9] 4096
+
+        assert_equal [r bitcount mybitmap7] {11}
+        assert_equal [r bitcount mybitmap8] {1}
+        assert_equal [r bitcount mybitmap9] {0}
+
+        r config set swap-bitmap-subkey-size $bak_bitmap_subkey_size
+    }
+
 }
 
 start_server {tags {persist} overrides {swap-persist-enabled yes swap-dirty-subkeys-enabled yes}} {

tests/swap/support/util.tcl

@@ -265,6 +265,19 @@ proc object_meta_pure_cold_subkeys_num {r key} {
     }
 }
 
+proc object_meta_subkey_size {r key} {
+    set str [$r swap object $key]
+    set subkey_size [swap_object_property $str hot_meta subkey_size]
+    if {$subkey_size == ""} {
+        set subkey_size [swap_object_property $str cold_meta subkey_size]
+    }
+    if {$subkey_size != ""} {
+        set _ $subkey_size
+    } else {
+        set _ 0
+    }
+}
+
 proc object_meta_version {r key} {
     if { [catch {$r swap object $key} e] } {
         set _ 0

tests/swap/unit/bitmap.tcl

@@ -1462,6 +1462,41 @@ start_server {
             }
         }
     }
+
+    test {modify swap-bitmap-subkey-size during swap} {
+        r flushdb
+        set bak_bitmap_subkey_size [lindex [r config get swap-bitmap-subkey-size] 1]
+
+        r CONFIG SET swap-bitmap-subkey-size 4096
+
+        build_cold_data mybitmap1
+        build_pure_hot_data mybitmap2
+
+        r CONFIG SET swap-bitmap-subkey-size 2048
+
+        r swap.evict mybitmap2
+        wait_key_cold r mybitmap2
+
+        # once bitmap become persisted, bitmap meta existed, subkey size is determined by configuration
+
+        assert_equal [object_meta_subkey_size r mybitmap1] 4096
+        assert_equal [object_meta_subkey_size r mybitmap2] 2048
+
+        # mybitmap1 turn pure hot, then turn cold
+        r getrange mybitmap1 0 1
+        r getbit mybitmap1 0
+        r swap.evict mybitmap1
+        wait_key_cold r mybitmap1
+
+        assert_equal [object_meta_subkey_size r mybitmap1] 2048
+
+        check_mybitmap_is_right mybitmap1 $notextend
+        check_mybitmap_is_right mybitmap2 $notextend
+
+        r flushdb
+        r config set swap-bitmap-subkey-size $bak_bitmap_subkey_size
+    }
+
 }
 
 start_server {
@@ -1527,7 +1562,6 @@ start_server {
         }
     }
 
-
     test {cold rdbsave and rdbload with swap-bitmap-subkey-size exchange 2048 to 4096} {
         r flushdb
         set bak_bitmap_subkey_size [lindex [r config get swap-bitmap-subkey-size] 1]
@@ -1562,6 +1596,36 @@ start_server {
         r config set swap-bitmap-subkey-size $bak_bitmap_subkey_size
     }
 
+    test {modify swap-bitmap-subkey-size during rdbsave and load} {
+        r flushdb
+        set bak_bitmap_subkey_size [lindex [r config get swap-bitmap-subkey-size] 1]
+
+        r CONFIG SET swap-bitmap-subkey-size 4096
+
+        build_cold_data mybitmap1
+        build_pure_hot_data mybitmap2
+
+        r CONFIG SET swap-bitmap-subkey-size 2048
+
+        # once bitmap become persisted, bitmap meta existed, subkey size is determined by configuration
+
+        assert_equal [object_meta_subkey_size r mybitmap1] 4096
+        assert_equal [object_meta_subkey_size r mybitmap2] 0
+
+        r SAVE
+        r CONFIG SET swap-bitmap-subkey-size 1024
+        r DEBUG RELOAD NOSAVE
+
+        assert_equal [object_meta_subkey_size r mybitmap1] 1024
+        assert_equal [object_meta_subkey_size r mybitmap1] 1024
+
+        check_mybitmap_is_right mybitmap1 $notextend
+        check_mybitmap_is_right mybitmap2 $notextend
+
+        r flushdb
+        r config set swap-bitmap-subkey-size $bak_bitmap_subkey_size
+    }
+
     set bak_rdb_bitmap_enable [lindex [r config get swap-rdb-bitmap-encode-enabled] 1]
     r CONFIG SET swap-rdb-bitmap-encode-enabled no
 
@@ -1958,17 +2022,28 @@ start_server {tags {"bitmap chaos test"} overrides {save ""}} {
             for {set round 0} {$round < $rounds} {incr round} {
                 puts "chaos load $bitmaps bitmaps with $loaders loaders in $duration seconds ($round/$rounds)"
                     # load with chaos bitmap operations
+
+                set min_subkey_size 2048
+                set max_subkey_size 4096
+
                 for {set loader 0} {$loader < $loaders} {incr loader} {
+                    set subkey_size_slave [expr { int(rand() * ($max_subkey_size - $min_subkey_size + 1)) + $min_subkey_size }]
+                    $slave config set swap-bitmap-subkey-size $subkey_size_slave
+
                     lappend load_handles [start_run_load $master_host $master_port $duration 0 {
                         set bitmaps 4
                         # in bit
                         set bitmap_max_length 335872
                         set block_timeout 0.1
-                                    set count 0
+                        set count 0
                         set mybitmap "mybitmap-[randomInt $bitmaps]"
                         # set mybitmap_len [$r1 llen $mybitmap]
                         set mybitmap_len [expr {[$r1 strlen $mybitmap] * 8}]
-                                    set otherbitmap "mybitmap-[randomInt $bitmaps]"
+                        set otherbitmap "mybitmap-[randomInt $bitmaps]"
+
+                        set min_subkey_size 2048
+                        set max_subkey_size 4096
+
                         set src_direction [randpath {return LEFT} {return RIGHT}]
                         set dst_direction [randpath {return LEFT} {return RIGHT}]
                         randpath {
@@ -1985,6 +2060,9 @@ start_server {tags {"bitmap chaos test"} overrides {save ""}} {
                         } {
                             set randIdx [randomInt $bitmap_max_length]
                             $r1 BITFIELD $mybitmap get u4 $randIdx
+                        } {
+                            set subkey_size_master [expr { int(rand() * ($max_subkey_size - $min_subkey_size + 1)) + $min_subkey_size }]
+                            $r1 config set swap-bitmap-subkey-size $subkey_size_master
                         } {
                             set randIdx [randomInt $bitmap_max_length]
                             $r1 BITFIELD_RO $mybitmap get u4 $randIdx