diff --git a/pgxn/neon/communicator/src/lib.rs b/pgxn/neon/communicator/src/lib.rs index 30fa698d83..050391051d 100644 --- a/pgxn/neon/communicator/src/lib.rs +++ b/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; diff --git a/pgxn/neon/communicator/src/neon_request.rs b/pgxn/neon/communicator/src/neon_request.rs index 2839b25257..20f896651d 100644 --- a/pgxn/neon/communicator/src/neon_request.rs +++ b/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)] diff --git a/pgxn/neon/communicator/src/worker_process/main_loop.rs b/pgxn/neon/communicator/src/worker_process/main_loop.rs index 70ed5f6eb6..e6fb05241c 100644 --- a/pgxn/neon/communicator/src/worker_process/main_loop.rs +++ b/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 _) } diff --git a/pgxn/neon/communicator_new.c b/pgxn/neon/communicator_new.c index a879a40e18..b14fd5a9f8 100644 --- a/pgxn/neon/communicator_new.c +++ b/pgxn/neon/communicator_new.c @@ -13,6 +13,7 @@ */ #include "postgres.h" +#include #include #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; diff --git a/pgxn/neon/communicator_new.h b/pgxn/neon/communicator_new.h index 5ddb2d303b..5101349687 100644 --- a/pgxn/neon/communicator_new.h +++ b/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 */ diff --git a/pgxn/neon/pagestore_smgr.c b/pgxn/neon/pagestore_smgr.c index 23841e9e91..79cd4685cb 100644 --- a/pgxn/neon/pagestore_smgr.c +++ b/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);