LCOV - code coverage report
Current view: top level - safekeeper/tests/walproposer_sim - walproposer_api.rs (source / functions) Coverage Total Hit
Test: 07bee600374ccd486c69370d0972d9035964fe68.info Lines: 95.5 % 486 464
Test Date: 2025-02-20 13:11:02 Functions: 99.3 % 138 137

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

Generated by: LCOV version 2.1-beta