Write directly to the file instead of memory and then a file

pgxn/neon/communicator/src/lib.rs

@@ -22,30 +22,6 @@ mod worker_process;
 mod global_allocator;
 
 // FIXME: get this from postgres headers somehow
-/// Size of a disk block: this also limits the size of a tuple. You can
-/// set it bigger if you need bigger tuples (although `TOAST` should
-/// reduce the need to have large tuples, since fields can be spread
-/// across multiple tuples). [`BLCKSZ`] must be a power of `2`. The
-/// maximum possible value of [`BLCKSZ`] is currently `2^15` (`32768`).
-/// This is determined by the `15-bit` widths of the `lp_off` and
-/// `lp_len` fields in ItemIdData (see `include/storage/itemid.h`).
-/// Changing [`BLCKSZ`] requires an `initdb`.
 pub const BLCKSZ: usize = 8192;
-
-/// Define SLRU segment size. A page is the same [`BLCKSZ``] and is used
-/// everywhere else in Postgres. The segment size can be chosen somewhat
-/// arbitrarily; we make it `32` pages by default, or `256Kb`, i.e. 1
-/// **million** transactions for `CLOG` or `64K` transactions for
-/// `SUBTRANS`.
-///
-/// # Note
-///
-/// Because TransactionIds are 32 bits and wrap around at `0xFFFFFFFF``,
-/// page numbering also wraps around at `0xFFFFFFFF/xxxx_XACTS_PER_PAGE`
-/// (where `xxxx` is `CLOG` or `SUBTRANS`, respectively), and segment
-/// numbering at
-/// `0xFFFFFFFF/xxxx_XACTS_PER_PAGE/SLRU_PAGES_PER_SEGMENT`. We need
-/// take no explicit notice of that fact in `slru.c`, except when
-/// comparing segment and page numbers in SimpleLruTruncate
-/// (see PagePrecedes()).
+// FIXME: get this from postgres headers somehow
 pub const SLRU_PAGES_PER_SEGMENT: usize = 32;

pgxn/neon/communicator/src/neon_request.rs

@@ -4,6 +4,8 @@ pub type COid = u32;
 // This conveniently matches PG_IOV_MAX
 pub const MAX_GETPAGEV_PAGES: usize = 32;
 
+use std::ffi::CStr;
+
 use pageserver_page_api::{self as page_api, SlruKind};
 
 #[allow(clippy::large_enum_variant)]
@@ -186,9 +188,18 @@ pub struct CReadSlruSegmentRequest {
     pub slru_kind: SlruKind,
     pub segment_number: u32,
     pub request_lsn: CLsn,
+    /// Must be a null-terminated C string containing **absolute**
+    /// file path.
+    pub destination_file_path: [std::ffi::c_char; libc::PATH_MAX as usize],
+}
 
-    // These fields define where the result is written. Must point into a buffer in shared memory!
-    pub dest: ShmemBuf,
+impl CReadSlruSegmentRequest {
+    /// Returns the path to the file where the segment is read into.
+    pub(crate) fn destination_file_path(&self) -> String {
+        unsafe { CStr::from_ptr(self.destination_file_path.as_ptr()) }
+            .to_string_lossy()
+            .into_owned()
+    }
 }
 
 #[repr(C)]

pgxn/neon/communicator/src/worker_process/main_loop.rs

@@ -24,7 +24,7 @@ use utils::id::{TenantId, TimelineId};
 
 use super::callbacks::{get_request_lsn, notify_proc};
 
-use tracing::{error, info, info_span, trace};
+use tracing::{debug, error, info, info_span, trace};
 
 use utils::lsn::Lsn;
 
@@ -107,9 +107,8 @@ pub(super) async fn init(
         .integrated_cache_init_struct
         .worker_process_init(last_lsn, file_cache);
 
-    info!("Initialised integrated cache: {cache:?}");
+    debug!("Initialised integrated cache: {cache:?}");
 
-    // TODO: plumb through the stripe size.
     let tenant_id = TenantId::from_str(&tenant_id).expect("invalid tenant ID");
     let timeline_id = TimelineId::from_str(&timeline_id).expect("invalid timeline ID");
     let shard_spec = ShardSpec::new(shard_map, stripe_size).expect("invalid shard spec");
@@ -428,6 +427,7 @@ impl<'t> CommunicatorWorkerProcessStruct<'t> {
             NeonIORequest::ReadSlruSegment(req) => {
                 self.request_read_slru_segment_counter.inc();
                 let lsn = Lsn(req.request_lsn);
+                let file_path = req.destination_file_path();
 
                 match self
                     .client
@@ -439,15 +439,12 @@ impl<'t> CommunicatorWorkerProcessStruct<'t> {
                     .await
                 {
                     Ok(slru_bytes) => {
-                        let src: &[u8] = &slru_bytes.as_ref();
-                        let dest = req.dest;
-                        let len = std::cmp::min(src.len(), dest.bytes_total());
+                        if let Err(e) = std::fs::write(&file_path, &slru_bytes) {
+                            info!("could not write slru segment to file {file_path}: {e}");
+                            return NeonIOResult::Error(e.raw_os_error().unwrap_or(libc::EIO));
+                        }
 
-                        unsafe {
-                            std::ptr::copy_nonoverlapping(src.as_ptr(), dest.as_mut_ptr(), len);
-                        };
-
-                        let blocks_count = len / crate::BLCKSZ;
+                        let blocks_count = slru_bytes.len() / crate::BLCKSZ;
 
                         NeonIOResult::ReadSlruSegment(blocks_count as _)
                     }

pgxn/neon/communicator_new.c

@@ -13,6 +13,7 @@
  */
 #include "postgres.h"
 
+#include <stdlib.h>
 #include <unistd.h>
 
 #include "access/xlog.h"
@@ -1001,7 +1002,7 @@ communicator_new_read_slru_segment(
 	SlruKind kind,
 	uint32_t segno,
 	neon_request_lsns *request_lsns,
-	void *buffer)
+	char* path)
 {
 	NeonIOResult result = {};
 	NeonIORequest request = {
@@ -1011,22 +1012,50 @@ communicator_new_read_slru_segment(
 			.slru_kind = kind,
 			.segment_number = segno,
 			.request_lsn = request_lsns->request_lsn,
-			.dest.ptr = bounce_buf(),
 		}
 	};
 
-	elog(DEBUG5, "readslrusegment called for kind=%u, segno=%u", kind, segno);
+	if (path == NULL) {
+		elog(ERROR, "read_slru_segment called with NULL path");
+		return -1;
+	}
+
+	// Scoping should help deallocate the absolute path buffer.
+	{
+		char abs_path[PATH_MAX];
+
+		if (path[0] != '/') {
+			char cwd[PATH_MAX];
+			getcwd(cwd, sizeof(cwd));
+
+			const int size = snprintf(NULL, 0, "%s/%s", cwd, path);
+			if (size < 0 || size >= PATH_MAX) {
+				elog(ERROR, "read_slru_segment failed to create an absolute path for \"%s\"", path);
+				return -1;
+			}
+
+			snprintf(abs_path, size + 1, "%s/%s", cwd, path);
+		} else {
+			strncpy(abs_path, path, sizeof(abs_path));
+		}
+		strncpy(request.read_slru_segment.destination_file_path, abs_path, sizeof(request.read_slru_segment.destination_file_path));
+	}
+
+	elog(DEBUG5, "readslrusegment called for kind=%u, segno=%u, file_path=\"%s\"",
+		kind, segno, request.read_slru_segment.destination_file_path);
 
 	/* FIXME: see `request_lsns` in main_loop.rs for why this is needed */
 	XLogSetAsyncXactLSN(request_lsns->request_lsn);
 
 	perform_request(&request, &result);
 
+	int nblocks = -1;
+
 	switch (result.tag)
 	{
 		case NeonIOResult_ReadSlruSegment:
-			memcpy(buffer, request.read_slru_segment.dest.ptr, 8192);
-			return result.read_slru_segment;
+			nblocks = result.read_slru_segment;
+			break;
 		case NeonIOResult_Error:
 			ereport(ERROR,
 					(errcode_for_file_access(),
@@ -1038,7 +1067,7 @@ communicator_new_read_slru_segment(
 			break;
 	}
 
-	return 0;
+	return nblocks;
 }
 
 /* Write requests */
@@ -1343,6 +1372,18 @@ print_neon_io_request(NeonIORequest *request)
 								r->spc_oid, r->db_oid, r->rel_number, r->fork_number, r->block_number, r->block_number + r->nblocks);
 				return buf;
 			}
+		case NeonIORequest_ReadSlruSegment:
+			{
+				CReadSlruSegmentRequest *r = &request->read_slru_segment;
+
+				snprintf(buf, sizeof(buf), "ReadSlruSegment: req " UINT64_FORMAT " slrukind=%u, segno=%u, lsn=%X/%X, file_path=\"%s\"",
+								r->request_id,
+								r->slru_kind,
+								r->segment_number,
+								LSN_FORMAT_ARGS(r->request_lsn),
+								r->destination_file_path);
+				return buf;
+			}
 		case NeonIORequest_PrefetchV:
 			{
 				CPrefetchVRequest *r = &request->prefetch_v;

pgxn/neon/communicator_new.h

@@ -42,7 +42,7 @@ extern int communicator_new_read_slru_segment(
 	SlruKind kind,
 	uint32_t segno,
 	neon_request_lsns *request_lsns,
-	void *buffer
+	char *path
 );
 
 /* Write requests, to keep the caches up-to-date */

pgxn/neon/pagestore_smgr.c

@@ -2385,7 +2385,7 @@ neon_read_slru_segment(SMgrRelation reln, const char* path, int segno, void* buf
 	request_lsns.effective_request_lsn = request_lsn;
 
 	if (neon_enable_new_communicator)
-		n_blocks = communicator_new_read_slru_segment(kind, (uint32_t)segno, &request_lsns, buffer);
+		n_blocks = communicator_new_read_slru_segment(kind, (uint32_t)segno, &request_lsns, path);
 	else
 		n_blocks = communicator_read_slru_segment(kind, segno, &request_lsns, buffer);