LCOV - code coverage report
Current view: top level - safekeeper/tests/walproposer_sim - walproposer_api.rs (source / functions) Coverage Total Hit
Test: 1b0a6a0c05cee5a7de360813c8034804e105ce1c.info Lines: 93.4 % 486 454
Test Date: 2025-03-12 00:01:28 Functions: 97.1 % 138 134

            Line data    Source code
       1              : use std::cell::{RefCell, RefMut, UnsafeCell};
       2              : use std::ffi::CStr;
       3              : use std::sync::Arc;
       4              : 
       5              : use bytes::Bytes;
       6              : use desim::executor::{self, PollSome};
       7              : use desim::network::TCP;
       8              : use desim::node_os::NodeOs;
       9              : use desim::proto::{AnyMessage, NetEvent, NodeEvent};
      10              : use desim::world::NodeId;
      11              : use tracing::debug;
      12              : use utils::lsn::Lsn;
      13              : use walproposer::api_bindings::Level;
      14              : use walproposer::bindings::{
      15              :     NeonWALReadResult, SafekeeperStateDesiredEvents, WL_SOCKET_READABLE, WL_SOCKET_WRITEABLE,
      16              : };
      17              : use walproposer::walproposer::{ApiImpl, Config};
      18              : 
      19              : use super::walproposer_disk::DiskWalProposer;
      20              : 
      21              : /// Special state for each wp->sk connection.
      22              : struct SafekeeperConn {
      23              :     host: String,
      24              :     port: String,
      25              :     node_id: NodeId,
      26              :     // socket is Some(..) equals to connection is established
      27              :     socket: Option<TCP>,
      28              :     // connection is in progress
      29              :     is_connecting: bool,
      30              :     // START_WAL_PUSH is in progress
      31              :     is_start_wal_push: bool,
      32              :     // pointer to Safekeeper in walproposer for callbacks
      33              :     raw_ptr: *mut walproposer::bindings::Safekeeper,
      34              : }
      35              : 
      36              : impl SafekeeperConn {
      37        26376 :     pub fn new(host: String, port: String) -> Self {
      38        26376 :         // port number is the same as NodeId
      39        26376 :         let port_num = port.parse::<u32>().unwrap();
      40        26376 :         Self {
      41        26376 :             host,
      42        26376 :             port,
      43        26376 :             node_id: port_num,
      44        26376 :             socket: None,
      45        26376 :             is_connecting: false,
      46        26376 :             is_start_wal_push: false,
      47        26376 :             raw_ptr: std::ptr::null_mut(),
      48        26376 :         }
      49        26376 :     }
      50              : }
      51              : 
      52              : /// Simulation version of a postgres WaitEventSet. At pos 0 there is always
      53              : /// a special NodeEvents channel, which is used as a latch.
      54              : struct EventSet {
      55              :     os: NodeOs,
      56              :     // all pollable channels, 0 is always NodeEvent channel
      57              :     chans: Vec<Box<dyn PollSome>>,
      58              :     // 0 is always nullptr
      59              :     sk_ptrs: Vec<*mut walproposer::bindings::Safekeeper>,
      60              :     // event mask for each channel
      61              :     masks: Vec<u32>,
      62              : }
      63              : 
      64              : impl EventSet {
      65         8792 :     pub fn new(os: NodeOs) -> Self {
      66         8792 :         let node_events = os.node_events();
      67         8792 :         Self {
      68         8792 :             os,
      69         8792 :             chans: vec![Box::new(node_events)],
      70         8792 :             sk_ptrs: vec![std::ptr::null_mut()],
      71         8792 :             masks: vec![WL_SOCKET_READABLE],
      72         8792 :         }
      73         8792 :     }
      74              : 
      75              :     /// Leaves all readable channels at the beginning of the array.
      76        27419 :     fn sort_readable(&mut self) -> usize {
      77        27419 :         let mut cnt = 1;
      78        64466 :         for i in 1..self.chans.len() {
      79        64466 :             if self.masks[i] & WL_SOCKET_READABLE != 0 {
      80        64466 :                 self.chans.swap(i, cnt);
      81        64466 :                 self.sk_ptrs.swap(i, cnt);
      82        64466 :                 self.masks.swap(i, cnt);
      83        64466 :                 cnt += 1;
      84        64466 :             }
      85              :         }
      86        27419 :         cnt
      87        27419 :     }
      88              : 
      89        63873 :     fn update_event_set(&mut self, conn: &SafekeeperConn, event_mask: u32) {
      90        63873 :         let index = self
      91        63873 :             .sk_ptrs
      92        63873 :             .iter()
      93       241905 :             .position(|&ptr| ptr == conn.raw_ptr)
      94        63873 :             .expect("safekeeper should exist in event set");
      95        63873 :         self.masks[index] = event_mask;
      96        63873 :     }
      97              : 
      98        58814 :     fn add_safekeeper(&mut self, sk: &SafekeeperConn, event_mask: u32) {
      99       142317 :         for ptr in self.sk_ptrs.iter() {
     100       142317 :             assert!(*ptr != sk.raw_ptr);
     101              :         }
     102              : 
     103        58814 :         self.chans.push(Box::new(
     104        58814 :             sk.socket
     105        58814 :                 .as_ref()
     106        58814 :                 .expect("socket should not be closed")
     107        58814 :                 .recv_chan(),
     108        58814 :         ));
     109        58814 :         self.sk_ptrs.push(sk.raw_ptr);
     110        58814 :         self.masks.push(event_mask);
     111        58814 :     }
     112              : 
     113        37252 :     fn remove_safekeeper(&mut self, sk: &SafekeeperConn) {
     114        79093 :         let index = self.sk_ptrs.iter().position(|&ptr| ptr == sk.raw_ptr);
     115        37252 :         if index.is_none() {
     116            3 :             debug!("remove_safekeeper: sk={:?} not found", sk.raw_ptr);
     117            3 :             return;
     118        37249 :         }
     119        37249 :         let index = index.unwrap();
     120        37249 : 
     121        37249 :         self.chans.remove(index);
     122        37249 :         self.sk_ptrs.remove(index);
     123        37249 :         self.masks.remove(index);
     124        37249 : 
     125        37249 :         // to simulate the actual behaviour
     126        37249 :         self.refresh_event_set();
     127        37252 :     }
     128              : 
     129              :     /// Updates all masks to match the result of a SafekeeperStateDesiredEvents.
     130        43307 :     fn refresh_event_set(&mut self) {
     131       122558 :         for (i, mask) in self.masks.iter_mut().enumerate() {
     132       122558 :             if i == 0 {
     133        43307 :                 continue;
     134        79251 :             }
     135        79251 : 
     136        79251 :             let mut mask_sk: u32 = 0;
     137        79251 :             let mut mask_nwr: u32 = 0;
     138        79251 :             unsafe { SafekeeperStateDesiredEvents(self.sk_ptrs[i], &mut mask_sk, &mut mask_nwr) };
     139        79251 : 
     140        79251 :             if mask_sk != *mask {
     141            0 :                 debug!(
     142            0 :                     "refresh_event_set: sk={:?}, old_mask={:#b}, new_mask={:#b}",
     143            0 :                     self.sk_ptrs[i], *mask, mask_sk
     144              :                 );
     145            0 :                 *mask = mask_sk;
     146        79251 :             }
     147              :         }
     148        43307 :     }
     149              : 
     150              :     /// Wait for events on all channels.
     151        27419 :     fn wait(&mut self, timeout_millis: i64) -> walproposer::walproposer::WaitResult {
     152              :         // all channels are always writeable
     153        91885 :         for (i, mask) in self.masks.iter().enumerate() {
     154        91885 :             if *mask & WL_SOCKET_WRITEABLE != 0 {
     155            0 :                 return walproposer::walproposer::WaitResult::Network(
     156            0 :                     self.sk_ptrs[i],
     157            0 :                     WL_SOCKET_WRITEABLE,
     158            0 :                 );
     159        91885 :             }
     160              :         }
     161              : 
     162        27419 :         let cnt = self.sort_readable();
     163        27419 : 
     164        27419 :         let slice = &self.chans[0..cnt];
     165        27419 :         match executor::epoll_chans(slice, timeout_millis) {
     166        11135 :             None => walproposer::walproposer::WaitResult::Timeout,
     167              :             Some(0) => {
     168          498 :                 let msg = self.os.node_events().must_recv();
     169          498 :                 match msg {
     170          498 :                     NodeEvent::Internal(AnyMessage::Just32(0)) => {
     171          498 :                         // got a notification about new WAL available
     172          498 :                     }
     173            0 :                     NodeEvent::Internal(_) => unreachable!(),
     174            0 :                     NodeEvent::Accept(_) => unreachable!(),
     175              :                 }
     176          498 :                 walproposer::walproposer::WaitResult::Latch
     177              :             }
     178        15786 :             Some(index) => walproposer::walproposer::WaitResult::Network(
     179        15786 :                 self.sk_ptrs[index],
     180        15786 :                 WL_SOCKET_READABLE,
     181        15786 :             ),
     182              :         }
     183        27419 :     }
     184              : }
     185              : 
     186              : /// This struct handles all calls from walproposer into walproposer_api.
     187              : pub struct SimulationApi {
     188              :     os: NodeOs,
     189              :     safekeepers: RefCell<Vec<SafekeeperConn>>,
     190              :     disk: Arc<DiskWalProposer>,
     191              :     redo_start_lsn: Option<Lsn>,
     192              :     last_logged_commit_lsn: u64,
     193              :     shmem: UnsafeCell<walproposer::bindings::WalproposerShmemState>,
     194              :     config: Config,
     195              :     event_set: RefCell<Option<EventSet>>,
     196              : }
     197              : 
     198              : pub struct Args {
     199              :     pub os: NodeOs,
     200              :     pub config: Config,
     201              :     pub disk: Arc<DiskWalProposer>,
     202              :     pub redo_start_lsn: Option<Lsn>,
     203              : }
     204              : 
     205              : impl SimulationApi {
     206         8792 :     pub fn new(args: Args) -> Self {
     207         8792 :         // initialize connection state for each safekeeper
     208         8792 :         let sk_conns = args
     209         8792 :             .config
     210         8792 :             .safekeepers_list
     211         8792 :             .iter()
     212        26376 :             .map(|s| {
     213        26376 :                 SafekeeperConn::new(
     214        26376 :                     s.split(':').next().unwrap().to_string(),
     215        26376 :                     s.split(':').nth(1).unwrap().to_string(),
     216        26376 :                 )
     217        26376 :             })
     218         8792 :             .collect::<Vec<_>>();
     219         8792 : 
     220         8792 :         Self {
     221         8792 :             os: args.os,
     222         8792 :             safekeepers: RefCell::new(sk_conns),
     223         8792 :             disk: args.disk,
     224         8792 :             redo_start_lsn: args.redo_start_lsn,
     225         8792 :             last_logged_commit_lsn: 0,
     226         8792 :             shmem: UnsafeCell::new(walproposer::api_bindings::empty_shmem()),
     227         8792 :             config: args.config,
     228         8792 :             event_set: RefCell::new(None),
     229         8792 :         }
     230         8792 :     }
     231              : 
     232              :     /// Get SafekeeperConn for the given Safekeeper.
     233       288568 :     fn get_conn(&self, sk: &mut walproposer::bindings::Safekeeper) -> RefMut<'_, SafekeeperConn> {
     234       288568 :         let sk_port = unsafe { CStr::from_ptr(sk.port).to_str().unwrap() };
     235       288568 :         let state = self.safekeepers.borrow_mut();
     236       288568 :         RefMut::map(state, |v| {
     237       288568 :             v.iter_mut()
     238       575586 :                 .find(|conn| conn.port == sk_port)
     239       288568 :                 .expect("safekeeper conn not found by port")
     240       288568 :         })
     241       288568 :     }
     242              : }
     243              : 
     244              : impl ApiImpl for SimulationApi {
     245       309356 :     fn get_current_timestamp(&self) -> i64 {
     246       309356 :         debug!("get_current_timestamp");
     247              :         // PG TimestampTZ is microseconds, but simulation unit is assumed to be
     248              :         // milliseconds, so add 10^3
     249       309356 :         self.os.now() as i64 * 1000
     250       309356 :     }
     251              : 
     252          903 :     fn update_donor(&self, donor: &mut walproposer::bindings::Safekeeper, donor_lsn: u64) {
     253          903 :         let mut shmem = unsafe { *self.get_shmem_state() };
     254          903 :         shmem.propEpochStartLsn.value = donor_lsn;
     255          903 :         shmem.donor_conninfo = donor.conninfo;
     256          903 :     }
     257              : 
     258        32612 :     fn conn_status(
     259        32612 :         &self,
     260        32612 :         _: &mut walproposer::bindings::Safekeeper,
     261        32612 :     ) -> walproposer::bindings::WalProposerConnStatusType {
     262        32612 :         debug!("conn_status");
     263              :         // break the connection with a 10% chance
     264        32612 :         if self.os.random(100) < 10 {
     265         3205 :             walproposer::bindings::WalProposerConnStatusType_WP_CONNECTION_BAD
     266              :         } else {
     267        29407 :             walproposer::bindings::WalProposerConnStatusType_WP_CONNECTION_OK
     268              :         }
     269        32612 :     }
     270              : 
     271        32612 :     fn conn_connect_start(&self, sk: &mut walproposer::bindings::Safekeeper) {
     272        32612 :         debug!("conn_connect_start");
     273        32612 :         let mut conn = self.get_conn(sk);
     274        32612 : 
     275        32612 :         assert!(conn.socket.is_none());
     276        32612 :         let socket = self.os.open_tcp(conn.node_id);
     277        32612 :         conn.socket = Some(socket);
     278        32612 :         conn.raw_ptr = sk;
     279        32612 :         conn.is_connecting = true;
     280        32612 :     }
     281              : 
     282        29407 :     fn conn_connect_poll(
     283        29407 :         &self,
     284        29407 :         _: &mut walproposer::bindings::Safekeeper,
     285        29407 :     ) -> walproposer::bindings::WalProposerConnectPollStatusType {
     286        29407 :         debug!("conn_connect_poll");
     287              :         // TODO: break the connection here
     288        29407 :         walproposer::bindings::WalProposerConnectPollStatusType_WP_CONN_POLLING_OK
     289        29407 :     }
     290              : 
     291        29407 :     fn conn_send_query(&self, sk: &mut walproposer::bindings::Safekeeper, query: &str) -> bool {
     292        29407 :         debug!("conn_send_query: {}", query);
     293        29407 :         self.get_conn(sk).is_start_wal_push = true;
     294        29407 :         true
     295        29407 :     }
     296              : 
     297        29407 :     fn conn_get_query_result(
     298        29407 :         &self,
     299        29407 :         _: &mut walproposer::bindings::Safekeeper,
     300        29407 :     ) -> walproposer::bindings::WalProposerExecStatusType {
     301        29407 :         debug!("conn_get_query_result");
     302              :         // TODO: break the connection here
     303        29407 :         walproposer::bindings::WalProposerExecStatusType_WP_EXEC_SUCCESS_COPYBOTH
     304        29407 :     }
     305              : 
     306        17760 :     fn conn_async_read(
     307        17760 :         &self,
     308        17760 :         sk: &mut walproposer::bindings::Safekeeper,
     309        17760 :         vec: &mut Vec<u8>,
     310        17760 :     ) -> walproposer::bindings::PGAsyncReadResult {
     311        17760 :         debug!("conn_async_read");
     312        17760 :         let mut conn = self.get_conn(sk);
     313              : 
     314        17760 :         let socket = if let Some(socket) = conn.socket.as_mut() {
     315        17760 :             socket
     316              :         } else {
     317              :             // socket is already closed
     318            0 :             return walproposer::bindings::PGAsyncReadResult_PG_ASYNC_READ_FAIL;
     319              :         };
     320              : 
     321        17760 :         let msg = socket.recv_chan().try_recv();
     322              : 
     323        15634 :         match msg {
     324              :             None => {
     325              :                 // no message is ready
     326         2126 :                 walproposer::bindings::PGAsyncReadResult_PG_ASYNC_READ_TRY_AGAIN
     327              :             }
     328              :             Some(NetEvent::Closed) => {
     329              :                 // connection is closed
     330         7588 :                 debug!("conn_async_read: connection is closed");
     331         7588 :                 conn.socket = None;
     332         7588 :                 walproposer::bindings::PGAsyncReadResult_PG_ASYNC_READ_FAIL
     333              :             }
     334         8046 :             Some(NetEvent::Message(msg)) => {
     335              :                 // got a message
     336         8046 :                 let b = match msg {
     337         8046 :                     desim::proto::AnyMessage::Bytes(b) => b,
     338            0 :                     _ => unreachable!(),
     339              :                 };
     340         8046 :                 vec.extend_from_slice(&b);
     341         8046 :                 walproposer::bindings::PGAsyncReadResult_PG_ASYNC_READ_SUCCESS
     342              :             }
     343              :         }
     344        17760 :     }
     345              : 
     346        32660 :     fn conn_blocking_write(&self, sk: &mut walproposer::bindings::Safekeeper, buf: &[u8]) -> bool {
     347        32660 :         let mut conn = self.get_conn(sk);
     348        32660 :         debug!("conn_blocking_write to {}: {:?}", conn.node_id, buf);
     349        32660 :         let socket = conn.socket.as_mut().unwrap();
     350        32660 :         socket.send(desim::proto::AnyMessage::Bytes(Bytes::copy_from_slice(buf)));
     351        32660 :         true
     352        32660 :     }
     353              : 
     354         4880 :     fn conn_async_write(
     355         4880 :         &self,
     356         4880 :         sk: &mut walproposer::bindings::Safekeeper,
     357         4880 :         buf: &[u8],
     358         4880 :     ) -> walproposer::bindings::PGAsyncWriteResult {
     359         4880 :         let mut conn = self.get_conn(sk);
     360         4880 :         debug!("conn_async_write to {}: {:?}", conn.node_id, buf);
     361         4880 :         if let Some(socket) = conn.socket.as_mut() {
     362         4880 :             socket.send(desim::proto::AnyMessage::Bytes(Bytes::copy_from_slice(buf)));
     363         4880 :         } else {
     364              :             // connection is already closed
     365            0 :             debug!("conn_async_write: writing to a closed socket!");
     366              :             // TODO: maybe we should return error here?
     367              :         }
     368         4880 :         walproposer::bindings::PGAsyncWriteResult_PG_ASYNC_WRITE_SUCCESS
     369         4880 :     }
     370              : 
     371          948 :     fn wal_reader_allocate(&self, _: &mut walproposer::bindings::Safekeeper) -> NeonWALReadResult {
     372          948 :         debug!("wal_reader_allocate");
     373          948 :         walproposer::bindings::NeonWALReadResult_NEON_WALREAD_SUCCESS
     374          948 :     }
     375              : 
     376          918 :     fn wal_read(
     377          918 :         &self,
     378          918 :         _sk: &mut walproposer::bindings::Safekeeper,
     379          918 :         buf: &mut [u8],
     380          918 :         startpos: u64,
     381          918 :     ) -> NeonWALReadResult {
     382          918 :         self.disk.lock().read(startpos, buf);
     383          918 :         walproposer::bindings::NeonWALReadResult_NEON_WALREAD_SUCCESS
     384          918 :     }
     385              : 
     386         8792 :     fn init_event_set(&self, _: &mut walproposer::bindings::WalProposer) {
     387         8792 :         debug!("init_event_set");
     388         8792 :         let new_event_set = EventSet::new(self.os.clone());
     389         8792 :         let old_event_set = self.event_set.replace(Some(new_event_set));
     390         8792 :         assert!(old_event_set.is_none());
     391         8792 :     }
     392              : 
     393        63873 :     fn update_event_set(&self, sk: &mut walproposer::bindings::Safekeeper, event_mask: u32) {
     394        63873 :         debug!(
     395            0 :             "update_event_set, sk={:?}, events_mask={:#b}",
     396            0 :             sk as *mut walproposer::bindings::Safekeeper, event_mask
     397              :         );
     398        63873 :         let conn = self.get_conn(sk);
     399        63873 : 
     400        63873 :         self.event_set
     401        63873 :             .borrow_mut()
     402        63873 :             .as_mut()
     403        63873 :             .unwrap()
     404        63873 :             .update_event_set(&conn, event_mask);
     405        63873 :     }
     406              : 
     407        58814 :     fn add_safekeeper_event_set(
     408        58814 :         &self,
     409        58814 :         sk: &mut walproposer::bindings::Safekeeper,
     410        58814 :         event_mask: u32,
     411        58814 :     ) {
     412        58814 :         debug!(
     413            0 :             "add_safekeeper_event_set, sk={:?}, events_mask={:#b}",
     414            0 :             sk as *mut walproposer::bindings::Safekeeper, event_mask
     415              :         );
     416              : 
     417        58814 :         self.event_set
     418        58814 :             .borrow_mut()
     419        58814 :             .as_mut()
     420        58814 :             .unwrap()
     421        58814 :             .add_safekeeper(&self.get_conn(sk), event_mask);
     422        58814 :     }
     423              : 
     424        37252 :     fn rm_safekeeper_event_set(&self, sk: &mut walproposer::bindings::Safekeeper) {
     425        37252 :         debug!(
     426            0 :             "rm_safekeeper_event_set, sk={:?}",
     427            0 :             sk as *mut walproposer::bindings::Safekeeper,
     428              :         );
     429              : 
     430        37252 :         self.event_set
     431        37252 :             .borrow_mut()
     432        37252 :             .as_mut()
     433        37252 :             .unwrap()
     434        37252 :             .remove_safekeeper(&self.get_conn(sk));
     435        37252 :     }
     436              : 
     437         6058 :     fn active_state_update_event_set(&self, sk: &mut walproposer::bindings::Safekeeper) {
     438         6058 :         debug!("active_state_update_event_set");
     439              : 
     440         6058 :         assert!(sk.state == walproposer::bindings::SafekeeperState_SS_ACTIVE);
     441         6058 :         self.event_set
     442         6058 :             .borrow_mut()
     443         6058 :             .as_mut()
     444         6058 :             .unwrap()
     445         6058 :             .refresh_event_set();
     446         6058 :     }
     447              : 
     448        16787 :     fn wal_reader_events(&self, _sk: &mut walproposer::bindings::Safekeeper) -> u32 {
     449        16787 :         0
     450        16787 :     }
     451              : 
     452        86233 :     fn wait_event_set(
     453        86233 :         &self,
     454        86233 :         _: &mut walproposer::bindings::WalProposer,
     455        86233 :         timeout_millis: i64,
     456        86233 :     ) -> walproposer::walproposer::WaitResult {
     457        86233 :         // TODO: handle multiple stages as part of the simulation (e.g. connect, start_wal_push, etc)
     458        86233 :         let mut conns = self.safekeepers.borrow_mut();
     459       199829 :         for conn in conns.iter_mut() {
     460       199829 :             if conn.socket.is_some() && conn.is_connecting {
     461        29407 :                 conn.is_connecting = false;
     462        29407 :                 debug!("wait_event_set, connecting to {}:{}", conn.host, conn.port);
     463        29407 :                 return walproposer::walproposer::WaitResult::Network(
     464        29407 :                     conn.raw_ptr,
     465        29407 :                     WL_SOCKET_READABLE | WL_SOCKET_WRITEABLE,
     466        29407 :                 );
     467       170422 :             }
     468       170422 :             if conn.socket.is_some() && conn.is_start_wal_push {
     469        29407 :                 conn.is_start_wal_push = false;
     470        29407 :                 debug!(
     471            0 :                     "wait_event_set, start wal push to {}:{}",
     472              :                     conn.host, conn.port
     473              :                 );
     474        29407 :                 return walproposer::walproposer::WaitResult::Network(
     475        29407 :                     conn.raw_ptr,
     476        29407 :                     WL_SOCKET_READABLE,
     477        29407 :                 );
     478       141015 :             }
     479              :         }
     480        27419 :         drop(conns);
     481        27419 : 
     482        27419 :         let res = self
     483        27419 :             .event_set
     484        27419 :             .borrow_mut()
     485        27419 :             .as_mut()
     486        27419 :             .unwrap()
     487        27419 :             .wait(timeout_millis);
     488        27419 : 
     489        27419 :         debug!(
     490            0 :             "wait_event_set, timeout_millis={}, res={:?}",
     491              :             timeout_millis, res,
     492              :         );
     493        19057 :         res
     494        77871 :     }
     495              : 
     496            0 :     fn strong_random(&self, buf: &mut [u8]) -> bool {
     497            0 :         debug!("strong_random");
     498            0 :         buf.fill(0);
     499            0 :         true
     500            0 :     }
     501              : 
     502          325 :     fn finish_sync_safekeepers(&self, lsn: u64) {
     503          325 :         debug!("finish_sync_safekeepers, lsn={}", lsn);
     504          325 :         executor::exit(0, Lsn(lsn).to_string());
     505          325 :     }
     506              : 
     507       144071 :     fn log_internal(&self, _wp: &mut walproposer::bindings::WalProposer, level: Level, msg: &str) {
     508       144071 :         debug!("wp_log[{}] {}", level, msg);
     509       144071 :         if level == Level::Fatal || level == Level::Panic {
     510           40 :             if msg.contains("rejects our connection request with term") {
     511           32 :                 // collected quorum with lower term, then got rejected by next connected safekeeper
     512           32 :                 executor::exit(1, msg.to_owned());
     513           32 :             }
     514           40 :             if msg.contains("collected propTermStartLsn") && msg.contains(", but basebackup LSN ") {
     515            0 :                 // sync-safekeepers collected wrong quorum, walproposer collected another quorum
     516            0 :                 executor::exit(1, msg.to_owned());
     517           40 :             }
     518           40 :             if msg.contains("failed to download WAL for logical replicaiton") {
     519            8 :                 // Recovery connection broken and recovery was failed
     520            8 :                 executor::exit(1, msg.to_owned());
     521           32 :             }
     522           40 :             if msg.contains("missing majority of votes, collected") {
     523            0 :                 // Voting bug when safekeeper disconnects after voting
     524            0 :                 executor::exit(1, msg.to_owned());
     525           40 :             }
     526           40 :             panic!("unknown FATAL error from walproposer: {}", msg);
     527       144031 :         }
     528       144031 :     }
     529              : 
     530          512 :     fn after_election(&self, wp: &mut walproposer::bindings::WalProposer) {
     531          512 :         let prop_lsn = wp.propTermStartLsn;
     532          512 :         let prop_term = wp.propTerm;
     533          512 : 
     534          512 :         let mut prev_lsn: u64 = 0;
     535          512 :         let mut prev_term: u64 = 0;
     536          512 : 
     537          512 :         unsafe {
     538          512 :             let history = wp.propTermHistory.entries;
     539          512 :             let len = wp.propTermHistory.n_entries as usize;
     540          512 :             if len > 1 {
     541          460 :                 let entry = *history.wrapping_add(len - 2);
     542          460 :                 prev_lsn = entry.lsn;
     543          460 :                 prev_term = entry.term;
     544          460 :             }
     545              :         }
     546              : 
     547          512 :         let msg = format!(
     548          512 :             "prop_elected;{};{};{};{}",
     549          512 :             prop_lsn, prop_term, prev_lsn, prev_term
     550          512 :         );
     551          512 : 
     552          512 :         debug!(msg);
     553          512 :         self.os.log_event(msg);
     554          512 :     }
     555              : 
     556          213 :     fn get_redo_start_lsn(&self) -> u64 {
     557          213 :         debug!("get_redo_start_lsn -> {:?}", self.redo_start_lsn);
     558          213 :         self.redo_start_lsn.expect("redo_start_lsn is not set").0
     559          213 :     }
     560              : 
     561         1927 :     fn get_shmem_state(&self) -> *mut walproposer::bindings::WalproposerShmemState {
     562         1927 :         self.shmem.get()
     563         1927 :     }
     564              : 
     565          160 :     fn start_streaming(
     566          160 :         &self,
     567          160 :         startpos: u64,
     568          160 :         callback: &walproposer::walproposer::StreamingCallback,
     569          160 :     ) {
     570          160 :         let disk = &self.disk;
     571          160 :         let disk_lsn = disk.lock().flush_rec_ptr().0;
     572          160 :         debug!("start_streaming at {} (disk_lsn={})", startpos, disk_lsn);
     573          160 :         if startpos < disk_lsn {
     574           37 :             debug!(
     575            0 :                 "startpos < disk_lsn, it means we wrote some transaction even before streaming started"
     576              :             );
     577          123 :         }
     578          160 :         assert!(startpos <= disk_lsn);
     579          160 :         let mut broadcasted = Lsn(startpos);
     580              : 
     581              :         loop {
     582          666 :             let available = disk.lock().flush_rec_ptr();
     583          666 :             assert!(available >= broadcasted);
     584          506 :             callback.broadcast(broadcasted, available);
     585          506 :             broadcasted = available;
     586          506 :             callback.poll();
     587              :         }
     588              :     }
     589              : 
     590         2286 :     fn process_safekeeper_feedback(
     591         2286 :         &mut self,
     592         2286 :         wp: &mut walproposer::bindings::WalProposer,
     593         2286 :         _sk: &mut walproposer::bindings::Safekeeper,
     594         2286 :     ) {
     595         2286 :         debug!("process_safekeeper_feedback, commit_lsn={}", wp.commitLsn);
     596         2286 :         if wp.commitLsn > self.last_logged_commit_lsn {
     597          518 :             self.os.log_event(format!("commit_lsn;{}", wp.commitLsn));
     598          518 :             self.last_logged_commit_lsn = wp.commitLsn;
     599         1768 :         }
     600         2286 :     }
     601              : 
     602          101 :     fn get_flush_rec_ptr(&self) -> u64 {
     603          101 :         let lsn = self.disk.lock().flush_rec_ptr();
     604          101 :         debug!("get_flush_rec_ptr: {}", lsn);
     605          101 :         lsn.0
     606          101 :     }
     607              : 
     608          512 :     fn recovery_download(
     609          512 :         &self,
     610          512 :         wp: &mut walproposer::bindings::WalProposer,
     611          512 :         sk: &mut walproposer::bindings::Safekeeper,
     612          512 :     ) -> bool {
     613          512 :         let mut startpos = wp.truncateLsn;
     614          512 :         let endpos = wp.propTermStartLsn;
     615          512 : 
     616          512 :         if startpos == endpos {
     617          252 :             debug!("recovery_download: nothing to download");
     618          252 :             return true;
     619          260 :         }
     620          260 : 
     621          260 :         debug!("recovery_download from {} to {}", startpos, endpos,);
     622              : 
     623          260 :         let replication_prompt = format!(
     624          260 :             "START_REPLICATION {} {} {} {}",
     625          260 :             self.config.ttid.tenant_id, self.config.ttid.timeline_id, startpos, endpos,
     626          260 :         );
     627          260 :         let async_conn = self.get_conn(sk);
     628          260 : 
     629          260 :         let conn = self.os.open_tcp(async_conn.node_id);
     630          260 :         conn.send(desim::proto::AnyMessage::Bytes(replication_prompt.into()));
     631          260 : 
     632          260 :         let chan = conn.recv_chan();
     633          447 :         while startpos < endpos {
     634          260 :             let event = chan.recv();
     635          252 :             match event {
     636              :                 NetEvent::Closed => {
     637            8 :                     debug!("connection closed in recovery");
     638            8 :                     break;
     639              :                 }
     640          252 :                 NetEvent::Message(AnyMessage::Bytes(b)) => {
     641          252 :                     debug!("got recovery bytes from safekeeper");
     642          187 :                     self.disk.lock().write(startpos, &b);
     643          187 :                     startpos += b.len() as u64;
     644              :                 }
     645            0 :                 NetEvent::Message(_) => unreachable!(),
     646              :             }
     647              :         }
     648              : 
     649          195 :         debug!("recovery finished at {}", startpos);
     650              : 
     651          195 :         startpos == endpos
     652          447 :     }
     653              : 
     654        11050 :     fn conn_finish(&self, sk: &mut walproposer::bindings::Safekeeper) {
     655        11050 :         let mut conn = self.get_conn(sk);
     656        11050 :         debug!("conn_finish to {}", conn.node_id);
     657        11050 :         if let Some(socket) = conn.socket.as_mut() {
     658         3459 :             socket.close();
     659         7591 :         } else {
     660         7591 :             // connection is already closed
     661         7591 :         }
     662        11050 :         conn.socket = None;
     663        11050 :     }
     664              : 
     665        10793 :     fn conn_error_message(&self, _sk: &mut walproposer::bindings::Safekeeper) -> String {
     666        10793 :         "connection is closed, probably".into()
     667        10793 :     }
     668              : }
        

Generated by: LCOV version 2.1-beta