parallel-processor 0.2.3

Framework to manage asynchronous execution of multiple compute units communicating using messages
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
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

use std::{
    io::{Read, Write},
    marker::PhantomData,
};

use bincode::{Decode, Encode};

pub trait BucketItemSerializer {
    type InputElementType<'a>: ?Sized;
    type ExtraData;
    type ReadBuffer: Default;
    type ExtraDataBuffer: Default;
    type ReadType<'a>;
    type InitData;

    type CheckpointData: Encode + Decode<()> + 'static;

    /// Creates a new instance
    fn new(init_data: Self::InitData) -> Self;
    /// Reset on non continuous data
    fn reset(&mut self);

    fn write_to(
        &mut self,
        element: &Self::InputElementType<'_>,
        bucket: &mut Vec<u8>,
        extra_data: &Self::ExtraData,
        extra_read_buffer: &Self::ExtraDataBuffer,
    );
    fn read_from<'a, S: Read>(
        &mut self,
        stream: S,
        read_buffer: &'a mut Self::ReadBuffer,
        extra_read_buffer: &mut Self::ExtraDataBuffer,
    ) -> Option<Self::ReadType<'a>>;

    fn get_size(&self, element: &Self::InputElementType<'_>, extra: &Self::ExtraData) -> usize;
}

#[repr(transparent)]
pub struct BytesArrayBuffer<const SIZE: usize>([u8; SIZE]);

impl<const SIZE: usize> Default for BytesArrayBuffer<SIZE> {
    fn default() -> Self {
        Self([0; SIZE])
    }
}

pub struct BytesArraySerializer<const SIZE: usize>(PhantomData<[(); SIZE]>);
impl<const SIZE: usize> BucketItemSerializer for BytesArraySerializer<SIZE> {
    type InputElementType<'a> = [u8; SIZE];
    type ExtraData = ();
    type ExtraDataBuffer = ();
    type ReadBuffer = BytesArrayBuffer<SIZE>;
    type ReadType<'a> = &'a [u8; SIZE];
    type InitData = ();

    type CheckpointData = ();

    #[inline(always)]
    fn new(_: ()) -> Self {
        Self(PhantomData)
    }

    #[inline(always)]
    fn reset(&mut self) {}

    #[inline(always)]
    fn write_to(
        &mut self,
        element: &Self::InputElementType<'_>,
        bucket: &mut Vec<u8>,
        _: &Self::ExtraData,
        _: &Self::ExtraDataBuffer,
    ) {
        bucket.write(element).unwrap();
    }

    fn read_from<'a, S: Read>(
        &mut self,
        mut stream: S,
        read_buffer: &'a mut Self::ReadBuffer,
        _: &mut Self::ExtraDataBuffer,
    ) -> Option<Self::ReadType<'a>> {
        stream.read_exact(&mut read_buffer.0).ok()?;
        Some(&read_buffer.0)
    }

    #[inline(always)]
    fn get_size(&self, element: &Self::InputElementType<'_>, _: &()) -> usize {
        element.len()
    }
}

pub struct BytesSliceSerializer;
impl BucketItemSerializer for BytesSliceSerializer {
    type InputElementType<'a> = [u8];
    type ExtraData = ();
    type ExtraDataBuffer = ();
    type ReadBuffer = ();
    type ReadType<'a> = ();
    type InitData = ();

    type CheckpointData = ();

    #[inline(always)]
    fn new(_: ()) -> Self {
        Self
    }

    #[inline(always)]
    fn reset(&mut self) {}

    #[inline(always)]
    fn write_to(
        &mut self,
        element: &Self::InputElementType<'_>,
        bucket: &mut Vec<u8>,
        _extra_data: &Self::ExtraData,
        _: &Self::ExtraDataBuffer,
    ) {
        bucket.write(element).unwrap();
    }

    fn read_from<'a, S: Read>(
        &mut self,
        _stream: S,
        _read_buffer: &'a mut Self::ReadBuffer,
        _: &mut Self::ExtraDataBuffer,
    ) -> Option<Self::ReadType<'a>> {
        unimplemented!("Cannot read slices of unknown size!")
    }

    #[inline(always)]
    fn get_size(&self, element: &Self::InputElementType<'_>, _: &()) -> usize {
        element.len()
    }
}