LCOV - code coverage report
Current view: top level - pageserver/src/tenant - tasks.rs (source / functions) Coverage Total Hit
Test: 90b23405d17e36048d3bb64e314067f397803f1b.info Lines: 7.4 % 403 30
Test Date: 2024-09-20 13:14:58 Functions: 11.1 % 36 4

            Line data    Source code
       1              : //! This module contains functions to serve per-tenant background processes,
       2              : //! such as compaction and GC
       3              : 
       4              : use std::ops::ControlFlow;
       5              : use std::str::FromStr;
       6              : use std::sync::Arc;
       7              : use std::time::{Duration, Instant};
       8              : 
       9              : use crate::context::{DownloadBehavior, RequestContext};
      10              : use crate::metrics::TENANT_TASK_EVENTS;
      11              : use crate::task_mgr;
      12              : use crate::task_mgr::{TaskKind, BACKGROUND_RUNTIME};
      13              : use crate::tenant::throttle::Stats;
      14              : use crate::tenant::timeline::CompactionError;
      15              : use crate::tenant::{Tenant, TenantState};
      16              : use rand::Rng;
      17              : use tokio_util::sync::CancellationToken;
      18              : use tracing::*;
      19              : use utils::{backoff, completion, pausable_failpoint};
      20              : 
      21              : static CONCURRENT_BACKGROUND_TASKS: once_cell::sync::Lazy<tokio::sync::Semaphore> =
      22           60 :     once_cell::sync::Lazy::new(|| {
      23           60 :         let total_threads = task_mgr::TOKIO_WORKER_THREADS.get();
      24           60 :         let permits = usize::max(
      25           60 :             1,
      26           60 :             // while a lot of the work is done on spawn_blocking, we still do
      27           60 :             // repartitioning in the async context. this should give leave us some workers
      28           60 :             // unblocked to be blocked on other work, hopefully easing any outside visible
      29           60 :             // effects of restarts.
      30           60 :             //
      31           60 :             // 6/8 is a guess; previously we ran with unlimited 8 and more from
      32           60 :             // spawn_blocking.
      33           60 :             (total_threads * 3).checked_div(4).unwrap_or(0),
      34           60 :         );
      35           60 :         assert_ne!(permits, 0, "we will not be adding in permits later");
      36           60 :         assert!(
      37           60 :             permits < total_threads,
      38            0 :             "need threads avail for shorter work"
      39              :         );
      40           60 :         tokio::sync::Semaphore::new(permits)
      41           60 :     });
      42              : 
      43         1080 : #[derive(Debug, PartialEq, Eq, Clone, Copy, strum_macros::IntoStaticStr, enum_map::Enum)]
      44              : #[strum(serialize_all = "snake_case")]
      45              : pub(crate) enum BackgroundLoopKind {
      46              :     Compaction,
      47              :     Gc,
      48              :     Eviction,
      49              :     IngestHouseKeeping,
      50              :     ConsumptionMetricsCollectMetrics,
      51              :     ConsumptionMetricsSyntheticSizeWorker,
      52              :     InitialLogicalSizeCalculation,
      53              :     HeatmapUpload,
      54              :     SecondaryDownload,
      55              : }
      56              : 
      57              : impl BackgroundLoopKind {
      58            0 :     fn as_static_str(&self) -> &'static str {
      59            0 :         self.into()
      60            0 :     }
      61              : }
      62              : 
      63              : /// Cancellation safe.
      64         1092 : pub(crate) async fn concurrent_background_tasks_rate_limit_permit(
      65         1092 :     loop_kind: BackgroundLoopKind,
      66         1092 :     _ctx: &RequestContext,
      67         1092 : ) -> tokio::sync::SemaphorePermit<'static> {
      68         1092 :     let _guard = crate::metrics::BACKGROUND_LOOP_SEMAPHORE.measure_acquisition(loop_kind);
      69         1092 : 
      70         1092 :     pausable_failpoint!(
      71              :         "initial-size-calculation-permit-pause",
      72         1092 :         loop_kind == BackgroundLoopKind::InitialLogicalSizeCalculation
      73              :     );
      74              : 
      75              :     // TODO: assert that we run on BACKGROUND_RUNTIME; requires tokio_unstable Handle::id();
      76         1092 :     match CONCURRENT_BACKGROUND_TASKS.acquire().await {
      77         1092 :         Ok(permit) => permit,
      78            0 :         Err(_closed) => unreachable!("we never close the semaphore"),
      79              :     }
      80         1092 : }
      81              : 
      82              : /// Start per tenant background loops: compaction and gc.
      83            0 : pub fn start_background_loops(
      84            0 :     tenant: &Arc<Tenant>,
      85            0 :     background_jobs_can_start: Option<&completion::Barrier>,
      86            0 : ) {
      87            0 :     let tenant_shard_id = tenant.tenant_shard_id;
      88            0 :     task_mgr::spawn(
      89            0 :         BACKGROUND_RUNTIME.handle(),
      90            0 :         TaskKind::Compaction,
      91            0 :         tenant_shard_id,
      92            0 :         None,
      93            0 :         &format!("compactor for tenant {tenant_shard_id}"),
      94            0 :         {
      95            0 :             let tenant = Arc::clone(tenant);
      96            0 :             let background_jobs_can_start = background_jobs_can_start.cloned();
      97            0 :             async move {
      98            0 :                 let cancel = task_mgr::shutdown_token();
      99            0 :                 tokio::select! {
     100            0 :                     _ = cancel.cancelled() => { return Ok(()) },
     101            0 :                     _ = completion::Barrier::maybe_wait(background_jobs_can_start) => {}
     102            0 :                 };
     103            0 :                 compaction_loop(tenant, cancel)
     104            0 :                     // If you rename this span, change the RUST_LOG env variable in test_runner/performance/test_branch_creation.py
     105            0 :                     .instrument(info_span!("compaction_loop", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug()))
     106            0 :                     .await;
     107            0 :                 Ok(())
     108            0 :             }
     109            0 :         },
     110            0 :     );
     111            0 :     task_mgr::spawn(
     112            0 :         BACKGROUND_RUNTIME.handle(),
     113            0 :         TaskKind::GarbageCollector,
     114            0 :         tenant_shard_id,
     115            0 :         None,
     116            0 :         &format!("garbage collector for tenant {tenant_shard_id}"),
     117            0 :         {
     118            0 :             let tenant = Arc::clone(tenant);
     119            0 :             let background_jobs_can_start = background_jobs_can_start.cloned();
     120            0 :             async move {
     121            0 :                 let cancel = task_mgr::shutdown_token();
     122            0 :                 tokio::select! {
     123            0 :                     _ = cancel.cancelled() => { return Ok(()) },
     124            0 :                     _ = completion::Barrier::maybe_wait(background_jobs_can_start) => {}
     125            0 :                 };
     126            0 :                 gc_loop(tenant, cancel)
     127            0 :                     .instrument(info_span!("gc_loop", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug()))
     128            0 :                     .await;
     129            0 :                 Ok(())
     130            0 :             }
     131            0 :         },
     132            0 :     );
     133            0 : 
     134            0 :     task_mgr::spawn(
     135            0 :         BACKGROUND_RUNTIME.handle(),
     136            0 :         TaskKind::IngestHousekeeping,
     137            0 :         tenant_shard_id,
     138            0 :         None,
     139            0 :         &format!("ingest housekeeping for tenant {tenant_shard_id}"),
     140            0 :         {
     141            0 :             let tenant = Arc::clone(tenant);
     142            0 :             let background_jobs_can_start = background_jobs_can_start.cloned();
     143            0 :             async move {
     144            0 :                 let cancel = task_mgr::shutdown_token();
     145            0 :                 tokio::select! {
     146            0 :                     _ = cancel.cancelled() => { return Ok(()) },
     147            0 :                     _ = completion::Barrier::maybe_wait(background_jobs_can_start) => {}
     148            0 :                 };
     149            0 :                 ingest_housekeeping_loop(tenant, cancel)
     150            0 :                     .instrument(info_span!("ingest_housekeeping_loop", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug()))
     151            0 :                     .await;
     152            0 :                 Ok(())
     153            0 :             }
     154            0 :         },
     155            0 :     );
     156            0 : }
     157              : 
     158              : ///
     159              : /// Compaction task's main loop
     160              : ///
     161            0 : async fn compaction_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
     162              :     const MAX_BACKOFF_SECS: f64 = 300.0;
     163              :     // How many errors we have seen consequtively
     164            0 :     let mut error_run_count = 0;
     165            0 : 
     166            0 :     let mut last_throttle_flag_reset_at = Instant::now();
     167            0 : 
     168            0 :     TENANT_TASK_EVENTS.with_label_values(&["start"]).inc();
     169            0 :     async {
     170            0 :         let ctx = RequestContext::todo_child(TaskKind::Compaction, DownloadBehavior::Download);
     171            0 :         let mut first = true;
     172              :         loop {
     173            0 :             tokio::select! {
     174            0 :                 _ = cancel.cancelled() => {
     175            0 :                     return;
     176              :                 },
     177            0 :                 tenant_wait_result = wait_for_active_tenant(&tenant) => match tenant_wait_result {
     178            0 :                     ControlFlow::Break(()) => return,
     179            0 :                     ControlFlow::Continue(()) => (),
     180            0 :                 },
     181            0 :             }
     182            0 : 
     183            0 :             let period = tenant.get_compaction_period();
     184            0 : 
     185            0 :             // TODO: we shouldn't need to await to find tenant and this could be moved outside of
     186            0 :             // loop, #3501. There are also additional "allowed_errors" in tests.
     187            0 :             if first {
     188            0 :                 first = false;
     189            0 :                 if random_init_delay(period, &cancel).await.is_err() {
     190            0 :                     break;
     191            0 :                 }
     192            0 :             }
     193              : 
     194              : 
     195              : 
     196              :             let sleep_duration;
     197            0 :             if period == Duration::ZERO {
     198              :                 #[cfg(not(feature = "testing"))]
     199              :                 info!("automatic compaction is disabled");
     200              :                 // check again in 10 seconds, in case it's been enabled again.
     201            0 :                 sleep_duration = Duration::from_secs(10)
     202              :             } else {
     203            0 :                 let iteration = Iteration {
     204            0 :                     started_at: Instant::now(),
     205            0 :                     period,
     206            0 :                     kind: BackgroundLoopKind::Compaction,
     207            0 :                 };
     208              : 
     209              :                 // Run compaction
     210            0 :                 let IterationResult { output, elapsed } = iteration.run(tenant.compaction_iteration(&cancel, &ctx)).await;
     211            0 :                 match output {
     212            0 :                     Ok(has_pending_task) => {
     213            0 :                         error_run_count = 0;
     214            0 :                         // schedule the next compaction immediately in case there is a pending compaction task
     215            0 :                         sleep_duration = if has_pending_task { Duration::ZERO } else { period };
     216              :                     }
     217            0 :                     Err(e) => {
     218            0 :                         let wait_duration = backoff::exponential_backoff_duration_seconds(
     219            0 :                             error_run_count + 1,
     220            0 :                             1.0,
     221            0 :                             MAX_BACKOFF_SECS,
     222            0 :                         );
     223            0 :                         error_run_count += 1;
     224            0 :                         let wait_duration = Duration::from_secs_f64(wait_duration);
     225            0 :                         log_compaction_error(
     226            0 :                             &e,
     227            0 :                             error_run_count,
     228            0 :                             &wait_duration,
     229            0 :                             cancel.is_cancelled(),
     230            0 :                         );
     231            0 :                         sleep_duration = wait_duration;
     232            0 :                     }
     233              :                 }
     234              : 
     235              :                 // the duration is recorded by performance tests by enabling debug in this function
     236            0 :                 tracing::debug!(elapsed_ms=elapsed.as_millis(), "compaction iteration complete");
     237              :             };
     238              : 
     239              : 
     240              :             // Perhaps we did no work and the walredo process has been idle for some time:
     241              :             // give it a chance to shut down to avoid leaving walredo process running indefinitely.
     242            0 :             if let Some(walredo_mgr) = &tenant.walredo_mgr {
     243            0 :                 walredo_mgr.maybe_quiesce(period * 10);
     244            0 :             }
     245              : 
     246              :             // TODO: move this (and walredo quiesce) to a separate task that isn't affected by the back-off,
     247              :             // so we get some upper bound guarantee on when walredo quiesce / this throttling reporting here happens.
     248            0 :             info_span!(parent: None, "timeline_get_throttle", tenant_id=%tenant.tenant_shard_id, shard_id=%tenant.tenant_shard_id.shard_slug()).in_scope(|| {
     249            0 :                 let now = Instant::now();
     250            0 :                 let prev = std::mem::replace(&mut last_throttle_flag_reset_at, now);
     251            0 :                 let Stats { count_accounted, count_throttled, sum_throttled_usecs } = tenant.timeline_get_throttle.reset_stats();
     252            0 :                 if count_throttled == 0 {
     253            0 :                     return;
     254            0 :                 }
     255            0 :                 let allowed_rps = tenant.timeline_get_throttle.steady_rps();
     256            0 :                 let delta = now - prev;
     257            0 :                 info!(
     258            0 :                     n_seconds=%format_args!("{:.3}",
     259            0 :                     delta.as_secs_f64()),
     260              :                     count_accounted,
     261              :                     count_throttled,
     262              :                     sum_throttled_usecs,
     263            0 :                     allowed_rps=%format_args!("{allowed_rps:.0}"),
     264            0 :                     "shard was throttled in the last n_seconds"
     265              :                 );
     266            0 :             });
     267            0 : 
     268            0 :             // Sleep
     269            0 :             if tokio::time::timeout(sleep_duration, cancel.cancelled())
     270            0 :                 .await
     271            0 :                 .is_ok()
     272              :             {
     273            0 :                 break;
     274            0 :             }
     275              :         }
     276            0 :     }
     277            0 :     .await;
     278            0 :     TENANT_TASK_EVENTS.with_label_values(&["stop"]).inc();
     279            0 : }
     280              : 
     281            0 : fn log_compaction_error(
     282            0 :     e: &CompactionError,
     283            0 :     error_run_count: u32,
     284            0 :     sleep_duration: &std::time::Duration,
     285            0 :     task_cancelled: bool,
     286            0 : ) {
     287              :     use crate::tenant::upload_queue::NotInitialized;
     288              :     use crate::tenant::PageReconstructError;
     289              :     use CompactionError::*;
     290              : 
     291              :     enum LooksLike {
     292              :         Info,
     293              :         Error,
     294              :     }
     295              : 
     296            0 :     let decision = match e {
     297            0 :         ShuttingDown => None,
     298            0 :         _ if task_cancelled => Some(LooksLike::Info),
     299            0 :         Other(e) => {
     300            0 :             let root_cause = e.root_cause();
     301              : 
     302            0 :             let is_stopping = {
     303            0 :                 let upload_queue = root_cause
     304            0 :                     .downcast_ref::<NotInitialized>()
     305            0 :                     .is_some_and(|e| e.is_stopping());
     306            0 : 
     307            0 :                 let timeline = root_cause
     308            0 :                     .downcast_ref::<PageReconstructError>()
     309            0 :                     .is_some_and(|e| e.is_stopping());
     310            0 : 
     311            0 :                 upload_queue || timeline
     312              :             };
     313              : 
     314            0 :             if is_stopping {
     315            0 :                 Some(LooksLike::Info)
     316              :             } else {
     317            0 :                 Some(LooksLike::Error)
     318              :             }
     319              :         }
     320              :     };
     321              : 
     322            0 :     match decision {
     323            0 :         Some(LooksLike::Info) => info!(
     324            0 :             "Compaction failed {error_run_count} times, retrying in {sleep_duration:?}: {e:#}",
     325              :         ),
     326            0 :         Some(LooksLike::Error) => error!(
     327            0 :             "Compaction failed {error_run_count} times, retrying in {sleep_duration:?}: {e:?}",
     328              :         ),
     329            0 :         None => {}
     330              :     }
     331            0 : }
     332              : 
     333              : ///
     334              : /// GC task's main loop
     335              : ///
     336            0 : async fn gc_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
     337              :     const MAX_BACKOFF_SECS: f64 = 300.0;
     338              :     // How many errors we have seen consequtively
     339            0 :     let mut error_run_count = 0;
     340            0 : 
     341            0 :     TENANT_TASK_EVENTS.with_label_values(&["start"]).inc();
     342            0 :     async {
     343            0 :         // GC might require downloading, to find the cutoff LSN that corresponds to the
     344            0 :         // cutoff specified as time.
     345            0 :         let ctx =
     346            0 :             RequestContext::todo_child(TaskKind::GarbageCollector, DownloadBehavior::Download);
     347            0 : 
     348            0 :         let mut first = true;
     349            0 :         tenant.gc_block.set_lsn_lease_deadline(tenant.get_lsn_lease_length());
     350              :         loop {
     351            0 :             tokio::select! {
     352            0 :                 _ = cancel.cancelled() => {
     353            0 :                     return;
     354              :                 },
     355            0 :                 tenant_wait_result = wait_for_active_tenant(&tenant) => match tenant_wait_result {
     356            0 :                     ControlFlow::Break(()) => return,
     357            0 :                     ControlFlow::Continue(()) => (),
     358            0 :                 },
     359            0 :             }
     360            0 : 
     361            0 :             let period = tenant.get_gc_period();
     362            0 : 
     363            0 :             if first {
     364            0 :                 first = false;
     365            0 : 
     366            0 :                 let delays = async {
     367            0 :                     random_init_delay(period, &cancel).await?;
     368            0 :                     Ok::<_, Cancelled>(())
     369            0 :                 };
     370              : 
     371            0 :                 if delays.await.is_err() {
     372            0 :                     break;
     373            0 :                 }
     374            0 :             }
     375              : 
     376            0 :             let gc_horizon = tenant.get_gc_horizon();
     377            0 :             let sleep_duration;
     378            0 :             if period == Duration::ZERO || gc_horizon == 0 {
     379            0 :                 #[cfg(not(feature = "testing"))]
     380            0 :                 info!("automatic GC is disabled");
     381            0 :                 // check again in 10 seconds, in case it's been enabled again.
     382            0 :                 sleep_duration = Duration::from_secs(10);
     383            0 :             } else {
     384            0 :                 let iteration = Iteration {
     385            0 :                     started_at: Instant::now(),
     386            0 :                     period,
     387            0 :                     kind: BackgroundLoopKind::Gc,
     388            0 :                 };
     389              :                 // Run gc
     390            0 :                 let IterationResult { output, elapsed: _ } =
     391            0 :                     iteration.run(tenant.gc_iteration(None, gc_horizon, tenant.get_pitr_interval(), &cancel, &ctx))
     392            0 :                     .await;
     393            0 :                 match output {
     394            0 :                     Ok(_) => {
     395            0 :                         error_run_count = 0;
     396            0 :                         sleep_duration = period;
     397            0 :                     }
     398              :                     Err(crate::tenant::GcError::TenantCancelled) => {
     399            0 :                         return;
     400              :                     }
     401            0 :                     Err(e) => {
     402            0 :                         let wait_duration = backoff::exponential_backoff_duration_seconds(
     403            0 :                             error_run_count + 1,
     404            0 :                             1.0,
     405            0 :                             MAX_BACKOFF_SECS,
     406            0 :                         );
     407            0 :                         error_run_count += 1;
     408            0 :                         let wait_duration = Duration::from_secs_f64(wait_duration);
     409              : 
     410            0 :                         if matches!(e, crate::tenant::GcError::TimelineCancelled) {
     411              :                             // Timeline was cancelled during gc. We might either be in an event
     412              :                             // that affects the entire tenant (tenant deletion, pageserver shutdown),
     413              :                             // or in one that affects the timeline only (timeline deletion).
     414              :                             // Therefore, don't exit the loop.
     415            0 :                             info!("Gc failed {error_run_count} times, retrying in {wait_duration:?}: {e:?}");
     416              :                         } else {
     417            0 :                             error!("Gc failed {error_run_count} times, retrying in {wait_duration:?}: {e:?}");
     418              :                         }
     419              : 
     420            0 :                         sleep_duration = wait_duration;
     421              :                     }
     422              :                 }
     423              :             };
     424              : 
     425            0 :             if tokio::time::timeout(sleep_duration, cancel.cancelled())
     426            0 :                 .await
     427            0 :                 .is_ok()
     428              :             {
     429            0 :                 break;
     430            0 :             }
     431              :         }
     432            0 :     }
     433            0 :     .await;
     434            0 :     TENANT_TASK_EVENTS.with_label_values(&["stop"]).inc();
     435            0 : }
     436              : 
     437            0 : async fn ingest_housekeeping_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
     438            0 :     TENANT_TASK_EVENTS.with_label_values(&["start"]).inc();
     439            0 :     async {
     440              :         loop {
     441            0 :             tokio::select! {
     442            0 :                 _ = cancel.cancelled() => {
     443            0 :                     return;
     444              :                 },
     445            0 :                 tenant_wait_result = wait_for_active_tenant(&tenant) => match tenant_wait_result {
     446            0 :                     ControlFlow::Break(()) => return,
     447            0 :                     ControlFlow::Continue(()) => (),
     448            0 :                 },
     449            0 :             }
     450            0 : 
     451            0 :             // We run ingest housekeeping with the same frequency as compaction: it is not worth
     452            0 :             // having a distinct setting.  But we don't run it in the same task, because compaction
     453            0 :             // blocks on acquiring the background job semaphore.
     454            0 :             let period = tenant.get_compaction_period();
     455              : 
     456              :             // If compaction period is set to zero (to disable it), then we will use a reasonable default
     457            0 :             let period = if period == Duration::ZERO {
     458            0 :                 humantime::Duration::from_str(
     459            0 :                     pageserver_api::config::tenant_conf_defaults::DEFAULT_COMPACTION_PERIOD,
     460            0 :                 )
     461            0 :                 .unwrap()
     462            0 :                 .into()
     463              :             } else {
     464            0 :                 period
     465              :             };
     466              : 
     467              :             // Jitter the period by +/- 5%
     468            0 :             let period =
     469            0 :                 rand::thread_rng().gen_range((period * (95)) / 100..(period * (105)) / 100);
     470            0 : 
     471            0 :             // Always sleep first: we do not need to do ingest housekeeping early in the lifetime of
     472            0 :             // a tenant, since it won't have started writing any ephemeral files yet.
     473            0 :             if tokio::time::timeout(period, cancel.cancelled())
     474            0 :                 .await
     475            0 :                 .is_ok()
     476              :             {
     477            0 :                 break;
     478            0 :             }
     479            0 : 
     480            0 :             let iteration = Iteration {
     481            0 :                 started_at: Instant::now(),
     482            0 :                 period,
     483            0 :                 kind: BackgroundLoopKind::IngestHouseKeeping,
     484            0 :             };
     485            0 :             iteration.run(tenant.ingest_housekeeping()).await;
     486              :         }
     487            0 :     }
     488            0 :     .await;
     489            0 :     TENANT_TASK_EVENTS.with_label_values(&["stop"]).inc();
     490            0 : }
     491              : 
     492            0 : async fn wait_for_active_tenant(tenant: &Arc<Tenant>) -> ControlFlow<()> {
     493            0 :     // if the tenant has a proper status already, no need to wait for anything
     494            0 :     if tenant.current_state() == TenantState::Active {
     495            0 :         ControlFlow::Continue(())
     496              :     } else {
     497            0 :         let mut tenant_state_updates = tenant.subscribe_for_state_updates();
     498              :         loop {
     499            0 :             match tenant_state_updates.changed().await {
     500              :                 Ok(()) => {
     501            0 :                     let new_state = &*tenant_state_updates.borrow();
     502            0 :                     match new_state {
     503              :                         TenantState::Active => {
     504            0 :                             debug!("Tenant state changed to active, continuing the task loop");
     505            0 :                             return ControlFlow::Continue(());
     506              :                         }
     507            0 :                         state => {
     508            0 :                             debug!("Not running the task loop, tenant is not active: {state:?}");
     509            0 :                             continue;
     510              :                         }
     511              :                     }
     512              :                 }
     513            0 :                 Err(_sender_dropped_error) => {
     514            0 :                     return ControlFlow::Break(());
     515              :                 }
     516              :             }
     517              :         }
     518              :     }
     519            0 : }
     520              : 
     521            0 : #[derive(thiserror::Error, Debug)]
     522              : #[error("cancelled")]
     523              : pub(crate) struct Cancelled;
     524              : 
     525              : /// Provide a random delay for background task initialization.
     526              : ///
     527              : /// This delay prevents a thundering herd of background tasks and will likely keep them running on
     528              : /// different periods for more stable load.
     529            0 : pub(crate) async fn random_init_delay(
     530            0 :     period: Duration,
     531            0 :     cancel: &CancellationToken,
     532            0 : ) -> Result<(), Cancelled> {
     533            0 :     if period == Duration::ZERO {
     534            0 :         return Ok(());
     535            0 :     }
     536            0 : 
     537            0 :     let d = {
     538            0 :         let mut rng = rand::thread_rng();
     539            0 :         rng.gen_range(Duration::ZERO..=period)
     540            0 :     };
     541            0 :     match tokio::time::timeout(d, cancel.cancelled()).await {
     542            0 :         Ok(_) => Err(Cancelled),
     543            0 :         Err(_) => Ok(()),
     544              :     }
     545            0 : }
     546              : 
     547              : struct Iteration {
     548              :     started_at: Instant,
     549              :     period: Duration,
     550              :     kind: BackgroundLoopKind,
     551              : }
     552              : 
     553              : struct IterationResult<O> {
     554              :     output: O,
     555              :     elapsed: Duration,
     556              : }
     557              : 
     558              : impl Iteration {
     559            0 :     #[instrument(skip_all)]
     560              :     pub(crate) async fn run<Fut, O>(self, fut: Fut) -> IterationResult<O>
     561              :     where
     562              :         Fut: std::future::Future<Output = O>,
     563              :     {
     564              :         let Self {
     565              :             started_at,
     566              :             period,
     567              :             kind,
     568              :         } = self;
     569              : 
     570              :         let mut fut = std::pin::pin!(fut);
     571              : 
     572              :         // Wrap `fut` into a future that logs a message every `period` so that we get a
     573              :         // very obvious breadcrumb in the logs _while_ a slow iteration is happening.
     574            0 :         let liveness_logger = async move {
     575              :             loop {
     576            0 :                 match tokio::time::timeout(period, &mut fut).await {
     577            0 :                     Ok(x) => return x,
     578              :                     Err(_) => {
     579              :                         // info level as per the same rationale why warn_when_period_overrun is info
     580              :                         // =>  https://github.com/neondatabase/neon/pull/5724
     581            0 :                         info!("still running");
     582              :                     }
     583              :                 }
     584              :             }
     585            0 :         };
     586              : 
     587              :         let output = liveness_logger.await;
     588              : 
     589              :         let elapsed = started_at.elapsed();
     590              :         warn_when_period_overrun(elapsed, period, kind);
     591              : 
     592              :         IterationResult { output, elapsed }
     593              :     }
     594              : }
     595              : /// Attention: the `task` and `period` beocme labels of a pageserver-wide prometheus metric.
     596            0 : pub(crate) fn warn_when_period_overrun(
     597            0 :     elapsed: Duration,
     598            0 :     period: Duration,
     599            0 :     task: BackgroundLoopKind,
     600            0 : ) {
     601            0 :     // Duration::ZERO will happen because it's the "disable [bgtask]" value.
     602            0 :     if elapsed >= period && period != Duration::ZERO {
     603              :         // humantime does no significant digits clamping whereas Duration's debug is a bit more
     604              :         // intelligent. however it makes sense to keep the "configuration format" for period, even
     605              :         // though there's no way to output the actual config value.
     606            0 :         info!(
     607              :             ?elapsed,
     608            0 :             period = %humantime::format_duration(period),
     609            0 :             ?task,
     610            0 :             "task iteration took longer than the configured period"
     611              :         );
     612            0 :         crate::metrics::BACKGROUND_LOOP_PERIOD_OVERRUN_COUNT
     613            0 :             .with_label_values(&[task.as_static_str(), &format!("{}", period.as_secs())])
     614            0 :             .inc();
     615            0 :     }
     616            0 : }
        

Generated by: LCOV version 2.1-beta