alumet 0.8.0

Modular framework for hardware and software measurement (including energy consumption and more).
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
102
103
104
105
106
107
108
109
110
111

//! Abstractions over different kinds of channel.

use futures::Stream;
use tokio::sync::{broadcast, mpsc};

use crate::measurement::MeasurementBuffer;

/// Trait that allows to receive measurements from different kinds of channel.
pub trait MeasurementReceiver {
    async fn recv(&mut self) -> Result<MeasurementBuffer, RecvError>;
    fn discard_pending(self) -> Self;
    fn into_stream(self) -> impl Stream<Item = Result<MeasurementBuffer, StreamRecvError>>;
}

pub enum ReceiverEnum {
    Broadcast(broadcast::Receiver<MeasurementBuffer>),
    Single(mpsc::Receiver<MeasurementBuffer>),
}

pub struct ReceiverProvider(ProviderEnum);

enum ProviderEnum {
    Broadcast(broadcast::Sender<MeasurementBuffer>),
    Single(Option<mpsc::Receiver<MeasurementBuffer>>),
}

// common error enum

pub enum RecvError {
    Lagged(u64),
    Closed,
}

#[non_exhaustive]
#[derive(Debug)]
pub enum StreamRecvError {
    Lagged(u64),
}

// receiver implementations

impl MeasurementReceiver for broadcast::Receiver<MeasurementBuffer> {
    async fn recv(&mut self) -> Result<MeasurementBuffer, RecvError> {
        broadcast::Receiver::recv(self).await.map_err(|e| match e {
            broadcast::error::RecvError::Closed => RecvError::Closed,
            broadcast::error::RecvError::Lagged(n) => RecvError::Lagged(n),
        })
    }

    fn discard_pending(self) -> Self {
        self.resubscribe()
    }

    fn into_stream(self) -> impl Stream<Item = Result<MeasurementBuffer, StreamRecvError>> {
        use tokio_stream::wrappers::{errors::BroadcastStreamRecvError, BroadcastStream};
        use tokio_stream::StreamExt;

        BroadcastStream::new(self).map(|item| {
            item.map_err(|e| match e {
                BroadcastStreamRecvError::Lagged(n) => StreamRecvError::Lagged(n),
            })
        })
    }
}

impl MeasurementReceiver for mpsc::Receiver<MeasurementBuffer> {
    async fn recv(&mut self) -> Result<MeasurementBuffer, RecvError> {
        match mpsc::Receiver::recv(self).await {
            Some(buf) => Ok(buf),
            None => Err(RecvError::Closed),
        }
    }

    fn discard_pending(mut self) -> Self {
        while let Ok(_) = self.try_recv() {
            ()
        }
        self
    }

    fn into_stream(self) -> impl Stream<Item = Result<MeasurementBuffer, StreamRecvError>> {
        use tokio_stream::{wrappers::ReceiverStream, StreamExt};
        ReceiverStream::new(self).map(Ok)
    }
}

// providers

impl ReceiverProvider {
    pub fn get(&mut self) -> ReceiverEnum {
        match &mut self.0 {
            ProviderEnum::Broadcast(tx) => ReceiverEnum::Broadcast(tx.subscribe()),
            ProviderEnum::Single(rx) => ReceiverEnum::Single(
                rx.take()
                    .expect("ProviderEnum::get called but the single MeasurementReceiver has already been taken"),
            ),
        }
    }
}

impl From<broadcast::Sender<MeasurementBuffer>> for ReceiverProvider {
    fn from(value: broadcast::Sender<MeasurementBuffer>) -> Self {
        Self(ProviderEnum::Broadcast(value))
    }
}

impl From<mpsc::Receiver<MeasurementBuffer>> for ReceiverProvider {
    fn from(value: mpsc::Receiver<MeasurementBuffer>) -> Self {
        Self(ProviderEnum::Single(Some(value)))
    }
}