feat(chunkserver): add HDD_CHUNK_BULK_SIZE option

Introduce a reloadable HDD_CHUNK_BULK_SIZE option for chunkserver.
Use it for chunk registration and damaged/lost/new report bulks.
Add default (1000) to config template, admin defaults, and docs.

Signed-off-by: guillex <guillex@leil.io>

Author: lgsilva3087

doc/sfschunkserver.cfg.5.adoc

@@ -102,6 +102,11 @@ maximum throughput.
 *HDD_ADVISE_NO_CACHE*:: whether to remove each chunk from page when closing it
 to reduce cache pressure generated by chunkserver (default is 0, i.e. no)
 
+*HDD_CHUNK_BULK_SIZE*:: number of chunks to process in a single bulk operation
+when reporting chunks to the master (e.g., when registering chunks, reporting
+new chunks, or reporting damaged or lost chunks).
+Reloadable. (default is 1000, minimum is 1, maximum is 1000000)
+
 *HDD_PUNCH_HOLES*:: if enabled then chunkserver detects zero values in chunk
 data and frees corresponding file blocks (decreasing file system usage). This
 option works only on Linux with file systems supporting punching holes (XFS,

src/admin/dump_config_command.cc

@@ -171,6 +171,7 @@ const static std::unordered_map<std::string, std::string> defaultOptionsCS = {
     {"HDD_CHECK_CRC_WHEN_READING", "1"},
     {"HDD_CHECK_CRC_WHEN_WRITING", "1"},
     {"HDD_ADVISE_NO_CACHE", "0"},
+    {"HDD_CHUNK_BULK_SIZE", "1000"},
     {"HDD_PUNCH_HOLES", "0"},
     {"ENABLE_LOAD_FACTOR", "0"},
     {"REPLICATION_BANDWIDTH_LIMIT_KBPS", "0"},

src/chunkserver/hddspacemgr.cc

@@ -2799,6 +2799,10 @@ void hddReload(void) {
 	gCheckCrcWhenWriting = cfg_getuint8("HDD_CHECK_CRC_WHEN_WRITING", 1) != 0U;
 	gPunchHolesInFiles = cfg_getuint32("HDD_PUNCH_HOLES", 0);
 
+	gChunkBulkSize.store(cfg_get_minmaxvalue<uint32_t>("HDD_CHUNK_BULK_SIZE", kDefaultChunkBulkSize,
+	                                                   kMinChunkBulkSize, kMaxChunkBulkSize),
+	                     std::memory_order_relaxed);
+
 	char *leaveFreeStr = cfg_getstr("HDD_LEAVE_SPACE_DEFAULT",
 	                                disk::gLeaveSpaceDefaultDefaultStrValue);
 	auto parsedLeaveFree = cfg_parse_size(leaveFreeStr);
@@ -2968,6 +2972,10 @@ int hddInit() {
 
 	gPunchHolesInFiles = cfg_getuint32("HDD_PUNCH_HOLES", 0);
 
+	gChunkBulkSize.store(cfg_get_minmaxvalue<uint32_t>("HDD_CHUNK_BULK_SIZE", kDefaultChunkBulkSize,
+	                                                   kMinChunkBulkSize, kMaxChunkBulkSize),
+	                     std::memory_order_relaxed);
+
 	eventloop_reloadregister(hddReload);
 	eventloop_timeregister(TIMEMODE_RUN_LATE, SECONDS_IN_ONE_MINUTE, 0,
 	                       hddDiskInfoRotateStats);

src/chunkserver/hddspacemgr.h

@@ -22,6 +22,7 @@
 
 #include "common/platform.h"
 
+#include <atomic>
 #include <vector>
 
 #include "chunkserver-common/chunk_interface.h"
@@ -44,11 +45,16 @@ void hddSerializeAllDiskInfosV2(uint8_t *buff);
 
 std::string hddGetDiskGroups();
 
-const std::size_t kChunkBulkSize = 1000;
+// Controls the number of chunks processed per bulk operation when reporting
+// chunks to the master (e.g. registration, damaged, lost, new chunks).
+inline constexpr uint32_t kDefaultChunkBulkSize = 1'000;
+inline constexpr uint32_t kMinChunkBulkSize = 1;
+inline constexpr uint32_t kMaxChunkBulkSize = 1'000'000;
+inline std::atomic_uint32_t gChunkBulkSize{kDefaultChunkBulkSize};
+
 using BulkFunction = std::function<void(std::vector<ChunkWithVersionAndType>&)>;
-/// Executes the callback for each bulk of at most \p kChunkBulkSize chunks.
-void hddForeachChunkInBulks(BulkFunction bulkCallback,
-                            std::size_t bulkSize = kChunkBulkSize);
+/// Executes the callback for each bulk of at most \p bulkSize chunks.
+void hddForeachChunkInBulks(BulkFunction bulkCallback, std::size_t bulkSize);
 
 int hddGetAndResetSpaceChanged();
 void hddGetTotalSpace(uint64_t *usedSpace, uint64_t *totalSpace,

src/chunkserver/master_connection.cc

@@ -184,9 +184,11 @@ void MasterConn::onRegistered(const std::vector<uint8_t> &data) {
 	isVersionLessThan5_ = false;
 	registrationAttempts_ = 0;
 
-	hddForeachChunkInBulks([this](const std::vector<ChunkWithVersionAndType> &chunksBulk) {
-		createAttachedPacket(cstoma::registerChunks::build(chunksBulk));
-	});
+	hddForeachChunkInBulks(
+	    [this](const std::vector<ChunkWithVersionAndType> &chunksBulk) {
+		    createAttachedPacket(cstoma::registerChunks::build(chunksBulk));
+	    },
+	    gChunkBulkSize.load(std::memory_order_relaxed));
 
 	uint64_t usedSpace;
 	uint64_t totalSpace;

src/chunkserver/masterconn.cc

@@ -117,19 +117,21 @@ void masterconn_check_hdd_reports() {
 			errorcounter--;
 		}
 
+		const auto chunkBulkSize = gChunkBulkSize.load(std::memory_order_relaxed);
+
 		std::vector<ChunkWithType> chunks_with_type;
-		hddGetDamagedChunks(chunks_with_type, 1000);
+		hddGetDamagedChunks(chunks_with_type, chunkBulkSize);
 		if (!chunks_with_type.empty()) {
 			eptr->createAttachedPacket(cstoma::chunkDamaged::build(chunks_with_type));
 		}
 
-		hddGetLostChunks(chunks_with_type, 1000);
+		hddGetLostChunks(chunks_with_type, chunkBulkSize);
 		if (!chunks_with_type.empty()) {
 			eptr->createAttachedPacket(cstoma::chunkLost::build(chunks_with_type));
 		}
 
 		std::vector<ChunkWithVersionAndType> chunks_with_version;
-		hddGetNewChunks(chunks_with_version, 1000);
+		hddGetNewChunks(chunks_with_version, chunkBulkSize);
 		if (!chunks_with_version.empty()) {
 			eptr->createAttachedPacket(cstoma::chunkNew::build(chunks_with_version));
 		}

src/data/sfschunkserver.cfg.in

@@ -113,6 +113,12 @@
 ## (Default: 0)
 # HDD_ADVISE_NO_CACHE = 0
 
+## Number of chunks to process in a single bulk operation when reporting
+## chunks to the master (e.g., when registering chunks, reporting new
+## chunks, or reporting damaged or lost chunks).
+## Reloadable. (Default: 1000, Minimum: 1, Maximum: 1000000)
+# HDD_CHUNK_BULK_SIZE = 1000
+
 ## If enabled then chunkserver detects zero values in chunk data and frees
 ## corresponding file blocks (decreasing file system usage).
 ## This option works only on Linux