harmos_runtime/streams/
series.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::any::Any;
18use std::ops::Bound;
19use std::sync::Arc;
20// Third Party Imports
21use harmos_domain::Aggregate;
22use harmos_storage::signal::Summary;
23use tokio::sync::{mpsc, oneshot};
24// Local Imports
25use super::BeneathWindow;
26use crate::aggregate;
27use crate::{state::At, state::Position, streams::Definition};
28
29/* === Constants === */
30
31/* === Macros === */
32
33/* === Type Aliases === */
34
35pub(crate) type Served = At<Arc<dyn Any + Send + Sync>>;
36
37/* === Structs === */
38
39#[derive(Clone, Copy, Debug, PartialEq)]
40pub struct Point {
41    pub axis: u64,
42    pub value: f64,
43}
44
45// One plot bucket of a field: the points a chart draws, and the running
46// summary every mergeable aggregate is answered from.
47#[derive(Clone, Debug, PartialEq)]
48pub struct Bucket {
49    pub field: String,
50    pub first: Point,
51    pub last: Point,
52    pub min: Point,
53    pub max: Point,
54    summary: Summary,
55}
56
57#[derive(Clone, Debug, PartialEq)]
58pub struct Aggregated {
59    pub field: String,
60    pub value: f64,
61}
62
63pub(crate) struct Opened {
64    pub(crate) source: Result<Box<dyn crate::streams::RowSource>, crate::streams::ReadError>,
65    pub(crate) permit: tokio::sync::OwnedSemaphorePermit,
66}
67
68#[derive(Clone)]
69pub(crate) struct Tier {
70    sender: mpsc::Sender<Request>,
71    readers: Arc<tokio::sync::Semaphore>,
72}
73
74/* === Enums === */
75
76pub(crate) enum Request {
77    Reader {
78        stream: Definition,
79        track: Vec<String>,
80        start: Bound<Position>,
81        end: Bound<Position>,
82        limits: crate::streams::ReadLimits,
83        permit: tokio::sync::OwnedSemaphorePermit,
84        reply: oneshot::Sender<Opened>,
85    },
86    Rows {
87        stream: Definition,
88        track: Vec<String>,
89        start: Bound<Position>,
90        end: Bound<Position>,
91        reply: oneshot::Sender<Result<Vec<Served>, BeneathWindow>>,
92    },
93    Buckets {
94        stream: Definition,
95        track: Vec<String>,
96        start: Bound<u64>,
97        end: Bound<u64>,
98        budget: usize,
99        reply: oneshot::Sender<Result<Vec<Bucket>, BeneathWindow>>,
100    },
101    Aggregate {
102        stream: Definition,
103        track: Vec<String>,
104        start: Bound<u64>,
105        end: Bound<u64>,
106        kind: Aggregate,
107        reply: oneshot::Sender<Result<Vec<Aggregated>, BeneathWindow>>,
108    },
109    Seal {
110        stream: Definition,
111        track: Vec<String>,
112        reply: oneshot::Sender<Position>,
113    },
114}
115
116/* === Traits === */
117
118/* === Implementations === */
119
120impl Bucket {
121    pub fn point(field: impl Into<String>, point: Point) -> Self {
122        let mut summary = Summary::new();
123        summary.push(point.value);
124
125        Self { field: field.into(), first: point, last: point, min: point, max: point, summary }
126    }
127
128    pub fn merge(&self, later: &Self) -> Self {
129        assert_eq!(self.field, later.field, "a bucket merges only with a bucket of its own field");
130
131        Self {
132            field: self.field.clone(),
133            first: earliest(self.first, later.first),
134            last: latest(self.last, later.last),
135            min: lower(self.min, later.min),
136            max: higher(self.max, later.max),
137            summary: self.summary.merge(&later.summary),
138        }
139    }
140
141    pub fn count(&self) -> u64 {
142        self.summary.count()
143    }
144
145    pub fn summary(&self) -> &Summary {
146        &self.summary
147    }
148
149    // Mergeable members only: a bucket keeps no values, so median and
150    // percentiles answer `None` and are asked of the values instead.
151    pub fn aggregate(&self, kind: Aggregate) -> Option<f64> {
152        aggregate::of_summary(&self.summary, kind)
153    }
154}
155
156impl Tier {
157    pub(crate) fn channel(capacity: usize) -> (Self, mpsc::Receiver<Request>) {
158        let (sender, receiver) = mpsc::channel(capacity.max(1));
159
160        (Self { sender, readers: Arc::new(tokio::sync::Semaphore::new(capacity.max(1))) }, receiver)
161    }
162
163    pub(crate) fn reader_permit(&self) -> Result<tokio::sync::OwnedSemaphorePermit, crate::streams::ReadError> {
164        self.readers.clone().try_acquire_owned().map_err(|_| crate::streams::ReadError::Busy)
165    }
166
167    pub(crate) async fn reader(
168        &self, stream: Definition, track: Vec<String>, start: Bound<Position>, end: Bound<Position>, limits: crate::streams::ReadLimits,
169        permit: tokio::sync::OwnedSemaphorePermit,
170    ) -> Result<Opened, crate::streams::ReadError> {
171        let (reply, answer) = oneshot::channel();
172        self.sender
173            .send(Request::Reader { stream, track, start, end, limits, permit, reply })
174            .await
175            .map_err(|_| crate::streams::ReadError::Closed)?;
176        answer.await.map_err(|_| crate::streams::ReadError::Closed)
177    }
178
179    pub(crate) async fn rows(
180        &self, stream: Definition, track: Vec<String>, start: Bound<Position>, end: Bound<Position>,
181    ) -> Result<Vec<Served>, BeneathWindow> {
182        let (reply, answer) = oneshot::channel();
183        self.sender.send(Request::Rows { stream, track, start, end, reply }).await.map_err(|_| BeneathWindow::Unserved)?;
184
185        answer.await.map_err(|_| BeneathWindow::Unserved)?
186    }
187
188    pub(crate) async fn buckets(
189        &self, stream: Definition, track: Vec<String>, start: Bound<u64>, end: Bound<u64>, budget: usize,
190    ) -> Result<Vec<Bucket>, BeneathWindow> {
191        let (reply, answer) = oneshot::channel();
192        self.sender.send(Request::Buckets { stream, track, start, end, budget, reply }).await.map_err(|_| BeneathWindow::Unserved)?;
193
194        answer.await.map_err(|_| BeneathWindow::Unserved)?
195    }
196
197    pub(crate) async fn aggregate(
198        &self, stream: Definition, track: Vec<String>, start: Bound<u64>, end: Bound<u64>, kind: Aggregate,
199    ) -> Result<Vec<Aggregated>, BeneathWindow> {
200        let (reply, answer) = oneshot::channel();
201        self.sender.send(Request::Aggregate { stream, track, start, end, kind, reply }).await.map_err(|_| BeneathWindow::Unserved)?;
202
203        answer.await.map_err(|_| BeneathWindow::Unserved)?
204    }
205
206    pub(crate) async fn seal(&self, stream: Definition, track: Vec<String>) -> Position {
207        let (reply, answer) = oneshot::channel();
208
209        if self.sender.send(Request::Seal { stream, track, reply }).await.is_err() {
210            return Position::ORIGIN;
211        }
212
213        answer.await.unwrap_or(Position::ORIGIN)
214    }
215}
216
217/* === Public Functions === */
218
219/* === Private Functions === */
220
221fn earliest(held: Point, next: Point) -> Point {
222    if next.axis < held.axis { next } else { held }
223}
224
225fn latest(held: Point, next: Point) -> Point {
226    if next.axis >= held.axis { next } else { held }
227}
228
229fn lower(held: Point, next: Point) -> Point {
230    if next.value < held.value { next } else { held }
231}
232
233fn higher(held: Point, next: Point) -> Point {
234    if next.value > held.value { next } else { held }
235}
236
237/* === Tests === */
harmos_runtime/streams/series.rs

harmos_runtime/streams/
series.rs
