Optimize skiplist query efficiency by embedding the skiplist header (#2867)

Embedding the skiplist header reduces memory jumps, thus optimizing
sorted set query speed.

```
// Before
typedef struct zskiplist {
        struct zskiplistNode *header, *tail;
        unsigned long length;
        int level; /* Height of the skiplist. */
} zskiplist;


// After
typedef struct zskiplist {
        unsigned long length; 
        struct zskiplistNode *tail;
        /* Since the span at level 0 is always 1 or 0 , 
         * this field is instead used for storing the height of the skiplist. */
        struct zskiplistLevel {
                struct zskiplistNode *forward;
                unsigned long span;
        } level[ZSKIPLIST_MAXLEVEL];
} zskiplist;
```

---------

Signed-off-by: chzhoo <[email protected]>

Author: chzhoo

src/defrag.c

@@ -239,16 +239,16 @@ robj *activeDefragStringOb(robj *ob) {
 /* Internal function used by zslDefrag */
 static void zslUpdateNode(zskiplist *zsl, zskiplistNode *oldnode, zskiplistNode *newnode, zskiplistNode **update) {
     int i;
-    for (i = 0; i < zsl->level; i++) {
+    for (i = 0; i < zslGetHeight(zsl); i++) {
         if (update[i]->level[i].forward == oldnode) update[i]->level[i].forward = newnode;
     }
-    serverAssert(zsl->header != oldnode);
+    serverAssert(zslGetHeader(zsl) != oldnode);
     if (newnode->level[0].forward) {
         serverAssert(newnode->level[0].forward->backward == oldnode);
         newnode->level[0].forward->backward = newnode;
     } else {
-        serverAssert(zsl->tail == oldnode);
-        zsl->tail = newnode;
+        serverAssert(zslGetTail(zsl) == oldnode);
+        zslSetTail(zsl, newnode);
     }
 }
 
@@ -269,8 +269,8 @@ static void activeDefragZsetNode(void *privdata, void *entry_ref) {
     /* find skiplist pointers that need to be updated if we end up moving the
      * skiplist node. */
     zskiplistNode *update[ZSKIPLIST_MAXLEVEL];
-    zskiplistNode *x = zsl->header;
-    for (int i = zsl->level - 1; i >= 0; i--) {
+    zskiplistNode *x = zslGetHeader(zsl);
+    for (int i = zslGetHeight(zsl) - 1; i >= 0; i--) {
         /* stop when we've reached the end of this level or the next node comes
          * after our target in sorted order */
         zskiplistNode *next = x->level[i].forward;
@@ -453,13 +453,11 @@ static void defragZsetSkiplist(robj *ob) {
 
     zset *newzs;
     zskiplist *newzsl;
-    struct zskiplistNode *newheader;
     if ((newzs = activeDefragAlloc(zs))) {
         objectSetVal(ob, newzs);
         zs = newzs;
     }
     if ((newzsl = activeDefragAlloc(zs->zsl))) zs->zsl = newzsl;
-    if ((newheader = activeDefragAlloc(zs->zsl->header))) zs->zsl->header = newheader;
 
     hashtable *newtable;
     if ((newtable = hashtableDefragTables(zs->ht, activeDefragAlloc))) zs->ht = newtable;

src/lazyfree.c

@@ -134,7 +134,7 @@ size_t lazyfreeGetFreeEffort(robj *key, robj *obj, int dbid) {
         return hashtableSize(ht);
     } else if (obj->type == OBJ_ZSET && obj->encoding == OBJ_ENCODING_SKIPLIST) {
         zset *zs = objectGetVal(obj);
-        return zs->zsl->length;
+        return zslGetLength(zs->zsl);
     } else if (obj->type == OBJ_HASH && obj->encoding == OBJ_ENCODING_HASHTABLE) {
         hashtable *ht = objectGetVal(obj);
         return hashtableSize(ht);

src/object.c

@@ -720,11 +720,11 @@ void dismissZsetObject(robj *o, size_t size_hint) {
     if (o->encoding == OBJ_ENCODING_SKIPLIST) {
         zset *zs = objectGetVal(o);
         zskiplist *zsl = zs->zsl;
-        serverAssert(zsl->length != 0);
+        serverAssert(zslGetLength(zsl) != 0);
         /* We iterate all nodes only when average member size is bigger than a
          * page size, and there's a high chance we'll actually dismiss something. */
-        if (size_hint / zsl->length >= server.page_size) {
-            zskiplistNode *zn = zsl->tail;
+        if (size_hint / zslGetLength(zsl) >= server.page_size) {
+            zskiplistNode *zn = zslGetTail(zsl);
             while (zn != NULL) {
                 zskiplistNode *next = zn->backward;
                 dismissMemory(zn, 0);
@@ -1244,9 +1244,9 @@ size_t objectComputeSize(robj *key, robj *o, size_t sample_size, int dbid) {
         } else if (o->encoding == OBJ_ENCODING_SKIPLIST) {
             hashtable *ht = ((zset *)objectGetVal(o))->ht;
             zskiplist *zsl = ((zset *)objectGetVal(o))->zsl;
-            zskiplistNode *znode = zsl->header->level[0].forward;
-            asize += sizeof(zset) + sizeof(zskiplist) +
-                     hashtableMemUsage(ht) + zmalloc_size(zsl->header);
+            zskiplistNode *zheader = zslGetHeader(zsl);
+            zskiplistNode *znode = zheader->level[0].forward;
+            asize += sizeof(zset) + zslGetAllocSize() + hashtableMemUsage(ht);
             while (znode != NULL && samples < sample_size) {
                 elesize += zmalloc_size(znode);
                 samples++;

src/rdb.c

@@ -953,7 +953,7 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key, int dbid, unsigned char rdbt
             zset *zs = objectGetVal(o);
             zskiplist *zsl = zs->zsl;
 
-            if ((n = rdbSaveLen(rdb, zsl->length)) == -1) return -1;
+            if ((n = rdbSaveLen(rdb, zslGetLength(zsl))) == -1) return -1;
             nwritten += n;
 
             /* We save the skiplist elements from the greatest to the smallest
@@ -962,7 +962,7 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key, int dbid, unsigned char rdbt
              * element will always be the smaller, so adding to the skiplist
              * will always immediately stop at the head, making the insertion
              * O(1) instead of O(log(N)). */
-            zskiplistNode *zn = zsl->tail;
+            zskiplistNode *zn = zslGetTail(zsl);
             while (zn != NULL) {
                 sds ele = zslGetNodeElement(zn);
                 if ((n = rdbSaveRawString(rdb, (unsigned char *)ele, sdslen(ele))) == -1) {

src/server.h

@@ -1482,23 +1482,34 @@ struct sharedObjectsStruct {
 
 /* ZSETs use a specialized version of Skiplists */
 typedef struct zskiplistNode {
-    double score;
-    struct zskiplistNode *backward;
+    union {
+        double score;         /* Sorting score for node ordering. */
+        unsigned long length; /* Number of elements in the skiplist. */
+    };
+    union {
+        struct zskiplistNode *backward; /* Pointer to previous node for reverse traversal. */
+        struct zskiplistNode *tail;     /* Tail element of the skiplist. */
+    };
     struct zskiplistLevel {
         struct zskiplistNode *forward;
         /* At each level we keep the span, which is the number of elements which are on the "subtree"
          * from this node at this level to the next node at the same level.
          * One exception is the value at level 0. In level 0 the span can only be 1 or 0 (in case the last elements in the list)
          * So we use it in order to hold the height of the node, which is the number of levels. */
         unsigned long span;
-    } level[];
-    /* After the level[], sds header length (1 byte) and an embedded sds element are stored. */
+    } level[1]; /* Flexible array member - actual levels determined at node creation. */
+    /* For non-header nodes, after the level[], sds header length (1 byte) and an embedded sds element are stored. */
 } zskiplistNode;
 
+/* The header node does not store actual data (no score, no backward pointer,
+ * and its node height is fixed at ZSKIPLIST_MAXLEVEL).
+ * To save memory, we reuse the memory space of these fields in the header node to store:
+ *   - skiplist length (number of elements)
+ *   - tail pointer to the last element
+ *   - maximum current level of the skiplist
+ * For detailed memory layout, refer to the zskiplistNode struct definition. */
 typedef struct zskiplist {
-    struct zskiplistNode *header, *tail;
-    unsigned long length;
-    int level;
+    zskiplistNode header;
 } zskiplist;
 
 typedef struct zset {
@@ -3345,6 +3356,12 @@ typedef struct {
 #define ERROR_COMMAND_FAILED (1 << 1)   /* Indicate to update the command failed stats */
 
 zskiplist *zslCreate(void);
+int zslGetHeight(const zskiplist *zsl);
+zskiplistNode *zslGetTail(const zskiplist *zsl);
+void zslSetTail(zskiplist *zsl, zskiplistNode *tail);
+unsigned long zslGetLength(const zskiplist *zsl);
+zskiplistNode *zslGetHeader(zskiplist *zsl);
+size_t zslGetAllocSize(void);
 void zslFree(zskiplist *zsl);
 zskiplistNode *zslInsert(zskiplist *zsl, double score, const_sds ele);
 zskiplistNode *zslNthInRange(zskiplist *zsl, zrangespec *range, long n, long *rank);

src/sort.c

@@ -426,10 +426,11 @@ void sortCommandGeneric(client *c, int readonly) {
         if (desc) {
             long zsetlen = hashtableSize(((zset *)objectGetVal(sortval))->ht);
 
-            ln = zsl->tail;
+            ln = zslGetTail(zsl);
             if (start > 0) ln = zslGetElementByRank(zsl, zsetlen - start);
         } else {
-            ln = zsl->header->level[0].forward;
+            zskiplistNode *zheader = zslGetHeader(zsl);
+            ln = zheader->level[0].forward;
             if (start > 0) ln = zslGetElementByRank(zsl, start + 1);
         }