Repository: valkey-io/valkey
File: src/hashtable.c (C)

Code before:
}

/* Attempts to defrag bucket 'b' using the defrag callback function. If the
 * defrag callback function returns a pointer to a new allocation, this pointer
 * is returned and the 'prev' bucket is updated to point to the new allocation.
 * Otherwise, the 'b' pointer is returned. */
static bucket *bucketDefrag(bucket *prev, bucket *b, void *(*defragfn)(void *)) {
    bucket *reallocated = defragfn(b);
    if (reallocated == NULL) return b;
    prev->entries[ENTRIES_PER_BUCKET - 1] = reallocated;
    return reallocated;
}

/* Rehashes a single entry from the old table to the new table. */
static void rehashEntry(hashtable *ht, void *entry, uint64_t hash, uint8_t h2) {
    int pos_in_dst_bucket;
    bucket *dst = findBucketForInsert(ht, hash, &pos_in_dst_bucket, NULL);
    dst->entries[pos_in_dst_bucket] = entry;
    dst->hashes[pos_in_dst_bucket] = h2;
    dst->presence |= (1 << pos_in_dst_bucket);
    ht->used[0]--;
    ht->used[1]++;
}

/* Fetches entries from a bucket chain for batch processing. */
static bucket *fetchEntryForExpand(bucket *b, void *buf[], int *size, int max_fetch_count) {
    *size = 0;
    for (int idx = 0; idx < max_fetch_count && b != NULL; idx += 1) {
        for (int pos = 0; pos < numBucketPositions(b); pos++) {
            if (!isPositionFilled(b, pos)) continue; /* empty */
            buf[*size] = b->entries[pos];
            (*size)++;
        }
        b = getChildBucket(b);
    }
    return b;
}

/* Processes one bucket chain during incremental table expansion.
 * Uses batch processing to optimize memory access patterns. */
static void rehashStepExpand(hashtable *ht) {
    void *entry_buf[FETCH_ENTRY_BUFFER_SIZE_WHEN_EXPAND];
    const void *key_buf[FETCH_ENTRY_BUFFER_SIZE_WHEN_EXPAND];
    size_t idx = ht->rehash_idx;
    bucket *b = ht->tables[0] + idx;
    int size = 0;
    while (b != NULL) {
        b = fetchEntryForExpand(b, entry_buf, &size, FETCH_BUCKET_COUNT_WHEN_EXPAND);

        /* Key optimization: no loop-carried dependency enables concurrent memory access,
         * reducing CPU stall cycles. */

Review comment (suggestion):
* fetchEntry indicates that it fetches only one entry, but it is actually multiple entries. Let's use plural to make the name more clear.

* `max_fetch_count` is the maximum number of buckets to traverse, not entries. Let's make this more clear too.

```suggestion
static bucket *fetchEntriesForExpand(bucket *b, void *buf[], int *size, int max_bucket_count) {
```

Code after:
}

/* Attempts to defrag bucket 'b' using the defrag callback function. If the
 * defrag callback function returns a pointer to a new allocation, this pointer
 * is returned and the 'prev' bucket is updated to point to the new allocation.
 * Otherwise, the 'b' pointer is returned. */
static bucket *bucketDefrag(bucket *prev, bucket *b, void *(*defragfn)(void *)) {
    bucket *reallocated = defragfn(b);
    if (reallocated == NULL) return b;
    prev->entries[ENTRIES_PER_BUCKET - 1] = reallocated;
    return reallocated;
}

/* Rehashes a single entry from the old table to the new table. */
static void rehashEntry(hashtable *ht, void *entry, uint64_t hash, uint8_t h2) {
    int pos_in_dst_bucket;
    bucket *dst = findBucketForInsert(ht, hash, &pos_in_dst_bucket, NULL);
    dst->entries[pos_in_dst_bucket] = entry;
    dst->hashes[pos_in_dst_bucket] = h2;
    dst->presence |= (1 << pos_in_dst_bucket);
    ht->used[0]--;
    ht->used[1]++;
}

/* After migrating entries in a bucket chain from the old table
 * to the new one, this function should be called immediately to
 * handle the cleanup of old buckets, such as clearing presence bits. */
static void rehashStepFinalize(hashtable *ht) {
    size_t idx = ht->rehash_idx;
    /* Free child bucket. */
    bucket *b = getChildBucket(ht->tables[0] + idx);
    while (b != NULL) {
        bucket *next = getChildBucket(b);
        zfree(b);
        if (ht->type->trackMemUsage) ht->type->trackMemUsage(ht, -sizeof(bucket));
        ht->child_buckets[0]--;
        b = next;
    }

    /* Reset the bucket after clearing its child buckets. */
    b = ht->tables[0] + idx;
    b->chained = 0;
    b->presence = 0;

    /* Advance to the next bucket. */
    ht->rehash_idx++;
    if ((size_t)ht->rehash_idx >= numBuckets(ht->bucket_exp[0])) {
        rehashingCompleted(ht);
    }
}