Release of fifo and mutexQueue to OSS with bio refactored

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

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

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 <stdatomic.h>
 
 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