LCOV - code coverage report
Current view: top level - safekeeper/tests/walproposer_sim - walproposer_api.rs (source / functions) Coverage Total Hit
Test: 5fe7fa8d483b39476409aee736d6d5e32728bfac.info Lines: 93.8 % 486 456
Test Date: 2025-03-12 16:10:49 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        26505 :     pub fn new(host: String, port: String) -> Self {
      38        26505 :         // port number is the same as NodeId
      39        26505 :         let port_num = port.parse::<u32>().unwrap();
      40        26505 :         Self {
      41        26505 :             host,
      42        26505 :             port,
      43        26505 :             node_id: port_num,
      44        26505 :             socket: None,
      45        26505 :             is_connecting: false,
      46        26505 :             is_start_wal_push: false,
      47        26505 :             raw_ptr: std::ptr::null_mut(),
      48        26505 :         }
      49        26505 :     }
      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         8835 :     pub fn new(os: NodeOs) -> Self {
      66         8835 :         let node_events = os.node_events();
      67         8835 :         Self {
      68         8835 :             os,
      69         8835 :             chans: vec![Box::new(node_events)],
      70         8835 :             sk_ptrs: vec![std::ptr::null_mut()],
      71         8835 :             masks: vec![WL_SOCKET_READABLE],
      72         8835 :         }
      73         8835 :     }
      74              : 
      75              :     /// Leaves all readable channels at the beginning of the array.
      76        26655 :     fn sort_readable(&mut self) -> usize {
      77        26655 :         let mut cnt = 1;
      78        62728 :         for i in 1..self.chans.len() {
      79        62728 :             if self.masks[i] & WL_SOCKET_READABLE != 0 {
      80        62728 :                 self.chans.swap(i, cnt);
      81        62728 :                 self.sk_ptrs.swap(i, cnt);
      82        62728 :                 self.masks.swap(i, cnt);
      83        62728 :                 cnt += 1;
      84        62728 :             }
      85              :         }
      86        26655 :         cnt
      87        26655 :     }
      88              : 
      89        64048 :     fn update_event_set(&mut self, conn: &SafekeeperConn, event_mask: u32) {
      90        64048 :         let index = self
      91        64048 :             .sk_ptrs
      92        64048 :             .iter()
      93       242738 :             .position(|&ptr| ptr == conn.raw_ptr)
      94        64048 :             .expect("safekeeper should exist in event set");
      95        64048 :         self.masks[index] = event_mask;
      96        64048 :     }
      97              : 
      98        58902 :     fn add_safekeeper(&mut self, sk: &SafekeeperConn, event_mask: u32) {
      99       142607 :         for ptr in self.sk_ptrs.iter() {
     100       142607 :             assert!(*ptr != sk.raw_ptr);
     101              :         }
     102              : 
     103        58902 :         self.chans.push(Box::new(
     104        58902 :             sk.socket
     105        58902 :                 .as_ref()
     106        58902 :                 .expect("socket should not be closed")
     107        58902 :                 .recv_chan(),
     108        58902 :         ));
     109        58902 :         self.sk_ptrs.push(sk.raw_ptr);
     110        58902 :         self.masks.push(event_mask);
     111        58902 :     }
     112              : 
     113        37008 :     fn remove_safekeeper(&mut self, sk: &SafekeeperConn) {
     114        78811 :         let index = self.sk_ptrs.iter().position(|&ptr| ptr == sk.raw_ptr);
     115        37008 :         if index.is_none() {
     116            1 :             debug!("remove_safekeeper: sk={:?} not found", sk.raw_ptr);
     117            1 :             return;
     118        37007 :         }
     119        37007 :         let index = index.unwrap();
     120        37007 : 
     121        37007 :         self.chans.remove(index);
     122        37007 :         self.sk_ptrs.remove(index);
     123        37007 :         self.masks.remove(index);
     124        37007 : 
     125        37007 :         // to simulate the actual behaviour
     126        37007 :         self.refresh_event_set();
     127        37008 :     }
     128              : 
     129              :     /// Updates all masks to match the result of a SafekeeperStateDesiredEvents.
     130        42112 :     fn refresh_event_set(&mut self) {
     131       118217 :         for (i, mask) in self.masks.iter_mut().enumerate() {
     132       118217 :             if i == 0 {
     133        42112 :                 continue;
     134        76105 :             }
     135        76105 : 
     136        76105 :             let mut mask_sk: u32 = 0;
     137        76105 :             let mut mask_nwr: u32 = 0;
     138        76105 :             unsafe { SafekeeperStateDesiredEvents(self.sk_ptrs[i], &mut mask_sk, &mut mask_nwr) };
     139        76105 : 
     140        76105 :             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        76105 :             }
     147              :         }
     148        42112 :     }
     149              : 
     150              :     /// Wait for events on all channels.
     151        26655 :     fn wait(&mut self, timeout_millis: i64) -> walproposer::walproposer::WaitResult {
     152              :         // all channels are always writeable
     153        89383 :         for (i, mask) in self.masks.iter().enumerate() {
     154        89383 :             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        89383 :             }
     160              :         }
     161              : 
     162        26655 :         let cnt = self.sort_readable();
     163        26655 : 
     164        26655 :         let slice = &self.chans[0..cnt];
     165        26655 :         match executor::epoll_chans(slice, timeout_millis) {
     166        11119 :             None => walproposer::walproposer::WaitResult::Timeout,
     167              :             Some(0) => {
     168          511 :                 let msg = self.os.node_events().must_recv();
     169          511 :                 match msg {
     170          511 :                     NodeEvent::Internal(AnyMessage::Just32(0)) => {
     171          511 :                         // got a notification about new WAL available
     172          511 :                     }
     173            0 :                     NodeEvent::Internal(_) => unreachable!(),
     174            0 :                     NodeEvent::Accept(_) => unreachable!(),
     175              :                 }
     176          511 :                 walproposer::walproposer::WaitResult::Latch
     177              :             }
     178        15025 :             Some(index) => walproposer::walproposer::WaitResult::Network(
     179        15025 :                 self.sk_ptrs[index],
     180        15025 :                 WL_SOCKET_READABLE,
     181        15025 :             ),
     182              :         }
     183        26655 :     }
     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         8835 :     pub fn new(args: Args) -> Self {
     207         8835 :         // initialize connection state for each safekeeper
     208         8835 :         let sk_conns = args
     209         8835 :             .config
     210         8835 :             .safekeepers_list
     211         8835 :             .iter()
     212        26505 :             .map(|s| {
     213        26505 :                 SafekeeperConn::new(
     214        26505 :                     s.split(':').next().unwrap().to_string(),
     215        26505 :                     s.split(':').nth(1).unwrap().to_string(),
     216        26505 :                 )
     217        26505 :             })
     218         8835 :             .collect::<Vec<_>>();
     219         8835 : 
     220         8835 :         Self {
     221         8835 :             os: args.os,
     222         8835 :             safekeepers: RefCell::new(sk_conns),
     223         8835 :             disk: args.disk,
     224         8835 :             redo_start_lsn: args.redo_start_lsn,
     225         8835 :             last_logged_commit_lsn: 0,
     226         8835 :             shmem: UnsafeCell::new(walproposer::api_bindings::empty_shmem()),
     227         8835 :             config: args.config,
     228         8835 :             event_set: RefCell::new(None),
     229         8835 :         }
     230         8835 :     }
     231              : 
     232              :     /// Get SafekeeperConn for the given Safekeeper.
     233       286614 :     fn get_conn(&self, sk: &mut walproposer::bindings::Safekeeper) -> RefMut<'_, SafekeeperConn> {
     234       286614 :         let sk_port = unsafe { CStr::from_ptr(sk.port).to_str().unwrap() };
     235       286614 :         let state = self.safekeepers.borrow_mut();
     236       286614 :         RefMut::map(state, |v| {
     237       286614 :             v.iter_mut()
     238       571826 :                 .find(|conn| conn.port == sk_port)
     239       286614 :                 .expect("safekeeper conn not found by port")
     240       286614 :         })
     241       286614 :     }
     242              : }
     243              : 
     244              : impl ApiImpl for SimulationApi {
     245       306818 :     fn get_current_timestamp(&self) -> i64 {
     246       306818 :         debug!("get_current_timestamp");
     247              :         // PG TimestampTZ is microseconds, but simulation unit is assumed to be
     248              :         // milliseconds, so add 10^3
     249       306818 :         self.os.now() as i64 * 1000
     250       306818 :     }
     251              : 
     252          706 :     fn update_donor(&self, donor: &mut walproposer::bindings::Safekeeper, donor_lsn: u64) {
     253          706 :         let mut shmem = unsafe { *self.get_shmem_state() };
     254          706 :         shmem.propEpochStartLsn.value = donor_lsn;
     255          706 :         shmem.donor_conninfo = donor.conninfo;
     256          706 :     }
     257              : 
     258        32683 :     fn conn_status(
     259        32683 :         &self,
     260        32683 :         _: &mut walproposer::bindings::Safekeeper,
     261        32683 :     ) -> walproposer::bindings::WalProposerConnStatusType {
     262        32683 :         debug!("conn_status");
     263              :         // break the connection with a 10% chance
     264        32683 :         if self.os.random(100) < 10 {
     265         3232 :             walproposer::bindings::WalProposerConnStatusType_WP_CONNECTION_BAD
     266              :         } else {
     267        29451 :             walproposer::bindings::WalProposerConnStatusType_WP_CONNECTION_OK
     268              :         }
     269        32683 :     }
     270              : 
     271        32683 :     fn conn_connect_start(&self, sk: &mut walproposer::bindings::Safekeeper) {
     272        32683 :         debug!("conn_connect_start");
     273        32683 :         let mut conn = self.get_conn(sk);
     274        32683 : 
     275        32683 :         assert!(conn.socket.is_none());
     276        32683 :         let socket = self.os.open_tcp(conn.node_id);
     277        32683 :         conn.socket = Some(socket);
     278        32683 :         conn.raw_ptr = sk;
     279        32683 :         conn.is_connecting = true;
     280        32683 :     }
     281              : 
     282        29451 :     fn conn_connect_poll(
     283        29451 :         &self,
     284        29451 :         _: &mut walproposer::bindings::Safekeeper,
     285        29451 :     ) -> walproposer::bindings::WalProposerConnectPollStatusType {
     286        29451 :         debug!("conn_connect_poll");
     287              :         // TODO: break the connection here
     288        29451 :         walproposer::bindings::WalProposerConnectPollStatusType_WP_CONN_POLLING_OK
     289        29451 :     }
     290              : 
     291        29451 :     fn conn_send_query(&self, sk: &mut walproposer::bindings::Safekeeper, query: &str) -> bool {
     292        29451 :         debug!("conn_send_query: {}", query);
     293        29451 :         self.get_conn(sk).is_start_wal_push = true;
     294        29451 :         true
     295        29451 :     }
     296              : 
     297        29451 :     fn conn_get_query_result(
     298        29451 :         &self,
     299        29451 :         _: &mut walproposer::bindings::Safekeeper,
     300        29451 :     ) -> walproposer::bindings::WalProposerExecStatusType {
     301        29451 :         debug!("conn_get_query_result");
     302              :         // TODO: break the connection here
     303        29451 :         walproposer::bindings::WalProposerExecStatusType_WP_EXEC_SUCCESS_COPYBOTH
     304        29451 :     }
     305              : 
     306        16698 :     fn conn_async_read(
     307        16698 :         &self,
     308        16698 :         sk: &mut walproposer::bindings::Safekeeper,
     309        16698 :         vec: &mut Vec<u8>,
     310        16698 :     ) -> walproposer::bindings::PGAsyncReadResult {
     311        16698 :         debug!("conn_async_read");
     312        16698 :         let mut conn = self.get_conn(sk);
     313              : 
     314        16698 :         let socket = if let Some(socket) = conn.socket.as_mut() {
     315        16698 :             socket
     316              :         } else {
     317              :             // socket is already closed
     318            0 :             return walproposer::bindings::PGAsyncReadResult_PG_ASYNC_READ_FAIL;
     319              :         };
     320              : 
     321        16698 :         let msg = socket.recv_chan().try_recv();
     322              : 
     323        14890 :         match msg {
     324              :             None => {
     325              :                 // no message is ready
     326         1808 :                 walproposer::bindings::PGAsyncReadResult_PG_ASYNC_READ_TRY_AGAIN
     327              :             }
     328              :             Some(NetEvent::Closed) => {
     329              :                 // connection is closed
     330         7317 :                 debug!("conn_async_read: connection is closed");
     331         7317 :                 conn.socket = None;
     332         7317 :                 walproposer::bindings::PGAsyncReadResult_PG_ASYNC_READ_FAIL
     333              :             }
     334         7573 :             Some(NetEvent::Message(msg)) => {
     335              :                 // got a message
     336         7573 :                 let b = match msg {
     337         7573 :                     desim::proto::AnyMessage::Bytes(b) => b,
     338            0 :                     _ => unreachable!(),
     339              :                 };
     340         7573 :                 vec.extend_from_slice(&b);
     341         7573 :                 walproposer::bindings::PGAsyncReadResult_PG_ASYNC_READ_SUCCESS
     342              :             }
     343              :         }
     344        16698 :     }
     345              : 
     346        32763 :     fn conn_blocking_write(&self, sk: &mut walproposer::bindings::Safekeeper, buf: &[u8]) -> bool {
     347        32763 :         let mut conn = self.get_conn(sk);
     348        32763 :         debug!("conn_blocking_write to {}: {:?}", conn.node_id, buf);
     349        32763 :         let socket = conn.socket.as_mut().unwrap();
     350        32763 :         socket.send(desim::proto::AnyMessage::Bytes(Bytes::copy_from_slice(buf)));
     351        32763 :         true
     352        32763 :     }
     353              : 
     354         4049 :     fn conn_async_write(
     355         4049 :         &self,
     356         4049 :         sk: &mut walproposer::bindings::Safekeeper,
     357         4049 :         buf: &[u8],
     358         4049 :     ) -> walproposer::bindings::PGAsyncWriteResult {
     359         4049 :         let mut conn = self.get_conn(sk);
     360         4049 :         debug!("conn_async_write to {}: {:?}", conn.node_id, buf);
     361         4049 :         if let Some(socket) = conn.socket.as_mut() {
     362         4049 :             socket.send(desim::proto::AnyMessage::Bytes(Bytes::copy_from_slice(buf)));
     363         4049 :         } 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         4049 :         walproposer::bindings::PGAsyncWriteResult_PG_ASYNC_WRITE_SUCCESS
     369         4049 :     }
     370              : 
     371          752 :     fn wal_reader_allocate(&self, _: &mut walproposer::bindings::Safekeeper) -> NeonWALReadResult {
     372          752 :         debug!("wal_reader_allocate");
     373          752 :         walproposer::bindings::NeonWALReadResult_NEON_WALREAD_SUCCESS
     374          752 :     }
     375              : 
     376          881 :     fn wal_read(
     377          881 :         &self,
     378          881 :         _sk: &mut walproposer::bindings::Safekeeper,
     379          881 :         buf: &mut [u8],
     380          881 :         startpos: u64,
     381          881 :     ) -> NeonWALReadResult {
     382          881 :         self.disk.lock().read(startpos, buf);
     383          881 :         walproposer::bindings::NeonWALReadResult_NEON_WALREAD_SUCCESS
     384          881 :     }
     385              : 
     386         8835 :     fn init_event_set(&self, _: &mut walproposer::bindings::WalProposer) {
     387         8835 :         debug!("init_event_set");
     388         8835 :         let new_event_set = EventSet::new(self.os.clone());
     389         8835 :         let old_event_set = self.event_set.replace(Some(new_event_set));
     390         8835 :         assert!(old_event_set.is_none());
     391         8835 :     }
     392              : 
     393        64048 :     fn update_event_set(&self, sk: &mut walproposer::bindings::Safekeeper, event_mask: u32) {
     394        64048 :         debug!(
     395            0 :             "update_event_set, sk={:?}, events_mask={:#b}",
     396            0 :             sk as *mut walproposer::bindings::Safekeeper, event_mask
     397              :         );
     398        64048 :         let conn = self.get_conn(sk);
     399        64048 : 
     400        64048 :         self.event_set
     401        64048 :             .borrow_mut()
     402        64048 :             .as_mut()
     403        64048 :             .unwrap()
     404        64048 :             .update_event_set(&conn, event_mask);
     405        64048 :     }
     406              : 
     407        58902 :     fn add_safekeeper_event_set(
     408        58902 :         &self,
     409        58902 :         sk: &mut walproposer::bindings::Safekeeper,
     410        58902 :         event_mask: u32,
     411        58902 :     ) {
     412        58902 :         debug!(
     413            0 :             "add_safekeeper_event_set, sk={:?}, events_mask={:#b}",
     414            0 :             sk as *mut walproposer::bindings::Safekeeper, event_mask
     415              :         );
     416              : 
     417        58902 :         self.event_set
     418        58902 :             .borrow_mut()
     419        58902 :             .as_mut()
     420        58902 :             .unwrap()
     421        58902 :             .add_safekeeper(&self.get_conn(sk), event_mask);
     422        58902 :     }
     423              : 
     424        37008 :     fn rm_safekeeper_event_set(&self, sk: &mut walproposer::bindings::Safekeeper) {
     425        37008 :         debug!(
     426            0 :             "rm_safekeeper_event_set, sk={:?}",
     427            0 :             sk as *mut walproposer::bindings::Safekeeper,
     428              :         );
     429              : 
     430        37008 :         self.event_set
     431        37008 :             .borrow_mut()
     432        37008 :             .as_mut()
     433        37008 :             .unwrap()
     434        37008 :             .remove_safekeeper(&self.get_conn(sk));
     435        37008 :     }
     436              : 
     437         5105 :     fn active_state_update_event_set(&self, sk: &mut walproposer::bindings::Safekeeper) {
     438         5105 :         debug!("active_state_update_event_set");
     439              : 
     440         5105 :         assert!(sk.state == walproposer::bindings::SafekeeperState_SS_ACTIVE);
     441         5105 :         self.event_set
     442         5105 :             .borrow_mut()
     443         5105 :             .as_mut()
     444         5105 :             .unwrap()
     445         5105 :             .refresh_event_set();
     446         5105 :     }
     447              : 
     448        14096 :     fn wal_reader_events(&self, _sk: &mut walproposer::bindings::Safekeeper) -> u32 {
     449        14096 :         0
     450        14096 :     }
     451              : 
     452        85557 :     fn wait_event_set(
     453        85557 :         &self,
     454        85557 :         _: &mut walproposer::bindings::WalProposer,
     455        85557 :         timeout_millis: i64,
     456        85557 :     ) -> walproposer::walproposer::WaitResult {
     457        85557 :         // TODO: handle multiple stages as part of the simulation (e.g. connect, start_wal_push, etc)
     458        85557 :         let mut conns = self.safekeepers.borrow_mut();
     459       197695 :         for conn in conns.iter_mut() {
     460       197695 :             if conn.socket.is_some() && conn.is_connecting {
     461        29451 :                 conn.is_connecting = false;
     462        29451 :                 debug!("wait_event_set, connecting to {}:{}", conn.host, conn.port);
     463        29451 :                 return walproposer::walproposer::WaitResult::Network(
     464        29451 :                     conn.raw_ptr,
     465        29451 :                     WL_SOCKET_READABLE | WL_SOCKET_WRITEABLE,
     466        29451 :                 );
     467       168244 :             }
     468       168244 :             if conn.socket.is_some() && conn.is_start_wal_push {
     469        29451 :                 conn.is_start_wal_push = false;
     470        29451 :                 debug!(
     471            0 :                     "wait_event_set, start wal push to {}:{}",
     472              :                     conn.host, conn.port
     473              :                 );
     474        29451 :                 return walproposer::walproposer::WaitResult::Network(
     475        29451 :                     conn.raw_ptr,
     476        29451 :                     WL_SOCKET_READABLE,
     477        29451 :                 );
     478       138793 :             }
     479              :         }
     480        26655 :         drop(conns);
     481        26655 : 
     482        26655 :         let res = self
     483        26655 :             .event_set
     484        26655 :             .borrow_mut()
     485        26655 :             .as_mut()
     486        26655 :             .unwrap()
     487        26655 :             .wait(timeout_millis);
     488        26655 : 
     489        26655 :         debug!(
     490            0 :             "wait_event_set, timeout_millis={}, res={:?}",
     491              :             timeout_millis, res,
     492              :         );
     493        18244 :         res
     494        77146 :     }
     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          311 :     fn finish_sync_safekeepers(&self, lsn: u64) {
     503          311 :         debug!("finish_sync_safekeepers, lsn={}", lsn);
     504          311 :         executor::exit(0, Lsn(lsn).to_string());
     505          311 :     }
     506              : 
     507       142783 :     fn log_internal(&self, _wp: &mut walproposer::bindings::WalProposer, level: Level, msg: &str) {
     508       142783 :         debug!("wp_log[{}] {}", level, msg);
     509       142783 :         if level == Level::Fatal || level == Level::Panic {
     510           62 :             if msg.contains("rejects our connection request with term") {
     511           33 :                 // collected quorum with lower term, then got rejected by next connected safekeeper
     512           33 :                 executor::exit(1, msg.to_owned());
     513           33 :             }
     514           62 :             if msg.contains("collected propTermStartLsn") && msg.contains(", but basebackup LSN ") {
     515            4 :                 // sync-safekeepers collected wrong quorum, walproposer collected another quorum
     516            4 :                 executor::exit(1, msg.to_owned());
     517           58 :             }
     518           62 :             if msg.contains("failed to download WAL for logical replicaiton") {
     519           25 :                 // Recovery connection broken and recovery was failed
     520           25 :                 executor::exit(1, msg.to_owned());
     521           37 :             }
     522           62 :             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           62 :             }
     526           62 :             panic!("unknown FATAL error from walproposer: {}", msg);
     527       142721 :         }
     528       142721 :     }
     529              : 
     530          413 :     fn after_election(&self, wp: &mut walproposer::bindings::WalProposer) {
     531          413 :         let prop_lsn = wp.propTermStartLsn;
     532          413 :         let prop_term = wp.propTerm;
     533          413 : 
     534          413 :         let mut prev_lsn: u64 = 0;
     535          413 :         let mut prev_term: u64 = 0;
     536          413 : 
     537          413 :         unsafe {
     538          413 :             let history = wp.propTermHistory.entries;
     539          413 :             let len = wp.propTermHistory.n_entries as usize;
     540          413 :             if len > 1 {
     541          362 :                 let entry = *history.wrapping_add(len - 2);
     542          362 :                 prev_lsn = entry.lsn;
     543          362 :                 prev_term = entry.term;
     544          362 :             }
     545              :         }
     546              : 
     547          413 :         let msg = format!(
     548          413 :             "prop_elected;{};{};{};{}",
     549          413 :             prop_lsn, prop_term, prev_lsn, prev_term
     550          413 :         );
     551          413 : 
     552          413 :         debug!(msg);
     553          413 :         self.os.log_event(msg);
     554          413 :     }
     555              : 
     556          188 :     fn get_redo_start_lsn(&self) -> u64 {
     557          188 :         debug!("get_redo_start_lsn -> {:?}", self.redo_start_lsn);
     558          188 :         self.redo_start_lsn.expect("redo_start_lsn is not set").0
     559          188 :     }
     560              : 
     561         1540 :     fn get_shmem_state(&self) -> *mut walproposer::bindings::WalproposerShmemState {
     562         1540 :         self.shmem.get()
     563         1540 :     }
     564              : 
     565          124 :     fn start_streaming(
     566          124 :         &self,
     567          124 :         startpos: u64,
     568          124 :         callback: &walproposer::walproposer::StreamingCallback,
     569          124 :     ) {
     570          124 :         let disk = &self.disk;
     571          124 :         let disk_lsn = disk.lock().flush_rec_ptr().0;
     572          124 :         debug!("start_streaming at {} (disk_lsn={})", startpos, disk_lsn);
     573          124 :         if startpos < disk_lsn {
     574           35 :             debug!(
     575            0 :                 "startpos < disk_lsn, it means we wrote some transaction even before streaming started"
     576              :             );
     577           89 :         }
     578          124 :         assert!(startpos <= disk_lsn);
     579          124 :         let mut broadcasted = Lsn(startpos);
     580              : 
     581              :         loop {
     582          588 :             let available = disk.lock().flush_rec_ptr();
     583          588 :             assert!(available >= broadcasted);
     584          464 :             callback.broadcast(broadcasted, available);
     585          464 :             broadcasted = available;
     586          464 :             callback.poll();
     587              :         }
     588              :     }
     589              : 
     590         1962 :     fn process_safekeeper_feedback(
     591         1962 :         &mut self,
     592         1962 :         wp: &mut walproposer::bindings::WalProposer,
     593         1962 :         _sk: &mut walproposer::bindings::Safekeeper,
     594         1962 :     ) {
     595         1962 :         debug!("process_safekeeper_feedback, commit_lsn={}", wp.commitLsn);
     596         1962 :         if wp.commitLsn > self.last_logged_commit_lsn {
     597          448 :             self.os.log_event(format!("commit_lsn;{}", wp.commitLsn));
     598          448 :             self.last_logged_commit_lsn = wp.commitLsn;
     599         1514 :         }
     600         1962 :     }
     601              : 
     602          112 :     fn get_flush_rec_ptr(&self) -> u64 {
     603          112 :         let lsn = self.disk.lock().flush_rec_ptr();
     604          112 :         debug!("get_flush_rec_ptr: {}", lsn);
     605          112 :         lsn.0
     606          112 :     }
     607              : 
     608          413 :     fn recovery_download(
     609          413 :         &self,
     610          413 :         wp: &mut walproposer::bindings::WalProposer,
     611          413 :         sk: &mut walproposer::bindings::Safekeeper,
     612          413 :     ) -> bool {
     613          413 :         let mut startpos = wp.truncateLsn;
     614          413 :         let endpos = wp.propTermStartLsn;
     615          413 : 
     616          413 :         if startpos == endpos {
     617          190 :             debug!("recovery_download: nothing to download");
     618          190 :             return true;
     619          223 :         }
     620          223 : 
     621          223 :         debug!("recovery_download from {} to {}", startpos, endpos,);
     622              : 
     623          223 :         let replication_prompt = format!(
     624          223 :             "START_REPLICATION {} {} {} {}",
     625          223 :             self.config.ttid.tenant_id, self.config.ttid.timeline_id, startpos, endpos,
     626          223 :         );
     627          223 :         let async_conn = self.get_conn(sk);
     628          223 : 
     629          223 :         let conn = self.os.open_tcp(async_conn.node_id);
     630          223 :         conn.send(desim::proto::AnyMessage::Bytes(replication_prompt.into()));
     631          223 : 
     632          223 :         let chan = conn.recv_chan();
     633          370 :         while startpos < endpos {
     634          223 :             let event = chan.recv();
     635          198 :             match event {
     636              :                 NetEvent::Closed => {
     637           25 :                     debug!("connection closed in recovery");
     638           25 :                     break;
     639              :                 }
     640          198 :                 NetEvent::Message(AnyMessage::Bytes(b)) => {
     641          198 :                     debug!("got recovery bytes from safekeeper");
     642          147 :                     self.disk.lock().write(startpos, &b);
     643          147 :                     startpos += b.len() as u64;
     644              :                 }
     645            0 :                 NetEvent::Message(_) => unreachable!(),
     646              :             }
     647              :         }
     648              : 
     649          172 :         debug!("recovery finished at {}", startpos);
     650              : 
     651          172 :         startpos == endpos
     652          362 :     }
     653              : 
     654        10789 :     fn conn_finish(&self, sk: &mut walproposer::bindings::Safekeeper) {
     655        10789 :         let mut conn = self.get_conn(sk);
     656        10789 :         debug!("conn_finish to {}", conn.node_id);
     657        10789 :         if let Some(socket) = conn.socket.as_mut() {
     658         3471 :             socket.close();
     659         7318 :         } else {
     660         7318 :             // connection is already closed
     661         7318 :         }
     662        10789 :         conn.socket = None;
     663        10789 :     }
     664              : 
     665        10549 :     fn conn_error_message(&self, _sk: &mut walproposer::bindings::Safekeeper) -> String {
     666        10549 :         "connection is closed, probably".into()
     667        10549 :     }
     668              : }
        

Generated by: LCOV version 2.1-beta