opendal_layer_tail_cut/
lib.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18#![doc = include_str!("../README.md")]
19#![cfg_attr(docsrs, feature(doc_cfg))]
20#![cfg_attr(docsrs, doc(auto_cfg))]
21#![deny(missing_docs)]
22use std::fmt::Debug;
23use std::sync::Arc;
24use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
25
26use opendal_core::raw::*;
27use opendal_core::*;
28
29/// Builder for TailCutLayer.
30///
31/// Use this to configure the layer, then call `build()` to create a layer
32/// that can be cloned and shared across multiple operators.
33///
34/// # Examples
35///
36/// ```no_run
37/// # use std::time::Duration;
38/// #
39/// # use opendal_core::services;
40/// # use opendal_core::Operator;
41/// # use opendal_core::Result;
42/// # use opendal_layer_tail_cut::TailCutLayer;
43/// #
44/// # fn main() -> Result<()> {
45/// let layer = TailCutLayer::builder()
46///     .percentile(95)
47///     .window(Duration::from_secs(60))
48///     .build();
49///
50/// let op = Operator::new(services::Memory::default())?
51///     .layer(layer);
52/// # Ok(())
53/// # }
54/// ```
55#[derive(Clone)]
56pub struct TailCutLayerBuilder {
57    percentile: u8,
58    safety_factor: f64,
59    window: Duration,
60    min_samples: usize,
61    min_deadline: Duration,
62    max_deadline: Duration,
63}
64
65impl Default for TailCutLayerBuilder {
66    fn default() -> Self {
67        Self {
68            percentile: 95,
69            safety_factor: 1.3,
70            window: Duration::from_secs(60),
71            min_samples: 200,
72            min_deadline: Duration::from_millis(500),
73            max_deadline: Duration::from_secs(30),
74        }
75    }
76}
77
78impl TailCutLayerBuilder {
79    /// Create a new [`TailCutLayerBuilder`] with default settings.
80    pub fn new() -> Self {
81        Self::default()
82    }
83
84    /// Set the percentile threshold (e.g., 95 for P95, 99 for P99).
85    ///
86    /// Requests slower than this percentile × safety_factor will be cancelled.
87    ///
88    /// Default: 95
89    ///
90    /// # Panics
91    ///
92    /// Panics if percentile is not between 50 and 99.
93    pub fn percentile(mut self, percentile: u8) -> Self {
94        assert!(
95            (50..=99).contains(&percentile),
96            "percentile must be between 50 and 99"
97        );
98        self.percentile = percentile;
99        self
100    }
101
102    /// Set the safety factor multiplier.
103    ///
104    /// The actual deadline is calculated as: P{percentile} × safety_factor.
105    /// A higher value reduces false positives but may miss some long tails.
106    ///
107    /// Default: 1.3 (30% buffer)
108    ///
109    /// # Panics
110    ///
111    /// Panics if factor is not between 1.0 and 5.0.
112    pub fn safety_factor(mut self, factor: f64) -> Self {
113        assert!(
114            (1.0..=5.0).contains(&factor),
115            "safety_factor must be between 1.0 and 5.0"
116        );
117        self.safety_factor = factor;
118        self
119    }
120
121    /// Set the sliding window duration for statistics collection.
122    ///
123    /// Longer windows provide more stable statistics but react slower to changes.
124    /// Shorter windows adapt faster but may be more noisy.
125    ///
126    /// Default: 60 seconds
127    ///
128    /// # Panics
129    ///
130    /// Panics if window is greater than 120 seconds.
131    pub fn window(mut self, window: Duration) -> Self {
132        assert!(
133            window <= Duration::from_secs(120),
134            "window must be <= 120 seconds"
135        );
136        self.window = window;
137        self
138    }
139
140    /// Set the minimum number of samples required before enabling adaptive cancellation.
141    ///
142    /// During cold start (when sample count < min_samples), the layer will not
143    /// cancel any requests to avoid false positives.
144    ///
145    /// Default: 200
146    pub fn min_samples(mut self, min_samples: usize) -> Self {
147        self.min_samples = min_samples;
148        self
149    }
150
151    /// Set the minimum deadline (floor).
152    ///
153    /// Even if calculated deadline is shorter, it will be clamped to this value.
154    /// This prevents overly aggressive cancellation on very fast backends.
155    ///
156    /// Default: 500ms
157    pub fn min_deadline(mut self, deadline: Duration) -> Self {
158        self.min_deadline = deadline;
159        self
160    }
161
162    /// Set the maximum deadline (ceiling).
163    ///
164    /// Even if calculated deadline is longer, it will be clamped to this value.
165    /// This acts as a safety fallback timeout.
166    ///
167    /// Default: 30s
168    pub fn max_deadline(mut self, deadline: Duration) -> Self {
169        self.max_deadline = deadline;
170        self
171    }
172
173    /// Build the layer.
174    ///
175    /// The returned layer can be cloned to share statistics across operators.
176    ///
177    /// # Examples
178    ///
179    /// ```no_run
180    /// # use std::time::Duration;
181    /// #
182    /// # use opendal_core::services;
183    /// # use opendal_core::Operator;
184    /// # use opendal_core::Result;
185    /// # use opendal_layer_tail_cut::TailCutLayer;
186    /// #
187    /// # fn main() -> Result<()> {
188    /// let layer = TailCutLayer::builder()
189    ///     .percentile(95)
190    ///     .window(Duration::from_secs(60))
191    ///     .build();
192    ///
193    /// // Share the layer across operators
194    /// let op1 = Operator::new(services::Memory::default())?
195    ///     .layer(layer.clone());
196    ///
197    /// let op2 = Operator::new(services::Memory::default())?
198    ///     .layer(layer.clone());
199    /// // op1 and op2 share the same statistics
200    /// # Ok(())
201    /// # }
202    /// ```
203    pub fn build(self) -> TailCutLayer {
204        TailCutLayer {
205            config: Arc::new(TailCutConfig {
206                percentile: self.percentile,
207                safety_factor: self.safety_factor,
208                window: self.window,
209                min_samples: self.min_samples,
210                min_deadline: self.min_deadline,
211                max_deadline: self.max_deadline,
212            }),
213            stats: Arc::new(TailCutStats::new()),
214        }
215    }
216}
217
218/// Configuration for TailCutLayer (immutable).
219#[derive(Debug)]
220struct TailCutConfig {
221    percentile: u8,
222    safety_factor: f64,
223    window: Duration,
224    min_samples: usize,
225    min_deadline: Duration,
226    max_deadline: Duration,
227}
228
229/// `TailCutLayer` monitors request latency and cancels requests that run
230/// significantly slower than the historical baseline (for example, slower than
231/// P95).
232///
233/// Create this layer with [`TailCutLayer::builder()`]. Clone it to share
234/// statistics across multiple operators.
235///
236/// # Examples
237///
238/// ```no_run
239/// # use std::time::Duration;
240/// #
241/// # use opendal_core::services;
242/// # use opendal_core::Operator;
243/// # use opendal_core::Result;
244/// # use opendal_layer_tail_cut::TailCutLayer;
245/// #
246/// # fn main() -> Result<()> {
247/// let layer = TailCutLayer::builder()
248///     .percentile(95)
249///     .safety_factor(1.3)
250///     .window(Duration::from_secs(60))
251///     .build();
252///
253/// let op = Operator::new(services::Memory::default())?
254///     .layer(layer);
255/// # Ok(())
256/// # }
257/// ```
258#[derive(Clone)]
259pub struct TailCutLayer {
260    config: Arc<TailCutConfig>,
261    stats: Arc<TailCutStats>,
262}
263
264impl Debug for TailCutLayer {
265    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
266        f.debug_struct("TailCutLayer")
267            .field("config", &self.config)
268            .finish_non_exhaustive()
269    }
270}
271
272impl Default for TailCutLayer {
273    fn default() -> Self {
274        Self::builder().build()
275    }
276}
277
278impl TailCutLayer {
279    /// Create a new [`TailCutLayerBuilder`] to configure the layer.
280    pub fn builder() -> TailCutLayerBuilder {
281        TailCutLayerBuilder::default()
282    }
283
284    /// Create a new [`TailCutLayer`].
285    ///
286    /// This is equivalent to `TailCutLayer::builder().build()`.
287    pub fn new() -> Self {
288        Self::default()
289    }
290}
291
292impl Layer for TailCutLayer {
293    fn apply_service(&self, inner: Servicer) -> Servicer {
294        Arc::new(self.layer(inner))
295    }
296}
297
298impl TailCutLayer {
299    fn layer(&self, inner: Servicer) -> TailCutService {
300        TailCutService {
301            inner,
302            config: self.config.clone(),
303            stats: self.stats.clone(),
304        }
305    }
306}
307
308#[doc(hidden)]
309/// Service wrapper that applies tail-cut deadlines to operation calls.
310pub struct TailCutService {
311    inner: Servicer,
312    config: Arc<TailCutConfig>,
313    stats: Arc<TailCutStats>,
314}
315
316impl Debug for TailCutService {
317    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
318        f.debug_struct("TailCutService")
319            .field("config", &self.config)
320            .finish_non_exhaustive()
321    }
322}
323
324impl TailCutService {
325    /// Calculate the deadline for a given operation and size.
326    fn calculate_deadline(&self, op: Operation, size: Option<u64>) -> Option<Duration> {
327        let op_stats = self.stats.stats_for(op);
328
329        if op_stats.total_samples(size, self.config.window) < self.config.min_samples {
330            return None;
331        }
332
333        let q = self.config.percentile as f64 / 100.0;
334        let pctl = op_stats.quantile(size, q, self.config.window)?;
335
336        let deadline = Duration::from_secs_f64(pctl.as_secs_f64() * self.config.safety_factor);
337        Some(deadline.clamp(self.config.min_deadline, self.config.max_deadline))
338    }
339
340    async fn with_deadline<F, T>(&self, op: Operation, size: Option<u64>, fut: F) -> Result<T>
341    where
342        F: Future<Output = Result<T>>,
343    {
344        let start = Instant::now();
345
346        let result = if let Some(deadline) = self.calculate_deadline(op, size) {
347            match tokio::time::timeout(deadline, fut).await {
348                Ok(res) => res,
349                Err(_) => Err(Error::new(ErrorKind::Unexpected, "cancelled by tail cut")
350                    .with_operation(op)
351                    .with_context("percentile", format!("P{}", self.config.percentile))
352                    .with_context("deadline", format!("{:?}", deadline))
353                    .set_temporary()),
354            }
355        } else {
356            fut.await
357        };
358
359        if result.is_ok() {
360            let latency = start.elapsed();
361            self.stats.stats_for(op).record(size, latency);
362        }
363
364        result
365    }
366}
367
368impl Service for TailCutService {
369    type Reader = TailCutWrapper<oio::Reader>;
370    type Writer = TailCutWrapper<oio::Writer>;
371    type Lister = TailCutWrapper<oio::Lister>;
372    type Deleter = TailCutWrapper<oio::Deleter>;
373    type Copier = TailCutWrapper<oio::Copier>;
374
375    fn info(&self) -> ServiceInfo {
376        self.inner.info()
377    }
378
379    fn capability(&self) -> Capability {
380        self.inner.capability()
381    }
382
383    async fn create_dir(
384        &self,
385        ctx: &OperationContext,
386        path: &str,
387        args: OpCreateDir,
388    ) -> Result<RpCreateDir> {
389        self.with_deadline(
390            Operation::CreateDir,
391            None,
392            self.inner.create_dir(ctx, path, args),
393        )
394        .await
395    }
396
397    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
398        self.inner
399            .read(ctx, path, args)
400            .map(|r| TailCutWrapper::new(r, None, self.config.clone(), self.stats.clone()))
401    }
402
403    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
404        self.inner
405            .write(ctx, path, args)
406            .map(|w| TailCutWrapper::new(w, None, self.config.clone(), self.stats.clone()))
407    }
408
409    fn copy(
410        &self,
411        ctx: &OperationContext,
412        from: &str,
413        to: &str,
414        args: OpCopy,
415        opts: OpCopier,
416    ) -> Result<Self::Copier> {
417        self.inner
418            .copy(ctx, from, to, args, opts)
419            .map(|c| TailCutWrapper::new(c, None, self.config.clone(), self.stats.clone()))
420    }
421
422    async fn rename(
423        &self,
424        ctx: &OperationContext,
425        from: &str,
426        to: &str,
427        args: OpRename,
428    ) -> Result<RpRename> {
429        self.with_deadline(
430            Operation::Rename,
431            None,
432            self.inner.rename(ctx, from, to, args),
433        )
434        .await
435    }
436
437    async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
438        self.with_deadline(Operation::Stat, None, self.inner.stat(ctx, path, args))
439            .await
440    }
441
442    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
443        self.inner
444            .delete(ctx)
445            .map(|d| TailCutWrapper::new(d, None, self.config.clone(), self.stats.clone()))
446    }
447
448    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
449        self.inner
450            .list(ctx, path, args)
451            .map(|l| TailCutWrapper::new(l, None, self.config.clone(), self.stats.clone()))
452    }
453
454    async fn presign(
455        &self,
456        ctx: &OperationContext,
457        path: &str,
458        args: OpPresign,
459    ) -> Result<RpPresign> {
460        self.with_deadline(
461            Operation::Presign,
462            None,
463            self.inner.presign(ctx, path, args),
464        )
465        .await
466    }
467}
468
469#[doc(hidden)]
470/// Body wrapper that applies tail-cut deadlines to streaming I/O calls.
471pub struct TailCutWrapper<R> {
472    inner: R,
473    size: Option<u64>,
474    config: Arc<TailCutConfig>,
475    stats: Arc<TailCutStats>,
476}
477
478impl<R> TailCutWrapper<R> {
479    fn new(
480        inner: R,
481        size: Option<u64>,
482        config: Arc<TailCutConfig>,
483        stats: Arc<TailCutStats>,
484    ) -> Self {
485        Self {
486            inner,
487            size,
488            config,
489            stats,
490        }
491    }
492
493    fn calculate_deadline(&self, op: Operation) -> Option<Duration> {
494        self.calculate_deadline_for(op, self.size)
495    }
496
497    fn calculate_deadline_for(&self, op: Operation, size: Option<u64>) -> Option<Duration> {
498        let op_stats = self.stats.stats_for(op);
499
500        if op_stats.total_samples(size, self.config.window) < self.config.min_samples {
501            return None;
502        }
503
504        let q = self.config.percentile as f64 / 100.0;
505        let pctl = op_stats.quantile(size, q, self.config.window)?;
506
507        let deadline = Duration::from_secs_f64(pctl.as_secs_f64() * self.config.safety_factor);
508        Some(deadline.clamp(self.config.min_deadline, self.config.max_deadline))
509    }
510
511    #[inline]
512    async fn with_io_deadline<F, T>(
513        deadline: Option<Duration>,
514        percentile: u8,
515        stats: &Arc<TailCutStats>,
516        size: Option<u64>,
517        op: Operation,
518        fut: F,
519    ) -> Result<T>
520    where
521        F: std::future::Future<Output = Result<T>>,
522    {
523        let start = Instant::now();
524
525        let result = if let Some(dl) = deadline {
526            match tokio::time::timeout(dl, fut).await {
527                Ok(res) => res,
528                Err(_) => Err(
529                    Error::new(ErrorKind::Unexpected, "io cancelled by tail cut")
530                        .with_operation(op)
531                        .with_context("percentile", format!("P{}", percentile))
532                        .with_context("deadline", format!("{:?}", dl))
533                        .set_temporary(),
534                ),
535            }
536        } else {
537            fut.await
538        };
539
540        if result.is_ok() {
541            let latency = start.elapsed();
542            stats.stats_for(op).record(size, latency);
543        }
544
545        result
546    }
547}
548
549impl<R: oio::ReadStream> oio::ReadStream for TailCutWrapper<R> {
550    async fn read(&mut self) -> Result<Buffer> {
551        let deadline = self.calculate_deadline_for(Operation::Read, self.size);
552        Self::with_io_deadline(
553            deadline,
554            self.config.percentile,
555            &self.stats,
556            self.size,
557            Operation::Read,
558            self.inner.read(),
559        )
560        .await
561    }
562}
563
564impl<R: oio::Read> oio::Read for TailCutWrapper<R> {
565    async fn open(&self, range: BytesRange) -> Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> {
566        let size = range.size();
567        let deadline = self.calculate_deadline_for(Operation::Read, size);
568        let (rp, stream) = Self::with_io_deadline(
569            deadline,
570            self.config.percentile,
571            &self.stats,
572            size,
573            Operation::Read,
574            self.inner.open(range),
575        )
576        .await?;
577
578        Ok((
579            rp,
580            Box::new(TailCutWrapper::new(
581                stream,
582                size,
583                self.config.clone(),
584                self.stats.clone(),
585            )) as Box<dyn oio::ReadStreamDyn>,
586        ))
587    }
588
589    async fn read(&self, range: BytesRange) -> Result<(RpRead, Buffer)> {
590        let size = range.size();
591        let deadline = self.calculate_deadline_for(Operation::Read, size);
592        Self::with_io_deadline(
593            deadline,
594            self.config.percentile,
595            &self.stats,
596            size,
597            Operation::Read,
598            self.inner.read(range),
599        )
600        .await
601    }
602}
603
604impl<R: oio::Write> oio::Write for TailCutWrapper<R> {
605    async fn write(&mut self, bs: Buffer) -> Result<()> {
606        let deadline = self.calculate_deadline(Operation::Write);
607        Self::with_io_deadline(
608            deadline,
609            self.config.percentile,
610            &self.stats,
611            self.size,
612            Operation::Write,
613            self.inner.write(bs),
614        )
615        .await
616    }
617
618    async fn close(&mut self) -> Result<Metadata> {
619        let deadline = self.calculate_deadline(Operation::Write);
620        Self::with_io_deadline(
621            deadline,
622            self.config.percentile,
623            &self.stats,
624            self.size,
625            Operation::Write,
626            self.inner.close(),
627        )
628        .await
629    }
630
631    async fn abort(&mut self) -> Result<()> {
632        let deadline = self.calculate_deadline(Operation::Write);
633        Self::with_io_deadline(
634            deadline,
635            self.config.percentile,
636            &self.stats,
637            self.size,
638            Operation::Write,
639            self.inner.abort(),
640        )
641        .await
642    }
643}
644
645impl<R: oio::List> oio::List for TailCutWrapper<R> {
646    async fn next(&mut self) -> Result<Option<oio::Entry>> {
647        let deadline = self.calculate_deadline(Operation::List);
648        Self::with_io_deadline(
649            deadline,
650            self.config.percentile,
651            &self.stats,
652            self.size,
653            Operation::List,
654            self.inner.next(),
655        )
656        .await
657    }
658}
659
660impl<R: oio::Delete> oio::Delete for TailCutWrapper<R> {
661    async fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
662        self.inner.delete(path, args).await
663    }
664
665    async fn close(&mut self) -> Result<()> {
666        let deadline = self.calculate_deadline(Operation::Delete);
667        Self::with_io_deadline(
668            deadline,
669            self.config.percentile,
670            &self.stats,
671            self.size,
672            Operation::Delete,
673            self.inner.close(),
674        )
675        .await
676    }
677}
678
679impl<C: oio::Copy> oio::Copy for TailCutWrapper<C> {
680    async fn next(&mut self) -> Result<Option<usize>> {
681        let deadline = self.calculate_deadline(Operation::Copy);
682        Self::with_io_deadline(
683            deadline,
684            self.config.percentile,
685            &self.stats,
686            self.size,
687            Operation::Copy,
688            self.inner.next(),
689        )
690        .await
691    }
692
693    async fn close(&mut self) -> Result<Metadata> {
694        let deadline = self.calculate_deadline(Operation::Copy);
695        Self::with_io_deadline(
696            deadline,
697            self.config.percentile,
698            &self.stats,
699            self.size,
700            Operation::Copy,
701            self.inner.close(),
702        )
703        .await
704    }
705
706    async fn abort(&mut self) -> Result<()> {
707        let deadline = self.calculate_deadline(Operation::Copy);
708        Self::with_io_deadline(
709            deadline,
710            self.config.percentile,
711            &self.stats,
712            self.size,
713            Operation::Copy,
714            self.inner.abort(),
715        )
716        .await
717    }
718}
719
720/// Statistics engine for tail cut layer.
721struct TailCutStats {
722    // Statistics for each operation type (7 operations)
723    operations: [Arc<OperationStats>; 7],
724}
725
726impl TailCutStats {
727    fn new() -> Self {
728        Self {
729            operations: std::array::from_fn(|_| Arc::new(OperationStats::new())),
730        }
731    }
732
733    fn stats_for(&self, op: Operation) -> &Arc<OperationStats> {
734        let idx = match op {
735            Operation::Read => 0,
736            Operation::Write => 1,
737            Operation::Stat => 2,
738            Operation::List => 3,
739            Operation::Delete => 4,
740            Operation::Copy => 5,
741            Operation::Rename => 6,
742            _ => 2, // fallback to Stat
743        };
744        &self.operations[idx]
745    }
746}
747
748/// Statistics for a single operation type.
749struct OperationStats {
750    buckets: Vec<SizeBucket>,
751}
752
753impl OperationStats {
754    fn new() -> Self {
755        Self {
756            buckets: vec![
757                SizeBucket::new(0, Some(4 * 1024)),                   // [0, 4KB)
758                SizeBucket::new(4 * 1024, Some(64 * 1024)),           // [4KB, 64KB)
759                SizeBucket::new(64 * 1024, Some(1024 * 1024)),        // [64KB, 1MB)
760                SizeBucket::new(1024 * 1024, Some(16 * 1024 * 1024)), // [1MB, 16MB)
761                SizeBucket::new(16 * 1024 * 1024, Some(256 * 1024 * 1024)), // [16MB, 256MB)
762                SizeBucket::new(256 * 1024 * 1024, None),             // [256MB, ∞)
763            ],
764        }
765    }
766
767    fn bucket_for(&self, size: Option<u64>) -> &SizeBucket {
768        let size = size.unwrap_or(u64::MAX);
769
770        self.buckets
771            .iter()
772            .find(|b| b.contains(size))
773            .unwrap_or(&self.buckets[self.buckets.len() - 1])
774    }
775
776    fn record(&self, size: Option<u64>, latency: Duration) {
777        self.bucket_for(size).histogram.record(latency);
778    }
779
780    fn quantile(&self, size: Option<u64>, q: f64, window: Duration) -> Option<Duration> {
781        self.bucket_for(size).histogram.quantile(q, window)
782    }
783
784    fn total_samples(&self, size: Option<u64>, window: Duration) -> usize {
785        self.bucket_for(size).histogram.total_samples(window)
786    }
787}
788
789/// Size bucket for categorizing operations by data size.
790struct SizeBucket {
791    min_size: u64,
792    max_size: Option<u64>,
793    histogram: WindowedHistogram,
794}
795
796impl SizeBucket {
797    fn new(min_size: u64, max_size: Option<u64>) -> Self {
798        Self {
799            min_size,
800            max_size,
801            histogram: WindowedHistogram::new(),
802        }
803    }
804
805    fn contains(&self, size: u64) -> bool {
806        size >= self.min_size && self.max_size.is_none_or(|max| size < max)
807    }
808}
809
810const SLICE_DURATION_MS: u64 = 10_000; // 10 seconds per slice
811const NUM_SLICES: usize = 12; // 12 slices = 120 seconds total window
812const NUM_BUCKETS: usize = 17; // 17 buckets covering 1ms to 64s
813
814/// Windowed histogram using lock-free atomic operations.
815struct WindowedHistogram {
816    slices: Box<[TimeSlice; NUM_SLICES]>,
817    current_idx: AtomicUsize,
818    last_rotate: AtomicU64,
819}
820
821impl WindowedHistogram {
822    fn new() -> Self {
823        Self {
824            slices: Box::new(std::array::from_fn(|_| TimeSlice::new())),
825            current_idx: AtomicUsize::new(0),
826            last_rotate: AtomicU64::new(Self::now_ms()),
827        }
828    }
829
830    fn record(&self, latency: Duration) {
831        self.maybe_rotate();
832
833        let bucket_idx = Self::latency_to_bucket(latency);
834        let slice_idx = self.current_idx.load(Ordering::Relaxed);
835
836        self.slices[slice_idx].buckets[bucket_idx].fetch_add(1, Ordering::Relaxed);
837    }
838
839    fn quantile(&self, q: f64, window: Duration) -> Option<Duration> {
840        debug_assert!((0.0..=1.0).contains(&q), "quantile must be in [0, 1]");
841
842        let snapshot = self.snapshot(window);
843        let total: u64 = snapshot.iter().sum();
844
845        if total == 0 {
846            return None;
847        }
848
849        let target = (total as f64 * q).ceil() as u64;
850        let mut cumsum = 0u64;
851
852        for (bucket_idx, &count) in snapshot.iter().enumerate() {
853            cumsum += count;
854            if cumsum >= target {
855                return Some(Self::bucket_to_latency(bucket_idx));
856            }
857        }
858
859        Some(Self::bucket_to_latency(NUM_BUCKETS - 1))
860    }
861
862    fn total_samples(&self, window: Duration) -> usize {
863        self.snapshot(window).iter().map(|&v| v as usize).sum()
864    }
865
866    fn snapshot(&self, window: Duration) -> [u64; NUM_BUCKETS] {
867        let mut result = [0u64; NUM_BUCKETS];
868        let now_ms = Self::now_ms();
869        let window_ms = window.as_millis() as u64;
870
871        for slice in self.slices.iter() {
872            let start = slice.start_epoch_ms.load(Ordering::Acquire);
873
874            if start > 0 && now_ms.saturating_sub(start) < window_ms + SLICE_DURATION_MS {
875                for (i, bucket) in slice.buckets.iter().enumerate() {
876                    result[i] += bucket.load(Ordering::Relaxed);
877                }
878            }
879        }
880
881        result
882    }
883
884    fn maybe_rotate(&self) {
885        let now = Self::now_ms();
886        let last_rotate = self.last_rotate.load(Ordering::Relaxed);
887
888        if now - last_rotate >= SLICE_DURATION_MS
889            && self
890                .last_rotate
891                .compare_exchange(last_rotate, now, Ordering::Release, Ordering::Relaxed)
892                .is_ok()
893        {
894            let old_idx = self.current_idx.load(Ordering::Relaxed);
895            let new_idx = (old_idx + 1) % NUM_SLICES;
896
897            let new_slice = &self.slices[new_idx];
898            new_slice.start_epoch_ms.store(now, Ordering::Release);
899            for bucket in &new_slice.buckets {
900                bucket.store(0, Ordering::Relaxed);
901            }
902
903            self.current_idx.store(new_idx, Ordering::Release);
904        }
905    }
906
907    fn latency_to_bucket(latency: Duration) -> usize {
908        let ms = latency.as_millis() as u64;
909
910        if ms == 0 {
911            return 0;
912        }
913
914        let bucket = 64 - ms.leading_zeros();
915        (bucket as usize).min(NUM_BUCKETS - 1)
916    }
917
918    fn bucket_to_latency(bucket_idx: usize) -> Duration {
919        if bucket_idx == 0 {
920            Duration::from_millis(1)
921        } else if bucket_idx >= NUM_BUCKETS - 1 {
922            Duration::from_secs(64)
923        } else {
924            Duration::from_millis(1u64 << bucket_idx)
925        }
926    }
927
928    fn now_ms() -> u64 {
929        // SAFETY: Unless System clock goes backwards before UNIX_EPOCH, this should never fail.
930        u64::try_from(Timestamp::now().into_inner().as_millisecond()).unwrap()
931    }
932}
933
934/// Time slice in the sliding window.
935struct TimeSlice {
936    // 17 buckets covering 1ms to 64s (logarithmic scale)
937    buckets: [AtomicU64; NUM_BUCKETS],
938    start_epoch_ms: AtomicU64,
939}
940
941impl TimeSlice {
942    fn new() -> Self {
943        Self {
944            buckets: std::array::from_fn(|_| AtomicU64::new(0)),
945            start_epoch_ms: AtomicU64::new(0),
946        }
947    }
948}
949
950#[cfg(test)]
951mod tests {
952    use super::*;
953
954    #[test]
955    fn test_latency_to_bucket() {
956        assert_eq!(
957            WindowedHistogram::latency_to_bucket(Duration::from_millis(0)),
958            0
959        );
960        assert_eq!(
961            WindowedHistogram::latency_to_bucket(Duration::from_millis(1)),
962            1
963        );
964        assert_eq!(
965            WindowedHistogram::latency_to_bucket(Duration::from_millis(2)),
966            2
967        );
968        assert_eq!(
969            WindowedHistogram::latency_to_bucket(Duration::from_millis(4)),
970            3
971        );
972        assert_eq!(
973            WindowedHistogram::latency_to_bucket(Duration::from_millis(8)),
974            4
975        );
976        assert_eq!(
977            WindowedHistogram::latency_to_bucket(Duration::from_millis(500)),
978            9
979        );
980        assert_eq!(
981            WindowedHistogram::latency_to_bucket(Duration::from_secs(1)),
982            10
983        );
984        assert_eq!(
985            WindowedHistogram::latency_to_bucket(Duration::from_secs(2)),
986            11
987        );
988        assert_eq!(
989            WindowedHistogram::latency_to_bucket(Duration::from_secs(64)),
990            16
991        );
992        assert_eq!(
993            WindowedHistogram::latency_to_bucket(Duration::from_secs(1000)),
994            16
995        );
996    }
997
998    #[test]
999    fn test_size_bucket_contains() {
1000        let bucket = SizeBucket::new(0, Some(4096));
1001        assert!(bucket.contains(0));
1002        assert!(bucket.contains(4095));
1003        assert!(!bucket.contains(4096));
1004
1005        let bucket = SizeBucket::new(4096, None);
1006        assert!(!bucket.contains(4095));
1007        assert!(bucket.contains(4096));
1008        assert!(bucket.contains(u64::MAX));
1009    }
1010
1011    #[tokio::test]
1012    async fn test_histogram_basic() {
1013        let hist = WindowedHistogram::new();
1014        let now = WindowedHistogram::now_ms();
1015        hist.slices[0].start_epoch_ms.store(now, Ordering::Release);
1016
1017        hist.record(Duration::from_millis(10));
1018        hist.record(Duration::from_millis(20));
1019        hist.record(Duration::from_millis(30));
1020
1021        let samples = hist.total_samples(Duration::from_secs(60));
1022        assert_eq!(samples, 3);
1023
1024        let p50 = hist.quantile(0.5, Duration::from_secs(60));
1025        assert!(p50.is_some());
1026    }
1027
1028    #[tokio::test]
1029    async fn test_tail_cut_layer_build() {
1030        let layer = TailCutLayer::builder()
1031            .percentile(95)
1032            .safety_factor(1.5)
1033            .window(Duration::from_secs(60))
1034            .min_samples(100)
1035            .min_deadline(Duration::from_millis(200))
1036            .max_deadline(Duration::from_secs(20))
1037            .build();
1038
1039        assert_eq!(layer.config.percentile, 95);
1040        assert_eq!(layer.config.safety_factor, 1.5);
1041        assert_eq!(layer.config.window, Duration::from_secs(60));
1042        assert_eq!(layer.config.min_samples, 100);
1043        assert_eq!(layer.config.min_deadline, Duration::from_millis(200));
1044        assert_eq!(layer.config.max_deadline, Duration::from_secs(20));
1045    }
1046
1047    #[tokio::test]
1048    async fn test_layer_clone_shares_stats() {
1049        let layer = TailCutLayer::new();
1050        let cloned = layer.clone();
1051
1052        assert!(Arc::ptr_eq(&layer.stats, &cloned.stats));
1053    }
1054}
opendal_layer_tail_cut/lib.rs

opendal_layer_tail_cut/
lib.rs
