Merge branch 'dev' into feat-add-gc-stats-to-chart

Author: GigaCronos

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,23 @@ void mainNetworkThreadWantExit(void) {
 	gDoTerminate.store(true);
 }
 
-int mainNetworkThreadCanExit(void) {
+bool networkThreadsCanExit() {
 	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 mainNetworkThreadCanExit() {
+	// Preserve this order:
+	// networkThreadsCanExit() must be checked before masterconn_canexit().
+	// If masterconn_canexit() is checked first, a network worker may still be processing an
+	// endChunkLock, which could add statuses to the masterconn job pool after masterconn_canexit()
+	// returns true. This could lead to the chunkserver exiting prematurely while holding chunk
+	// locks that have not been replied to the master.
+	return networkThreadsCanExit() && masterconn_canexit();
 }
 
 void mainNetworkThreadTerm(void) {

src/chunkserver/network_worker_thread.cc

@@ -122,10 +122,15 @@ void NetworkWorkerThread::operator()() {
 }
 
 bool NetworkWorkerThread::updateAndCheckTerminationStatus() {
+	// Don't even check the rest if we already know we can terminate.
+	if (canTerminate_.load()) {
+		return true;
+	}
+
 	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;

tests/test_suites/SingleMachineTests/test_helgrind_basic.sh

@@ -53,3 +53,5 @@ generateFiles
 # finishes validating files and sometimes it gets stuck
 # drop_caches
 # validateFiles
+
+saunafs_chunkserver_daemon 0 stop