Line data    Source code

       1              : /// Layer of indirection previously used to support multiple implementations.
       2              : /// Subject to removal: <https://github.com/neondatabase/neon/issues/7753>
       3              : use std::time::Duration;
       4              : 
       5              : use bytes::Bytes;
       6              : use pageserver_api::{reltag::RelTag, shard::TenantShardId};
       7              : use tracing::warn;
       8              : use utils::lsn::Lsn;
       9              : 
      10              : use crate::{config::PageServerConf, walrecord::NeonWalRecord};
      11              : 
      12              : mod no_leak_child;
      13              : /// The IPC protocol that pageserver and walredo process speak over their shared pipe.
      14              : mod protocol;
      15              : 
      16              : mod process_impl {
      17              :     pub(super) mod process_async;
      18              : }
      19              : 
      20              : #[derive(
      21              :     Clone,
      22              :     Copy,
      23              :     Debug,
      24              :     PartialEq,
      25              :     Eq,
      26          160 :     strum_macros::EnumString,
      27            0 :     strum_macros::Display,
      28            0 :     strum_macros::IntoStaticStr,
      29            0 :     serde_with::DeserializeFromStr,
      30              :     serde_with::SerializeDisplay,
      31              : )]
      32              : #[strum(serialize_all = "kebab-case")]
      33              : #[repr(u8)]
      34              : pub enum Kind {
      35              :     Sync,
      36              :     Async,
      37              : }
      38              : 
      39              : pub(crate) struct Process(process_impl::process_async::WalRedoProcess);
      40              : 
      41              : impl Process {
      42              :     #[inline(always)]
      43            8 :     pub fn launch(
      44            8 :         conf: &'static PageServerConf,
      45            8 :         tenant_shard_id: TenantShardId,
      46            8 :         pg_version: u32,
      47            8 :     ) -> anyhow::Result<Self> {
      48            8 :         if conf.walredo_process_kind != Kind::Async {
      49            0 :             warn!(
      50              :                 configured = %conf.walredo_process_kind,
      51            0 :                 "the walredo_process_kind setting has been turned into a no-op, using async implementation"
      52              :             );
      53            8 :         }
      54            8 :         Ok(Self(process_impl::process_async::WalRedoProcess::launch(
      55            8 :             conf,
      56            8 :             tenant_shard_id,
      57            8 :             pg_version,
      58            8 :         )?))
      59            8 :     }
      60              : 
      61              :     #[inline(always)]
      62            8 :     pub(crate) async fn apply_wal_records(
      63            8 :         &self,
      64            8 :         rel: RelTag,
      65            8 :         blknum: u32,
      66            8 :         base_img: &Option<Bytes>,
      67            8 :         records: &[(Lsn, NeonWalRecord)],
      68            8 :         wal_redo_timeout: Duration,
      69            8 :     ) -> anyhow::Result<Bytes> {
      70            8 :         self.0
      71            8 :             .apply_wal_records(rel, blknum, base_img, records, wal_redo_timeout)
      72           16 :             .await
      73            8 :     }
      74              : 
      75            8 :     pub(crate) fn id(&self) -> u32 {
      76            8 :         self.0.id()
      77            8 :     }
      78              : }