fix(mount): fix not released read cache entries

This PR targets fixing unexpected high memory consumption while
reading and client shows "Maximum requested offset should be greater
than or equal current offset" message.

The found case can be described the following way:
- assume some previous reads have left behind some read buffers in
the buffer pool.
- previous reads have read the interval currently requested.
- immediately previous reads have generated readahead requests
currently in progress in lower offsets of the file.
- the request is instantly replied from cache, but adding extra
requests is triggered.
- the maximum offset requested is set then to those lower offsets.
- the forced insert adds an entry from the buffers pool with zero size
but actually has the space reserved.
- the above message is then shown and addExtraRequests_ returns without
adding the inserted entry to any other container.
- that entry is unable to get its done set to true, which implies it
can only be released while destroying the ReadRecord instance, often
at file descriptor close.

The proposed solution prevents the entry insertion in the cases of an
early return. The maximumRequestedOffset update was also modified that
avoids early returns while keeping the same meaning.

The test test_slow_rereads was added to check the previously described
issues. A new mode "only_slow" was added to the chunk_operations_eio
library. The util create-and-reread-file was created specifically for
this test.

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

Author: dmga44

src/mount/readdata.cc

@@ -418,8 +418,9 @@ void ReadaheadOperationsManager::addExtraRequests_(
 	}
 
 	if (!rrec->readaheadRequests.empty()) {
-		maximumRequestedOffset = rrec->readaheadRequests.lastPendingRequest()
-		                             ->requestPtr->endOffset();
+		maximumRequestedOffset =
+		    std::max(maximumRequestedOffset,
+		             rrec->readaheadRequests.lastPendingRequest()->requestPtr->endOffset());
 	}
 
 	uint64_t throughputWindow = rrec->readahead_adviser.throughputWindow();
@@ -438,7 +439,6 @@ void ReadaheadOperationsManager::addExtraRequests_(
 		// Try to align extra requests to SFSCHUNKSIZE
 		uint64_t extraRequestSize = std::min<uint64_t>(
 		    satisfyingSize, SFSCHUNKSIZE - (maximumRequestedOffset % SFSCHUNKSIZE));
-		ReadCache::Entry *entry = rrec->cache.forceInsert(maximumRequestedOffset, extraRequestSize);
 
 		if (maximumRequestedOffset < currentOffset) {
 			safs::log_warn(
@@ -449,6 +449,9 @@ void ReadaheadOperationsManager::addExtraRequests_(
 			// Next subtraction will overflow, so let's just return here
 			return;
 		}
+
+		ReadCache::Entry *entry = rrec->cache.forceInsert(maximumRequestedOffset, extraRequestSize);
+
 		int64_t extraPriority =
 		    rrec->readahead_adviser.expectedNeededTime_us(maximumRequestedOffset - currentOffset);
 

tests/test_suites/ShortSystemTests/test_slow_rereads.sh

@@ -0,0 +1,55 @@
+timeout_set 3 minutes
+
+cacheexpirationtime_ms=15000
+readbuffersexpirationtime_ms=60000
+
+CHUNKSERVERS=1 \
+	DISK_PER_CHUNKSERVER=1 \
+	CHUNKSERVER_0_DISK_0="$RAMDISK_DIR/pread_only_slow_hdd_0" \
+	USE_RAMDISK=YES \
+	MOUNT_EXTRA_CONFIG="sfscachemode=NEVER|maxreadaheadrequests=4`
+		`|cacheexpirationtime=${cacheexpirationtime_ms}|readworkers=1`
+		`|readaheadmaxwindowsize=16384`
+		`|readbuffersexpirationtime=${readbuffersexpirationtime_ms}" \
+	MASTER_EXTRA_CONFIG="CHUNKS_LOOP_MIN_TIME = 1`
+			`|CHUNKS_LOOP_MAX_CPU = 90`
+			`|CHUNKS_LOOP_PERIOD = 10`
+			`|OPERATIONS_DELAY_INIT = 0`
+			`|OPERATIONS_DELAY_DISCONNECT = 0" \
+	setup_local_empty_saunafs info
+
+cd ${info[mount0]}
+
+# Create a file with 30 chunks, and read it to make sure client creates all necessary read buffers
+dd if=/dev/zero of=file bs=1M count=$((30 * 64)) status=none
+dd if=file of=/dev/null bs=1M count=$((30 * 64)) status=none
+
+saunafs settrashtime 0 file
+rm file
+
+# Restart the first chunkserver preloading pread with slow version of reads
+LD_PRELOAD="${SAUNAFS_INSTALL_FULL_LIBDIR}/libchunk_operations_eio.so" \
+		assert_success saunafs_chunkserver_daemon 0 restart
+saunafs_wait_for_all_ready_chunkservers
+
+create-and-reread-file "file" $(( (cacheexpirationtime_ms + readbuffersexpirationtime_ms) / 1000)) &
+
+while ! [[ -f notify_file_reread ]]; do sleep 0.1; done
+
+# At least 75s must have passed since the last read of the file (see create_and_reread_file.cc),
+# so all read buffers should have been expired by now. Check that sfsmount is not using too much 
+# memory, which would indicate that it is keeping all read buffers in memory instead of expiring
+# them.
+
+sfsmount_pid="$(pgrep -f "sfsmount.*${info[mount0]}" | head -n1)"
+assert_success test -n "${sfsmount_pid}"
+
+rss_kb="$(ps -o rss= -p "${sfsmount_pid}" | tr -d '[:space:]')"
+vsz_kb="$(ps -o vsz= -p "${sfsmount_pid}" | tr -d '[:space:]')"
+
+echo "sfsmount pid=${sfsmount_pid} rss=${rss_kb}KB vsz=${vsz_kb}KB"
+
+# 1 GB RSS is a very high limit that should be never reached if read buffers are expired properly
+assert_less_than "${rss_kb}" $((1024 * 1024)) 
+
+wait

utils/CMakeLists.txt

@@ -72,6 +72,10 @@ install(TARGETS readdir-unlink-test RUNTIME DESTINATION ${BIN_SUBDIR})
 add_executable(big-session-metadata-benchmark big_session_metadata_benchmark.cc)
 install(TARGETS big-session-metadata-benchmark RUNTIME DESTINATION ${BIN_SUBDIR})
 
+# create_and_reread_file test for testing some rereads after creating a file
+add_executable(create-and-reread-file create_and_reread_file.cc)
+install(TARGETS create-and-reread-file RUNTIME DESTINATION ${BIN_SUBDIR})
+
 add_executable(metadata-notifier metadata_notifier.cc)
 target_link_libraries(metadata-notifier sfscommon)
 install(TARGETS metadata-notifier RUNTIME DESTINATION ${BIN_SUBDIR})

utils/chunk_operations_eio.c

@@ -32,6 +32,12 @@ int EIO_replies = 0;
 // * fsync always fails with EIO if file name contains "fsync_EIO"
 // * pread fails with EIO if offset>FAR_OFFSET_THRESHOLD and file name contains "pread_far_EIO"
 // * pwrite fails with EIO if offset>FAR_OFFSET_THRESHOLD and file name contains "pwrite_far_EIO"
+// * pread fails with EIO if the first operation and takes 2s more if file name contains
+// "pread_slow_and_one_eio_trigger"
+// * pwrite fails with EIO if the first operation and takes 2s more if file name contains
+// "pwrite_slow_and_one_eio_trigger"
+// * pread takes 10ms + 1us per 250B if file name contains "pread_only_slow"
+// * pwrite takes 10ms + 1us per 250B if file name contains "pwrite_only_slow"
 
 // returns -1 on failure and sets errno (via readlink call)
 ssize_t read_filename(int fd, char *buf, int bufsize) {
@@ -42,11 +48,12 @@ ssize_t read_filename(int fd, char *buf, int bufsize) {
 	return readlink(fdpath, buf, bufsize);
 }
 
-static int err_on_operation(int fd, const char* opname, size_t offset) {
+static int err_on_operation(int fd, const char* opname, size_t offset, size_t size) {
 	char filename[FILENAME_BUFSIZE] = {0};
 	char always_eio_trigger[COMMAND_BUFSIZE] = {0};
 	char far_eio_trigger[COMMAND_BUFSIZE] = {0};
 	char slow_and_one_eio_trigger[COMMAND_BUFSIZE] = {0};
+	char only_slow[COMMAND_BUFSIZE] = {0};
 
 	ssize_t result = read_filename(fd, filename, FILENAME_BUFSIZE);
 	if (result == -1) {
@@ -61,6 +68,7 @@ static int err_on_operation(int fd, const char* opname, size_t offset) {
 	sprintf(always_eio_trigger, "%s_EIO", opname);
 	sprintf(far_eio_trigger, "%s_far_EIO", opname);
 	sprintf(slow_and_one_eio_trigger, "%s_slow_and_one_EIO", opname);
+	sprintf(only_slow, "%s_only_slow", opname);
 
 	// TODO: remove fixed pattern for SMRs after the basic support
 	if (strstr(filename, always_eio_trigger) || strstr(filename, "sauna_nullb0")) {
@@ -75,6 +83,12 @@ static int err_on_operation(int fd, const char* opname, size_t offset) {
 			return EIO;
 		}
 		return 0;
+	} else if (strstr(filename, only_slow)) {
+		// sleep for a while to simulate a slow operation on hdd of 250MB/s peak throughput
+		int kBaseLatency_us = 10 * 1000;  // 10ms
+		int bytesPerUs = 250;  // 250B per 1us
+		usleep(kBaseLatency_us + size / bytesPerUs);
+		return 0;
 	} else {
 		return 0;
 	}
@@ -86,7 +100,7 @@ ssize_t pread(int fd, void *buf, size_t count, off_t offset) {
 	int err;
 	static pread_t _pread = NULL;
 
-	err = err_on_operation(fd, "pread", offset);
+	err = err_on_operation(fd, "pread", offset, count);
 	if (err) {
 		errno = err;
 		return -1;
@@ -101,7 +115,7 @@ ssize_t pwrite(int fd, const void *buf, size_t count, off_t offset) {
 	int err;
 	static pwrite_t _pwrite = NULL;
 
-	err = err_on_operation(fd, "pwrite", offset);
+	err = err_on_operation(fd, "pwrite", offset, count);
 	if (err) {
 		errno = err;
 		return -1;
@@ -116,7 +130,7 @@ int close(int fd) {
 	int err;
 	static close_t _close = NULL;
 
-	err = err_on_operation(fd, "close", 0);
+	err = err_on_operation(fd, "close", 0, 0);
 	if (err) {
 		errno = err;
 		return -1;
@@ -131,7 +145,7 @@ int fsync(int fd) {
 	int err;
 	static fsync_t _fsync = NULL;
 
-	err = err_on_operation(fd, "fsync", 0);
+	err = err_on_operation(fd, "fsync", 0, 0);
 	if (err) {
 		errno = err;
 		return -1;

utils/create_and_reread_file.cc

@@ -0,0 +1,87 @@
+/*
+   Copyright 2026      Leil Storage OÜ
+
+   This file is part of SaunaFS.
+
+   SaunaFS is free software: you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation, version 3.
+
+   SaunaFS is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with SaunaFS. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "common/aligned_allocator.h"
+
+#include "utils/data_generator.h"
+
+#ifndef SFSCHUNKSIZE
+#define SFSCHUNKSIZE (SFSBLOCKSIZE * SFSBLOCKSINCHUNK)
+#endif
+
+int main(int argc, char** argv) {
+	if (argc != 3) {
+		std::cerr << "Usage:\n"
+		             "    "
+		          << argv[0]
+		          << " <file> <SLEEP_TIME>\n"
+		             "Creates a file with the specified name and fills it with generated data.\n"
+		             "Afterwards reads and reads again some chunks from it.\n"
+		             "Sleeps for the specified time in seconds and creates a notification file.\n"
+					 "At the end closes the file descriptor.\n";
+		return 1;
+	}
+
+	uint64_t fileSize = 30 * SFSCHUNKSIZE;  // 1920 MiB (30 chunks of 64 MiB)
+	DataGenerator generator(0);
+	generator.createFile(argv[1], fileSize);
+	uint32_t sleepTime = std::stoul(argv[2]);
+	std::cerr << "File created successfully.\n";
+
+	int fd = open(argv[1], O_RDONLY | O_DIRECT);
+	utils_passert(fd >= 0);
+
+	// Read the last two chunks
+	std::vector<char, AlignedAllocator<char, SFSBLOCKSIZE>> buffer(SFSBLOCKSIZE);
+	for (uint64_t offset = fileSize - 2 * SFSCHUNKSIZE; offset < fileSize; offset += SFSBLOCKSIZE) {
+		ssize_t bytesRead = pread(fd, buffer.data(), buffer.size(), offset);
+		utils_passert(bytesRead == (ssize_t)buffer.size());
+	}
+	std::cerr << "Last two chunks read successfully.\n";
+
+	// Read first three chunks
+	for (uint64_t offset = 0; offset < 3 * SFSCHUNKSIZE; offset += SFSBLOCKSIZE) {
+		ssize_t bytesRead = pread(fd, buffer.data(), buffer.size(), offset);
+		utils_passert(bytesRead == (ssize_t)buffer.size());
+	}
+
+	std::cerr << "First three and last two chunks read successfully.\n";
+	// Sleep for a while to process some of the readahead requests
+	usleep(600000);
+
+	// Read last chunk again
+	for (uint64_t offset = fileSize - SFSCHUNKSIZE; offset < fileSize; offset += SFSBLOCKSIZE) {
+		ssize_t bytesRead = pread(fd, buffer.data(), buffer.size(), offset);
+		utils_passert(bytesRead == (ssize_t)buffer.size());
+	}
+	std::cerr << "Reads completed successfully.\n";
+
+	sleep(sleepTime);
+
+	auto notify_file_reread = open("notify_file_reread", O_CREAT | O_WRONLY, 0644);
+	utils_passert(notify_file_reread >= 0);
+	utils_zassert(close(notify_file_reread));
+
+	sleep(1);
+
+	utils_zassert(close(fd));
+	std::cerr << "FD closed successfully.\n";
+	unlink("notify_file_reread");
+
+	return 0;
+}