Add pwrite() in FdHandle to support concurrent reads and writes at specific offsets

Author: komozoi

include/fs/FdHandle.h

@@ -301,6 +301,56 @@ class FdHandle {
 	 */
 	ssize_t read(void* value, size_t size) const;
 
+	/**
+	 * @brief Reads a value of type T from a specific file offset.
+	 *
+	 * This is a thread-safe positional read.  Unlike read(), it does not
+	 * mutate or rely on the file's seek position and does not hold the file
+	 * mutex for the duration of the syscall, allowing concurrent readers.
+	 *
+	 * @tparam T Type of value.
+	 * @param value Reference to store the read value.
+	 * @param offset Offset in the file to read from.
+	 * @return Number of bytes read.
+	 */
+	template<class T>
+	inline ssize_t pread(T& value, off_t offset) const {
+		return pread(&value, sizeof(value), offset);
+	}
+
+	/**
+	 * @brief Reads raw bytes from a specific file offset.
+	 * @param value Pointer to buffer.
+	 * @param size Number of bytes to read.
+	 * @param offset Offset in the file to read from.
+	 * @return Number of bytes read.
+	 */
+	ssize_t pread(void* value, size_t size, off_t offset) const;
+
+	/**
+	 * @brief Writes a value of type T to a specific file offset.
+	 *
+	 * Positional write that does not move the file's seek position.
+	 *
+	 * @tparam T Type of value.
+	 * @param value Value to write.
+	 * @param offset Offset in the file to write to.
+	 * @return Number of bytes written.
+	 */
+	template<class T>
+	inline ssize_t pwrite(const T& value, off_t offset) const {
+		return pwrite(&value, sizeof(value), offset);
+	}
+
+	/**
+	 * @brief Writes raw bytes to a specific file offset.
+	 * @param value Pointer to data.
+	 * @param size Number of bytes to write.
+	 * @param offset Offset in the file to write to.
+	 * @return Number of bytes written.
+	 */
+	ssize_t pwrite(const void* value, size_t size, off_t offset) const;
+
 	/**
 	 * @brief Waits until data is available for reading.
 	 * @return true if ready.
@@ -382,7 +432,7 @@ class FdHandle {
 	 * @brief Returns a lock guard for the file's mutex.
 	 * @return lock_guard.
 	 */
-	std::lock_guard<std::mutex> getLock() const;
+	std::lock_guard<std::recursive_mutex> getLock() const;
 
 	/**
 	 * @brief Destructor (decreases reference count).
@@ -506,7 +556,7 @@ class FdTransaction {
 	bool isStream() const;
 
 private:
-	std::lock_guard<std::mutex> guard;
+	std::lock_guard<std::recursive_mutex> guard;
 	FdHandleData& handleData; /**< Underlying file handle data. */
 };
 

src/fs/FdHandle.cpp

@@ -231,7 +231,7 @@ class FdHandleData {
 	}
 
 	int16_t fd;
-	std::mutex mutex;
+	std::recursive_mutex mutex;
 
 	int numReferences() const {
 		return refs;
@@ -471,6 +471,53 @@ ssize_t FdHandle::read(void* value, size_t size) const {
 	return getHandleData(fd).read(value, size);
 }
 
+ssize_t FdHandle::pread(void* value, size_t size, off_t offset) const {
+	FdHandleData& handle = getHandleData(fd);
+
+	// Drain any queued positional writes so the read sees the latest data,
+	// then release the handle mutex before issuing the syscall.  Multiple
+	// readers can then execute concurrently inside the kernel.
+	{
+		std::lock_guard<std::recursive_mutex> _(handle.mutex);
+		handle.flushWrites();
+	}
+
+	char* buffer = (char*)value;
+	size_t totalRead = 0;
+	while (totalRead < size) {
+		ssize_t res = ::pread(handle.fd, &buffer[totalRead], size - totalRead, offset + (off_t)totalRead);
+		if (res < 0) {
+			if (totalRead > 0) return (ssize_t)totalRead;
+			return res;
+		} else if (res == 0)
+			break;
+		totalRead += (size_t)res;
+	}
+	return (ssize_t)totalRead;
+}
+
+ssize_t FdHandle::pwrite(const void* value, size_t size, off_t offset) const {
+	FdHandleData& handle = getHandleData(fd);
+
+	{
+		std::lock_guard<std::recursive_mutex> _(handle.mutex);
+		handle.flushWrites();
+	}
+
+	const char* buffer = (const char*)value;
+	size_t totalWritten = 0;
+	while (totalWritten < size) {
+		ssize_t res = ::pwrite(handle.fd, &buffer[totalWritten], size - totalWritten, offset + (off_t)totalWritten);
+		if (res < 0) {
+			if (totalWritten > 0) return (ssize_t)totalWritten;
+			return res;
+		} else if (res == 0)
+			break;
+		totalWritten += (size_t)res;
+	}
+	return (ssize_t)totalWritten;
+}
+
 bool FdHandle::waitForRead() const {
 	return getHandleData(fd).waitForRead();
 }
@@ -547,8 +594,8 @@ void FdHandle::close() {
 	}
 }
 
-std::lock_guard<std::mutex> FdHandle::getLock() const {
-	return std::lock_guard<std::mutex>(getHandleData(fd).mutex);
+std::lock_guard<std::recursive_mutex> FdHandle::getLock() const {
+	return std::lock_guard<std::recursive_mutex>(getHandleData(fd).mutex);
 }
 
 void FdHandle::markToClose() const {
@@ -561,7 +608,7 @@ MmapHandle FdHandle::getMmapHandle(off_t offset, size_t size, int prot, int flag
 	// Make sure the file is big enough, then
 	// restore the previous cursor position
 	{
-		std::lock_guard<std::mutex> _(handle.mutex);
+		std::lock_guard<std::recursive_mutex> _(handle.mutex);
 
 		// Get current file offset to restore later
 		off_t previousOffset = lseek(fd, 0, SEEK_CUR);