diff --git a/libs/walproposer/build.rs b/libs/walproposer/build.rs index e3713d20b2..97614c7019 100644 --- a/libs/walproposer/build.rs +++ b/libs/walproposer/build.rs @@ -115,6 +115,7 @@ fn main() -> anyhow::Result<()> { .allowlist_var("am_wal_proposer") .allowlist_var("neon_timeline_walproposer") .allowlist_var("neon_tenant_walproposer") + .allowlist_var("syncSafekeepers") .clang_arg(format!("-I{inc_server_path}")) .clang_arg(format!("-I{inc_pgxn_path}")) // Finish the builder and generate the bindings. diff --git a/libs/walproposer/src/sim.rs b/libs/walproposer/src/sim.rs index 1b58238189..9b9f1cb935 100644 --- a/libs/walproposer/src/sim.rs +++ b/libs/walproposer/src/sim.rs @@ -107,6 +107,18 @@ pub extern "C" fn sim_epoll_rcv(timeout: i64) -> Event { any_message: anymessage_tag(&message), } } + NodeEvent::Internal(message) => { + // store message in thread local storage, C code should use + // sim_msg_* functions to access it. + MESSAGE_BUF.with(|cell| { + *cell.borrow_mut() = message.clone(); + }); + Event { + tag: EventTag::Internal, + tcp: 0, + any_message: anymessage_tag(&message), + } + } NodeEvent::WakeTimeout(_) => { // can't happen unreachable!() diff --git a/libs/walproposer/src/sim_proto.rs b/libs/walproposer/src/sim_proto.rs index 9fccfd9b3b..f1616c0384 100644 --- a/libs/walproposer/src/sim_proto.rs +++ b/libs/walproposer/src/sim_proto.rs @@ -87,6 +87,7 @@ pub enum EventTag { Accept, Closed, Message, + Internal, } #[repr(u8)] diff --git a/libs/walproposer/src/simtest/wp_sk.rs b/libs/walproposer/src/simtest/wp_sk.rs index 5bc7896832..7dbbfcf8ca 100644 --- a/libs/walproposer/src/simtest/wp_sk.rs +++ b/libs/walproposer/src/simtest/wp_sk.rs @@ -10,7 +10,7 @@ use utils::{id::TenantTimelineId, logging, lsn::Lsn}; use crate::{ bindings::{ neon_tenant_walproposer, neon_timeline_walproposer, wal_acceptor_connection_timeout, - wal_acceptor_reconnect_timeout, wal_acceptors_list, WalProposerRust, WalProposerCleanup, + wal_acceptor_reconnect_timeout, wal_acceptors_list, WalProposerRust, WalProposerCleanup, syncSafekeepers, }, c_context, simtest::safekeeper::run_server, @@ -96,6 +96,7 @@ impl Test { unsafe { WalProposerCleanup(); + syncSafekeepers = true; wal_acceptors_list = list.into_raw(); wal_acceptor_reconnect_timeout = 1000; wal_acceptor_connection_timeout = 5000; @@ -125,6 +126,52 @@ impl Test { let lsn = Lsn::from_str(&res.1)?; Ok(lsn) } + + fn launch_walproposer(&self, lsn: Lsn) -> WalProposer { + let client_node = self.world.new_node(); + + // start the client thread + let guc = self.safekeepers_guc.clone(); + let ttid = self.ttid.clone(); + client_node.launch(move |_| { + let list = CString::new(guc).unwrap(); + + unsafe { + WalProposerCleanup(); + + // TODO: set LSN to a variable + + syncSafekeepers = false; + wal_acceptors_list = list.into_raw(); + wal_acceptor_reconnect_timeout = 1000; + wal_acceptor_connection_timeout = 5000; + neon_tenant_walproposer = + CString::new(ttid.tenant_id.to_string()).unwrap().into_raw(); + neon_timeline_walproposer = CString::new(ttid.timeline_id.to_string()) + .unwrap() + .into_raw(); + WalProposerRust(); + } + }); + + self.world.await_all(); + + WalProposer { + node: client_node, + } + } +} + +struct WalProposer { + node: Arc, +} + +impl WalProposer { + fn gen_wal_record(&self) -> Lsn { + // TODO: + + panic!("implement me") + } } #[test] @@ -142,3 +189,20 @@ fn sync_empty_safekeepers() { assert_eq!(lsn, Lsn(0)); println!("Sucessfully synced empty safekeepers at 0/0"); } + +#[test] +fn run_walproposer_generate_wal() { + logging::init(logging::LogFormat::Plain).unwrap(); + + let config = TestConfig::new(); + let test = config.start(1337); + + let lsn = test.sync_safekeepers().unwrap(); + assert_eq!(lsn, Lsn(0)); + println!("Sucessfully synced empty safekeepers at 0/0"); + + let wp = test.launch_walproposer(lsn); + let rec1 = wp.gen_wal_record(); + + // TODO: +} \ No newline at end of file diff --git a/pgxn/neon/walproposer.c b/pgxn/neon/walproposer.c index 69007269bf..ad5807d548 100644 --- a/pgxn/neon/walproposer.c +++ b/pgxn/neon/walproposer.c @@ -71,7 +71,7 @@ #include "walproposer.h" #include "walproposer_utils.h" -static bool syncSafekeepers = false; +bool syncSafekeepers = false; char *wal_acceptors_list; int wal_acceptor_reconnect_timeout; @@ -350,8 +350,48 @@ void WalProposerCleanup() void WalProposerRust() { + struct stat stat_buf; + walprop_log(LOG, "WalProposerRust"); - WalProposerSync(0, NULL); +#if PG_VERSION_NUM < 150000 + ThisTimeLineID = 1; +#endif + + /* + * Initialize postmaster_alive_fds as WaitEventSet checks them. + * + * Copied from InitPostmasterDeathWatchHandle() + */ + if (pipe(postmaster_alive_fds) < 0) + ereport(FATAL, + (errcode_for_file_access(), + errmsg_internal("could not create pipe to monitor postmaster death: %m"))); + if (fcntl(postmaster_alive_fds[POSTMASTER_FD_WATCH], F_SETFL, O_NONBLOCK) == -1) + ereport(FATAL, + (errcode_for_socket_access(), + errmsg_internal("could not set postmaster death monitoring pipe to nonblocking mode: %m"))); + + ChangeToDataDir(); + + /* Create pg_wal directory, if it doesn't exist */ + if (stat(XLOGDIR, &stat_buf) != 0) + { + ereport(LOG, (errmsg("creating missing WAL directory \"%s\"", XLOGDIR))); + if (MakePGDirectory(XLOGDIR) < 0) + { + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not create directory \"%s\": %m", + XLOGDIR))); + exit(1); + } + } + + WalProposerInit(0, 0); + + BackgroundWorkerUnblockSignals(); + + WalProposerStart(); } /* @@ -442,6 +482,7 @@ SimWaitEventSetWait(Safekeeper **sk, long timeout, WaitEvent *occurred_events) } else { Assert(false); } + // TODO: handle notification about new LSN available } #endif diff --git a/pgxn/neon/walproposer.h b/pgxn/neon/walproposer.h index d122a9e58a..29c07eeac2 100644 --- a/pgxn/neon/walproposer.h +++ b/pgxn/neon/walproposer.h @@ -46,6 +46,8 @@ */ #define WL_NO_EVENTS 0 +extern bool syncSafekeepers; + extern char *wal_acceptors_list; extern int wal_acceptor_reconnect_timeout; extern int wal_acceptor_connection_timeout; diff --git a/safekeeper/src/simlib/world.rs b/safekeeper/src/simlib/world.rs index 5d105cba97..e7792d91ee 100644 --- a/safekeeper/src/simlib/world.rs +++ b/safekeeper/src/simlib/world.rs @@ -380,6 +380,7 @@ pub enum NodeEvent { Accept(TCP), Closed(TCP), Message((AnyMessage, TCP)), + Internal(AnyMessage), WakeTimeout(u64), // TODO: close? }