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..0db08f60c97 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. @@ -63,6 +69,7 @@ #include "server.h" #include "connection.h" #include "bio.h" +#include "mutexqueue.h" #include static unsigned int bio_job_to_worker[] = { @@ -76,9 +83,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 +101,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 +148,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 +173,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 +229,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 +237,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 +248,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 +295,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(100); } - 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..bdef7dac4f3 --- /dev/null +++ b/src/fifo.c @@ -0,0 +1,353 @@ +/* + * Copyright (c) Valkey Contributors + * All rights reserved. + * SPDX-License-Identifier: BSD-3-Clause + */ + +/* FIFO - A high-performance First-In, First-Out queue implementation */ + +#include +#include +#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 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; /* Number of items */ + fifoBlock *first; + fifoBlock *last; +}; + + +fifo *fifoCreate(void) { + fifo *q = zmalloc(sizeof(fifo)); + q->length = 0; + q->first = q->last = NULL; + return q; +} + + +void fifoPush(fifo *q, void *ptr) { + if (q->first == NULL) { + /* 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 */ + 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++; +} + + +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; + *item = q->first->items[firstIdx]; + return true; +} + + +bool fifoPop(fifo *q, void **item) { + if (q->length == 0) return false; + + if (q->first == q->last) { + /* 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. + */ + *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. + * 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. Release 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 true; +} + + +long fifoLength(fifo *q) { + return q->length; +} + + +void fifoRelease(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); +} + + +/* Overwrites target from source. */ +static void overwriteFifoContents(fifo *target, fifo *source) { + target->length = source->length; + target->first = source->first; + target->last = source->last; + source->length = 0; + source->first = source->last = NULL; +} + + +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. + * + * 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. */ + 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. */ + 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 to the right in the last block if it is partially full */ + int shift = (ITEMS_PER_BLOCK - 1) - lastIdx; + if (shift > 0) { + 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 */ + 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; +} + + +fifo *fifoPopAll(fifo *q) { + fifo *newQ = zmalloc(sizeof(fifo)); + overwriteFifoContents(newQ, q); + return newQ; +} diff --git a/src/fifo.h b/src/fifo.h new file mode 100644 index 00000000000..32976035247 --- /dev/null +++ b/src/fifo.h @@ -0,0 +1,74 @@ +/* + * Copyright (c) Valkey Contributors + * All rights reserved. + * SPDX-License-Identifier: BSD-3-Clause + */ + +/* 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. + * + * 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. */ +fifo *fifoCreate(void); + +/* Push an item onto the end of the fifo. */ +void fifoPush(fifo *q, void *ptr); + +/* 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 fifo. + * 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. */ +long fifoLength(fifo *q); + +/* Release the fifo. + * NOTE: this does not free items which may be referenced by inserted pointers. */ +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. + * + * 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); + +/* 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 fifoRelease(batch) when done */ +fifo *fifoPopAll(fifo *q); + +#endif diff --git a/src/mutexqueue.c b/src/mutexqueue.c new file mode 100644 index 00000000000..37aee3f5f36 --- /dev/null +++ b/src/mutexqueue.c @@ -0,0 +1,159 @@ +/* + * 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 */ +}; + + +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; +} + + +/* 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 mutexQueue. */ +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); + + fifoRelease(mq->priority_fifo); + fifoRelease(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); +} + + +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; +} + + +void mutexQueuePushPriority(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); +} + + +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); +} + + +/* Note: 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); +} + + +/* 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; + + 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) { + fifoPop(mq->priority_fifo, &value); + } else if (fifoLength(mq->normal_fifo) > 0) { + fifoPop(mq->normal_fifo, &value); + } + + pthread_mutex_unlock(&mq->mutex); + return value; +} + + +/* 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; + + 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..975ea66b0a7 --- /dev/null +++ b/src/mutexqueue.h @@ -0,0 +1,66 @@ +/* + * 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 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 + * + * 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 +#define __MUTEXQUEUE_H + +#include +#include "fifo.h" + +/* The mutexQueue is an opaque structure. */ +typedef struct mutexQueue mutexQueue; + +/* Create an empty mutexQueue. */ +mutexQueue *mutexQueueCreate(void); + +/* Release an empty mutexQueue. */ +void mutexQueueRelease(mutexQueue *theQueue); + +/* Number of items in the mutexQueue. */ +unsigned long mutexQueueLength(mutexQueue *theQueue); + +/* 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 mutexQueue. */ +void mutexQueueAdd(mutexQueue *theQueue, void *value); + +/* Insert multiple items (from a fifo) to the end of the mutexQueue. */ +void mutexQueueAddMultiple(mutexQueue *theQueue, fifo *valueFifo); + +/* Retrieves the first item off the mutexQueue (or NULL if mutexQueue is empty). */ +void *mutexQueuePop(mutexQueue *theQueue, bool blocking); + +/* 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 new file mode 100644 index 00000000000..254a3062ba7 --- /dev/null +++ b/src/unit/test_fifo.c @@ -0,0 +1,247 @@ +/* + * Copyright (c) Valkey Contributors + * All rights reserved. + * SPDX-License-Identifier: BSD-3-Clause + */ + +#include "../fifo.h" +#include "test_help.h" +#include +#include +#include + +static inline void *intToPointer(intptr_t i) { + return (void *)i; +} + +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, 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 = pointerToInt(peekPtr); + TEST_EXPECT(peekValue == expected); + int len = fifoLength(q); + void *popPtr; + TEST_EXPECT(fifoPop(q, &popPtr)); + intptr_t value = pointerToInt(popPtr); + TEST_EXPECT(fifoLength(q) == len - 1); + TEST_EXPECT(value == expected); + return value; +} + +/* 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); + + fifo *q = fifoCreate(); + TEST_EXPECT(fifoLength(q) == 0); + void *result; + TEST_EXPECT(fifoPop(q, &result) == false); + fifoRelease(q); + return 0; +} + +/* 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); + + fifo *q = fifoCreate(); + TEST_EXPECT(fifoLength(q) == 0); + void *result; + TEST_EXPECT(fifoPeek(q, &result) == false); + fifoRelease(q); + 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); + fifoRelease(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); + } + fifoRelease(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++); + } + } + fifoRelease(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 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++) { + 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, intToPointer(pushValue++)); + for (int i = 0; i < qPop; i++) { + void *ptr; + TEST_EXPECT(fifoPop(q, &ptr) && pointerToInt(ptr) == popQValue++); + } + + intptr_t popQ2Value = 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) && pointerToInt(ptr) == 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++) { + void *ptr; + TEST_EXPECT(fifoPop(temp, &ptr) && pointerToInt(ptr) == popQValue++); + } + for (int i = 0; i < (q2Len - q2Pop); i++) { + void *ptr; + TEST_EXPECT(fifoPop(temp, &ptr) && pointerToInt(ptr) == popQ2Value++); + } + TEST_EXPECT(fifoLength(temp) == 0); + + fifoRelease(temp); + } + } + } + } + + fifoRelease(q); + fifoRelease(q2); + return 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, 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) == intToPointer(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, 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 == intToPointer(i)); + } + TEST_EXPECT(fifoLength(q) == 0); + fifoRelease(q); +} + +int test_fifoComparePerformance(int argc, char *argv[], int flags) { + UNUSED(flags); + + /* 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; + + 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); + + 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 ab75aa10d45..a2eae5cddcf 100644 --- a/src/unit/test_files.h +++ b/src/unit/test_files.h @@ -26,6 +26,13 @@ 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); +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); @@ -115,6 +122,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); @@ -269,10 +285,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_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}}; 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}}; @@ -298,10 +316,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..5feceaff4cb --- /dev/null +++ b/src/unit/test_mutexqueue.c @@ -0,0 +1,294 @@ +/* + * 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 priorityAdd(mutexQueue *q, long value) { + unsigned long len = mutexQueueLength(q); + mutexQueuePushPriority(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); + priorityAdd(q, 1); + add(q, 11); + priorityAdd(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); + priorityAdd(q, 1); + add(q, 11); + priorityAdd(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); + + 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); + + fifoRelease(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); + fifoRelease(f); + + add(q, 4); + priorityAdd(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); + fifoRelease(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; +}