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

Generated by: LCOV version 2.1-beta