actordb 0.7.0

An actor-first distributed database
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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

//! Aggregate encoded-body admission shared by every physical peer connection.

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use tokio::sync::{OwnedSemaphorePermit, Semaphore, TryAcquireError};
use tokio::time::{timeout_at, Instant};

use super::MAX_FRAME_BYTES;

pub(crate) const NODE_PEER_QUEUE_BYTE_CAPACITY: usize = 256 * 1024 * 1024;

/// Approximate node-wide encoded peer-writer admission usage.
///
/// Counts include reservations before encoding and bodies queued or currently
/// being written. They exclude protobuf objects, frame headers, allocator and
/// task overhead, cache/storage/replication payloads, and kernel socket buffers.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PeerWriterDiagnostics {
    capacity_bytes: usize,
    admitted_bytes: usize,
    producer_admitted_bytes: usize,
    saturation_events: u64,
}
impl PeerWriterDiagnostics {
    /// Total encoded-body allowance across all physical peer connections.
    pub fn capacity_bytes(&self) -> usize {
        self.capacity_bytes
    }
    /// Encoded bytes reserved, queued, or being written across the Node.
    pub fn admitted_bytes(&self) -> usize {
        self.admitted_bytes
    }
    /// Bytes reserved by ordinary producers, excluding priority responses.
    pub fn producer_admitted_bytes(&self) -> usize {
        self.producer_admitted_bytes
    }
    /// Failed immediate aggregate-byte acquisitions, including attempts that
    /// subsequently wait for capacity within their original deadline.
    pub fn saturation_events(&self) -> u64 {
        self.saturation_events
    }
}

pub(crate) struct GlobalBytePermits {
    pub(super) total: OwnedSemaphorePermit,
    pub(super) producer: Option<OwnedSemaphorePermit>,
}

pub(crate) struct NodePeerByteBudget {
    capacity: usize,
    total: Arc<Semaphore>,
    producers: Arc<Semaphore>,
    saturations: AtomicU64,
}
impl NodePeerByteBudget {
    /// Called only after Node runtime option validation; two maximum bodies
    /// fit so one full response remains reserved for readers.
    pub(crate) fn new(capacity: usize) -> Arc<Self> {
        assert!(capacity >= 2 * MAX_FRAME_BYTES);
        Arc::new(Self {
            capacity,
            total: Arc::new(Semaphore::new(capacity)),
            producers: Arc::new(Semaphore::new(capacity - MAX_FRAME_BYTES)),
            saturations: AtomicU64::new(0),
        })
    }
    fn saturated(&self) {
        let mut observed = self.saturations.load(Ordering::Relaxed);
        loop {
            match self.saturations.compare_exchange_weak(
                observed,
                observed.saturating_add(1),
                Ordering::Relaxed,
                Ordering::Relaxed,
            ) {
                Ok(_) => return,
                Err(actual) => observed = actual,
            }
        }
    }
    pub(crate) fn diagnostics(&self) -> PeerWriterDiagnostics {
        PeerWriterDiagnostics {
            capacity_bytes: self.capacity,
            admitted_bytes: self.capacity.saturating_sub(self.total.available_permits()),
            producer_admitted_bytes: (self.capacity - MAX_FRAME_BYTES)
                .saturating_sub(self.producers.available_permits()),
            saturation_events: self.saturations.load(Ordering::Relaxed),
        }
    }
    pub(crate) fn try_response(&self, bytes: u32) -> Result<GlobalBytePermits, ()> {
        match self.total.clone().try_acquire_many_owned(bytes) {
            Ok(total) => Ok(GlobalBytePermits {
                total,
                producer: None,
            }),
            Err(TryAcquireError::NoPermits) => {
                self.saturated();
                Err(())
            }
            Err(TryAcquireError::Closed) => Err(()),
        }
    }
    async fn acquire(
        &self,
        semaphore: &Arc<Semaphore>,
        bytes: u32,
        deadline: Instant,
    ) -> Result<OwnedSemaphorePermit, ()> {
        if Instant::now() >= deadline {
            return Err(());
        }
        match semaphore.clone().try_acquire_many_owned(bytes) {
            Ok(permit) => Ok(permit),
            Err(TryAcquireError::Closed) => Err(()),
            Err(TryAcquireError::NoPermits) => {
                self.saturated();
                timeout_at(deadline, semaphore.clone().acquire_many_owned(bytes))
                    .await
                    .map_err(|_| ())?
                    .map_err(|_| ())
            }
        }
    }
    pub(crate) async fn acquire_producer(
        &self,
        bytes: u32,
        deadline: Instant,
    ) -> Result<GlobalBytePermits, ()> {
        // Producers cannot consume the priority-response reserve. Acquire both
        // leases before allocating an encoded body, and keep them until Drop.
        let producer = self.acquire(&self.producers, bytes, deadline).await?;
        let total = self.acquire(&self.total, bytes, deadline).await?;
        Ok(GlobalBytePermits {
            total,
            producer: Some(producer),
        })
    }
}