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

Code before:
                atomic_store_explicit(&server.fsynced_reploff_pending, job->fd_args.offset, memory_order_relaxed);
            }

            if (job->fd_args.need_reclaim_cache) {
                if (reclaimFilePageCache(job->fd_args.fd, 0, 0) == -1) {
                    serverLog(LL_NOTICE, "Unable to reclaim page cache: %s", strerror(errno));
                }
            }
            if (job_type == BIO_CLOSE_AOF) close(job->fd_args.fd);
        } else if (job_type == BIO_LAZY_FREE) {
            job->free_args.free_fn(job->free_args.free_args);
        } else if (job_type == BIO_RDB_SAVE) {
            replicaReceiveRDBFromPrimaryToDisk(job->save_to_disk_args.conn, job->save_to_disk_args.is_dual_channel);
        } else if (job_type == BIO_TLS_RELOAD) {
#if defined(USE_OPENSSL) && USE_OPENSSL == 1 /* BUILD_YES */
            tlsConfigureAsync();
#else
            serverPanic("BIO_TLS_RELOAD job type requires built-in TLS (BUILD_TLS=yes).");
#endif
        } else {
            serverPanic("Wrong job type in bioProcessBackgroundJobs().");
        }
        /* Remove the exact processed job: priority inserts may have changed the global queue head. */
        void *removed_job = NULL;
        serverAssert(mutexQueueRemoveExpectedHead(bwd->bio_jobs, job, &removed_job));
        serverAssert(removed_job == job);
        zfree(job);
        atomic_fetch_sub(&bio_jobs_counter[job_type], 1);
    }
}

/* Return the number of pending jobs of the specified type. */
unsigned long bioPendingJobsOfType(int type) {
    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 type) {
    while (bioPendingJobsOfType(type) > 0) {
        usleep(100);
    }
}

/* Kill the running bio threads in an unclean way. This function should be
 * used only when it's critical to stop the threads for some reason.
 * Currently the server does this only on crash (for instance on SIGSEGV) in order
 * to perform a fast memory check without other threads messing with memory. */
void bioKillThreads(void) {
    int err;

    for (bio_worker_data *bwd = bio_workers; bwd != bio_worker_end; ++bwd) {

Review comment (refactor):
```suggestion
        void *removed_job = mutexQueuePeek(bwd->bio_jobs, job, false);
        serverAssert(removed_job == job);
```
Simplify

Code after:
                atomic_store_explicit(&server.fsynced_reploff_pending, job->fd_args.offset, memory_order_relaxed);
            }

            if (job->fd_args.need_reclaim_cache) {
                if (reclaimFilePageCache(job->fd_args.fd, 0, 0) == -1) {
                    serverLog(LL_NOTICE, "Unable to reclaim page cache: %s", strerror(errno));
                }
            }
            if (job_type == BIO_CLOSE_AOF) close(job->fd_args.fd);
        } else if (job_type == BIO_LAZY_FREE) {
            job->free_args.free_fn(job->free_args.free_args);
        } else if (job_type == BIO_RDB_SAVE) {
            replicaReceiveRDBFromPrimaryToDisk(job->save_to_disk_args.conn, job->save_to_disk_args.is_dual_channel);
        } else if (job_type == BIO_TLS_RELOAD) {
#if defined(USE_OPENSSL) && USE_OPENSSL == 1 /* BUILD_YES */
            tlsConfigureAsync();
#else
            serverPanic("BIO_TLS_RELOAD job type requires built-in TLS (BUILD_TLS=yes).");
#endif
        } else {
            serverPanic("Wrong job type in bioProcessBackgroundJobs().");
        }
        void *removed_job = mutexQueuePop(bwd->bio_jobs, false);
        serverAssert(removed_job == job);
        zfree(job);
        atomic_fetch_sub(&bio_jobs_counter[job_type], 1);
    }
}

/* Return the number of pending jobs of the specified type. */
unsigned long bioPendingJobsOfType(int type) {
    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 type) {
    while (bioPendingJobsOfType(type) > 0) {
        usleep(100);
    }
}

/* Kill the running bio threads in an unclean way. This function should be
 * used only when it's critical to stop the threads for some reason.
 * Currently the server does this only on crash (for instance on SIGSEGV) in order
 * to perform a fast memory check without other threads messing with memory. */
void bioKillThreads(void) {
    int err;

    for (bio_worker_data *bwd = bio_workers; bwd != bio_worker_end; ++bwd) {
        if (pthread_equal(bwd->bio_thread_id, pthread_self())) continue;
        if (bwd->bio_thread_id && pthread_cancel(bwd->bio_thread_id) == 0) {