From 5a829a40b804f8f59e40dd32779567caf7d444b9 Mon Sep 17 00:00:00 2001 From: Alina Liu Date: Thu, 4 Dec 2025 01:16:14 +0000 Subject: [PATCH 1/6] Add FIFO and mutexQueue with bio.c refactored Signed-off-by: Alina Liu --- cmake/Modules/SourceFiles.cmake | 4 +- src/Makefile | 2 +- src/bio.c | 94 ++-------- src/fifo.c | 305 ++++++++++++++++++++++++++++++++ src/fifo.h | 48 +++++ src/mutexqueue.c | 166 +++++++++++++++++ src/mutexqueue.h | 54 ++++++ src/unit/test_fifo.c | 252 ++++++++++++++++++++++++++ src/unit/test_files.h | 20 +++ src/unit/test_mutexqueue.c | 293 ++++++++++++++++++++++++++++++ 10 files changed, 1161 insertions(+), 77 deletions(-) create mode 100644 src/fifo.c create mode 100644 src/fifo.h create mode 100644 src/mutexqueue.c create mode 100644 src/mutexqueue.h create mode 100644 src/unit/test_fifo.c create mode 100644 src/unit/test_mutexqueue.c diff --git a/cmake/Modules/SourceFiles.cmake b/cmake/Modules/SourceFiles.cmake index 74ed04d58cf..08540490b7b 100644 --- a/cmake/Modules/SourceFiles.cmake +++ b/cmake/Modules/SourceFiles.cmake @@ -118,7 +118,9 @@ set(VALKEY_SERVER_SRCS ${CMAKE_SOURCE_DIR}/src/server.c ${CMAKE_SOURCE_DIR}/src/logreqres.c ${CMAKE_SOURCE_DIR}/src/entry.c - ${CMAKE_SOURCE_DIR}/src/vset.c) + ${CMAKE_SOURCE_DIR}/src/vset.c + ${CMAKE_SOURCE_DIR}/src/fifo.c + ${CMAKE_SOURCE_DIR}/src/mutexqueue.c) # valkey-cli diff --git a/src/Makefile b/src/Makefile index 4254b12d2ae..15ce2520ea8 100644 --- a/src/Makefile +++ b/src/Makefile @@ -443,7 +443,7 @@ ENGINE_NAME=valkey SERVER_NAME=$(ENGINE_NAME)-server$(PROG_SUFFIX) ENGINE_SENTINEL_NAME=$(ENGINE_NAME)-sentinel$(PROG_SUFFIX) ENGINE_TRACE_OBJ=trace/trace.o trace/trace_commands.o trace/trace_db.o trace/trace_cluster.o trace/trace_server.o trace/trace_rdb.o trace/trace_aof.o -ENGINE_SERVER_OBJ=threads_mngr.o adlist.o vector.o quicklist.o ae.o anet.o dict.o hashtable.o kvstore.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o memory_prefetch.o io_threads.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o cluster_legacy.o cluster_slot_stats.o crc16.o cluster_migrateslots.o endianconv.o commandlog.o eval.o bio.o rio.o rand.o memtest.o syscheck.o crcspeed.o crccombine.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o valkey-check-rdb.o valkey-check-aof.o geo.o lazyfree.o module.o lrulfu.o evict.o expire.o geohash.o geohash_helper.o childinfo.o allocator_defrag.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o lolwut6.o lolwut9.o acl.o tracking.o socket.o tls.o sha256.o timeout.o setcpuaffinity.o monotonic.o mt19937-64.o resp_parser.o call_reply.o script.o functions.o commands.o strl.o connection.o unix.o logreqres.o rdma.o scripting_engine.o entry.o vset.o +ENGINE_SERVER_OBJ=threads_mngr.o adlist.o vector.o quicklist.o ae.o anet.o dict.o hashtable.o kvstore.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o memory_prefetch.o io_threads.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o cluster_legacy.o cluster_slot_stats.o crc16.o cluster_migrateslots.o endianconv.o commandlog.o eval.o bio.o rio.o rand.o memtest.o syscheck.o crcspeed.o crccombine.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o valkey-check-rdb.o valkey-check-aof.o geo.o lazyfree.o module.o lrulfu.o evict.o expire.o geohash.o geohash_helper.o childinfo.o allocator_defrag.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o lolwut6.o lolwut9.o acl.o tracking.o socket.o tls.o sha256.o timeout.o setcpuaffinity.o monotonic.o mt19937-64.o resp_parser.o call_reply.o script.o functions.o commands.o strl.o connection.o unix.o logreqres.o rdma.o scripting_engine.o entry.o vset.o fifo.o mutexqueue.o ENGINE_SERVER_OBJ+=$(ENGINE_TRACE_OBJ) ENGINE_CLI_NAME=$(ENGINE_NAME)-cli$(PROG_SUFFIX) ENGINE_CLI_OBJ=anet.o adlist.o dict.o valkey-cli.o zmalloc.o release.o ae.o serverassert.o crcspeed.o crccombine.o crc64.o siphash.o crc16.o monotonic.o cli_common.o mt19937-64.o strl.o cli_commands.o sds.o util.o sha256.o diff --git a/src/bio.c b/src/bio.c index b044f9124cc..a8c2995d561 100644 --- a/src/bio.c +++ b/src/bio.c @@ -1,3 +1,9 @@ +/* + * Copyright (c) Valkey Contributors + * All rights reserved. + * SPDX-License-Identifier: BSD-3-Clause + */ + /* Background I/O service for the server. * * This file implements operations that we need to perform in the background. @@ -31,38 +37,12 @@ * * ---------------------------------------------------------------------------- * - * Copyright (c) 2009-2012, Redis Ltd. - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * * Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * * Redistributions in binary form must reproduce the above copyright - * notice, this list of conditions and the following disclaimer in the - * documentation and/or other materials provided with the distribution. - * * Neither the name of Redis nor the names of its contributors may be used - * to endorse or promote products derived from this software without - * specific prior written permission. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. */ - #include "server.h" #include "connection.h" #include "bio.h" +#include "mutexqueue.h" #include static unsigned int bio_job_to_worker[] = { @@ -76,9 +56,7 @@ static unsigned int bio_job_to_worker[] = { typedef struct { const char *const bio_worker_title; pthread_t bio_thread_id; - pthread_mutex_t bio_mutex; - pthread_cond_t bio_newjob_cond; - list *bio_jobs; + mutexQueue *bio_jobs; } bio_worker_data; static bio_worker_data bio_workers[] = { @@ -96,7 +74,7 @@ static size_t bioWorkerNum(const bio_worker_data *const bwd) { return (size_t)(bwd - bio_workers); } -static unsigned long bio_jobs_counter[BIO_NUM_OPS] = {0}; +static _Atomic unsigned long bio_jobs_counter[BIO_NUM_OPS] = {0}; static _Thread_local size_t bio_worker_num = 0; /* This structure represents a background Job. It is only used locally to this @@ -143,9 +121,7 @@ void bioInit(void) { /* Initialization of state vars and objects */ for (bio_worker_data *bwd = bio_workers; bwd != bio_worker_end; ++bwd) { - pthread_mutex_init(&bwd->bio_mutex, NULL); - pthread_cond_init(&bwd->bio_newjob_cond, NULL); - bwd->bio_jobs = listCreate(); + bwd->bio_jobs = mutexQueueCreate(); } /* Set the stack size as by default it may be small in some system */ @@ -170,11 +146,8 @@ void bioInit(void) { void bioSubmitJob(int type, bio_job *job) { job->header.type = type; bio_worker_data *const bwd = &bio_workers[bio_job_to_worker[type]]; - pthread_mutex_lock(&bwd->bio_mutex); - listAddNodeTail(bwd->bio_jobs, job); - bio_jobs_counter[type]++; - pthread_cond_signal(&bwd->bio_newjob_cond); - pthread_mutex_unlock(&bwd->bio_mutex); + mutexQueueAdd(bwd->bio_jobs, job); + atomic_fetch_add(&bio_jobs_counter[type], 1); } void bioCreateLazyFreeJob(lazy_free_fn free_fn, int arg_count, ...) { @@ -229,7 +202,6 @@ void bioCreateSaveRDBToDiskJob(connection *conn, int is_dual_channel) { void *bioProcessBackgroundJobs(void *arg) { bio_worker_data *const bwd = arg; - bio_job *job; sigset_t sigset; valkey_set_thread_title(bwd->bio_worker_title); @@ -238,7 +210,6 @@ void *bioProcessBackgroundJobs(void *arg) { makeThreadKillable(); - pthread_mutex_lock(&bwd->bio_mutex); /* Block SIGALRM so we are sure that only the main thread will * receive the watchdog signal. */ sigemptyset(&sigset); @@ -250,19 +221,8 @@ void *bioProcessBackgroundJobs(void *arg) { bio_worker_num = bioWorkerNum(bwd); while (1) { - listNode *ln; - - /* The loop always starts with the lock hold. */ - if (listLength(bwd->bio_jobs) == 0) { - pthread_cond_wait(&bwd->bio_newjob_cond, &bwd->bio_mutex); - continue; - } - /* Get the job from the queue. */ - ln = listFirst(bwd->bio_jobs); - job = ln->value; - /* It is now possible to unlock the background system as we know have - * a stand alone job structure to process.*/ - pthread_mutex_unlock(&bwd->bio_mutex); + /* Get job - blocking until available */ + bio_job *job = mutexQueuePop(bwd->bio_jobs, true); /* Process the job accordingly to its type. */ int job_type = job->header.type; @@ -308,36 +268,20 @@ void *bioProcessBackgroundJobs(void *arg) { serverPanic("Wrong job type in bioProcessBackgroundJobs()."); } zfree(job); - - /* Lock again before reiterating the loop, if there are no longer - * jobs to process we'll block again in pthread_cond_wait(). */ - pthread_mutex_lock(&bwd->bio_mutex); - listDelNode(bwd->bio_jobs, ln); - bio_jobs_counter[job_type]--; - pthread_cond_signal(&bwd->bio_newjob_cond); + atomic_fetch_sub(&bio_jobs_counter[job_type], 1); } } /* Return the number of pending jobs of the specified type. */ unsigned long bioPendingJobsOfType(int type) { - bio_worker_data *const bwd = &bio_workers[bio_job_to_worker[type]]; - - pthread_mutex_lock(&bwd->bio_mutex); - unsigned long val = bio_jobs_counter[type]; - pthread_mutex_unlock(&bwd->bio_mutex); - - return val; + return atomic_load(&bio_jobs_counter[type]); } /* Wait for the job queue of the worker for jobs of specified type to become empty. */ -void bioDrainWorker(int job_type) { - bio_worker_data *const bwd = &bio_workers[bio_job_to_worker[job_type]]; - - pthread_mutex_lock(&bwd->bio_mutex); - while (listLength(bwd->bio_jobs) > 0) { - pthread_cond_wait(&bwd->bio_newjob_cond, &bwd->bio_mutex); +void bioDrainWorker(int type) { + while (bioPendingJobsOfType(type) > 0) { + usleep(1000); /* Sleep for 1ms and check again*/ } - pthread_mutex_unlock(&bwd->bio_mutex); } /* Kill the running bio threads in an unclean way. This function should be diff --git a/src/fifo.c b/src/fifo.c new file mode 100644 index 00000000000..a91442f577f --- /dev/null +++ b/src/fifo.c @@ -0,0 +1,305 @@ +/* + * Copyright (c) Valkey Contributors + * All rights reserved. + * SPDX-License-Identifier: BSD-3-Clause + */ + +/* FIFO - A high-performance FIFO queue implementation */ + +#include + +#include "fifo.h" +#include "serverassert.h" +#include "zmalloc.h" + +/* Items per block was chosen as 7 because, including the next pointer, this gives us a nice even + * 64-byte block. Conveniently, the index values 0..6 will fit nicely in the 3 unused bits at the + * bottom of the next pointer, creating a very compact block. */ +#define ITEMS_PER_BLOCK 7 +static const uintptr_t IDX_MASK = 0x0007; + + +/* The FifoBlock contains up to 7 items (pointers). When compared with adlist, this results in + * roughly 60% memory reduction and 7x fewer memory allocations. Memory reduction is guaranteed + * with 5+ items in queue. + * + * In each block, there are 7 slots for item pointers (pointers to the caller's FIFO item). + * We need to keep track of the first & last slot used. Contextually, we will only need + * a single index - either the first slot used or the last slot used. Based on context, + * we can determine what is needed. + * + * Blocks are linked together in a chain. If the list is empty, there are no blocks. + * For non-empty lists, we will either have a single block OR a chain of blocks. + * + * For a SINGLE BLOCK containing (for example) 4 items, the layout looks like this: + * +--------+--------+--------+--------+--------+--------+--------+--------+ + * SINGLE BLOCK: | slot 0 | slot 1 | slot 2 | slot 3 | slot 4 | slot 5 | slot 6 | next/ | + * | item | item | item | item | - | - | - | lastIdx| + * +--------+--------+--------+--------+--------+--------+--------+--------+ + * ^ + * lastIdx (3) + * In single blocks, the items are always shifted so that the first item is in slot 0. + * We need to keep track of the lastIdx so that we will know where to push the next item. + * The last index is stored in the final 3 bits of the (unused) next pointer + * + * When MULTIPLE BLOCKS are chained together, items will be popped from the first block, and + * pushed onto the last block. All blocks in the middle are full. In the first block, we keep + * the firstIdx (so we know where to pop) ... on the last block, we keep lastIdx (so we know + * where to push). + * + * Example FIRST BLOCK with 2 items remaining: + * +--------+--------+--------+--------+--------+--------+--------+--------+ + * FIRST BLOCK: | slot 0 | slot 1 | slot 2 | slot 3 | slot 4 | slot 5 | slot 6 | next/ | + * | - | - | - | - | - | item | item |firstIdx| + * +--------+--------+--------+--------+--------+--------+--------+--------+ + * ^ + * firstIdx (5) + * Example LAST BLOCK with 3 items pushed so far: + * +--------+--------+--------+--------+--------+--------+--------+--------+ + * LAST BLOCK: | slot 0 | slot 1 | slot 2 | slot 3 | slot 4 | slot 5 | slot 6 | next/ | + * | item | item | item | - | - | - | - | lastIdx| + * +--------+--------+--------+--------+--------+--------+--------+--------+ + * ^ + * lastIdx (2) + */ +typedef struct fifoBlock fifoBlock; + +struct fifoBlock { + void *items[ITEMS_PER_BLOCK]; + union { + /* The last 3 bits of a pointer to a block allocated by malloc must always be zero as a + * minimum of 8-byte alignment is required for all such blocks. These bits are used as + * an index into the block indicating the first or last item in the block, depending on + * context. + * + * This UNION overlays a pointer with an integral value. This allows us to look at the + * pointer OR the integer without casting - but they use the same memory. + * + * If there is MORE THAN ONE block in the chain, the first block has a pointer/index that + * looks like this. However, if there is only a single block, it looks like the LAST block. + * +-----------------------------------------------------------+ + * | next pointer | firstIdx | + * | (61 bits) | (3 bits) | + * +-----------------------------------------------------------+ + * * The next pointer is only valid after zeroing out the last 3 bits. + * * "lastIdx" is implied to be 6 (because there are additional blocks). + * * "firstIdx" represents the first filled index (0..6). POP occurs here. + * + * Any blocks in the middle of the chain have a regular pointer like this: + * +-----------------------------------------------------------+ + * | next pointer | 0* | + * | (61 bits) | (3 bits) | + * +-----------------------------------------------------------+ + * * The next pointer is valid as-is + * * "lastIdx" is implied to be 6 in all middle blocks. + * * "firstIdx" is implied to be 0 in all middle blocks. + * * NOTE: In middle blocks, the index bits(0) are really still the firstIdx value. + * When Fifo's are joined, the O(1) operation may result in a partially + * full middle block. In this case, the items are "right-justified" and + * firstIdx indicates where the items start. + * + * The last (or only) block in the chain contains only the lastIndex, the pointer is unused. + * +-----------------------------------------------------------+ + * | 0 | lastIdx | + * | (61 bits) | (3 bits) | + * +-----------------------------------------------------------+ + * * The next pointer is unused and guaranteed NULL. + * * "lastIdx" represents the last filled index (0..6). + * * "firstIdx" is implied to be zero on the last (or only) block. + */ + uintptr_t last_or_first_idx; + fifoBlock *next; + } u; +}; + +struct fifo { + long length; /* Total number of items in queue */ + fifoBlock *first; + fifoBlock *last; +}; + + +/* Create a new FIFO queue. */ +fifo *fifoCreate(void) { + fifo *q = zmalloc(sizeof(fifo)); + q->length = 0; + q->first = q->last = NULL; + return q; +} + + +/* Push an item onto the end of the queue. */ +void fifoPush(fifo *q, void *ptr) { + if (q->first == NULL) { + /* Queue was empty - create block */ + assert(q->last == NULL && q->length == 0); + q->last = q->first = zmalloc(sizeof(fifoBlock)); + q->last->u.last_or_first_idx = 0; /* Item 0 is the last item in this block */ + q->last->items[0] = ptr; + } else { + int lastIdx = q->last->u.last_or_first_idx; /* pointer portion is 0 on last (or only) block */ + assert(lastIdx < ITEMS_PER_BLOCK); + + if (lastIdx < ITEMS_PER_BLOCK - 1) { + /* If the last block has space, just add the item */ + q->last->items[lastIdx + 1] = ptr; + q->last->u.last_or_first_idx++; + } else { + /* Otherwise, last block is full - add a new block */ + fifoBlock *newblock = zmalloc(sizeof(fifoBlock)); + newblock->u.last_or_first_idx = 0; + newblock->items[0] = ptr; + q->last->u.next = newblock; /* overwrites the index, setting it to 0 */ + q->last = newblock; + } + } + + q->length++; +} + + +/* Look at the first item in the queue (without removing it). + * NOTE: asserts if the queue is empty. */ +void *fifoPeek(fifo *q) { + assert(q->length > 0); + int firstIdx = (q->first == q->last) ? 0 : q->first->u.last_or_first_idx & IDX_MASK; + return q->first->items[firstIdx]; +} + + +/* Return and remove the first item from the queue. + * NOTE: asserts if the queue is empty. */ +void *fifoPop(fifo *q) { + assert(q->length > 0); + void *item; + + if (q->first == q->last) { + /* With only 1 block, POP occurs at index 0 and items 1..6 are shifted. */ + item = q->last->items[0]; + + int lastIdx = q->last->u.last_or_first_idx; /* pointer portion is 0 on last (or only) block */ + assert(lastIdx < ITEMS_PER_BLOCK); + + if (lastIdx > 0) { + /* With only 1 block, shift the items rather than eventually needing new block. + * (This is cheap, shifting a max of 6 pointers.) */ + for (int i = 0; i < lastIdx; i++) q->last->items[i] = q->last->items[i + 1]; + q->last->u.last_or_first_idx--; /* Decrement the last index */ + } else { + /* Just finished the only block. Delete it. */ + zfree(q->last); + q->first = q->last = NULL; + } + } else { + /* With more than 1 block, POP occurs at firstIdx, and firstIdx is incremented. */ + int firstIdx = q->first->u.last_or_first_idx & IDX_MASK; + item = q->first->items[firstIdx]; + + if (firstIdx < ITEMS_PER_BLOCK - 1) { + /* Just increment the first index to the next slot. */ + q->first->u.last_or_first_idx++; + } else { + /* Finished with this block, move to next */ + q->first->u.last_or_first_idx &= ~IDX_MASK; /* restores the next pointer */ + fifoBlock *next = q->first->u.next; + zfree(q->first); + q->first = next; + } + } + + q->length--; + + return item; +} + + +/* Return the number of items in the queue. */ +long fifoLength(fifo *q) { + return q->length; +} + + +/* Delete the queue. + * NOTE: this does not free items which may be referenced by inserted pointers. */ +void fifoDelete(fifo *q) { + if (q->length > 0) { + fifoBlock *cur = q->first; + while (cur != NULL) { + cur->u.last_or_first_idx &= ~IDX_MASK; /* zero out the last 3 bits */ + fifoBlock *next = cur->u.next; + zfree(cur); + cur = next; + } + } + zfree(q); +} + + +/* Blindly overwrites target from source. */ +static void blindlyMoveFifoContents(fifo *target, fifo *source) { + target->length = source->length; + target->first = source->first; + target->last = source->last; + source->length = 0; + source->first = source->last = NULL; +} + + +/* Join an "other" fifo onto this one (emptying "other") */ +void fifoJoin(fifo *q, fifo *other) { + /* When joining a fifo onto an existing fifo, we might be left with partially full blocks in the + * middle of the list. In the usual case, any blocks in the middle of the list have the index + * bits set to zero. This actually represents the firstIdx - which would normally be zero for + * blocks in the middle of the list. In the case of joining lists, we allow partially full + * blocks in the middle, but the values are "right-justified" and the firstIdx is set. + * + * To perform the join, we take the current last (or only) block - which is "left-justified" and + * shift the items so that the block becomes right-justified. Then the index is corrected, + * replacing the lastIdx with the firstIdx. + * + * The "other" list is correct as-is. If there is only a single block, it becomes the last + * block and remains left-justified. If there are multiple blocks, the first block of the + * "other" list is already right-justified and becomes a partially full middle block. + */ + if (other->length == 0) return; + + if (q->length == 0) { + /* If "q" is empty, it's a simple operation. */ + blindlyMoveFifoContents(q, other); + return; + } + + if (other->length < ITEMS_PER_BLOCK) { + /* In the case of a short "other" fifo, move each item. This prevents creation of a string + * of half-empty blocks if fifoJoin is repeatedly used on small fifos. */ + while (other->length > 0) fifoPush(q, fifoPop(other)); + return; + } + + fifoBlock *curLast = q->last; + int lastIdx = curLast->u.last_or_first_idx; + /* Shift the items in the last block if it is partially full */ + int shift = (ITEMS_PER_BLOCK - 1) - lastIdx; + if (shift > 0) { + for (int i = lastIdx; i >= 0; i--) + curLast->items[i + shift] = curLast->items[i]; + } + + /* Now fix up the next pointer to point to the next block */ + curLast->u.next = other->first; + curLast->u.last_or_first_idx += shift; /* Mask on the firstIdx for the shifted block */ + + /* Finally, clean up the main list structures */ + q->length += other->length; + q->last = other->last; + other->length = 0; + other->first = other->last = NULL; +} + + +/* Copy all of the items into a new fifo (emptying the original) */ +fifo *fifoPopAll(fifo *q) { + fifo *newQ = zmalloc(sizeof(fifo)); + blindlyMoveFifoContents(newQ, q); + return newQ; +} diff --git a/src/fifo.h b/src/fifo.h new file mode 100644 index 00000000000..9e4dfd6d07a --- /dev/null +++ b/src/fifo.h @@ -0,0 +1,48 @@ +/* + * Copyright (c) Valkey Contributors + * All rights reserved. + * SPDX-License-Identifier: BSD-3-Clause + */ + +/* A space/time efficient FIFO queue of pointers. + * + * Implemented with an unrolled single-linked list, the implementation packs multiple pointers into + * a single block. This increases space efficiency and cache locality over the Valkey `list` for the + * purpose of a simple FIFO queue. + */ + +#ifndef __FIFO_H_ +#define __FIFO_H_ + +typedef struct fifo fifo; + +/* Create a new FIFO queue. */ +fifo *fifoCreate(void); + +/* Push an item onto the end of the queue. */ +void fifoPush(fifo *q, void *ptr); + +/* Look at the first item in the queue (without removing it). + * NOTE: asserts if the queue is empty. */ +void *fifoPeek(fifo *q); + +/* Return and remove the first item from the queue. + * NOTE: asserts if the queue is empty. */ +void *fifoPop(fifo *q); + +/* Return the number of items in the queue. */ +long fifoLength(fifo *q); + +/* Delete the queue. + * NOTE: this does not free items which may be referenced by inserted pointers. */ +void fifoDelete(fifo *q); + +/* Joins the fifo "other" to the end of "q". "other" becomes empty, but remains valid. + * This is an O(1) operation. */ +void fifoJoin(fifo *q, fifo *other); + +/* Returns a new fifo, containing all of the items from "q". "q" remains valid, but becomes empty. + * This is an O(1) operation. */ +fifo *fifoPopAll(fifo *q); + +#endif diff --git a/src/mutexqueue.c b/src/mutexqueue.c new file mode 100644 index 00000000000..7a34a589bfb --- /dev/null +++ b/src/mutexqueue.c @@ -0,0 +1,166 @@ +/* + * Copyright (c) Valkey Contributors + * All rights reserved. + * SPDX-License-Identifier: BSD-3-Clause + */ + +/* mutexQueue - A thread-safe wrapper around FIFO */ + +#include "mutexqueue.h" +#include "serverassert.h" +#include "zmalloc.h" +#include + + +struct mutexQueue { + fifo *priority_fifo; /* Ordered list of priority insertions */ + fifo *normal_fifo; /* Ordered list of normal insertions */ + pthread_mutex_t mutex; /* Mutex to lock shared access */ + pthread_cond_t notify_cv; /* Condition variable to notify waiting threads */ +}; + + +/* Create an empty queue. */ +mutexQueue *mutexQueueCreate(void) { + mutexQueue *mq; + mq = zmalloc(sizeof(*mq)); + + pthread_mutex_init(&mq->mutex, NULL); + pthread_cond_init(&mq->notify_cv, NULL); + mq->priority_fifo = fifoCreate(); + mq->normal_fifo = fifoCreate(); + return mq; +} + + +/* Release an empty queue. + * Note: The queue must be empty before calling release. The quickest way to empty the queue is to + * call mutexQueuePopAll - which returns the items in a new fifo. It is the caller's + * responsibility to free memory (as necessary) for any items. + * Note: Behavior is undefined if other threads are accessing the queue. */ +void mutexQueueRelease(mutexQueue *theQueue) { + assert(mutexQueueLength(theQueue) == 0); + mutexQueue *mq = theQueue; + + pthread_mutex_destroy(&mq->mutex); + pthread_cond_broadcast(&mq->notify_cv); + pthread_cond_destroy(&mq->notify_cv); + + fifoDelete(mq->priority_fifo); + fifoDelete(mq->normal_fifo); + + zfree(mq); +} + + +/* Internal routine assumes mutex already locked. */ +static inline unsigned long mutexQueueLengthInternal(mutexQueue *mq) { + return fifoLength(mq->priority_fifo) + fifoLength(mq->normal_fifo); +} + + +/* Number of items in the queue. */ +unsigned long mutexQueueLength(mutexQueue *theQueue) { + mutexQueue *mq = theQueue; + + pthread_mutex_lock(&mq->mutex); + + unsigned long len = mutexQueueLengthInternal(mq); + + pthread_mutex_unlock(&mq->mutex); + return len; +} + + +/* Insert a priority item at the beginning of the queue (but after existing priority items). */ +void mutexQueueAddPriority(mutexQueue *theQueue, void *value) { + mutexQueue *mq = theQueue; + + pthread_mutex_lock(&mq->mutex); + + bool mustSignal = (mutexQueueLengthInternal(theQueue) == 0); + fifoPush(mq->priority_fifo, value); + if (mustSignal) pthread_cond_broadcast(&mq->notify_cv); + + pthread_mutex_unlock(&mq->mutex); +} + + +/* Insert an item at the end of the queue. */ +void mutexQueueAdd(mutexQueue *theQueue, void *value) { + mutexQueue *mq = theQueue; + + pthread_mutex_lock(&mq->mutex); + + bool mustSignal = (mutexQueueLengthInternal(theQueue) == 0); + fifoPush(mq->normal_fifo, value); + if (mustSignal) pthread_cond_broadcast(&mq->notify_cv); + + pthread_mutex_unlock(&mq->mutex); +} + + +/* Insert a fifo of items at the end of the queue. This removes the items from the source fifo! */ +void mutexQueueAddMultiple(mutexQueue *theQueue, fifo *valueFifo) { + mutexQueue *mq = theQueue; + + if (fifoLength(valueFifo) == 0) return; + + pthread_mutex_lock(&mq->mutex); + + bool mustSignal = (mutexQueueLengthInternal(theQueue) == 0); + fifoJoin(mq->normal_fifo, valueFifo); + if (mustSignal) pthread_cond_broadcast(&mq->notify_cv); + + pthread_mutex_unlock(&mq->mutex); +} + + +/* Retrieves the first item off the queue (or NULL if queue is empty). + * If 'blocking' is true, this method will block until an item is available. */ +void *mutexQueuePop(mutexQueue *theQueue, bool blocking) { + mutexQueue *mq = theQueue; + void *value = NULL; + + pthread_mutex_lock(&mq->mutex); + + if (blocking) { + while (mutexQueueLengthInternal(mq) == 0) { + pthread_cond_wait(&mq->notify_cv, &mq->mutex); + } + } + + if (fifoLength(mq->priority_fifo) > 0) { + value = fifoPop(mq->priority_fifo); + } else if (fifoLength(mq->normal_fifo) > 0) { + value = fifoPop(mq->normal_fifo); + } + + pthread_mutex_unlock(&mq->mutex); + return value; +} + + +/* Retrieves all items from the queue as a fifo (or NULL if the queue is empty). + * If 'blocking' is true, this method will block until an item is available. */ +fifo *mutexQueuePopAll(mutexQueue *theQueue, bool blocking) { + mutexQueue *mq = theQueue; + fifo *result = NULL; + + pthread_mutex_lock(&mq->mutex); + + if (blocking) { + while (mutexQueueLengthInternal(mq) == 0) { + pthread_cond_wait(&mq->notify_cv, &mq->mutex); + } + } + + if (mutexQueueLengthInternal(mq) > 0) { + result = fifoCreate(); + fifoJoin(result, mq->priority_fifo); + fifoJoin(result, mq->normal_fifo); + } + + pthread_mutex_unlock(&mq->mutex); + return result; +} diff --git a/src/mutexqueue.h b/src/mutexqueue.h new file mode 100644 index 00000000000..2a57e2c448d --- /dev/null +++ b/src/mutexqueue.h @@ -0,0 +1,54 @@ +/* + * Copyright (c) Valkey Contributors + * All rights reserved. + * SPDX-License-Identifier: BSD-3-Clause + */ + +/* + * A thread-safe queue, protected by a mutex. + * + * Supports: + * - Adding an item to the end of the queue + * - Adding a list of items (fifo) to the end of the queue + * - Insertion of a priority item at the beginning of the queue (but after existing priority items) + * - Removing an item from the beginning of the queue + * - Removing ALL items as (as new fifo) from the queue + * - Synchronous waiting on the queue for new items + * + * The caller is responsible for memory management for items in the queue. + */ + +#ifndef __MUTEXQUEUE_H +#define __MUTEXQUEUE_H + +#include +#include "fifo.h" + +/* The mutexQueue is an opaque structure. */ +typedef struct mutexQueue mutexQueue; + +/* Create an empty queue. */ +mutexQueue *mutexQueueCreate(void); + +/* Release an empty queue. */ +void mutexQueueRelease(mutexQueue *theQueue); + +/* Number of items in the queue. */ +unsigned long mutexQueueLength(mutexQueue *theQueue); + +/* Insert a priority item at the beginning of the queue (but after existing priority items). */ +void mutexQueueAddPriority(mutexQueue *theQueue, void *value); + +/* Insert an item at the end of the queue. */ +void mutexQueueAdd(mutexQueue *theQueue, void *value); + +/* Insert multiple items (from a fifo) to the end of the queue. */ +void mutexQueueAddMultiple(mutexQueue *theQueue, fifo *valueFifo); + +/* Retrieves the first item off the queue (or NULL if queue is empty). */ +void *mutexQueuePop(mutexQueue *theQueue, bool blocking); + +/* Retrieves all items from the queue as a fifo (or NULL if the queue is empty). */ +fifo *mutexQueuePopAll(mutexQueue *theQueue, bool blocking); + +#endif diff --git a/src/unit/test_fifo.c b/src/unit/test_fifo.c new file mode 100644 index 00000000000..747aacebe14 --- /dev/null +++ b/src/unit/test_fifo.c @@ -0,0 +1,252 @@ +/* + * Copyright (c) Valkey Contributors + * All rights reserved. + * SPDX-License-Identifier: BSD-3-Clause + */ + +#include "../fifo.h" +#include "test_help.h" +#include +#include + +/* Macro to detect if child process crashes as expected */ +#define DETECT_CRASH \ + pid_t pid = fork(); \ + if (pid < 0) { \ + /* Fork failed */ \ + TEST_EXPECT(0); \ + } else if (pid > 0) { \ + /* Parent process - wait for child to crash */ \ + int status; \ + waitpid(pid, &status, 0); \ + /* Verify child exited abnormally (assertion failure) */ \ + TEST_EXPECT(WIFSIGNALED(status) || (WIFEXITED(status) && WEXITSTATUS(status) != 0)); \ + } else /* macro is followed by the else clause */ + +static inline void *INT_TO_PTR(intptr_t i) { + return (void *)i; +} + +static inline intptr_t PTR_TO_INT(void *p) { + return (intptr_t)p; +} + +/* Helper functions */ +static void push(fifo *q, intptr_t value) { + int len = fifoLength(q); + fifoPush(q, INT_TO_PTR(value)); + TEST_EXPECT(fifoLength(q) == len + 1); +} + +static intptr_t popTest(fifo *q, intptr_t expected) { + intptr_t peekValue = PTR_TO_INT(fifoPeek(q)); + TEST_EXPECT(peekValue == expected); + int len = fifoLength(q); + intptr_t value = PTR_TO_INT(fifoPop(q)); + TEST_EXPECT(fifoLength(q) == len - 1); + TEST_EXPECT(value == expected); + return value; +} + +/* Test: emptyPop - verify that popping from empty queue triggers assertion */ +int test_fifoEmptyPop(int argc, char *argv[], int flags) { + UNUSED(argc); + UNUSED(argv); + UNUSED(flags); + + DETECT_CRASH { + /* Child process - this should crash with assertion */ + fifo *q = fifoCreate(); + TEST_EXPECT(fifoLength(q) == 0); + fifoPop(q); /* This should assert and crash */ + /* Should never reach here */ + TEST_ASSERT(0); + } + return 0; +} + +/* Test: emptyPeek - verify that peeking at empty queue triggers assertion */ +int test_fifoEmptyPeek(int argc, char *argv[], int flags) { + UNUSED(argc); + UNUSED(argv); + UNUSED(flags); + + DETECT_CRASH { + /* Child process - this should crash with assertion */ + fifo *q = fifoCreate(); + TEST_EXPECT(fifoLength(q) == 0); + fifoPeek(q); /* This should assert and crash */ + /* Should never reach here */ + TEST_ASSERT(0); + } + return 0; +} + +/* Test: simplePushPop */ +int test_fifoSimplePushPop(int argc, char *argv[], int flags) { + UNUSED(argc); + UNUSED(argv); + UNUSED(flags); + + fifo *q = fifoCreate(); + TEST_EXPECT(fifoLength(q) == 0); + push(q, 1); + popTest(q, 1); + TEST_EXPECT(fifoLength(q) == 0); + fifoDelete(q); + return 0; +} + +/* Test: tryVariousSizes */ +int test_fifoTryVariousSizes(int argc, char *argv[], int flags) { + UNUSED(argc); + UNUSED(argv); + UNUSED(flags); + + fifo *q = fifoCreate(); + for (int items = 1; items < 50; items++) { + TEST_EXPECT(fifoLength(q) == 0); + for (int value = 1; value <= items; value++) push(q, value); + for (int value = 1; value <= items; value++) popTest(q, value); + TEST_EXPECT(fifoLength(q) == 0); + } + fifoDelete(q); + return 0; +} + +/* Test: pushPopTest */ +int test_fifoPushPopTest(int argc, char *argv[], int flags) { + UNUSED(argc); + UNUSED(argv); + UNUSED(flags); + + fifo *q = fifoCreate(); + /* In this test, we repeatedly push 2 and pop 1. This hits the list differently than + * other tests which push a bunch and then pop them all off. */ + int pushVal = 1; + int popVal = 1; + for (int i = 0; i < 200; i++) { + if (i % 3 == 0 || i % 3 == 1) { + push(q, pushVal++); + } else { + popTest(q, popVal++); + } + } + fifoDelete(q); + return 0; +} + +/* Test: joinTest */ +int test_fifoJoinTest(int argc, char *argv[], int flags) { + UNUSED(argc); + UNUSED(argv); + UNUSED(flags); + + fifo *q = fifoCreate(); + fifo *q2 = fifoCreate(); + + /* In this test, there are 2 queues Q and Q2. + * Various sizes are tested. For each size, various amounts are popped off the front first. + * Q2 is appended to Q. */ + for (int qLen = 0; qLen <= 21; qLen++) { + for (int qPop = 0; qPop < 6 && qPop <= qLen; qPop++) { + for (int q2Len = 0; q2Len <= 21; q2Len++) { + for (int q2Pop = 0; q2Pop < 6 && q2Pop <= q2Len; q2Pop++) { + intptr_t pushValue = 1; + intptr_t popQValue = 1; + + for (int i = 0; i < qLen; i++) fifoPush(q, INT_TO_PTR(pushValue++)); + for (int i = 0; i < qPop; i++) TEST_EXPECT(PTR_TO_INT(fifoPop(q)) == popQValue++); + + intptr_t popQ2Value = pushValue; + for (int i = 0; i < q2Len; i++) fifoPush(q2, INT_TO_PTR(pushValue++)); + for (int i = 0; i < q2Pop; i++) TEST_EXPECT(PTR_TO_INT(fifoPop(q2)) == popQ2Value++); + + fifoJoin(q, q2); + TEST_EXPECT(fifoLength(q) == (qLen - qPop) + (q2Len - q2Pop)); + TEST_EXPECT(fifoLength(q2) == 0); + + fifo *temp = fifoPopAll(q); /* Exercise fifoPopAll also */ + TEST_EXPECT(fifoLength(temp) == (qLen - qPop) + (q2Len - q2Pop)); + TEST_EXPECT(fifoLength(q) == 0); + + for (int i = 0; i < (qLen - qPop); i++) TEST_EXPECT(PTR_TO_INT(fifoPop(temp)) == popQValue++); + for (int i = 0; i < (q2Len - q2Pop); i++) TEST_EXPECT(PTR_TO_INT(fifoPop(temp)) == popQ2Value++); + TEST_EXPECT(fifoLength(temp) == 0); + + fifoDelete(temp); + } + } + } + } + + fifoDelete(q); + fifoDelete(q2); + return 0; +} + +/* Set to 1 to prove that FIFO outperforms ADLIST. This test will exercise both. + * The test will (intentionally) fail, printing the results as failed assertions. */ +#define COMPARE_PERFORMANCE_TO_ADLIST 0 + +#include "../adlist.h" +#include "../monotonic.h" + +const int LIST_ITEMS = 10000; + +static void exerciseList(void) { + list *q = listCreate(); + for (intptr_t i = 0; i < LIST_ITEMS; i++) { + listAddNodeTail(q, INT_TO_PTR(i)); + } + TEST_EXPECT(listLength(q) == (unsigned)LIST_ITEMS); + for (intptr_t i = 0; i < LIST_ITEMS; i++) { + listNode *node = listFirst(q); + listDelNode(q, node); + TEST_EXPECT(listNodeValue(node) == INT_TO_PTR(i)); + } + TEST_EXPECT(listLength(q) == 0u); + listRelease(q); +} + +static void exerciseFifo(void) { + fifo *q = fifoCreate(); + for (intptr_t i = 0; i < LIST_ITEMS; i++) { + fifoPush(q, INT_TO_PTR(i)); + } + TEST_EXPECT(fifoLength(q) == LIST_ITEMS); + for (intptr_t i = 0; i < LIST_ITEMS; i++) { + TEST_EXPECT(fifoPop(q) == INT_TO_PTR(i)); + } + TEST_EXPECT(fifoLength(q) == 0); + fifoDelete(q); +} + +int test_fifoComparePerformance(int argc, char *argv[], int flags) { + UNUSED(argc); + UNUSED(argv); + UNUSED(flags); + + if (COMPARE_PERFORMANCE_TO_ADLIST) { + monotonicInit(); + monotime timer; + const int iterations = 500; + + exerciseList(); /* Warm up the list before timing */ + elapsedStart(&timer); + for (int i = 0; i < iterations; i++) exerciseList(); + long listMs = elapsedMs(timer); + + exerciseFifo(); /* Warm up the fifo before timing */ + elapsedStart(&timer); + for (int i = 0; i < iterations; i++) exerciseFifo(); + long fifoMs = elapsedMs(timer); + + TEST_EXPECT(listMs == fifoMs); /* This will fail, printing result */ + double percentImprovement = (double)(listMs - fifoMs) * 100.0 / listMs; + TEST_PRINT_INFO("List: %ld ms, FIFO: %ld ms, Improvement: %.2f%%", listMs, fifoMs, percentImprovement); + TEST_EXPECT(percentImprovement == 0.0); /* This will fail, printing result */ + } + + return 0; +} diff --git a/src/unit/test_files.h b/src/unit/test_files.h index 09dbfe59042..1de0df84d25 100644 --- a/src/unit/test_files.h +++ b/src/unit/test_files.h @@ -25,6 +25,13 @@ int test_entryUpdate(int argc, char **argv, int flags); int test_entryHasexpiry_entrySetExpiry(int argc, char **argv, int flags); int test_entryIsExpired(int argc, char **argv, int flags); int test_entryMemUsage_entrySetExpiry_entrySetValue(int argc, char **argv, int flags); +int test_fifoEmptyPop(int argc, char *argv[], int flags); +int test_fifoEmptyPeek(int argc, char *argv[], int flags); +int test_fifoSimplePushPop(int argc, char *argv[], int flags); +int test_fifoTryVariousSizes(int argc, char *argv[], int flags); +int test_fifoPushPopTest(int argc, char *argv[], int flags); +int test_fifoJoinTest(int argc, char *argv[], int flags); +int test_fifoComparePerformance(int argc, char *argv[], int flags); int test_cursor(int argc, char **argv, int flags); int test_set_hash_function_seed(int argc, char **argv, int flags); int test_add_find_delete(int argc, char **argv, int flags); @@ -114,6 +121,15 @@ int test_listpackBenchmarkLpValidateIntegrity(int argc, char **argv, int flags); int test_listpackBenchmarkLpCompareWithString(int argc, char **argv, int flags); int test_listpackBenchmarkLpCompareWithNumber(int argc, char **argv, int flags); int test_listpackBenchmarkFree(int argc, char **argv, int flags); +int test_mutexQueueSimplePushPop(int argc, char *argv[], int flags); +int test_mutexQueueDoublePushPop(int argc, char *argv[], int flags); +int test_mutexQueuePriorityOrdering(int argc, char *argv[], int flags); +int test_mutexQueueFifoPopAll(int argc, char *argv[], int flags); +int test_mutexQueueFifoAddMultiple(int argc, char *argv[], int flags); +int test_mutexQueueSimpleThread(int argc, char *argv[], int flags); +int test_mutexQueueParallelWriters(int argc, char *argv[], int flags); +int test_mutexQueueParallelReaders(int argc, char *argv[], int flags); +int test_mutexQueueParallelReadWrite(int argc, char *argv[], int flags); int test_writeToReplica(int argc, char **argv, int flags); int test_postWriteToReplica(int argc, char **argv, int flags); int test_backupAndUpdateClientArgv(int argc, char **argv, int flags); @@ -268,10 +284,12 @@ unitTest __test_crc64combine_c[] = {{"test_crc64combine", test_crc64combine}, {N unitTest __test_dict_c[] = {{"test_dictCreate", test_dictCreate}, {"test_dictAdd16Keys", test_dictAdd16Keys}, {"test_dictDisableResize", test_dictDisableResize}, {"test_dictAddOneKeyTriggerResize", test_dictAddOneKeyTriggerResize}, {"test_dictDeleteKeys", test_dictDeleteKeys}, {"test_dictDeleteOneKeyTriggerResize", test_dictDeleteOneKeyTriggerResize}, {"test_dictEmptyDirAdd128Keys", test_dictEmptyDirAdd128Keys}, {"test_dictDisableResizeReduceTo3", test_dictDisableResizeReduceTo3}, {"test_dictDeleteOneKeyTriggerResizeAgain", test_dictDeleteOneKeyTriggerResizeAgain}, {"test_dictBenchmark", test_dictBenchmark}, {NULL, NULL}}; unitTest __test_endianconv_c[] = {{"test_endianconv", test_endianconv}, {NULL, NULL}}; unitTest __test_entry_c[] = {{"test_entryCreate", test_entryCreate}, {"test_entryUpdate", test_entryUpdate}, {"test_entryHasexpiry_entrySetExpiry", test_entryHasexpiry_entrySetExpiry}, {"test_entryIsExpired", test_entryIsExpired}, {"test_entryMemUsage_entrySetExpiry_entrySetValue", test_entryMemUsage_entrySetExpiry_entrySetValue}, {NULL, NULL}}; +unitTest __test_fifo_c[] = {{"test_fifoEmptyPop", test_fifoEmptyPop}, {"test_fifoEmptyPeek", test_fifoEmptyPeek}, {"test_fifoSimplePushPop", test_fifoSimplePushPop}, {"test_fifoTryVariousSizes", test_fifoTryVariousSizes}, {"test_fifoPushPopTest", test_fifoPushPopTest}, {"test_fifoJoinTest", test_fifoJoinTest}, {"test_fifoComparePerformance", test_fifoComparePerformance}, {NULL, NULL}}; unitTest __test_hashtable_c[] = {{"test_cursor", test_cursor}, {"test_set_hash_function_seed", test_set_hash_function_seed}, {"test_add_find_delete", test_add_find_delete}, {"test_add_find_delete_avoid_resize", test_add_find_delete_avoid_resize}, {"test_instant_rehashing", test_instant_rehashing}, {"test_bucket_chain_length", test_bucket_chain_length}, {"test_two_phase_insert_and_pop", test_two_phase_insert_and_pop}, {"test_replace_reallocated_entry", test_replace_reallocated_entry}, {"test_incremental_find", test_incremental_find}, {"test_scan", test_scan}, {"test_iterator", test_iterator}, {"test_safe_iterator", test_safe_iterator}, {"test_compact_bucket_chain", test_compact_bucket_chain}, {"test_random_entry", test_random_entry}, {"test_random_entry_with_long_chain", test_random_entry_with_long_chain}, {"test_random_entry_sparse_table", test_random_entry_sparse_table}, {"test_safe_iterator_invalidation", test_safe_iterator_invalidation}, {"test_safe_iterator_empty_no_invalidation", test_safe_iterator_empty_no_invalidation}, {"test_safe_iterator_reset_invalidation", test_safe_iterator_reset_invalidation}, {"test_safe_iterator_reset_untracking", test_safe_iterator_reset_untracking}, {"test_safe_iterator_pause_resume_tracking", test_safe_iterator_pause_resume_tracking}, {"test_null_hashtable_iterator", test_null_hashtable_iterator}, {"test_hashtable_retarget_iterator", test_hashtable_retarget_iterator}, {NULL, NULL}}; unitTest __test_intset_c[] = {{"test_intsetValueEncodings", test_intsetValueEncodings}, {"test_intsetBasicAdding", test_intsetBasicAdding}, {"test_intsetLargeNumberRandomAdd", test_intsetLargeNumberRandomAdd}, {"test_intsetUpgradeFromint16Toint32", test_intsetUpgradeFromint16Toint32}, {"test_intsetUpgradeFromint16Toint64", test_intsetUpgradeFromint16Toint64}, {"test_intsetUpgradeFromint32Toint64", test_intsetUpgradeFromint32Toint64}, {"test_intsetStressLookups", test_intsetStressLookups}, {"test_intsetStressAddDelete", test_intsetStressAddDelete}, {NULL, NULL}}; unitTest __test_kvstore_c[] = {{"test_kvstoreAdd16Keys", test_kvstoreAdd16Keys}, {"test_kvstoreIteratorRemoveAllKeysNoDeleteEmptyHashtable", test_kvstoreIteratorRemoveAllKeysNoDeleteEmptyHashtable}, {"test_kvstoreIteratorRemoveAllKeysDeleteEmptyHashtable", test_kvstoreIteratorRemoveAllKeysDeleteEmptyHashtable}, {"test_kvstoreHashtableIteratorRemoveAllKeysNoDeleteEmptyHashtable", test_kvstoreHashtableIteratorRemoveAllKeysNoDeleteEmptyHashtable}, {"test_kvstoreHashtableIteratorRemoveAllKeysDeleteEmptyHashtable", test_kvstoreHashtableIteratorRemoveAllKeysDeleteEmptyHashtable}, {"test_kvstoreHashtableExpand", test_kvstoreHashtableExpand}, {NULL, NULL}}; unitTest __test_listpack_c[] = {{"test_listpackCreateIntList", test_listpackCreateIntList}, {"test_listpackCreateList", test_listpackCreateList}, {"test_listpackLpPrepend", test_listpackLpPrepend}, {"test_listpackLpPrependInteger", test_listpackLpPrependInteger}, {"test_listpackGetELementAtIndex", test_listpackGetELementAtIndex}, {"test_listpackPop", test_listpackPop}, {"test_listpackGetELementAtIndex2", test_listpackGetELementAtIndex2}, {"test_listpackIterate0toEnd", test_listpackIterate0toEnd}, {"test_listpackIterate1toEnd", test_listpackIterate1toEnd}, {"test_listpackIterate2toEnd", test_listpackIterate2toEnd}, {"test_listpackIterateBackToFront", test_listpackIterateBackToFront}, {"test_listpackIterateBackToFrontWithDelete", test_listpackIterateBackToFrontWithDelete}, {"test_listpackDeleteWhenNumIsMinusOne", test_listpackDeleteWhenNumIsMinusOne}, {"test_listpackDeleteWithNegativeIndex", test_listpackDeleteWithNegativeIndex}, {"test_listpackDeleteInclusiveRange0_0", test_listpackDeleteInclusiveRange0_0}, {"test_listpackDeleteInclusiveRange0_1", test_listpackDeleteInclusiveRange0_1}, {"test_listpackDeleteInclusiveRange1_2", test_listpackDeleteInclusiveRange1_2}, {"test_listpackDeleteWitStartIndexOutOfRange", test_listpackDeleteWitStartIndexOutOfRange}, {"test_listpackDeleteWitNumOverflow", test_listpackDeleteWitNumOverflow}, {"test_listpackBatchDelete", test_listpackBatchDelete}, {"test_listpackDeleteFooWhileIterating", test_listpackDeleteFooWhileIterating}, {"test_listpackReplaceWithSameSize", test_listpackReplaceWithSameSize}, {"test_listpackReplaceWithDifferentSize", test_listpackReplaceWithDifferentSize}, {"test_listpackRegressionGt255Bytes", test_listpackRegressionGt255Bytes}, {"test_listpackCreateLongListAndCheckIndices", test_listpackCreateLongListAndCheckIndices}, {"test_listpackCompareStrsWithLpEntries", test_listpackCompareStrsWithLpEntries}, {"test_listpackLpMergeEmptyLps", test_listpackLpMergeEmptyLps}, {"test_listpackLpMergeLp1Larger", test_listpackLpMergeLp1Larger}, {"test_listpackLpMergeLp2Larger", test_listpackLpMergeLp2Larger}, {"test_listpackLpNextRandom", test_listpackLpNextRandom}, {"test_listpackLpNextRandomCC", test_listpackLpNextRandomCC}, {"test_listpackRandomPairWithOneElement", test_listpackRandomPairWithOneElement}, {"test_listpackRandomPairWithManyElements", test_listpackRandomPairWithManyElements}, {"test_listpackRandomPairsWithOneElement", test_listpackRandomPairsWithOneElement}, {"test_listpackRandomPairsWithManyElements", test_listpackRandomPairsWithManyElements}, {"test_listpackRandomPairsUniqueWithOneElement", test_listpackRandomPairsUniqueWithOneElement}, {"test_listpackRandomPairsUniqueWithManyElements", test_listpackRandomPairsUniqueWithManyElements}, {"test_listpackPushVariousEncodings", test_listpackPushVariousEncodings}, {"test_listpackLpFind", test_listpackLpFind}, {"test_listpackLpValidateIntegrity", test_listpackLpValidateIntegrity}, {"test_listpackNumberOfElementsExceedsLP_HDR_NUMELE_UNKNOWN", test_listpackNumberOfElementsExceedsLP_HDR_NUMELE_UNKNOWN}, {"test_listpackStressWithRandom", test_listpackStressWithRandom}, {"test_listpackSTressWithVariableSize", test_listpackSTressWithVariableSize}, {"test_listpackBenchmarkInit", test_listpackBenchmarkInit}, {"test_listpackBenchmarkLpAppend", test_listpackBenchmarkLpAppend}, {"test_listpackBenchmarkLpFindString", test_listpackBenchmarkLpFindString}, {"test_listpackBenchmarkLpFindNumber", test_listpackBenchmarkLpFindNumber}, {"test_listpackBenchmarkLpSeek", test_listpackBenchmarkLpSeek}, {"test_listpackBenchmarkLpValidateIntegrity", test_listpackBenchmarkLpValidateIntegrity}, {"test_listpackBenchmarkLpCompareWithString", test_listpackBenchmarkLpCompareWithString}, {"test_listpackBenchmarkLpCompareWithNumber", test_listpackBenchmarkLpCompareWithNumber}, {"test_listpackBenchmarkFree", test_listpackBenchmarkFree}, {NULL, NULL}}; +unitTest __test_mutexqueue_c[] = {{"test_mutexQueueSimplePushPop", test_mutexQueueSimplePushPop}, {"test_mutexQueueDoublePushPop", test_mutexQueueDoublePushPop}, {"test_mutexQueuePriorityOrdering", test_mutexQueuePriorityOrdering}, {"test_mutexQueueFifoPopAll", test_mutexQueueFifoPopAll}, {"test_mutexQueueFifoAddMultiple", test_mutexQueueFifoAddMultiple}, {"test_mutexQueueSimpleThread", test_mutexQueueSimpleThread}, {"test_mutexQueueParallelWriters", test_mutexQueueParallelWriters}, {"test_mutexQueueParallelReaders", test_mutexQueueParallelReaders}, {"test_mutexQueueParallelReadWrite", test_mutexQueueParallelReadWrite}, {NULL, NULL}}; unitTest __test_networking_c[] = {{"test_writeToReplica", test_writeToReplica}, {"test_postWriteToReplica", test_postWriteToReplica}, {"test_backupAndUpdateClientArgv", test_backupAndUpdateClientArgv}, {"test_rewriteClientCommandArgument", test_rewriteClientCommandArgument}, {"test_addRepliesWithOffloadsToBuffer", test_addRepliesWithOffloadsToBuffer}, {"test_addRepliesWithOffloadsToList", test_addRepliesWithOffloadsToList}, {"test_addBufferToReplyIOV", test_addBufferToReplyIOV}, {NULL, NULL}}; unitTest __test_object_c[] = {{"test_object_with_key", test_object_with_key}, {"test_embedded_string_with_key", test_embedded_string_with_key}, {NULL, NULL}}; unitTest __test_quicklist_c[] = {{"test_quicklistCreateList", test_quicklistCreateList}, {"test_quicklistAddToTailOfEmptyList", test_quicklistAddToTailOfEmptyList}, {"test_quicklistAddToHeadOfEmptyList", test_quicklistAddToHeadOfEmptyList}, {"test_quicklistAddToTail5xAtCompress", test_quicklistAddToTail5xAtCompress}, {"test_quicklistAddToHead5xAtCompress", test_quicklistAddToHead5xAtCompress}, {"test_quicklistAddToTail500xAtCompress", test_quicklistAddToTail500xAtCompress}, {"test_quicklistAddToHead500xAtCompress", test_quicklistAddToHead500xAtCompress}, {"test_quicklistRotateEmpty", test_quicklistRotateEmpty}, {"test_quicklistComprassionPlainNode", test_quicklistComprassionPlainNode}, {"test_quicklistNextPlainNode", test_quicklistNextPlainNode}, {"test_quicklistRotatePlainNode", test_quicklistRotatePlainNode}, {"test_quicklistRotateOneValOnce", test_quicklistRotateOneValOnce}, {"test_quicklistRotate500Val5000TimesAtCompress", test_quicklistRotate500Val5000TimesAtCompress}, {"test_quicklistPopEmpty", test_quicklistPopEmpty}, {"test_quicklistPop1StringFrom1", test_quicklistPop1StringFrom1}, {"test_quicklistPopHead1NumberFrom1", test_quicklistPopHead1NumberFrom1}, {"test_quicklistPopHead500From500", test_quicklistPopHead500From500}, {"test_quicklistPopHead5000From500", test_quicklistPopHead5000From500}, {"test_quicklistIterateForwardOver500List", test_quicklistIterateForwardOver500List}, {"test_quicklistIterateReverseOver500List", test_quicklistIterateReverseOver500List}, {"test_quicklistInsertAfter1Element", test_quicklistInsertAfter1Element}, {"test_quicklistInsertBefore1Element", test_quicklistInsertBefore1Element}, {"test_quicklistInsertHeadWhileHeadNodeIsFull", test_quicklistInsertHeadWhileHeadNodeIsFull}, {"test_quicklistInsertTailWhileTailNodeIsFull", test_quicklistInsertTailWhileTailNodeIsFull}, {"test_quicklistInsertOnceInElementsWhileIteratingAtCompress", test_quicklistInsertOnceInElementsWhileIteratingAtCompress}, {"test_quicklistInsertBefore250NewInMiddleOf500ElementsAtCompress", test_quicklistInsertBefore250NewInMiddleOf500ElementsAtCompress}, {"test_quicklistInsertAfter250NewInMiddleOf500ElementsAtCompress", test_quicklistInsertAfter250NewInMiddleOf500ElementsAtCompress}, {"test_quicklistDuplicateEmptyList", test_quicklistDuplicateEmptyList}, {"test_quicklistDuplicateListOf1Element", test_quicklistDuplicateListOf1Element}, {"test_quicklistDuplicateListOf500", test_quicklistDuplicateListOf500}, {"test_quicklistIndex1200From500ListAtFill", test_quicklistIndex1200From500ListAtFill}, {"test_quicklistIndex12From500ListAtFill", test_quicklistIndex12From500ListAtFill}, {"test_quicklistIndex100From500ListAtFill", test_quicklistIndex100From500ListAtFill}, {"test_quicklistIndexTooBig1From50ListAtFill", test_quicklistIndexTooBig1From50ListAtFill}, {"test_quicklistDeleteRangeEmptyList", test_quicklistDeleteRangeEmptyList}, {"test_quicklistDeleteRangeOfEntireNodeInListOfOneNode", test_quicklistDeleteRangeOfEntireNodeInListOfOneNode}, {"test_quicklistDeleteRangeOfEntireNodeWithOverflowCounts", test_quicklistDeleteRangeOfEntireNodeWithOverflowCounts}, {"test_quicklistDeleteMiddle100Of500List", test_quicklistDeleteMiddle100Of500List}, {"test_quicklistDeleteLessThanFillButAcrossNodes", test_quicklistDeleteLessThanFillButAcrossNodes}, {"test_quicklistDeleteNegative1From500List", test_quicklistDeleteNegative1From500List}, {"test_quicklistDeleteNegative1From500ListWithOverflowCounts", test_quicklistDeleteNegative1From500ListWithOverflowCounts}, {"test_quicklistDeleteNegative100From500List", test_quicklistDeleteNegative100From500List}, {"test_quicklistDelete10Count5From50List", test_quicklistDelete10Count5From50List}, {"test_quicklistNumbersOnlyListRead", test_quicklistNumbersOnlyListRead}, {"test_quicklistNumbersLargerListRead", test_quicklistNumbersLargerListRead}, {"test_quicklistNumbersLargerListReadB", test_quicklistNumbersLargerListReadB}, {"test_quicklistLremTestAtCompress", test_quicklistLremTestAtCompress}, {"test_quicklistIterateReverseDeleteAtCompress", test_quicklistIterateReverseDeleteAtCompress}, {"test_quicklistIteratorAtIndexTestAtCompress", test_quicklistIteratorAtIndexTestAtCompress}, {"test_quicklistLtrimTestAAtCompress", test_quicklistLtrimTestAAtCompress}, {"test_quicklistLtrimTestBAtCompress", test_quicklistLtrimTestBAtCompress}, {"test_quicklistLtrimTestCAtCompress", test_quicklistLtrimTestCAtCompress}, {"test_quicklistLtrimTestDAtCompress", test_quicklistLtrimTestDAtCompress}, {"test_quicklistVerifySpecificCompressionOfInteriorNodes", test_quicklistVerifySpecificCompressionOfInteriorNodes}, {"test_quicklistBookmarkGetUpdatedToNextItem", test_quicklistBookmarkGetUpdatedToNextItem}, {"test_quicklistBookmarkLimit", test_quicklistBookmarkLimit}, {"test_quicklistCompressAndDecompressQuicklistListpackNode", test_quicklistCompressAndDecompressQuicklistListpackNode}, {"test_quicklistCompressAndDecomressQuicklistPlainNodeLargeThanUINT32MAX", test_quicklistCompressAndDecomressQuicklistPlainNodeLargeThanUINT32MAX}, {NULL, NULL}}; @@ -297,10 +315,12 @@ struct unitTestSuite { {"test_dict.c", __test_dict_c}, {"test_endianconv.c", __test_endianconv_c}, {"test_entry.c", __test_entry_c}, + {"test_fifo.c", __test_fifo_c}, {"test_hashtable.c", __test_hashtable_c}, {"test_intset.c", __test_intset_c}, {"test_kvstore.c", __test_kvstore_c}, {"test_listpack.c", __test_listpack_c}, + {"test_mutexqueue.c", __test_mutexqueue_c}, {"test_networking.c", __test_networking_c}, {"test_object.c", __test_object_c}, {"test_quicklist.c", __test_quicklist_c}, diff --git a/src/unit/test_mutexqueue.c b/src/unit/test_mutexqueue.c new file mode 100644 index 00000000000..930ad32f423 --- /dev/null +++ b/src/unit/test_mutexqueue.c @@ -0,0 +1,293 @@ +/* + * Copyright (c) Valkey Contributors + * All rights reserved. + * SPDX-License-Identifier: BSD-3-Clause + */ + +#include "../mutexqueue.h" +#include "test_help.h" +#include +#include + +/* Helper functions */ +static void add(mutexQueue *q, long value) { + unsigned long len = mutexQueueLength(q); + mutexQueueAdd(q, (void *)value); + TEST_EXPECT(mutexQueueLength(q) == len + 1); +} + +static void pAdd(mutexQueue *q, long value) { + unsigned long len = mutexQueueLength(q); + mutexQueueAddPriority(q, (void *)value); + TEST_EXPECT(mutexQueueLength(q) == len + 1); +} + +static void popTest(mutexQueue *q, long expected) { + unsigned long len = mutexQueueLength(q); + long value = (long)mutexQueuePop(q, false); + TEST_EXPECT(mutexQueueLength(q) == len - 1); + TEST_EXPECT(value == expected); +} + +/* Test: simplePushPop */ +int test_mutexQueueSimplePushPop(int argc, char *argv[], int flags) { + UNUSED(argc); + UNUSED(argv); + UNUSED(flags); + + mutexQueue *q = mutexQueueCreate(); + TEST_EXPECT(mutexQueueLength(q) == 0ul); + add(q, 1); + popTest(q, 1); + TEST_EXPECT(mutexQueuePop(q, false) == NULL); + + mutexQueueRelease(q); + return 0; +} + +/* Test: doublePushPop */ +int test_mutexQueueDoublePushPop(int argc, char *argv[], int flags) { + UNUSED(argc); + UNUSED(argv); + UNUSED(flags); + + mutexQueue *q = mutexQueueCreate(); + add(q, 1); + add(q, 2); + popTest(q, 1); + popTest(q, 2); + + mutexQueueRelease(q); + return 0; +} + +/* Test: priorityOrdering */ +int test_mutexQueuePriorityOrdering(int argc, char *argv[], int flags) { + UNUSED(argc); + UNUSED(argv); + UNUSED(flags); + + mutexQueue *q = mutexQueueCreate(); + add(q, 10); + pAdd(q, 1); + add(q, 11); + pAdd(q, 2); + popTest(q, 1); + popTest(q, 2); + popTest(q, 10); + popTest(q, 11); + TEST_EXPECT(mutexQueuePop(q, false) == NULL); + + mutexQueueRelease(q); + return 0; +} + +/* Test: fifoPopAll */ +int test_mutexQueueFifoPopAll(int argc, char *argv[], int flags) { + UNUSED(argc); + UNUSED(argv); + UNUSED(flags); + + mutexQueue *q = mutexQueueCreate(); + add(q, 10); + pAdd(q, 1); + add(q, 11); + pAdd(q, 2); + + fifo *f = mutexQueuePopAll(q, false); + TEST_ASSERT(f != NULL); /* Fatal - can't continue if NULL */ + TEST_EXPECT(mutexQueuePop(q, false) == NULL); + TEST_EXPECT(mutexQueuePopAll(q, false) == NULL); + TEST_EXPECT(mutexQueueLength(q) == 0ul); + + TEST_EXPECT((unsigned long)fifoPop(f) == 1ul); + TEST_EXPECT((unsigned long)fifoPop(f) == 2ul); + TEST_EXPECT((unsigned long)fifoPop(f) == 10ul); + TEST_EXPECT((unsigned long)fifoPop(f) == 11ul); + TEST_EXPECT(fifoLength(f) == 0); + + fifoDelete(f); + mutexQueueRelease(q); + return 0; +} + +/* Test: fifoAddMultiple */ +int test_mutexQueueFifoAddMultiple(int argc, char *argv[], int flags) { + UNUSED(argc); + UNUSED(argv); + UNUSED(flags); + + mutexQueue *q = mutexQueueCreate(); + add(q, 1); + + fifo *f = fifoCreate(); + fifoPush(f, (void *)2); + fifoPush(f, (void *)3); + mutexQueueAddMultiple(q, f); + TEST_EXPECT(fifoLength(f) == 0u); + fifoDelete(f); + + add(q, 4); + pAdd(q, 0); + popTest(q, 0); + popTest(q, 1); + popTest(q, 2); + popTest(q, 3); + popTest(q, 4); + TEST_EXPECT(mutexQueuePop(q, false) == NULL); + TEST_EXPECT(mutexQueueLength(q) == 0ul); + + mutexQueueRelease(q); + return 0; +} + +/* Thread functions for concurrent tests */ +static void *queue_writer(void *arg) { + mutexQueue *queue = (mutexQueue *)arg; + for (int i = 1; i <= 1000; i++) { + mutexQueueAdd(queue, (void *)(long)i); + } + return NULL; +} + +static void *queue_reader(void *arg) { + mutexQueue *queue = (mutexQueue *)arg; + int count = 0; + while (count < 1000) { + long value = (long)mutexQueuePop(queue, true); + TEST_EXPECT(value != 0); /* Should never be null if blocking */ + count++; + } + return NULL; +} + +/* Test: simpleThread */ +int test_mutexQueueSimpleThread(int argc, char *argv[], int flags) { + UNUSED(argc); + UNUSED(argv); + UNUSED(flags); + + mutexQueue *q = mutexQueueCreate(); + int rc; + pthread_t writer, reader; + + rc = pthread_create(&writer, NULL, &queue_writer, q); + TEST_ASSERT(rc == 0); /* Fatal - can't continue if thread creation fails */ + rc = pthread_create(&reader, NULL, &queue_reader, q); + TEST_ASSERT(rc == 0); + rc = pthread_join(writer, NULL); + TEST_ASSERT(rc == 0); + rc = pthread_join(reader, NULL); + TEST_ASSERT(rc == 0); + TEST_EXPECT(mutexQueueLength(q) == 0ul); + + rc = pthread_create(&reader, NULL, &queue_reader, q); + TEST_ASSERT(rc == 0); + rc = pthread_create(&writer, NULL, &queue_writer, q); + TEST_ASSERT(rc == 0); + rc = pthread_join(writer, NULL); + TEST_ASSERT(rc == 0); + rc = pthread_join(reader, NULL); + TEST_ASSERT(rc == 0); + TEST_EXPECT(mutexQueueLength(q) == 0ul); + + mutexQueueRelease(q); + return 0; +} + +/* Test: parallelWriters */ +int test_mutexQueueParallelWriters(int argc, char *argv[], int flags) { + UNUSED(argc); + UNUSED(argv); + UNUSED(flags); + + mutexQueue *q = mutexQueueCreate(); + const int num_threads = 20; + int rc; + pthread_t writer[num_threads]; + + for (int i = 0; i < num_threads; i++) { + rc = pthread_create(&writer[i], NULL, &queue_writer, q); + TEST_ASSERT(rc == 0); + } + + for (int i = 0; i < num_threads; i++) { + rc = pthread_join(writer[i], NULL); + TEST_ASSERT(rc == 0); + } + + TEST_EXPECT(mutexQueueLength(q) == (unsigned long)(num_threads * 1000)); + + fifo *f = mutexQueuePopAll(q, false); + fifoDelete(f); + mutexQueueRelease(q); + return 0; +} + +/* Test: parallelReaders */ +int test_mutexQueueParallelReaders(int argc, char *argv[], int flags) { + UNUSED(argc); + UNUSED(argv); + UNUSED(flags); + + mutexQueue *q = mutexQueueCreate(); + const int num_threads = 20; + int rc; + pthread_t reader[num_threads]; + + /* Start readers in advance - we want them fighting... */ + for (int i = 0; i < num_threads; i++) { + rc = pthread_create(&reader[i], NULL, &queue_reader, q); + TEST_ASSERT(rc == 0); /* Fatal - can't continue if thread creation fails */ + } + + /* Now perform writes serially... */ + for (int i = 0; i < num_threads; i++) { + queue_writer(q); + /* make sure other threads get to fight with a short sleep (don't write all at once!) */ + nanosleep((const struct timespec[]){{0, 10000000L}}, NULL); /* 10ms */ + } + + /* Readers should finish */ + for (int i = 0; i < num_threads; i++) { + rc = pthread_join(reader[i], NULL); + TEST_ASSERT(rc == 0); + } + + TEST_EXPECT(mutexQueueLength(q) == 0ul); + + mutexQueueRelease(q); + return 0; +} + +/* Test: parallelReadWrite */ +int test_mutexQueueParallelReadWrite(int argc, char *argv[], int flags) { + UNUSED(argc); + UNUSED(argv); + UNUSED(flags); + + mutexQueue *q = mutexQueueCreate(); + const int num_threads = 20; + int rc; + pthread_t reader[num_threads]; + pthread_t writer[num_threads]; + + for (int i = 0; i < num_threads; i++) { + rc = pthread_create(&writer[i], NULL, &queue_writer, q); + TEST_ASSERT(rc == 0); + rc = pthread_create(&reader[i], NULL, &queue_reader, q); + TEST_ASSERT(rc == 0); + } + + for (int i = 0; i < num_threads; i++) { + rc = pthread_join(writer[i], NULL); + TEST_ASSERT(rc == 0); + rc = pthread_join(reader[i], NULL); + TEST_ASSERT(rc == 0); + } + + TEST_EXPECT(mutexQueueLength(q) == 0ul); + + mutexQueueRelease(q); + return 0; +} From 6383bf33a6d5a94bc07125efe2de65f3a09f6b6b Mon Sep 17 00:00:00 2001 From: Alina Liu Date: Thu, 18 Dec 2025 19:33:59 +0000 Subject: [PATCH 2/6] Modify FIFO and mutexQueue with bio.c updated Signed-off-by: Alina Liu --- src/bio.c | 29 ++++++++- src/fifo.c | 128 ++++++++++++++++++++++++++----------- src/fifo.h | 56 +++++++++++----- src/mutexqueue.c | 23 +++---- src/mutexqueue.h | 44 ++++++++----- src/unit/test_fifo.c | 78 +++++++++++----------- src/unit/test_mutexqueue.c | 11 ++-- 7 files changed, 237 insertions(+), 132 deletions(-) diff --git a/src/bio.c b/src/bio.c index a8c2995d561..3ee0a53afb3 100644 --- a/src/bio.c +++ b/src/bio.c @@ -37,8 +37,35 @@ * * ---------------------------------------------------------------------------- * + * Copyright (c) 2009-2012, Redis Ltd. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * * Redistributions of source code must retain the above copyright notice, + * this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * * Neither the name of Redis nor the names of its contributors may be used + * to endorse or promote products derived from this software without + * specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" + * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE + * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR + * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF + * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS + * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN + * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) + * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE + * POSSIBILITY OF SUCH DAMAGE. */ + #include "server.h" #include "connection.h" #include "bio.h" @@ -280,7 +307,7 @@ unsigned long bioPendingJobsOfType(int type) { /* Wait for the job queue of the worker for jobs of specified type to become empty. */ void bioDrainWorker(int type) { while (bioPendingJobsOfType(type) > 0) { - usleep(1000); /* Sleep for 1ms and check again*/ + usleep(100); /* Sleep for 1ms and check again*/ } } diff --git a/src/fifo.c b/src/fifo.c index a91442f577f..630a1cc6dfe 100644 --- a/src/fifo.c +++ b/src/fifo.c @@ -4,9 +4,11 @@ * SPDX-License-Identifier: BSD-3-Clause */ -/* FIFO - A high-performance FIFO queue implementation */ +/* FIFO - A high-performance First-In, First-Out queue implementation */ +#include #include +#include #include "fifo.h" #include "serverassert.h" @@ -21,7 +23,7 @@ static const uintptr_t IDX_MASK = 0x0007; /* The FifoBlock contains up to 7 items (pointers). When compared with adlist, this results in * roughly 60% memory reduction and 7x fewer memory allocations. Memory reduction is guaranteed - * with 5+ items in queue. + * with 5+ items. * * In each block, there are 7 slots for item pointers (pointers to the caller's FIFO item). * We need to keep track of the first & last slot used. Contextually, we will only need @@ -113,13 +115,12 @@ struct fifoBlock { }; struct fifo { - long length; /* Total number of items in queue */ + long length; /* Number of items */ fifoBlock *first; fifoBlock *last; }; -/* Create a new FIFO queue. */ fifo *fifoCreate(void) { fifo *q = zmalloc(sizeof(fifo)); q->length = 0; @@ -128,10 +129,9 @@ fifo *fifoCreate(void) { } -/* Push an item onto the end of the queue. */ void fifoPush(fifo *q, void *ptr) { if (q->first == NULL) { - /* Queue was empty - create block */ + /* fifo was empty - create block */ assert(q->last == NULL && q->length == 0); q->last = q->first = zmalloc(sizeof(fifoBlock)); q->last->u.last_or_first_idx = 0; /* Item 0 is the last item in this block */ @@ -158,32 +158,49 @@ void fifoPush(fifo *q, void *ptr) { } -/* Look at the first item in the queue (without removing it). - * NOTE: asserts if the queue is empty. */ -void *fifoPeek(fifo *q) { - assert(q->length > 0); +bool fifoPeek(fifo *q, void **item) { + if (q->length == 0) return false; int firstIdx = (q->first == q->last) ? 0 : q->first->u.last_or_first_idx & IDX_MASK; - return q->first->items[firstIdx]; + *item = q->first->items[firstIdx]; + return true; } -/* Return and remove the first item from the queue. - * NOTE: asserts if the queue is empty. */ -void *fifoPop(fifo *q) { - assert(q->length > 0); - void *item; +bool fifoPop(fifo *q, void **item) { + if (q->length == 0) return false; + void *value; if (q->first == q->last) { - /* With only 1 block, POP occurs at index 0 and items 1..6 are shifted. */ - item = q->last->items[0]; + /* With only 1 block, POP occurs at index 0 and items 1..6 are shifted. + * + * Example SINGLE BLOCK with 4 items BEFORE pop: + * +--------+--------+--------+--------+--------+--------+--------+--------+ + * BEFORE POP: | slot 0 | slot 1 | slot 2 | slot 3 | slot 4 | slot 5 | slot 6 | next/ | + * | item | item | item | item | - | - | - | lastIdx| + * +--------+--------+--------+--------+--------+--------+--------+--------+ + * ^ ^ + * POP here lastIdx (3) + * + * Example SINGLE BLOCK with 3 items AFTER pop (items shifted left): + * +--------+--------+--------+--------+--------+--------+--------+--------+ + * AFTER POP: | slot 0 | slot 1 | slot 2 | slot 3 | slot 4 | slot 5 | slot 6 | next/ | + * | item | item | item | - | - | - | - | lastIdx| + * +--------+--------+--------+--------+--------+--------+--------+--------+ + * ^ + * lastIdx (2) + * + * Items are shifted left to keep them left-justified in the single block. + * This avoids needing to allocate a new block when pushing more items. + */ + value = q->last->items[0]; int lastIdx = q->last->u.last_or_first_idx; /* pointer portion is 0 on last (or only) block */ assert(lastIdx < ITEMS_PER_BLOCK); if (lastIdx > 0) { /* With only 1 block, shift the items rather than eventually needing new block. - * (This is cheap, shifting a max of 6 pointers.) */ - for (int i = 0; i < lastIdx; i++) q->last->items[i] = q->last->items[i + 1]; + * Use memmove to shift pointers to the left by 1 */ + memmove(q->last->items, q->last->items + 1, lastIdx * sizeof(q->last->items[0])); q->last->u.last_or_first_idx--; /* Decrement the last index */ } else { /* Just finished the only block. Delete it. */ @@ -193,7 +210,7 @@ void *fifoPop(fifo *q) { } else { /* With more than 1 block, POP occurs at firstIdx, and firstIdx is incremented. */ int firstIdx = q->first->u.last_or_first_idx & IDX_MASK; - item = q->first->items[firstIdx]; + value = q->first->items[firstIdx]; if (firstIdx < ITEMS_PER_BLOCK - 1) { /* Just increment the first index to the next slot. */ @@ -208,19 +225,16 @@ void *fifoPop(fifo *q) { } q->length--; - - return item; + *item = value; + return true; } -/* Return the number of items in the queue. */ long fifoLength(fifo *q) { return q->length; } -/* Delete the queue. - * NOTE: this does not free items which may be referenced by inserted pointers. */ void fifoDelete(fifo *q) { if (q->length > 0) { fifoBlock *cur = q->first; @@ -235,8 +249,8 @@ void fifoDelete(fifo *q) { } -/* Blindly overwrites target from source. */ -static void blindlyMoveFifoContents(fifo *target, fifo *source) { +/* Overwrites target from source. */ +static void overwriteFifoContents(fifo *target, fifo *source) { target->length = source->length; target->first = source->first; target->last = source->last; @@ -245,7 +259,6 @@ static void blindlyMoveFifoContents(fifo *target, fifo *source) { } -/* Join an "other" fifo onto this one (emptying "other") */ void fifoJoin(fifo *q, fifo *other) { /* When joining a fifo onto an existing fifo, we might be left with partially full blocks in the * middle of the list. In the usual case, any blocks in the middle of the list have the index @@ -257,32 +270,70 @@ void fifoJoin(fifo *q, fifo *other) { * shift the items so that the block becomes right-justified. Then the index is corrected, * replacing the lastIdx with the firstIdx. * - * The "other" list is correct as-is. If there is only a single block, it becomes the last - * block and remains left-justified. If there are multiple blocks, the first block of the - * "other" list is already right-justified and becomes a partially full middle block. + * Example: Joining two fifos where "q" has 3 items in its last block: + * + * BEFORE JOIN - "q" fifo (last block is left-justified): + * +--------+--------+--------+--------+--------+--------+--------+--------+ + * q->last: | slot 0 | slot 1 | slot 2 | slot 3 | slot 4 | slot 5 | slot 6 | next/ | + * | item | item | item | - | - | - | - | lastIdx| + * +--------+--------+--------+--------+--------+--------+--------+--------+ + * ^ (2) + * + * BEFORE JOIN - "other" fifo (first block is right-justified if multiple blocks): + * +--------+--------+--------+--------+--------+--------+--------+--------+ + * other->first: | slot 0 | slot 1 | slot 2 | slot 3 | slot 4 | slot 5 | slot 6 | next/ | + * | - | - | - | - | item | item | item |firstIdx| + * +--------+--------+--------+--------+--------+--------+--------+--------+ + * ^ (4) + * + * AFTER JOIN - q's last block is shifted right and linked to other: + * +--------+--------+--------+--------+--------+--------+--------+--------+ + * (was q->last) | slot 0 | slot 1 | slot 2 | slot 3 | slot 4 | slot 5 | slot 6 | next/ | + * now middle: | - | - | - | - | item | item | item |firstIdx| --+ + * +--------+--------+--------+--------+--------+--------+--------+--------+ | + * ^ (4) | + * | + * +------------------------------------------------------------------------+ + * | + * v + * +--------+--------+--------+--------+--------+--------+--------+--------+ + * other->first: | slot 0 | slot 1 | slot 2 | slot 3 | slot 4 | slot 5 | slot 6 | next/ | + * (now linked): | - | - | - | - | item | item | item |firstIdx| + * +--------+--------+--------+--------+--------+--------+--------+--------+ + * ^ (4) + * | + * firstIdx + * + * The shift ensures q's last block becomes a valid middle block (right-justified with firstIdx). + * + * The "other" list maintains its structure when appended: + * - If "other" has a single block (left-justified), it becomes q's new last block + * - If "other" has multiple blocks, its first block (right-justified if partial) becomes + * a middle block, and its last block becomes q's new last block + * This is essentially a "pop all from other into q" operation that preserves invariants. */ if (other->length == 0) return; if (q->length == 0) { /* If "q" is empty, it's a simple operation. */ - blindlyMoveFifoContents(q, other); + overwriteFifoContents(q, other); return; } if (other->length < ITEMS_PER_BLOCK) { /* In the case of a short "other" fifo, move each item. This prevents creation of a string * of half-empty blocks if fifoJoin is repeatedly used on small fifos. */ - while (other->length > 0) fifoPush(q, fifoPop(other)); + void *value; + while (fifoPop(other, &value)) fifoPush(q, value); return; } fifoBlock *curLast = q->last; int lastIdx = curLast->u.last_or_first_idx; - /* Shift the items in the last block if it is partially full */ + /* Shift the items to the right in the last block if it is partially full */ int shift = (ITEMS_PER_BLOCK - 1) - lastIdx; if (shift > 0) { - for (int i = lastIdx; i >= 0; i--) - curLast->items[i + shift] = curLast->items[i]; + memmove(q->last->items + shift, q->last->items, (lastIdx + 1) * sizeof(q->last->items[0])); } /* Now fix up the next pointer to point to the next block */ @@ -297,9 +348,8 @@ void fifoJoin(fifo *q, fifo *other) { } -/* Copy all of the items into a new fifo (emptying the original) */ fifo *fifoPopAll(fifo *q) { fifo *newQ = zmalloc(sizeof(fifo)); - blindlyMoveFifoContents(newQ, q); + overwriteFifoContents(newQ, q); return newQ; } diff --git a/src/fifo.h b/src/fifo.h index 9e4dfd6d07a..c62337d27d6 100644 --- a/src/fifo.h +++ b/src/fifo.h @@ -4,45 +4,71 @@ * SPDX-License-Identifier: BSD-3-Clause */ -/* A space/time efficient FIFO queue of pointers. +/* A space/time efficient First-In, First-Out queue of pointers. * * Implemented with an unrolled single-linked list, the implementation packs multiple pointers into * a single block. This increases space efficiency and cache locality over the Valkey `list` for the - * purpose of a simple FIFO queue. + * purpose of a simple fifo. + * + * IMPORTANT: NULL fifo are NOT supported by these APIs. + * All functions expect a valid fifo created by fifoCreate(). + * Passing NULL to any function will result in undefined behavior (likely a crash). + * + * STORED VALUES: The fifo supports arbitrary void* values with the following guarantees: + * - All bit patterns are preserved exactly as provided (no modifications) + * - NULL pointers can be stored as items (distinct from NULL fifo above) + * - Integer values (e.g., intptr_t) can be stored by casting to void* + * - Pointers with custom bit flags/tags are fully supported + * The implementation makes no assumptions about the stored values and keeps them intact. */ #ifndef __FIFO_H_ #define __FIFO_H_ +#include + typedef struct fifo fifo; -/* Create a new FIFO queue. */ +/* Create a new fifo. */ fifo *fifoCreate(void); -/* Push an item onto the end of the queue. */ +/* Push an item onto the end of the fifo. */ void fifoPush(fifo *q, void *ptr); -/* Look at the first item in the queue (without removing it). - * NOTE: asserts if the queue is empty. */ -void *fifoPeek(fifo *q); +/* Look at the first item in the fifo (without removing it). + * Returns true if an item exists. If false, `item` is undefined. */ +bool fifoPeek(fifo *q, void **item); -/* Return and remove the first item from the queue. - * NOTE: asserts if the queue is empty. */ -void *fifoPop(fifo *q); +/* Return and remove the first item from the fifo. + * Returns true if an item exists. If false, `item` is undefined. */ +bool fifoPop(fifo *q, void **item); -/* Return the number of items in the queue. */ +/* Return the number of items in the fifo. */ long fifoLength(fifo *q); -/* Delete the queue. +/* Delete the fifo. * NOTE: this does not free items which may be referenced by inserted pointers. */ void fifoDelete(fifo *q); /* Joins the fifo "other" to the end of "q". "other" becomes empty, but remains valid. - * This is an O(1) operation. */ + * This is an O(1) operation. + * + * Use case: Appending items from one fifo to another existing fifo. + * Example: fifoJoin(destination_q, source_q); + * // destination_q now contains all items from both fifos + * // source_q is now empty but still valid */ void fifoJoin(fifo *q, fifo *other); -/* Returns a new fifo, containing all of the items from "q". "q" remains valid, but becomes empty. - * This is an O(1) operation. */ +/* Copy all of the items into a new fifo (emptying the original) + * Returns a new fifo, containing all of the items from "q". "q" remains valid, but becomes empty. + * This is an O(1) operation. + * + * Use case: Retrieving/removing all items from a fifo as a batch. + * This is cleaner and more descriptive than using fifoJoin with an empty fifo when the intent + * is to extract items rather than merge fifos. + * Example: fifo *batch = fifoPopAll(source_q); + * // batch contains all items, source_q is now empty + * // Process batch, then fifoDelete(batch) when done */ fifo *fifoPopAll(fifo *q); #endif diff --git a/src/mutexqueue.c b/src/mutexqueue.c index 7a34a589bfb..ce0c4329e2d 100644 --- a/src/mutexqueue.c +++ b/src/mutexqueue.c @@ -20,7 +20,6 @@ struct mutexQueue { }; -/* Create an empty queue. */ mutexQueue *mutexQueueCreate(void) { mutexQueue *mq; mq = zmalloc(sizeof(*mq)); @@ -33,11 +32,10 @@ mutexQueue *mutexQueueCreate(void) { } -/* Release an empty queue. - * Note: The queue must be empty before calling release. The quickest way to empty the queue is to +/* Note: The mutexQueue must be empty before calling release. The quickest way to empty the mutexQueue is to * call mutexQueuePopAll - which returns the items in a new fifo. It is the caller's * responsibility to free memory (as necessary) for any items. - * Note: Behavior is undefined if other threads are accessing the queue. */ + * Note: Behavior is undefined if other threads are accessing the mutexQueue. */ void mutexQueueRelease(mutexQueue *theQueue) { assert(mutexQueueLength(theQueue) == 0); mutexQueue *mq = theQueue; @@ -59,7 +57,6 @@ static inline unsigned long mutexQueueLengthInternal(mutexQueue *mq) { } -/* Number of items in the queue. */ unsigned long mutexQueueLength(mutexQueue *theQueue) { mutexQueue *mq = theQueue; @@ -72,8 +69,7 @@ unsigned long mutexQueueLength(mutexQueue *theQueue) { } -/* Insert a priority item at the beginning of the queue (but after existing priority items). */ -void mutexQueueAddPriority(mutexQueue *theQueue, void *value) { +void mutexQueuePushPriority(mutexQueue *theQueue, void *value) { mutexQueue *mq = theQueue; pthread_mutex_lock(&mq->mutex); @@ -86,7 +82,6 @@ void mutexQueueAddPriority(mutexQueue *theQueue, void *value) { } -/* Insert an item at the end of the queue. */ void mutexQueueAdd(mutexQueue *theQueue, void *value) { mutexQueue *mq = theQueue; @@ -100,7 +95,7 @@ void mutexQueueAdd(mutexQueue *theQueue, void *value) { } -/* Insert a fifo of items at the end of the queue. This removes the items from the source fifo! */ +/* Note: This removes the items from the source fifo! */ void mutexQueueAddMultiple(mutexQueue *theQueue, fifo *valueFifo) { mutexQueue *mq = theQueue; @@ -116,8 +111,7 @@ void mutexQueueAddMultiple(mutexQueue *theQueue, fifo *valueFifo) { } -/* Retrieves the first item off the queue (or NULL if queue is empty). - * If 'blocking' is true, this method will block until an item is available. */ +/* Note: If 'blocking' is true, this method will block until an item is available. */ void *mutexQueuePop(mutexQueue *theQueue, bool blocking) { mutexQueue *mq = theQueue; void *value = NULL; @@ -131,9 +125,9 @@ void *mutexQueuePop(mutexQueue *theQueue, bool blocking) { } if (fifoLength(mq->priority_fifo) > 0) { - value = fifoPop(mq->priority_fifo); + fifoPop(mq->priority_fifo, &value); } else if (fifoLength(mq->normal_fifo) > 0) { - value = fifoPop(mq->normal_fifo); + fifoPop(mq->normal_fifo, &value); } pthread_mutex_unlock(&mq->mutex); @@ -141,8 +135,7 @@ void *mutexQueuePop(mutexQueue *theQueue, bool blocking) { } -/* Retrieves all items from the queue as a fifo (or NULL if the queue is empty). - * If 'blocking' is true, this method will block until an item is available. */ +/* Note: If 'blocking' is true, this method will block until an item is available. */ fifo *mutexQueuePopAll(mutexQueue *theQueue, bool blocking) { mutexQueue *mq = theQueue; fifo *result = NULL; diff --git a/src/mutexqueue.h b/src/mutexqueue.h index 2a57e2c448d..975ea66b0a7 100644 --- a/src/mutexqueue.h +++ b/src/mutexqueue.h @@ -8,14 +8,26 @@ * A thread-safe queue, protected by a mutex. * * Supports: - * - Adding an item to the end of the queue - * - Adding a list of items (fifo) to the end of the queue - * - Insertion of a priority item at the beginning of the queue (but after existing priority items) - * - Removing an item from the beginning of the queue - * - Removing ALL items as (as new fifo) from the queue - * - Synchronous waiting on the queue for new items + * - Adding an item to the end of the mutexQueue + * - Adding a list of items (fifo) to the end of the mutexQueue + * - Insertion of a priority item at the beginning of the mutexQueue (but after existing priority items) + * - Removing an item from the beginning of the mutexQueue + * - Removing ALL items as (as new fifo) from the mutexQueue + * - Synchronous waiting on the mutexQueue for new items * - * The caller is responsible for memory management for items in the queue. + * Priority Use Case: + * The priority feature provides a 2-level priority system (priority vs normal) without + * requiring separate fifos with their own mutexes and condition variables. This simplifies + * synchronization when you need to occasionally process urgent items ahead of routine work. + * + * Example: In a background worker thread processing file operations, you might want to + * prioritize critical shutdown tasks or urgent fsync operations over routine lazy-free jobs. + * Priority items are processed in FIFO order, followed by normal items in FIFO order. + * + * Implementation: Uses two internal FIFOs (priority_fifo and normal_fifo). Items are always + * popped from priority_fifo first. + * + * The caller is responsible for memory management for items in the mutexQueue. */ #ifndef __MUTEXQUEUE_H @@ -27,28 +39,28 @@ /* The mutexQueue is an opaque structure. */ typedef struct mutexQueue mutexQueue; -/* Create an empty queue. */ +/* Create an empty mutexQueue. */ mutexQueue *mutexQueueCreate(void); -/* Release an empty queue. */ +/* Release an empty mutexQueue. */ void mutexQueueRelease(mutexQueue *theQueue); -/* Number of items in the queue. */ +/* Number of items in the mutexQueue. */ unsigned long mutexQueueLength(mutexQueue *theQueue); -/* Insert a priority item at the beginning of the queue (but after existing priority items). */ -void mutexQueueAddPriority(mutexQueue *theQueue, void *value); +/* Insert a priority item at the beginning of the mutexQueue (but after existing priority items). */ +void mutexQueuePushPriority(mutexQueue *theQueue, void *value); -/* Insert an item at the end of the queue. */ +/* Insert an item at the end of the mutexQueue. */ void mutexQueueAdd(mutexQueue *theQueue, void *value); -/* Insert multiple items (from a fifo) to the end of the queue. */ +/* Insert multiple items (from a fifo) to the end of the mutexQueue. */ void mutexQueueAddMultiple(mutexQueue *theQueue, fifo *valueFifo); -/* Retrieves the first item off the queue (or NULL if queue is empty). */ +/* Retrieves the first item off the mutexQueue (or NULL if mutexQueue is empty). */ void *mutexQueuePop(mutexQueue *theQueue, bool blocking); -/* Retrieves all items from the queue as a fifo (or NULL if the queue is empty). */ +/* Retrieves all items from the mutexQueue as a fifo (or NULL if the mutexQueue is empty). */ fifo *mutexQueuePopAll(mutexQueue *theQueue, bool blocking); #endif diff --git a/src/unit/test_fifo.c b/src/unit/test_fifo.c index 747aacebe14..fd36e58e68a 100644 --- a/src/unit/test_fifo.c +++ b/src/unit/test_fifo.c @@ -9,20 +9,6 @@ #include #include -/* Macro to detect if child process crashes as expected */ -#define DETECT_CRASH \ - pid_t pid = fork(); \ - if (pid < 0) { \ - /* Fork failed */ \ - TEST_EXPECT(0); \ - } else if (pid > 0) { \ - /* Parent process - wait for child to crash */ \ - int status; \ - waitpid(pid, &status, 0); \ - /* Verify child exited abnormally (assertion failure) */ \ - TEST_EXPECT(WIFSIGNALED(status) || (WIFEXITED(status) && WEXITSTATUS(status) != 0)); \ - } else /* macro is followed by the else clause */ - static inline void *INT_TO_PTR(intptr_t i) { return (void *)i; } @@ -39,46 +25,44 @@ static void push(fifo *q, intptr_t value) { } static intptr_t popTest(fifo *q, intptr_t expected) { - intptr_t peekValue = PTR_TO_INT(fifoPeek(q)); + void *peekPtr; + TEST_EXPECT(fifoPeek(q, &peekPtr)); + intptr_t peekValue = PTR_TO_INT(peekPtr); TEST_EXPECT(peekValue == expected); int len = fifoLength(q); - intptr_t value = PTR_TO_INT(fifoPop(q)); + void *popPtr; + TEST_EXPECT(fifoPop(q, &popPtr)); + intptr_t value = PTR_TO_INT(popPtr); TEST_EXPECT(fifoLength(q) == len - 1); TEST_EXPECT(value == expected); return value; } -/* Test: emptyPop - verify that popping from empty queue triggers assertion */ +/* Test: emptyPop - verify that popping from empty fifo returns false */ int test_fifoEmptyPop(int argc, char *argv[], int flags) { UNUSED(argc); UNUSED(argv); UNUSED(flags); - DETECT_CRASH { - /* Child process - this should crash with assertion */ - fifo *q = fifoCreate(); - TEST_EXPECT(fifoLength(q) == 0); - fifoPop(q); /* This should assert and crash */ - /* Should never reach here */ - TEST_ASSERT(0); - } + fifo *q = fifoCreate(); + TEST_EXPECT(fifoLength(q) == 0); + void *result; + TEST_EXPECT(fifoPop(q, &result) == false); + fifoDelete(q); return 0; } -/* Test: emptyPeek - verify that peeking at empty queue triggers assertion */ +/* Test: emptyPeek - verify that peeking at empty fifo returns false */ int test_fifoEmptyPeek(int argc, char *argv[], int flags) { UNUSED(argc); UNUSED(argv); UNUSED(flags); - DETECT_CRASH { - /* Child process - this should crash with assertion */ - fifo *q = fifoCreate(); - TEST_EXPECT(fifoLength(q) == 0); - fifoPeek(q); /* This should assert and crash */ - /* Should never reach here */ - TEST_ASSERT(0); - } + fifo *q = fifoCreate(); + TEST_EXPECT(fifoLength(q) == 0); + void *result; + TEST_EXPECT(fifoPeek(q, &result) == false); + fifoDelete(q); return 0; } @@ -145,7 +129,7 @@ int test_fifoJoinTest(int argc, char *argv[], int flags) { fifo *q = fifoCreate(); fifo *q2 = fifoCreate(); - /* In this test, there are 2 queues Q and Q2. + /* In this test, there are 2 fifos Q and Q2. * Various sizes are tested. For each size, various amounts are popped off the front first. * Q2 is appended to Q. */ for (int qLen = 0; qLen <= 21; qLen++) { @@ -156,11 +140,17 @@ int test_fifoJoinTest(int argc, char *argv[], int flags) { intptr_t popQValue = 1; for (int i = 0; i < qLen; i++) fifoPush(q, INT_TO_PTR(pushValue++)); - for (int i = 0; i < qPop; i++) TEST_EXPECT(PTR_TO_INT(fifoPop(q)) == popQValue++); + for (int i = 0; i < qPop; i++) { + void *ptr; + TEST_EXPECT(fifoPop(q, &ptr) && PTR_TO_INT(ptr) == popQValue++); + } intptr_t popQ2Value = pushValue; for (int i = 0; i < q2Len; i++) fifoPush(q2, INT_TO_PTR(pushValue++)); - for (int i = 0; i < q2Pop; i++) TEST_EXPECT(PTR_TO_INT(fifoPop(q2)) == popQ2Value++); + for (int i = 0; i < q2Pop; i++) { + void *ptr; + TEST_EXPECT(fifoPop(q2, &ptr) && PTR_TO_INT(ptr) == popQ2Value++); + } fifoJoin(q, q2); TEST_EXPECT(fifoLength(q) == (qLen - qPop) + (q2Len - q2Pop)); @@ -170,8 +160,14 @@ int test_fifoJoinTest(int argc, char *argv[], int flags) { TEST_EXPECT(fifoLength(temp) == (qLen - qPop) + (q2Len - q2Pop)); TEST_EXPECT(fifoLength(q) == 0); - for (int i = 0; i < (qLen - qPop); i++) TEST_EXPECT(PTR_TO_INT(fifoPop(temp)) == popQValue++); - for (int i = 0; i < (q2Len - q2Pop); i++) TEST_EXPECT(PTR_TO_INT(fifoPop(temp)) == popQ2Value++); + for (int i = 0; i < (qLen - qPop); i++) { + void *ptr; + TEST_EXPECT(fifoPop(temp, &ptr) && PTR_TO_INT(ptr) == popQValue++); + } + for (int i = 0; i < (q2Len - q2Pop); i++) { + void *ptr; + TEST_EXPECT(fifoPop(temp, &ptr) && PTR_TO_INT(ptr) == popQ2Value++); + } TEST_EXPECT(fifoLength(temp) == 0); fifoDelete(temp); @@ -216,7 +212,8 @@ static void exerciseFifo(void) { } TEST_EXPECT(fifoLength(q) == LIST_ITEMS); for (intptr_t i = 0; i < LIST_ITEMS; i++) { - TEST_EXPECT(fifoPop(q) == INT_TO_PTR(i)); + void *ptr; + TEST_EXPECT(fifoPop(q, &ptr) && ptr == INT_TO_PTR(i)); } TEST_EXPECT(fifoLength(q) == 0); fifoDelete(q); @@ -242,7 +239,6 @@ int test_fifoComparePerformance(int argc, char *argv[], int flags) { for (int i = 0; i < iterations; i++) exerciseFifo(); long fifoMs = elapsedMs(timer); - TEST_EXPECT(listMs == fifoMs); /* This will fail, printing result */ double percentImprovement = (double)(listMs - fifoMs) * 100.0 / listMs; TEST_PRINT_INFO("List: %ld ms, FIFO: %ld ms, Improvement: %.2f%%", listMs, fifoMs, percentImprovement); TEST_EXPECT(percentImprovement == 0.0); /* This will fail, printing result */ diff --git a/src/unit/test_mutexqueue.c b/src/unit/test_mutexqueue.c index 930ad32f423..4d5340b5a54 100644 --- a/src/unit/test_mutexqueue.c +++ b/src/unit/test_mutexqueue.c @@ -18,7 +18,7 @@ static void add(mutexQueue *q, long value) { static void pAdd(mutexQueue *q, long value) { unsigned long len = mutexQueueLength(q); - mutexQueueAddPriority(q, (void *)value); + mutexQueuePushPriority(q, (void *)value); TEST_EXPECT(mutexQueueLength(q) == len + 1); } @@ -100,10 +100,11 @@ int test_mutexQueueFifoPopAll(int argc, char *argv[], int flags) { TEST_EXPECT(mutexQueuePopAll(q, false) == NULL); TEST_EXPECT(mutexQueueLength(q) == 0ul); - TEST_EXPECT((unsigned long)fifoPop(f) == 1ul); - TEST_EXPECT((unsigned long)fifoPop(f) == 2ul); - TEST_EXPECT((unsigned long)fifoPop(f) == 10ul); - TEST_EXPECT((unsigned long)fifoPop(f) == 11ul); + void *ptr; + TEST_EXPECT(fifoPop(f, &ptr) && (unsigned long)ptr == 1ul); + TEST_EXPECT(fifoPop(f, &ptr) && (unsigned long)ptr == 2ul); + TEST_EXPECT(fifoPop(f, &ptr) && (unsigned long)ptr == 10ul); + TEST_EXPECT(fifoPop(f, &ptr) && (unsigned long)ptr == 11ul); TEST_EXPECT(fifoLength(f) == 0); fifoDelete(f); From 9530fbda6bf5a6d58709dd270520435f77918b16 Mon Sep 17 00:00:00 2001 From: Alina Liu Date: Thu, 18 Dec 2025 14:17:24 -0800 Subject: [PATCH 3/6] Update comment in src/fifo.h Co-authored-by: Jim Brunner Signed-off-by: Alina Liu --- src/fifo.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/fifo.h b/src/fifo.h index c62337d27d6..77ee08b67b0 100644 --- a/src/fifo.h +++ b/src/fifo.h @@ -40,7 +40,7 @@ void fifoPush(fifo *q, void *ptr); bool fifoPeek(fifo *q, void **item); /* Return and remove the first item from the fifo. - * Returns true if an item exists. If false, `item` is undefined. */ + * Returns true if an item exists. If false, `item` is not updated. */ bool fifoPop(fifo *q, void **item); /* Return the number of items in the fifo. */ From ef36a4e1e24f6235a79a9ec6adfe52920447fe3b Mon Sep 17 00:00:00 2001 From: Alina Liu Date: Thu, 18 Dec 2025 14:24:46 -0800 Subject: [PATCH 4/6] Update fifoPop in src/fifo.c Co-authored-by: Jim Brunner Signed-off-by: Alina Liu --- src/fifo.c | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/src/fifo.c b/src/fifo.c index 630a1cc6dfe..9051abbe88a 100644 --- a/src/fifo.c +++ b/src/fifo.c @@ -168,7 +168,6 @@ bool fifoPeek(fifo *q, void **item) { bool fifoPop(fifo *q, void **item) { if (q->length == 0) return false; - void *value; if (q->first == q->last) { /* With only 1 block, POP occurs at index 0 and items 1..6 are shifted. @@ -192,7 +191,7 @@ bool fifoPop(fifo *q, void **item) { * Items are shifted left to keep them left-justified in the single block. * This avoids needing to allocate a new block when pushing more items. */ - value = q->last->items[0]; + *item = q->last->items[0]; int lastIdx = q->last->u.last_or_first_idx; /* pointer portion is 0 on last (or only) block */ assert(lastIdx < ITEMS_PER_BLOCK); @@ -210,7 +209,7 @@ bool fifoPop(fifo *q, void **item) { } else { /* With more than 1 block, POP occurs at firstIdx, and firstIdx is incremented. */ int firstIdx = q->first->u.last_or_first_idx & IDX_MASK; - value = q->first->items[firstIdx]; + *item = q->first->items[firstIdx]; if (firstIdx < ITEMS_PER_BLOCK - 1) { /* Just increment the first index to the next slot. */ @@ -225,7 +224,6 @@ bool fifoPop(fifo *q, void **item) { } q->length--; - *item = value; return true; } From 31306c305875aacfe7f59747b15d084d28533f2b Mon Sep 17 00:00:00 2001 From: Alina Liu Date: Wed, 31 Dec 2025 21:59:14 +0000 Subject: [PATCH 5/6] Updated FIFO and mutexQueue Signed-off-by: Alina Liu --- src/bio.c | 2 +- src/fifo.c | 4 +-- src/fifo.h | 6 ++-- src/mutexqueue.c | 4 +-- src/unit/test_fifo.c | 61 +++++++++++++++++++------------------- src/unit/test_mutexqueue.c | 18 +++++------ 6 files changed, 47 insertions(+), 48 deletions(-) diff --git a/src/bio.c b/src/bio.c index 3ee0a53afb3..0db08f60c97 100644 --- a/src/bio.c +++ b/src/bio.c @@ -307,7 +307,7 @@ unsigned long bioPendingJobsOfType(int type) { /* Wait for the job queue of the worker for jobs of specified type to become empty. */ void bioDrainWorker(int type) { while (bioPendingJobsOfType(type) > 0) { - usleep(100); /* Sleep for 1ms and check again*/ + usleep(100); } } diff --git a/src/fifo.c b/src/fifo.c index 9051abbe88a..bdef7dac4f3 100644 --- a/src/fifo.c +++ b/src/fifo.c @@ -202,7 +202,7 @@ bool fifoPop(fifo *q, void **item) { memmove(q->last->items, q->last->items + 1, lastIdx * sizeof(q->last->items[0])); q->last->u.last_or_first_idx--; /* Decrement the last index */ } else { - /* Just finished the only block. Delete it. */ + /* Just finished the only block. Release it. */ zfree(q->last); q->first = q->last = NULL; } @@ -233,7 +233,7 @@ long fifoLength(fifo *q) { } -void fifoDelete(fifo *q) { +void fifoRelease(fifo *q) { if (q->length > 0) { fifoBlock *cur = q->first; while (cur != NULL) { diff --git a/src/fifo.h b/src/fifo.h index 77ee08b67b0..32976035247 100644 --- a/src/fifo.h +++ b/src/fifo.h @@ -46,9 +46,9 @@ bool fifoPop(fifo *q, void **item); /* Return the number of items in the fifo. */ long fifoLength(fifo *q); -/* Delete the fifo. +/* Release the fifo. * NOTE: this does not free items which may be referenced by inserted pointers. */ -void fifoDelete(fifo *q); +void fifoRelease(fifo *q); /* Joins the fifo "other" to the end of "q". "other" becomes empty, but remains valid. * This is an O(1) operation. @@ -68,7 +68,7 @@ void fifoJoin(fifo *q, fifo *other); * is to extract items rather than merge fifos. * Example: fifo *batch = fifoPopAll(source_q); * // batch contains all items, source_q is now empty - * // Process batch, then fifoDelete(batch) when done */ + * // Process batch, then fifoRelease(batch) when done */ fifo *fifoPopAll(fifo *q); #endif diff --git a/src/mutexqueue.c b/src/mutexqueue.c index ce0c4329e2d..37aee3f5f36 100644 --- a/src/mutexqueue.c +++ b/src/mutexqueue.c @@ -44,8 +44,8 @@ void mutexQueueRelease(mutexQueue *theQueue) { pthread_cond_broadcast(&mq->notify_cv); pthread_cond_destroy(&mq->notify_cv); - fifoDelete(mq->priority_fifo); - fifoDelete(mq->normal_fifo); + fifoRelease(mq->priority_fifo); + fifoRelease(mq->normal_fifo); zfree(mq); } diff --git a/src/unit/test_fifo.c b/src/unit/test_fifo.c index fd36e58e68a..254a3062ba7 100644 --- a/src/unit/test_fifo.c +++ b/src/unit/test_fifo.c @@ -6,33 +6,34 @@ #include "../fifo.h" #include "test_help.h" +#include #include #include -static inline void *INT_TO_PTR(intptr_t i) { +static inline void *intToPointer(intptr_t i) { return (void *)i; } -static inline intptr_t PTR_TO_INT(void *p) { +static inline intptr_t pointerToInt(void *p) { return (intptr_t)p; } /* Helper functions */ static void push(fifo *q, intptr_t value) { int len = fifoLength(q); - fifoPush(q, INT_TO_PTR(value)); + fifoPush(q, intToPointer(value)); TEST_EXPECT(fifoLength(q) == len + 1); } static intptr_t popTest(fifo *q, intptr_t expected) { void *peekPtr; TEST_EXPECT(fifoPeek(q, &peekPtr)); - intptr_t peekValue = PTR_TO_INT(peekPtr); + intptr_t peekValue = pointerToInt(peekPtr); TEST_EXPECT(peekValue == expected); int len = fifoLength(q); void *popPtr; TEST_EXPECT(fifoPop(q, &popPtr)); - intptr_t value = PTR_TO_INT(popPtr); + intptr_t value = pointerToInt(popPtr); TEST_EXPECT(fifoLength(q) == len - 1); TEST_EXPECT(value == expected); return value; @@ -48,7 +49,7 @@ int test_fifoEmptyPop(int argc, char *argv[], int flags) { TEST_EXPECT(fifoLength(q) == 0); void *result; TEST_EXPECT(fifoPop(q, &result) == false); - fifoDelete(q); + fifoRelease(q); return 0; } @@ -62,7 +63,7 @@ int test_fifoEmptyPeek(int argc, char *argv[], int flags) { TEST_EXPECT(fifoLength(q) == 0); void *result; TEST_EXPECT(fifoPeek(q, &result) == false); - fifoDelete(q); + fifoRelease(q); return 0; } @@ -77,7 +78,7 @@ int test_fifoSimplePushPop(int argc, char *argv[], int flags) { push(q, 1); popTest(q, 1); TEST_EXPECT(fifoLength(q) == 0); - fifoDelete(q); + fifoRelease(q); return 0; } @@ -94,7 +95,7 @@ int test_fifoTryVariousSizes(int argc, char *argv[], int flags) { for (int value = 1; value <= items; value++) popTest(q, value); TEST_EXPECT(fifoLength(q) == 0); } - fifoDelete(q); + fifoRelease(q); return 0; } @@ -116,7 +117,7 @@ int test_fifoPushPopTest(int argc, char *argv[], int flags) { popTest(q, popVal++); } } - fifoDelete(q); + fifoRelease(q); return 0; } @@ -139,17 +140,17 @@ int test_fifoJoinTest(int argc, char *argv[], int flags) { intptr_t pushValue = 1; intptr_t popQValue = 1; - for (int i = 0; i < qLen; i++) fifoPush(q, INT_TO_PTR(pushValue++)); + for (int i = 0; i < qLen; i++) fifoPush(q, intToPointer(pushValue++)); for (int i = 0; i < qPop; i++) { void *ptr; - TEST_EXPECT(fifoPop(q, &ptr) && PTR_TO_INT(ptr) == popQValue++); + TEST_EXPECT(fifoPop(q, &ptr) && pointerToInt(ptr) == popQValue++); } intptr_t popQ2Value = pushValue; - for (int i = 0; i < q2Len; i++) fifoPush(q2, INT_TO_PTR(pushValue++)); + for (int i = 0; i < q2Len; i++) fifoPush(q2, intToPointer(pushValue++)); for (int i = 0; i < q2Pop; i++) { void *ptr; - TEST_EXPECT(fifoPop(q2, &ptr) && PTR_TO_INT(ptr) == popQ2Value++); + TEST_EXPECT(fifoPop(q2, &ptr) && pointerToInt(ptr) == popQ2Value++); } fifoJoin(q, q2); @@ -162,29 +163,25 @@ int test_fifoJoinTest(int argc, char *argv[], int flags) { for (int i = 0; i < (qLen - qPop); i++) { void *ptr; - TEST_EXPECT(fifoPop(temp, &ptr) && PTR_TO_INT(ptr) == popQValue++); + TEST_EXPECT(fifoPop(temp, &ptr) && pointerToInt(ptr) == popQValue++); } for (int i = 0; i < (q2Len - q2Pop); i++) { void *ptr; - TEST_EXPECT(fifoPop(temp, &ptr) && PTR_TO_INT(ptr) == popQ2Value++); + TEST_EXPECT(fifoPop(temp, &ptr) && pointerToInt(ptr) == popQ2Value++); } TEST_EXPECT(fifoLength(temp) == 0); - fifoDelete(temp); + fifoRelease(temp); } } } } - fifoDelete(q); - fifoDelete(q2); + fifoRelease(q); + fifoRelease(q2); return 0; } -/* Set to 1 to prove that FIFO outperforms ADLIST. This test will exercise both. - * The test will (intentionally) fail, printing the results as failed assertions. */ -#define COMPARE_PERFORMANCE_TO_ADLIST 0 - #include "../adlist.h" #include "../monotonic.h" @@ -193,13 +190,13 @@ const int LIST_ITEMS = 10000; static void exerciseList(void) { list *q = listCreate(); for (intptr_t i = 0; i < LIST_ITEMS; i++) { - listAddNodeTail(q, INT_TO_PTR(i)); + listAddNodeTail(q, intToPointer(i)); } TEST_EXPECT(listLength(q) == (unsigned)LIST_ITEMS); for (intptr_t i = 0; i < LIST_ITEMS; i++) { listNode *node = listFirst(q); listDelNode(q, node); - TEST_EXPECT(listNodeValue(node) == INT_TO_PTR(i)); + TEST_EXPECT(listNodeValue(node) == intToPointer(i)); } TEST_EXPECT(listLength(q) == 0u); listRelease(q); @@ -208,23 +205,25 @@ static void exerciseList(void) { static void exerciseFifo(void) { fifo *q = fifoCreate(); for (intptr_t i = 0; i < LIST_ITEMS; i++) { - fifoPush(q, INT_TO_PTR(i)); + fifoPush(q, intToPointer(i)); } TEST_EXPECT(fifoLength(q) == LIST_ITEMS); for (intptr_t i = 0; i < LIST_ITEMS; i++) { void *ptr; - TEST_EXPECT(fifoPop(q, &ptr) && ptr == INT_TO_PTR(i)); + TEST_EXPECT(fifoPop(q, &ptr) && ptr == intToPointer(i)); } TEST_EXPECT(fifoLength(q) == 0); - fifoDelete(q); + fifoRelease(q); } int test_fifoComparePerformance(int argc, char *argv[], int flags) { - UNUSED(argc); - UNUSED(argv); UNUSED(flags); - if (COMPARE_PERFORMANCE_TO_ADLIST) { + /* To run the performance comparison test, use: + * ./valkey-unit-tests --single test_fifo.c --compare-performance-to-adlist + * This test will exercise both FIFO and ADLIST to compare performance. + * The test will (intentionally) fail, printing the results as failed assertions. */ + if (argc > 3 && !strcasecmp(argv[3], "--compare-performance-to-adlist")) { monotonicInit(); monotime timer; const int iterations = 500; diff --git a/src/unit/test_mutexqueue.c b/src/unit/test_mutexqueue.c index 4d5340b5a54..5feceaff4cb 100644 --- a/src/unit/test_mutexqueue.c +++ b/src/unit/test_mutexqueue.c @@ -16,7 +16,7 @@ static void add(mutexQueue *q, long value) { TEST_EXPECT(mutexQueueLength(q) == len + 1); } -static void pAdd(mutexQueue *q, long value) { +static void priorityAdd(mutexQueue *q, long value) { unsigned long len = mutexQueueLength(q); mutexQueuePushPriority(q, (void *)value); TEST_EXPECT(mutexQueueLength(q) == len + 1); @@ -69,9 +69,9 @@ int test_mutexQueuePriorityOrdering(int argc, char *argv[], int flags) { mutexQueue *q = mutexQueueCreate(); add(q, 10); - pAdd(q, 1); + priorityAdd(q, 1); add(q, 11); - pAdd(q, 2); + priorityAdd(q, 2); popTest(q, 1); popTest(q, 2); popTest(q, 10); @@ -90,9 +90,9 @@ int test_mutexQueueFifoPopAll(int argc, char *argv[], int flags) { mutexQueue *q = mutexQueueCreate(); add(q, 10); - pAdd(q, 1); + priorityAdd(q, 1); add(q, 11); - pAdd(q, 2); + priorityAdd(q, 2); fifo *f = mutexQueuePopAll(q, false); TEST_ASSERT(f != NULL); /* Fatal - can't continue if NULL */ @@ -107,7 +107,7 @@ int test_mutexQueueFifoPopAll(int argc, char *argv[], int flags) { TEST_EXPECT(fifoPop(f, &ptr) && (unsigned long)ptr == 11ul); TEST_EXPECT(fifoLength(f) == 0); - fifoDelete(f); + fifoRelease(f); mutexQueueRelease(q); return 0; } @@ -126,10 +126,10 @@ int test_mutexQueueFifoAddMultiple(int argc, char *argv[], int flags) { fifoPush(f, (void *)3); mutexQueueAddMultiple(q, f); TEST_EXPECT(fifoLength(f) == 0u); - fifoDelete(f); + fifoRelease(f); add(q, 4); - pAdd(q, 0); + priorityAdd(q, 0); popTest(q, 0); popTest(q, 1); popTest(q, 2); @@ -220,7 +220,7 @@ int test_mutexQueueParallelWriters(int argc, char *argv[], int flags) { TEST_EXPECT(mutexQueueLength(q) == (unsigned long)(num_threads * 1000)); fifo *f = mutexQueuePopAll(q, false); - fifoDelete(f); + fifoRelease(f); mutexQueueRelease(q); return 0; } From f0409eb848c18e2896f7dadba5c0c6953a2bff17 Mon Sep 17 00:00:00 2001 From: Alina Liu Date: Wed, 31 Dec 2025 14:48:08 -0800 Subject: [PATCH 6/6] Update and reorder test_files.h Signed-off-by: Alina Liu --- src/unit/test_files.h | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/unit/test_files.h b/src/unit/test_files.h index 4006a5a219e..a2eae5cddcf 100644 --- a/src/unit/test_files.h +++ b/src/unit/test_files.h @@ -24,6 +24,8 @@ int test_entryCreate(int argc, char **argv, int flags); int test_entryUpdate(int argc, char **argv, int flags); int test_entryHasexpiry_entrySetExpiry(int argc, char **argv, int flags); int test_entryIsExpired(int argc, char **argv, int flags); +int test_entryMemUsage_entrySetExpiry_entryUpdate(int argc, char **argv, int flags); +int test_entryStringRef(int argc, char **argv, int flags); int test_fifoEmptyPop(int argc, char *argv[], int flags); int test_fifoEmptyPeek(int argc, char *argv[], int flags); int test_fifoSimplePushPop(int argc, char *argv[], int flags); @@ -31,8 +33,6 @@ int test_fifoTryVariousSizes(int argc, char *argv[], int flags); int test_fifoPushPopTest(int argc, char *argv[], int flags); int test_fifoJoinTest(int argc, char *argv[], int flags); int test_fifoComparePerformance(int argc, char *argv[], int flags); -int test_entryMemUsage_entrySetExpiry_entryUpdate(int argc, char **argv, int flags); -int test_entryStringRef(int argc, char **argv, int flags); int test_cursor(int argc, char **argv, int flags); int test_set_hash_function_seed(int argc, char **argv, int flags); int test_add_find_delete(int argc, char **argv, int flags); @@ -284,8 +284,8 @@ unitTest __test_crc64_c[] = {{"test_crc64", test_crc64}, {NULL, NULL}}; unitTest __test_crc64combine_c[] = {{"test_crc64combine", test_crc64combine}, {NULL, NULL}}; unitTest __test_dict_c[] = {{"test_dictCreate", test_dictCreate}, {"test_dictAdd16Keys", test_dictAdd16Keys}, {"test_dictDisableResize", test_dictDisableResize}, {"test_dictAddOneKeyTriggerResize", test_dictAddOneKeyTriggerResize}, {"test_dictDeleteKeys", test_dictDeleteKeys}, {"test_dictDeleteOneKeyTriggerResize", test_dictDeleteOneKeyTriggerResize}, {"test_dictEmptyDirAdd128Keys", test_dictEmptyDirAdd128Keys}, {"test_dictDisableResizeReduceTo3", test_dictDisableResizeReduceTo3}, {"test_dictDeleteOneKeyTriggerResizeAgain", test_dictDeleteOneKeyTriggerResizeAgain}, {"test_dictBenchmark", test_dictBenchmark}, {NULL, NULL}}; unitTest __test_endianconv_c[] = {{"test_endianconv", test_endianconv}, {NULL, NULL}}; -unitTest __test_fifo_c[] = {{"test_fifoEmptyPop", test_fifoEmptyPop}, {"test_fifoEmptyPeek", test_fifoEmptyPeek}, {"test_fifoSimplePushPop", test_fifoSimplePushPop}, {"test_fifoTryVariousSizes", test_fifoTryVariousSizes}, {"test_fifoPushPopTest", test_fifoPushPopTest}, {"test_fifoJoinTest", test_fifoJoinTest}, {"test_fifoComparePerformance", test_fifoComparePerformance}, {NULL, NULL}}; unitTest __test_entry_c[] = {{"test_entryCreate", test_entryCreate}, {"test_entryUpdate", test_entryUpdate}, {"test_entryHasexpiry_entrySetExpiry", test_entryHasexpiry_entrySetExpiry}, {"test_entryIsExpired", test_entryIsExpired}, {"test_entryMemUsage_entrySetExpiry_entryUpdate", test_entryMemUsage_entrySetExpiry_entryUpdate}, {"test_entryStringRef", test_entryStringRef}, {NULL, NULL}}; +unitTest __test_fifo_c[] = {{"test_fifoEmptyPop", test_fifoEmptyPop}, {"test_fifoEmptyPeek", test_fifoEmptyPeek}, {"test_fifoSimplePushPop", test_fifoSimplePushPop}, {"test_fifoTryVariousSizes", test_fifoTryVariousSizes}, {"test_fifoPushPopTest", test_fifoPushPopTest}, {"test_fifoJoinTest", test_fifoJoinTest}, {"test_fifoComparePerformance", test_fifoComparePerformance}, {NULL, NULL}}; unitTest __test_hashtable_c[] = {{"test_cursor", test_cursor}, {"test_set_hash_function_seed", test_set_hash_function_seed}, {"test_add_find_delete", test_add_find_delete}, {"test_add_find_delete_avoid_resize", test_add_find_delete_avoid_resize}, {"test_instant_rehashing", test_instant_rehashing}, {"test_bucket_chain_length", test_bucket_chain_length}, {"test_two_phase_insert_and_pop", test_two_phase_insert_and_pop}, {"test_replace_reallocated_entry", test_replace_reallocated_entry}, {"test_incremental_find", test_incremental_find}, {"test_scan", test_scan}, {"test_iterator", test_iterator}, {"test_safe_iterator", test_safe_iterator}, {"test_compact_bucket_chain", test_compact_bucket_chain}, {"test_random_entry", test_random_entry}, {"test_random_entry_with_long_chain", test_random_entry_with_long_chain}, {"test_random_entry_sparse_table", test_random_entry_sparse_table}, {"test_safe_iterator_invalidation", test_safe_iterator_invalidation}, {"test_safe_iterator_empty_no_invalidation", test_safe_iterator_empty_no_invalidation}, {"test_safe_iterator_reset_invalidation", test_safe_iterator_reset_invalidation}, {"test_safe_iterator_reset_untracking", test_safe_iterator_reset_untracking}, {"test_safe_iterator_pause_resume_tracking", test_safe_iterator_pause_resume_tracking}, {"test_null_hashtable_iterator", test_null_hashtable_iterator}, {"test_hashtable_retarget_iterator", test_hashtable_retarget_iterator}, {NULL, NULL}}; unitTest __test_intset_c[] = {{"test_intsetValueEncodings", test_intsetValueEncodings}, {"test_intsetBasicAdding", test_intsetBasicAdding}, {"test_intsetLargeNumberRandomAdd", test_intsetLargeNumberRandomAdd}, {"test_intsetUpgradeFromint16Toint32", test_intsetUpgradeFromint16Toint32}, {"test_intsetUpgradeFromint16Toint64", test_intsetUpgradeFromint16Toint64}, {"test_intsetUpgradeFromint32Toint64", test_intsetUpgradeFromint32Toint64}, {"test_intsetStressLookups", test_intsetStressLookups}, {"test_intsetStressAddDelete", test_intsetStressAddDelete}, {NULL, NULL}}; unitTest __test_kvstore_c[] = {{"test_kvstoreAdd16Keys", test_kvstoreAdd16Keys}, {"test_kvstoreIteratorRemoveAllKeysNoDeleteEmptyHashtable", test_kvstoreIteratorRemoveAllKeysNoDeleteEmptyHashtable}, {"test_kvstoreIteratorRemoveAllKeysDeleteEmptyHashtable", test_kvstoreIteratorRemoveAllKeysDeleteEmptyHashtable}, {"test_kvstoreHashtableIteratorRemoveAllKeysNoDeleteEmptyHashtable", test_kvstoreHashtableIteratorRemoveAllKeysNoDeleteEmptyHashtable}, {"test_kvstoreHashtableIteratorRemoveAllKeysDeleteEmptyHashtable", test_kvstoreHashtableIteratorRemoveAllKeysDeleteEmptyHashtable}, {"test_kvstoreHashtableExpand", test_kvstoreHashtableExpand}, {NULL, NULL}};