fix: avoid stopping CS with pending replies

It was noticed that could happen that CS gracefully stops without
sending all the expected responses. Those missing responses could cause
the master to invalidate some chunk parts when it was expected
everything to be ok.

The cause for such behavior comes from the span of time from the
instant in which a worker thread (from a jobPool) gets a new job and
the instant the job is processed. In that period of time, the job is
not counted when calling getJobCount, so the jobPool may appear empty
but there were still some jobs in need of processing. This was
happening the in jobPools attending requests from clients, and a
similar issue happened to the jobPools attending requests from the
master.

The solution targets improving the accuracy of the stop conditions
considering the previous issue.

Signed-off-by: Dave <dave@leil.io>

Author: dmga44

src/chunkserver/bgjobs.cc

@@ -154,6 +154,7 @@ uint32_t JobPool::addJob(ChunkOperation operation, JobCallback callback, void *e
 	        : 1;
 	jobsQueue->put(jobId, operation, reinterpret_cast<uint8_t *>(listenerInfo.jobHash[jobId].get()),
 	               1, priority);
+	unprocessedJobs_.fetch_add(1, std::memory_order_relaxed);
 	return jobId;
 }
 
@@ -180,6 +181,20 @@ uint32_t JobPool::addLockJob(JobCallback callback, void *extra, uint32_t listene
 	return jobId;
 }
 
+bool JobPool::allJobsProcessed() const {
+	return unprocessedJobs_.load(std::memory_order_relaxed) == 0;
+}
+
+bool JobPool::isEmpty() {
+	if (jobsQueue->elements() > 0) { return false; }
+
+	for (auto &listenerInfo : listenerInfos_) {
+		std::lock_guard lock(listenerInfo.notifierMutex);
+		if (!listenerInfo.statusQueue.empty()) { return false; }
+	}
+	return true;
+}
+
 uint32_t JobPool::getJobCount() const {
 	TRACETHIS();
 	return jobsQueue->elements();
@@ -264,6 +279,7 @@ void JobPool::processCompletedJobs(uint32_t listenerId) {
 			auto callback = jobIterator->second->callback;
 			if (callback) { callback(status, jobIterator->second->extra); }
 			listenerInfo.jobHash.erase(jobIterator);
+			unprocessedJobs_.fetch_sub(1, std::memory_order_relaxed);
 		}
 	}
 }

src/chunkserver/bgjobs.h

@@ -139,6 +139,17 @@ class JobPool {
 	/// @return The ID of the added lock job.
 	uint32_t addLockJob(JobCallback callback, void *extra, uint32_t listenerId = 0);
 
+	/// @brief Returns whether all jobs in the JobPool have been processed by the worker threads.
+	/// This is a very accurate way to check if there are pending jobs in the JobPool, as it counts
+	/// the number of jobs that have been added but not yet passed by processCompletedJobs. Must
+	/// not be used for the masterConn's jobPool due to the special behavior of the lock jobs.
+	bool allJobsProcessed() const;
+
+	/// @brief Checks if the JobPool has no jobs and no status to be sent.
+	/// This function is a lighter version of allJobsProcessed that can be used for the masterConn's
+	/// jobPool to check if it is idle.
+	bool isEmpty();
+
 	/// @brief Gets the number of jobs in the JobPool.
 	uint32_t getJobCount() const;
 
@@ -303,6 +314,10 @@ class JobPool {
 	uint8_t workers;                                   /// Number of worker threads in the pool.
 	std::vector<std::thread> workerThreads;            /// Vector of worker threads.
 	std::unique_ptr<ProducerConsumerQueue> jobsQueue;  /// Queue for jobs.
+	/// Counter for unprocessed jobs, i.e jobs that have been added to the JobPool but have not yet
+	/// been passed by processCompletedJobs and had their callbacks called. This is used to make
+	/// sure the JobPool is truly empty when stopping the chunkserver.
+	std::atomic<uint32_t> unprocessedJobs_{0};
 };
 
 /// @brief Adds an open job to the JobPool.

src/chunkserver/masterconn.cc

@@ -144,12 +144,10 @@ JobPool* masterconn_get_job_pool() {
 	return gJobPool.get();
 }
 
-int masterconn_canexit(void) {
-	if (gJobPool->getJobCount() == 0 && gReplicationJobPool->getJobCount() == 0 &&
-	    gMasterConnSingleton->isOutputQueueEmpty()) {
-		return 1;
-	}
-	return 0;
+bool masterconn_canexit() {
+	return gMasterConnSingleton->mode() != ConnectionMode::CONNECTED ||
+	       (gJobPool->isEmpty() && gReplicationJobPool->isEmpty() &&
+	        gMasterConnSingleton->isOutputQueueEmpty());
 }
 
 void masterconn_term(void) {
@@ -321,7 +319,6 @@ int masterconn_init(void) {
 	    eventloop_timeregister(TIMEMODE_RUN_LATE, reconnectionDelay,
 	                           rnd_ranged<uint32_t>(reconnectionDelay), masterconn_reconnect);
 
-	eventloop_canexitregister(masterconn_canexit);
 	eventloop_destructregister(masterconn_term);
 	eventloop_pollregister(masterconn_desc, masterconn_serve);
 	eventloop_reloadregister(masterconn_reload);

src/chunkserver/masterconn.h

@@ -29,3 +29,4 @@ void masterconn_stats(uint64_t *bin, uint64_t *bout, uint32_t *maxjobscnt);
 int masterconn_init(void);
 int masterconn_init_threads(void);
 JobPool* masterconn_get_job_pool();
+bool masterconn_canexit();

src/chunkserver/network_main_thread.cc

@@ -35,6 +35,7 @@
 #include "chunkserver/chunk_replicator.h"
 #include "chunkserver/g_limiters.h"
 #include "chunkserver/hdd_readahead.h"
+#include "chunkserver/masterconn.h"
 #include "chunkserver/network_main_thread.h"
 #include "chunkserver/network_worker_thread.h"
 #include "common/cwrap.h"
@@ -210,15 +211,18 @@ void mainNetworkThreadWantExit(void) {
 	gDoTerminate.store(true);
 }
 
-int mainNetworkThreadCanExit(void) {
+bool mainNetworkThreadCanExit() {
 	TRACETHIS();
 	bool allTerminated = true;
 	for (auto &threadObject : networkThreadObjects) {
-		if (!threadObject.updateAndCheckTerminationStatus()) {
-			allTerminated = false;
-		}
+		if (!threadObject.updateAndCheckTerminationStatus()) { allTerminated = false; }
 	}
-	return allTerminated ? 1 : 0;
+	return allTerminated;
+}
+
+int canExit() {
+	// This order must be preserved
+	return mainNetworkThreadCanExit() && masterconn_canexit();
 }
 
 void mainNetworkThreadTerm(void) {
@@ -313,7 +317,7 @@ int mainNetworkThreadInit(void) {
 
 	eventloop_reloadregister(mainNetworkThreadReload);
 	eventloop_wantexitregister(mainNetworkThreadWantExit);
-	eventloop_canexitregister(mainNetworkThreadCanExit);
+	eventloop_canexitregister(canExit);
 	eventloop_destructregister(mainNetworkThreadTerm);
 	eventloop_pollregister(mainNetworkThreadDesc, mainNetworkThreadServe);
 

src/chunkserver/network_worker_thread.cc

@@ -125,7 +125,7 @@ bool NetworkWorkerThread::updateAndCheckTerminationStatus() {
 	std::lock_guard lock(csservheadLock);
 	bool canTerminate =
 	    doTerminate.load() && ((csservEntries.empty() &&
-	                            (bgJobPool_.get() == nullptr || bgJobPool_->getJobCount() == 0)) ||
+	                            (bgJobPool_.get() == nullptr || bgJobPool_->allJobsProcessed())) ||
 	                           terminationTimer_.elapsed_ms() > kNWForcefulTerminationTimeout_ms);
 	canTerminate_.store(canTerminate);
 	return canTerminate;