perf: CS write buffering

doc/sfschunkserver.cfg.5.adoc

@@ -157,8 +157,9 @@ how much of skipped data to read if an offset of some read operation is greater
 than the offset where the previous operation finished (default is 0, i.e. don't
 read any skipped data; the value is aligned down to 64 KiB)
 
-*PERFORM_FSYNC*:: call fsync() after a chunk is modified (default is 1, i.e.
-enabled)
+*PERFORM_FSYNC*:: call fsync() after a chunk is modified. In the case of individual
+write jobs it does not apply, but when closing the chunks (default is 1, i.e.
+enabled).
 
 *STAT_CHUNKS_AT_DISK_SCAN*:: verify that chunk metadata and data parts exists at
 scan time. If set to 1, the data part is also checked to be multiple of
@@ -223,6 +224,20 @@ prioritization scheme. The available modes are:
 starving the other.
 Not reloadable. (Default: FIFO)
 
+*WRITE_BUFFERING_SIZE_MB (EXPERIMENTAL)*:: Caps the amount of data in MB the chunkserver
+"buffers", i.e fast replies to the client as if it was synced to the drives but it was
+only read from network and stored in memory. Similar to page cache. This feature has
+the following constraints:
+- does not work on standard and replica chunks
+- must have chunkserver side chunk locking enabled (see USE_CHUNKSERVER_SIDE_CHUNK_LOCK
+in master config)
+It is also suggested to set the IO_PRIORITY_MODE to SWITCH. The PERFORM_FSYNC
+continues working as intended, as it affects the time spent closing of the
+written chunk.
+Can provide better results for writes in general, but may significantly improve
+performance for small files.
+Reloadable. (Default: 0)
+
 *TLS_CERT_FILE (EXPERIMENTAL)*::
 Path to the TLS certificate file the chunk server will use for TLS connections
 (there is no default value).

src/admin/dump_config_command.cc

@@ -178,6 +178,7 @@ const static std::unordered_map<std::string, std::string> defaultOptionsCS = {
     {"NR_OF_HDD_WORKERS_PER_NETWORK_WORKER", "16"},
     {"BGJOBSCNT_PER_NETWORK_WORKER", "4000"},
     {"IO_PRIORITY_MODE", "FIFO"},
+    {"WRITE_BUFFERING_SIZE_MB", "0"},
     {"MAX_BLOCKS_PER_HDD_WRITE_JOB", "16"},
     {"MAX_BLOCKS_PER_HDD_READ_JOB", "16"},
     {"MAX_PARALLEL_HDD_READ_JOBS_PER_CS_ENTRY", "1"},

src/chunkserver/bgjobs.cc

@@ -435,6 +435,28 @@ uint32_t MasterJobPool::addLockJob(JobCallback callback, void *extra, uint32_t l
 	return jobId;
 }
 
+void MasterJobPool::changeLockJobsCallback(const CallbackMaker &callbackMaker,
+                                           uint32_t listenerId) {
+	// Check if the listenerId is valid
+	if (listenerId >= listenerInfos_.size()) {
+		safs::log_warn(
+		    "{} job pool: {}: Invalid listenerId {} for changing lock jobs callback, resetting to 0",
+		    name_, __func__, listenerId);
+		listenerId = 0;  // Reset to the first listener
+	}
+
+	auto &listenerInfo = listenerInfos_[listenerId];
+	std::scoped_lock lock(chunkToJobReplyMapMutex_, listenerInfo.jobsMutex);
+	for (auto &[chunkWithType, lockedChunkData] : chunkToJobReplyMap_) {
+		if (lockedChunkData.listenerId == listenerId) {
+			auto jobIterator = listenerInfo.jobHash.find(lockedChunkData.lockJobId);
+			if (jobIterator != listenerInfo.jobHash.end()) {
+				jobIterator->second->callback = callbackMaker(chunkWithType, listenerId);
+			}
+		}
+	}
+}
+
 bool MasterJobPool::startChunkLock(const JobPool::JobCallback &callback, void *packet,
                                    uint64_t chunkId, ChunkPartType chunkType, uint32_t listenerId) {
 	std::unique_lock lock(chunkToJobReplyMapMutex_);

src/chunkserver/bgjobs.h

@@ -280,6 +280,10 @@ class JobPool {
  */
 class MasterJobPool : public JobPool {
 public:
+	/// @brief Function type for creating job callbacks based on chunk information and listener ID.
+	using CallbackMaker =
+	    std::function<JobCallback(ChunkWithType chunkWithType, uint32_t listenerId)>;
+
 	/// @brief Constructor for MasterJobPool.
 	/// @param name Human readable name for this pool, useful for debugging.
 	/// @param workers The number of worker threads in the pool.
@@ -305,6 +309,13 @@ class MasterJobPool : public JobPool {
 	                           JobCallback callback, void *extra, ProcessJobCallback processJob,
 	                           uint32_t listenerId = 0);
 
+	/// @brief Changes the callback function for all lock jobs associated with a specific listener.
+	/// This function is used to update the callback for all lock jobs when the master server
+	/// disconnects to ensure that no chunks with broken data remain registered.
+	/// @param callbackMaker The function to create new callback functions for all lock jobs.
+	/// @param listenerId The ID of the listener associated with the lock jobs.
+	void changeLockJobsCallback(const CallbackMaker &callbackMaker, uint32_t listenerId = 0);
+
 	/// @brief Starts a chunk lock job for a specific chunk and type.
 	/// This function is triggered when the master server sends a chunk lock request for a chunk
 	/// that is not currently locked. It adds a lock job to the JobPool and associates it with the