LCOV - code coverage report
Current view: top level - safekeeper/tests/walproposer_sim - simulation.rs (source / functions) Coverage Total Hit
Test: 5fe7fa8d483b39476409aee736d6d5e32728bfac.info Lines: 97.7 % 299 292
Test Date: 2025-03-12 16:10:49 Functions: 88.9 % 63 56

            Line data    Source code
       1              : use std::cell::Cell;
       2              : use std::str::FromStr;
       3              : use std::sync::Arc;
       4              : 
       5              : use desim::executor::{self, ExternalHandle};
       6              : use desim::node_os::NodeOs;
       7              : use desim::options::{Delay, NetworkOptions};
       8              : use desim::proto::{AnyMessage, NodeEvent};
       9              : use desim::world::{Node, World};
      10              : use rand::{Rng, SeedableRng};
      11              : use tracing::{debug, info_span, warn};
      12              : use utils::id::TenantTimelineId;
      13              : use utils::lsn::Lsn;
      14              : use walproposer::walproposer::{Config, Wrapper};
      15              : 
      16              : use super::log::SimClock;
      17              : use super::safekeeper_disk::SafekeeperDisk;
      18              : use super::walproposer_api;
      19              : use super::walproposer_disk::DiskWalProposer;
      20              : use crate::walproposer_sim::safekeeper::run_server;
      21              : use crate::walproposer_sim::walproposer_api::SimulationApi;
      22              : 
      23              : /// Simulated safekeeper node.
      24              : pub struct SafekeeperNode {
      25              :     pub node: Arc<Node>,
      26              :     pub id: u32,
      27              :     pub disk: Arc<SafekeeperDisk>,
      28              :     pub thread: Cell<ExternalHandle>,
      29              : }
      30              : 
      31              : impl SafekeeperNode {
      32              :     /// Create and start a safekeeper at the specified Node.
      33         1524 :     pub fn new(node: Arc<Node>) -> Self {
      34         1524 :         let disk = Arc::new(SafekeeperDisk::new());
      35         1524 :         let thread = Cell::new(SafekeeperNode::launch(disk.clone(), node.clone()));
      36         1524 : 
      37         1524 :         Self {
      38         1524 :             id: node.id,
      39         1524 :             node,
      40         1524 :             disk,
      41         1524 :             thread,
      42         1524 :         }
      43         1524 :     }
      44              : 
      45         9707 :     fn launch(disk: Arc<SafekeeperDisk>, node: Arc<Node>) -> ExternalHandle {
      46         9707 :         // start the server thread
      47         9707 :         node.launch(move |os| {
      48         9554 :             run_server(os, disk).expect("server should finish without errors");
      49         9707 :         })
      50         9707 :     }
      51              : 
      52              :     /// Restart the safekeeper.
      53         8183 :     pub fn restart(&self) {
      54         8183 :         let new_thread = SafekeeperNode::launch(self.disk.clone(), self.node.clone());
      55         8183 :         let old_thread = self.thread.replace(new_thread);
      56         8183 :         old_thread.crash_stop();
      57         8183 :     }
      58              : }
      59              : 
      60              : /// Simulated walproposer node.
      61              : pub struct WalProposer {
      62              :     thread: ExternalHandle,
      63              :     node: Arc<Node>,
      64              :     disk: Arc<DiskWalProposer>,
      65              :     sync_safekeepers: bool,
      66              : }
      67              : 
      68              : impl WalProposer {
      69              :     /// Generic start function for both modes.
      70         8835 :     fn start(
      71         8835 :         os: NodeOs,
      72         8835 :         disk: Arc<DiskWalProposer>,
      73         8835 :         ttid: TenantTimelineId,
      74         8835 :         addrs: Vec<String>,
      75         8835 :         lsn: Option<Lsn>,
      76         8835 :     ) {
      77         8835 :         let sync_safekeepers = lsn.is_none();
      78              : 
      79         8835 :         let _enter = if sync_safekeepers {
      80         8527 :             info_span!("sync", started = executor::now()).entered()
      81              :         } else {
      82          308 :             info_span!("walproposer", started = executor::now()).entered()
      83              :         };
      84              : 
      85         8835 :         os.log_event(format!("started;walproposer;{}", sync_safekeepers as i32));
      86         8835 : 
      87         8835 :         let config = Config {
      88         8835 :             ttid,
      89         8835 :             safekeepers_list: addrs,
      90         8835 :             safekeeper_reconnect_timeout: 1000,
      91         8835 :             safekeeper_connection_timeout: 5000,
      92         8835 :             sync_safekeepers,
      93         8835 :         };
      94         8835 :         let args = walproposer_api::Args {
      95         8835 :             os,
      96         8835 :             config: config.clone(),
      97         8835 :             disk,
      98         8835 :             redo_start_lsn: lsn,
      99         8835 :         };
     100         8835 :         let api = SimulationApi::new(args);
     101         8835 :         let wp = Wrapper::new(Box::new(api), config);
     102         8835 :         wp.start();
     103         8835 :     }
     104              : 
     105              :     /// Start walproposer in a sync_safekeepers mode.
     106         8729 :     pub fn launch_sync(ttid: TenantTimelineId, addrs: Vec<String>, node: Arc<Node>) -> Self {
     107         8729 :         debug!("sync_safekeepers started at node {}", node.id);
     108         8729 :         let disk = DiskWalProposer::new();
     109         8729 :         let disk_wp = disk.clone();
     110         8729 : 
     111         8729 :         // start the client thread
     112         8729 :         let handle = node.launch(move |os| {
     113         8527 :             WalProposer::start(os, disk_wp, ttid, addrs, None);
     114         8729 :         });
     115         8729 : 
     116         8729 :         Self {
     117         8729 :             thread: handle,
     118         8729 :             node,
     119         8729 :             disk,
     120         8729 :             sync_safekeepers: true,
     121         8729 :         }
     122         8729 :     }
     123              : 
     124              :     /// Start walproposer in a normal mode.
     125          308 :     pub fn launch_walproposer(
     126          308 :         ttid: TenantTimelineId,
     127          308 :         addrs: Vec<String>,
     128          308 :         node: Arc<Node>,
     129          308 :         lsn: Lsn,
     130          308 :     ) -> Self {
     131          308 :         debug!("walproposer started at node {}", node.id);
     132          308 :         let disk = DiskWalProposer::new();
     133          308 :         disk.lock().reset_to(lsn);
     134          308 :         let disk_wp = disk.clone();
     135          308 : 
     136          308 :         // start the client thread
     137          308 :         let handle = node.launch(move |os| {
     138          308 :             WalProposer::start(os, disk_wp, ttid, addrs, Some(lsn));
     139          308 :         });
     140          308 : 
     141          308 :         Self {
     142          308 :             thread: handle,
     143          308 :             node,
     144          308 :             disk,
     145          308 :             sync_safekeepers: false,
     146          308 :         }
     147          308 :     }
     148              : 
     149          523 :     pub fn write_tx(&mut self, cnt: usize) {
     150          523 :         let start_lsn = self.disk.lock().flush_rec_ptr();
     151          523 : 
     152        11501 :         for _ in 0..cnt {
     153        11501 :             self.disk
     154        11501 :                 .lock()
     155        11501 :                 .insert_logical_message(c"prefix", b"message");
     156        11501 :         }
     157              : 
     158          523 :         let end_lsn = self.disk.lock().flush_rec_ptr();
     159          523 : 
     160          523 :         // log event
     161          523 :         self.node
     162          523 :             .log_event(format!("write_wal;{};{};{}", start_lsn.0, end_lsn.0, cnt));
     163          523 : 
     164          523 :         // now we need to set "Latch" in walproposer
     165          523 :         self.node
     166          523 :             .node_events()
     167          523 :             .send(NodeEvent::Internal(AnyMessage::Just32(0)));
     168          523 :     }
     169              : 
     170         8166 :     pub fn stop(&self) {
     171         8166 :         self.thread.crash_stop();
     172         8166 :     }
     173              : }
     174              : 
     175              : /// Holds basic simulation settings, such as network options.
     176              : pub struct TestConfig {
     177              :     pub network: NetworkOptions,
     178              :     pub timeout: u64,
     179              :     pub clock: Option<SimClock>,
     180              : }
     181              : 
     182              : impl TestConfig {
     183              :     /// Create a new TestConfig with default settings.
     184            9 :     pub fn new(clock: Option<SimClock>) -> Self {
     185            9 :         Self {
     186            9 :             network: NetworkOptions {
     187            9 :                 keepalive_timeout: Some(2000),
     188            9 :                 connect_delay: Delay {
     189            9 :                     min: 1,
     190            9 :                     max: 5,
     191            9 :                     fail_prob: 0.0,
     192            9 :                 },
     193            9 :                 send_delay: Delay {
     194            9 :                     min: 1,
     195            9 :                     max: 5,
     196            9 :                     fail_prob: 0.0,
     197            9 :                 },
     198            9 :             },
     199            9 :             timeout: 1_000 * 10,
     200            9 :             clock,
     201            9 :         }
     202            9 :     }
     203              : 
     204              :     /// Start a new simulation with the specified seed.
     205          508 :     pub fn start(&self, seed: u64) -> Test {
     206          508 :         let world = Arc::new(World::new(seed, Arc::new(self.network.clone())));
     207              : 
     208          508 :         if let Some(clock) = &self.clock {
     209          508 :             clock.set_clock(world.clock());
     210          508 :         }
     211              : 
     212          508 :         let servers = [
     213          508 :             SafekeeperNode::new(world.new_node()),
     214          508 :             SafekeeperNode::new(world.new_node()),
     215          508 :             SafekeeperNode::new(world.new_node()),
     216          508 :         ];
     217          508 : 
     218          508 :         let server_ids = [servers[0].id, servers[1].id, servers[2].id];
     219         1524 :         let safekeepers_addrs = server_ids.map(|id| format!("node:{}", id)).to_vec();
     220          508 : 
     221          508 :         let ttid = TenantTimelineId::generate();
     222          508 : 
     223          508 :         Test {
     224          508 :             world,
     225          508 :             servers,
     226          508 :             sk_list: safekeepers_addrs,
     227          508 :             ttid,
     228          508 :             timeout: self.timeout,
     229          508 :         }
     230          508 :     }
     231              : }
     232              : 
     233              : /// Holds simulation state.
     234              : pub struct Test {
     235              :     pub world: Arc<World>,
     236              :     pub servers: [SafekeeperNode; 3],
     237              :     pub sk_list: Vec<String>,
     238              :     pub ttid: TenantTimelineId,
     239              :     pub timeout: u64,
     240              : }
     241              : 
     242              : impl Test {
     243              :     /// Start a sync_safekeepers thread and wait for it to finish.
     244            6 :     pub fn sync_safekeepers(&self) -> anyhow::Result<Lsn> {
     245            6 :         let wp = self.launch_sync_safekeepers();
     246            6 : 
     247            6 :         // poll until exit or timeout
     248            6 :         let time_limit = self.timeout;
     249          230 :         while self.world.step() && self.world.now() < time_limit && !wp.thread.is_finished() {}
     250              : 
     251            6 :         if !wp.thread.is_finished() {
     252            0 :             anyhow::bail!("timeout or idle stuck");
     253            6 :         }
     254            6 : 
     255            6 :         let res = wp.thread.result();
     256            6 :         if res.0 != 0 {
     257            0 :             anyhow::bail!("non-zero exitcode: {:?}", res);
     258            6 :         }
     259            6 :         let lsn = Lsn::from_str(&res.1)?;
     260            6 :         Ok(lsn)
     261            6 :     }
     262              : 
     263              :     /// Spawn a new sync_safekeepers thread.
     264         8729 :     pub fn launch_sync_safekeepers(&self) -> WalProposer {
     265         8729 :         WalProposer::launch_sync(self.ttid, self.sk_list.clone(), self.world.new_node())
     266         8729 :     }
     267              : 
     268              :     /// Spawn a new walproposer thread.
     269          308 :     pub fn launch_walproposer(&self, lsn: Lsn) -> WalProposer {
     270          308 :         let lsn = if lsn.0 == 0 {
     271              :             // usual LSN after basebackup
     272          187 :             Lsn(21623024)
     273              :         } else {
     274          121 :             lsn
     275              :         };
     276              : 
     277          308 :         WalProposer::launch_walproposer(self.ttid, self.sk_list.clone(), self.world.new_node(), lsn)
     278          308 :     }
     279              : 
     280              :     /// Execute the simulation for the specified duration.
     281          105 :     pub fn poll_for_duration(&self, duration: u64) {
     282          105 :         let time_limit = std::cmp::min(self.world.now() + duration, self.timeout);
     283         1762 :         while self.world.step() && self.world.now() < time_limit {}
     284          105 :     }
     285              : 
     286              :     /// Execute the simulation together with events defined in some schedule.
     287          504 :     pub fn run_schedule(&self, schedule: &Schedule) -> anyhow::Result<()> {
     288          504 :         // scheduling empty events so that world will stop in those points
     289          504 :         {
     290          504 :             let clock = self.world.clock();
     291          504 : 
     292          504 :             let now = self.world.now();
     293        25077 :             for (time, _) in schedule {
     294        24573 :                 if *time < now {
     295            0 :                     continue;
     296        24573 :                 }
     297        24573 :                 clock.schedule_fake(*time - now);
     298              :             }
     299              :         }
     300              : 
     301          504 :         let mut wp = self.launch_sync_safekeepers();
     302          504 : 
     303          504 :         let mut skipped_tx = 0;
     304          504 :         let mut started_tx = 0;
     305          504 : 
     306          504 :         let mut schedule_ptr = 0;
     307              : 
     308              :         loop {
     309        25447 :             if wp.sync_safekeepers && wp.thread.is_finished() {
     310          359 :                 let res = wp.thread.result();
     311          359 :                 if res.0 != 0 {
     312           54 :                     warn!("sync non-zero exitcode: {:?}", res);
     313           54 :                     debug!("restarting sync_safekeepers");
     314              :                     // restart the sync_safekeepers
     315           54 :                     wp = self.launch_sync_safekeepers();
     316           54 :                     continue;
     317          305 :                 }
     318          305 :                 let lsn = Lsn::from_str(&res.1)?;
     319          305 :                 debug!("sync_safekeepers finished at LSN {}", lsn);
     320          305 :                 wp = self.launch_walproposer(lsn);
     321          305 :                 debug!("walproposer started at thread {}", wp.thread.id());
     322        25088 :             }
     323              : 
     324        25393 :             let now = self.world.now();
     325        49966 :             while schedule_ptr < schedule.len() && schedule[schedule_ptr].0 <= now {
     326        24573 :                 if now != schedule[schedule_ptr].0 {
     327            0 :                     warn!("skipped event {:?} at {}", schedule[schedule_ptr], now);
     328        24573 :                 }
     329              : 
     330        24573 :                 let action = &schedule[schedule_ptr].1;
     331        24573 :                 match action {
     332         8226 :                     TestAction::WriteTx(size) => {
     333         8226 :                         if !wp.sync_safekeepers && !wp.thread.is_finished() {
     334          421 :                             started_tx += *size;
     335          421 :                             wp.write_tx(*size);
     336          421 :                             debug!("written {} transactions", size);
     337              :                         } else {
     338         7805 :                             skipped_tx += size;
     339         7805 :                             debug!("skipped {} transactions", size);
     340              :                         }
     341              :                     }
     342         8182 :                     TestAction::RestartSafekeeper(id) => {
     343         8182 :                         debug!("restarting safekeeper {}", id);
     344         8182 :                         self.servers[*id].restart();
     345              :                     }
     346              :                     TestAction::RestartWalProposer => {
     347         8165 :                         debug!("restarting sync_safekeepers");
     348         8165 :                         wp.stop();
     349         8165 :                         wp = self.launch_sync_safekeepers();
     350              :                     }
     351              :                 }
     352        24573 :                 schedule_ptr += 1;
     353              :             }
     354              : 
     355        25393 :             if schedule_ptr == schedule.len() {
     356          504 :                 break;
     357        24889 :             }
     358        24889 :             let next_event_time = schedule[schedule_ptr].0;
     359        24889 : 
     360        24889 :             // poll until the next event
     361        24889 :             if wp.thread.is_finished() {
     362          299 :                 while self.world.step() && self.world.now() < next_event_time {}
     363              :             } else {
     364       379582 :                 while self.world.step()
     365       379582 :                     && self.world.now() < next_event_time
     366       355081 :                     && !wp.thread.is_finished()
     367       354714 :                 {}
     368              :             }
     369              :         }
     370              : 
     371          504 :         debug!(
     372            0 :             "finished schedule, total steps: {}",
     373            0 :             self.world.get_thread_step_count()
     374              :         );
     375          504 :         debug!("skipped_tx: {}", skipped_tx);
     376          504 :         debug!("started_tx: {}", started_tx);
     377              : 
     378          504 :         Ok(())
     379          504 :     }
     380              : }
     381              : 
     382              : #[derive(Debug, Clone)]
     383              : pub enum TestAction {
     384              :     WriteTx(usize),
     385              :     RestartSafekeeper(usize),
     386              :     RestartWalProposer,
     387              : }
     388              : 
     389              : pub type Schedule = Vec<(u64, TestAction)>;
     390              : 
     391          502 : pub fn generate_schedule(seed: u64) -> Schedule {
     392          502 :     let mut rng = rand::rngs::StdRng::seed_from_u64(seed);
     393          502 :     let mut schedule = Vec::new();
     394          502 :     let mut time = 0;
     395          502 : 
     396          502 :     let cnt = rng.gen_range(1..100);
     397          502 : 
     398          502 :     for _ in 0..cnt {
     399        24458 :         time += rng.gen_range(0..500);
     400        24458 :         let action = match rng.gen_range(0..3) {
     401         8119 :             0 => TestAction::WriteTx(rng.gen_range(1..10)),
     402         8176 :             1 => TestAction::RestartSafekeeper(rng.gen_range(0..3)),
     403         8163 :             2 => TestAction::RestartWalProposer,
     404            0 :             _ => unreachable!(),
     405              :         };
     406        24458 :         schedule.push((time, action));
     407              :     }
     408              : 
     409          502 :     schedule
     410          502 : }
     411              : 
     412          502 : pub fn generate_network_opts(seed: u64) -> NetworkOptions {
     413          502 :     let mut rng = rand::rngs::StdRng::seed_from_u64(seed);
     414          502 : 
     415          502 :     let timeout = rng.gen_range(100..2000);
     416          502 :     let max_delay = rng.gen_range(1..2 * timeout);
     417          502 :     let min_delay = rng.gen_range(1..=max_delay);
     418          502 : 
     419          502 :     let max_fail_prob = rng.gen_range(0.0..0.9);
     420          502 :     let connect_fail_prob = rng.gen_range(0.0..max_fail_prob);
     421          502 :     let send_fail_prob = rng.gen_range(0.0..connect_fail_prob);
     422          502 : 
     423          502 :     NetworkOptions {
     424          502 :         keepalive_timeout: Some(timeout),
     425          502 :         connect_delay: Delay {
     426          502 :             min: min_delay,
     427          502 :             max: max_delay,
     428          502 :             fail_prob: connect_fail_prob,
     429          502 :         },
     430          502 :         send_delay: Delay {
     431          502 :             min: min_delay,
     432          502 :             max: max_delay,
     433          502 :             fail_prob: send_fail_prob,
     434          502 :         },
     435          502 :     }
     436          502 : }
        

Generated by: LCOV version 2.1-beta