millipede-core 0.1.0

Core primitives for the Millipede web crawler: request model, storage traits, events, errors, configuration.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};

struct UpdateGuard<'a>(&'a AtomicBool);

impl Drop for UpdateGuard<'_> {
    fn drop(&mut self) {
        self.0.store(false, Ordering::Release);
    }
}

/// An atomic additive-increase, multiplicative-decrease concurrency controller.
pub struct AimdController {
    min: usize,
    max: usize,
    increase_after_successes: usize,
    decrease_factor: f32,
    desired: AtomicUsize,
    success_streak: AtomicUsize,
    updating: AtomicBool,
}

impl AimdController {
    /// Creates a controller with normalized bounds and tuning values.
    pub fn new(
        min_concurrency: usize,
        max_concurrency: usize,
        initial_concurrency: usize,
        increase_after_successes: usize,
        decrease_factor: f32,
    ) -> Self {
        let min = min_concurrency.max(1);
        let max = max_concurrency.max(min);
        let decrease_factor = if decrease_factor > 0.0 && decrease_factor <= 1.0 {
            decrease_factor
        } else {
            0.5
        };
        Self {
            min,
            max,
            increase_after_successes: increase_after_successes.max(1),
            decrease_factor,
            desired: AtomicUsize::new(initial_concurrency.clamp(min, max)),
            success_streak: AtomicUsize::new(0),
            updating: AtomicBool::new(false),
        }
    }

    /// Returns the currently desired concurrency.
    pub fn desired_concurrency(&self) -> usize {
        self.desired.load(Ordering::Acquire)
    }

    /// Records a successful attempt, additively increasing after a sustained streak.
    pub fn record_success(&self) {
        let _update = self.begin_update();
        let previous = self
            .success_streak
            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |streak| {
                let next = streak.saturating_add(1);
                Some(if next >= self.increase_after_successes {
                    next - self.increase_after_successes
                } else {
                    next
                })
            })
            .expect("success streak update always produces a value");
        if previous >= self.increase_after_successes - 1 {
            let _ = self
                .desired
                .fetch_update(Ordering::AcqRel, Ordering::Acquire, |desired| {
                    Some(desired.saturating_add(1).min(self.max))
                });
        }
    }

    /// Records a setback, clearing the streak and multiplicatively decreasing concurrency.
    pub fn record_setback(&self) {
        let _update = self.begin_update();
        self.success_streak.store(0, Ordering::Release);
        let _ = self
            .desired
            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |desired| {
                Some(
                    ((desired as f32 * self.decrease_factor).round() as usize)
                        .clamp(self.min, self.max),
                )
            });
    }

    fn begin_update(&self) -> UpdateGuard<'_> {
        while self
            .updating
            .compare_exchange_weak(false, true, Ordering::Acquire, Ordering::Relaxed)
            .is_err()
        {
            std::hint::spin_loop();
        }
        UpdateGuard(&self.updating)
    }
}