Line data    Source code

       1              : use std::ffi::CStr;
       2              : use std::sync::Arc;
       3              : 
       4              : use parking_lot::{Mutex, MutexGuard};
       5              : use postgres_ffi::v16::wal_generator::{LogicalMessageGenerator, WalGenerator};
       6              : use utils::lsn::Lsn;
       7              : 
       8              : use super::block_storage::BlockStorage;
       9              : 
      10              : /// Simulation implementation of walproposer WAL storage.
      11              : pub struct DiskWalProposer {
      12              :     state: Mutex<State>,
      13              : }
      14              : 
      15              : impl DiskWalProposer {
      16         9037 :     pub fn new() -> Arc<DiskWalProposer> {
      17         9037 :         Arc::new(DiskWalProposer {
      18         9037 :             state: Mutex::new(State {
      19         9037 :                 internal_available_lsn: Lsn(0),
      20         9037 :                 prev_lsn: Lsn(0),
      21         9037 :                 disk: BlockStorage::new(),
      22         9037 :                 wal_generator: WalGenerator::new(LogicalMessageGenerator::new(c"", &[]), Lsn(0)),
      23         9037 :             }),
      24         9037 :         })
      25         9037 :     }
      26              : 
      27        14583 :     pub fn lock(&self) -> MutexGuard<State> {
      28        14583 :         self.state.lock()
      29        14583 :     }
      30              : }
      31              : 
      32              : pub struct State {
      33              :     // flush_lsn
      34              :     internal_available_lsn: Lsn,
      35              :     // needed for WAL generation
      36              :     prev_lsn: Lsn,
      37              :     // actual WAL storage
      38              :     disk: BlockStorage,
      39              :     // WAL record generator
      40              :     wal_generator: WalGenerator<LogicalMessageGenerator>,
      41              : }
      42              : 
      43              : impl State {
      44          881 :     pub fn read(&self, pos: u64, buf: &mut [u8]) {
      45          881 :         self.disk.read(pos, buf);
      46          881 :         // TODO: fail on reading uninitialized data
      47          881 :     }
      48              : 
      49          147 :     pub fn write(&mut self, pos: u64, buf: &[u8]) {
      50          147 :         self.disk.write(pos, buf);
      51          147 :     }
      52              : 
      53              :     /// Update the internal available LSN to the given value.
      54          308 :     pub fn reset_to(&mut self, lsn: Lsn) {
      55          308 :         self.internal_available_lsn = lsn;
      56          308 :         self.prev_lsn = Lsn(0); // Safekeeper doesn't care if this is omitted
      57          308 :         self.wal_generator.lsn = self.internal_available_lsn;
      58          308 :         self.wal_generator.prev_lsn = self.prev_lsn;
      59          308 :     }
      60              : 
      61              :     /// Get current LSN.
      62         1746 :     pub fn flush_rec_ptr(&self) -> Lsn {
      63         1746 :         self.internal_available_lsn
      64         1746 :     }
      65              : 
      66              :     /// Inserts a logical record in the WAL at the current LSN.
      67        11501 :     pub fn insert_logical_message(&mut self, prefix: &CStr, msg: &[u8]) {
      68        11501 :         let (_, record) = self.wal_generator.append_logical_message(prefix, msg);
      69        11501 :         self.disk.write(self.internal_available_lsn.into(), &record);
      70        11501 :         self.prev_lsn = self.internal_available_lsn;
      71        11501 :         self.internal_available_lsn += record.len() as u64;
      72        11501 :     }
      73              : }