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

Generated by: LCOV version 2.1-beta