fix: Make a threadpool for trash-remover threads.

Add a ProducerConsumerQueue and ThreadPool
with up to 5 runners, for executing the per disk deletion jobs.

Signed-off-by: GigaCronos <jorge.cabrera@leil.io>

Author: GigaCronos

src/chunkserver/chunkserver-common/chunk_trash_manager.cc

@@ -84,6 +84,24 @@ int ChunkTrashManager::init(const std::string &diskPath) {
 	return impl->init(diskPath);
 }
 
+void ChunkTrashManager::terminate(){
+    if (!isEnabled) { return; }
+
+	// Protect against concurrent access
+	ImplementationPtr impl;
+	{
+		std::lock_guard<std::mutex> lock(implMutex);
+		impl = getImpl();
+	}
+	if (!impl) {
+		safs::log_error_code(SAUNAFS_ERROR_EINVAL,
+		                     "ChunkTrashManager implementation not initialized");
+		return;
+	}
+	impl->terminate();
+
+}
+
 void ChunkTrashManager::collectGarbage() {
 	if (!isEnabled) { return; }
 

src/chunkserver/chunkserver-common/chunk_trash_manager.h

@@ -34,6 +34,8 @@ class IChunkTrashManagerImpl {
 
 	virtual int init(const std::string &) = 0;
 
+	virtual void terminate() = 0;		
+
 	virtual void collectGarbage() = 0;
 
 	virtual void reloadConfig() = 0;
@@ -73,6 +75,11 @@ class ChunkTrashManager {
 	 */
 	static int init(const std::string &diskPath);
 
+	/**
+	 * @brief Join the background threads.
+	 */
+	static void terminate();
+
 	/**
 	 * @brief Moves a file to the trash directory.
 	 *

src/chunkserver/chunkserver-common/chunk_trash_manager_impl.cc

@@ -20,14 +20,17 @@
 
 #include <sys/statvfs.h>
 #include <algorithm>
+#include <atomic>
 #include <cstdint>
 #include <ctime>
+#include <utility>
 
 #include "chunkserver-common/chunk_trash_manager_impl.h"
 #include "config/cfg.h"
 #include "errors/saunafs_error_codes.h"
 #include "global_shared_resources.h"
 #include "hdd_stats.h"
+#include "hdd_utils.h"
 #include "slogger/slogger.h"
 
 namespace fs = std::filesystem;
@@ -47,6 +50,12 @@ uint64_t ChunkTrashManagerImpl::previousBytesWritePerDisk = 1024 * 1024; //1MiB
 const std::string ChunkTrashManagerImpl::kTrashGuardString =
     std::string("/") + ChunkTrashManager::kTrashDirname + "/";
 
+std::vector<std::thread> ChunkTrashManagerImpl::removeFromTrashThreads{};
+std::unique_ptr<ProducerConsumerQueue> ChunkTrashManagerImpl::removeFromTrashJobQueue =
+    std::make_unique<ProducerConsumerQueue>();
+
+std::atomic<uint32_t> ChunkTrashManagerImpl::NotIdleThreadCount{0};
+
 void ChunkTrashManagerImpl::reloadConfig() {
 	availableThresholdGB =
 	    cfg_get("CHUNK_TRASH_FREE_SPACE_THRESHOLD_GB", kDefaultAvailableThresholdGB);
@@ -167,19 +176,50 @@ int ChunkTrashManagerImpl::moveToTrash(const fs::path &filePath, const fs::path
 
 void ChunkTrashManagerImpl::removeTrashFiles(
     const ChunkTrashIndex::TrashIndexDiskEntries &filesToRemove) const {
-	std::vector<std::jthread> parallelRemovers;
-
 	for (const auto &[diskPath, fileEntries] : filesToRemove) {
-		parallelRemovers.emplace_back(removeTrashFilesFromDisk, fileEntries, diskPath);
+		removeFromTrashJobQueue->put(
+		    0, 1,
+		    reinterpret_cast<uint8_t *>(
+		        new std::pair<ChunkTrashIndex::TrashIndexFileEntries, std::string>(fileEntries,
+		                                                                           diskPath)),
+		    1);
+	}
+
+	while(NotIdleThreadCount && !removeFromTrashJobQueue->isEmpty()){
+
 	}
 }
 
-void ChunkTrashManagerImpl::removeTrashFilesFromDisk(
-    const ChunkTrashIndex::TrashIndexFileEntries &filesToRemove, const std::string &diskPath) {
-	for (const auto &fileEntry : filesToRemove) {
-		if (removeFileFromTrash(fileEntry.second) != SAUNAFS_STATUS_OK) { continue; }
+void ChunkTrashManagerImpl::removeTrashFilesFromDiskThread(uint8_t workerId) {
+	std::string threadName ="removeTrashFilesFromDisk_worker_" + std::to_string(workerId);
+	pthread_setname_np(pthread_self(), threadName.c_str());
+	
+	uint32_t jobId;
+	uint32_t operation;
+	uint8_t *jobPtrArg;
+
+	while (true) {
+		removeFromTrashJobQueue->get(&jobId, &operation, &jobPtrArg, nullptr);
+		
+		if(operation == 0){
+			break;
+		}
+
+		NotIdleThreadCount ++;
+		auto tempTuple = reinterpret_cast<std::pair<ChunkTrashIndex::TrashIndexFileEntries, std::string> *>(jobPtrArg);
+
+		ChunkTrashIndex::TrashIndexFileEntries filesToRemove = tempTuple->first;
+		std::string diskPath = tempTuple->second;
+
+		for (const auto &fileEntry : filesToRemove) {
+			if (removeFileFromTrash(fileEntry.second) != SAUNAFS_STATUS_OK) { continue; }
 			HddStats::gStatsOperationsGCPurge++;
-		getTrashIndex().remove(fileEntry.first, fileEntry.second, diskPath);
+			getTrashIndex().remove(fileEntry.first, fileEntry.second, diskPath);
+		}
+
+		delete tempTuple;
+
+		NotIdleThreadCount --;
 	}
 }
 
@@ -225,9 +265,30 @@ int ChunkTrashManagerImpl::init(const std::string &diskPath) {
 		}
 	}
 
+	if (ChunkTrashManagerImpl::removeFromTrashThreads.size() < 5) {
+		ChunkTrashManagerImpl::removeFromTrashThreads.emplace_back(
+		    &ChunkTrashManagerImpl::removeTrashFilesFromDiskThread,
+		    uint8_t(ChunkTrashManagerImpl::removeFromTrashThreads.size()));
+	}
+
 	return SAUNAFS_STATUS_OK;
 }
 
+void ChunkTrashManagerImpl::terminate() {
+	for(uint8_t i=0; i < ChunkTrashManagerImpl::removeFromTrashThreads.size(); i++){
+		ChunkTrashManagerImpl::removeFromTrashJobQueue->put(0, 0, nullptr, 1);
+	}
+	
+	
+	for (auto &thread : ChunkTrashManagerImpl::removeFromTrashThreads) {
+		if (thread.joinable()) { thread.join(); }
+	}
+	
+	ChunkTrashManagerImpl::removeFromTrashThreads.clear();
+
+}
+
+
 bool ChunkTrashManagerImpl::isValidTimestampFormat(const std::string &timestamp) {
 	return timestamp.size() == kTimeStampLength && std::ranges::all_of(timestamp, ::isdigit);
 }

src/chunkserver/chunkserver-common/chunk_trash_manager_impl.h

@@ -20,10 +20,14 @@
 
 #include "common/platform.h"
 
+#include <cstdint>
 #include <string>
+#include <thread>
+
 
 #include "chunkserver-common/chunk_trash_index.h"
 #include "chunkserver-common/chunk_trash_manager.h"
+#include "common/pcqueue.h"
 #include "errors/saunafs_error_codes.h"
 
 class ChunkTrashManagerImpl : public IChunkTrashManagerImpl {
@@ -49,6 +53,9 @@ class ChunkTrashManagerImpl : public IChunkTrashManagerImpl {
 	 */
 	int init(const std::string &diskPath) override;
 
+
+	void terminate() override;	
+
 	/**
 	 * @brief Moves a specified file to the trash directory.
 	 * @param filePath The path of the file to be moved.
@@ -85,8 +92,7 @@ class ChunkTrashManagerImpl : public IChunkTrashManagerImpl {
 	 * @param filesToRemove The list of files to be permanently deleted.
 	 * @param diskPath The path of the disk of the files.
 	 */
-	static void removeTrashFilesFromDisk(
-	    const ChunkTrashIndex::TrashIndexFileEntries &filesToRemove, const std::string &diskPath);
+	static	void removeTrashFilesFromDiskThread(uint8_t workerId);
 
 	/**
 	 * @brief Checks if a given timestamp string matches the expected format.
@@ -172,6 +178,10 @@ class ChunkTrashManagerImpl : public IChunkTrashManagerImpl {
 	static size_t garbageCollectorSpaceRecoveryStep;
 	static constexpr size_t kDefaultGarbageCollectorSpaceRecoveryStep = 100;
 
+	static std::vector<std::thread> removeFromTrashThreads;            /// Vector of worker threads.
+	static std::unique_ptr<ProducerConsumerQueue> removeFromTrashJobQueue;  /// Queue for jobs.	
+
+	static std::atomic<uint32_t> NotIdleThreadCount;	
 	// Use a function to safely get the ChunkTrashIndex instance
 	// This prevents issues during program shutdown
 	/**

src/chunkserver/chunkserver-common/chunk_trash_manager_unittest.cc

@@ -29,6 +29,7 @@ class MockChunkTrashManagerImpl : public IChunkTrashManagerImpl {
 	            (const std::filesystem::path &, const std::filesystem::path &, const std::time_t &),
 	            ());
 	MOCK_METHOD(int, init, (const std::string &), ());
+	MOCK_METHOD(void, terminate, (), ());
 	MOCK_METHOD(void, collectGarbage, (), ());
 	MOCK_METHOD(void, reloadConfig, (), ());
 };

src/chunkserver/hddspacemgr.cc

@@ -2541,6 +2541,10 @@ void hddFreeResourcesThread() {
 		releaseOldIoBuffers(kOldIoBuffersExpirationTimeMs);
 		usleep(timeout.remaining_us());
 	}
+
+
+	ChunkTrashManager::terminate();
+
 }
 
 void hddTerminate(void) {