Optimize skiplist query efficiency by embedding the skiplist header

src/defrag.c

@@ -245,16 +245,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);
     }
 }
 
@@ -275,8 +275,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;
@@ -456,10 +456,8 @@ static void defragZsetSkiplist(robj *ob) {
 
     zset *newzs;
     zskiplist *newzsl;
-    struct zskiplistNode *newheader;
     if ((newzs = activeDefragAlloc(zs))) ob->ptr = 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

@@ -125,7 +125,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 = obj->ptr;
-        return zs->zsl->length;
+        return zslGetLength(zs->zsl);
     } else if (obj->type == OBJ_HASH && obj->encoding == OBJ_ENCODING_HASHTABLE) {
         hashtable *ht = obj->ptr;
         return hashtableSize(ht);

src/object.c

@@ -661,11 +661,11 @@ void dismissZsetObject(robj *o, size_t size_hint) {
     if (o->encoding == OBJ_ENCODING_SKIPLIST) {
         zset *zs = o->ptr;
         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);
@@ -1187,9 +1187,9 @@ size_t objectComputeSize(robj *key, robj *o, size_t sample_size, int dbid) {
         } else if (o->encoding == OBJ_ENCODING_SKIPLIST) {
             hashtable *ht = ((zset *)o->ptr)->ht;
             zskiplist *zsl = ((zset *)o->ptr)->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

@@ -949,7 +949,7 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key, int dbid) {
             zset *zs = o->ptr;
             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
@@ -958,7 +958,7 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key, int dbid) {
              * 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

@@ -1424,25 +1424,34 @@ struct sharedObjectsStruct {
 
 /* ZSETs use a specialized version of Skiplists */
 typedef struct zskiplistNode {
-    double score;
-    struct zskiplistNode *backward;
+    double score;                   /* Sorting score for node ordering */
+    struct zskiplistNode *backward; /* Pointer to previous node for reverse traversal */
+    /* level[] is declared with a size of 1 for compilation, but its actual size is
+     * determined at runtime based on the node's randomly generated height. */
     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];
+    /* For non-header nodes, after the level[], sds header length (1 byte) and an embedded sds element are stored. */
 } zskiplistNode;
 
 typedef struct zskiplist {
-    struct zskiplistNode *header, *tail;
-    unsigned long length;
-    int level;
+    union {
+        struct zskiplistMeta {
+            unsigned long length;       /* Number of elements in the skiplist */
+            struct zskiplistNode *tail; /* Tail element of the skiplist */
+        } meta;
+        struct zskiplistNode header;
+    };
 } zskiplist;
 
+static_assert(sizeof(struct zskiplistMeta) <= offsetof(zskiplistNode, level),
+              "Union overlay requirement: metadata must fit in node header without overlapping level array");
+
 typedef struct zset {
     hashtable *ht;
     zskiplist *zsl;
@@ -3275,6 +3284,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

@@ -425,10 +425,11 @@ void sortCommandGeneric(client *c, int readonly) {
         if (desc) {
             long zsetlen = hashtableSize(((zset *)sortval->ptr)->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);
         }