rstango 0.4.0

Implementation of the Tango distributed control system
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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250

//! IO extensions for the client and server modules.

use std::io::{self, Read, Write};

use crate::buf::Buf;

use super::cdr::CdrSender;


/// The least a read is ever offered room for.
///
/// Small enough that a buffer of it is nothing to hold for the length of a request, and
/// large enough that the messages of this protocol arrive in a single read: a Tango request
/// is a header, an object key and an operation name, which is a few hundred octets, and so
/// is the reply of most single-attribute operations. Past that it is the header of the
/// message itself that says how much to make room for, and this stops mattering, see
/// [`ReadExt::read_nb_to_cdr`].
const RECV_MIN_LEN: usize = 512;
/// How far the origin of a sender is let run before the octets that have already gone out
/// are moved off the front of it: up to here the room they hold is cheaper than the copy
/// that reclaims it, see [`CdrSender::drop_off_origin`].
const SENDER_DROP_OF_THRESHOLD: usize = 1 * 1024;


pub trait ReadExt: Read {

    /// Read everything this reader has to give into the octets of a buffer, behind whatever
    /// they already hold, and say how much arrived and whether the peer is through sending.
    ///
    /// What a read is offered is whatever the buffer holds room for, never less, so a
    /// caller that has read a header announcing a message of some length and has made room
    /// for the whole of it has the whole of it in a single read. Making that room is the
    /// caller's to do and it lasts, an allocation only ever growing, see
    /// [`crate::buf::BufMut::try_reserve`]. Short of that the room is [`RECV_MIN_LEN`],
    /// which is what a buffer holding nothing but the front of a message gets.
    ///
    /// The buffer is written through copy-on-write, see [`Buf::make_mut`]: when its
    /// storage is still shared with the messages split out of it, only its own octets are
    /// copied, which are only ever the front of the messages behind them.
    ///
    /// There is no reading part of what has arrived: the socket behind this reader is
    /// watched edge-triggered, so it announces octets arriving and not octets waiting, and
    /// what is left on it here is announced no further. It is read until it says it has
    /// nothing, whether that comes to less than a message or to several of them.
    ///
    /// The end of the stream is not the end of what arrived: a peer that writes its last
    /// message and closes right behind it is one whose octets turn up in the very read that
    /// finds that end. So the two are given back together, and it is for the caller to take
    /// what came before it gives the connection up.
    fn read_nb_to_cdr(&mut self, buf: &mut Buf<u8, 1>) -> io::Result<(usize, bool)> {

        let buffer = buf.make_mut();
        let mut total_len = 0;

        loop {

            let at = buffer.len();

            if buffer.capacity() < RECV_MIN_LEN {
                buffer.reserve(RECV_MIN_LEN);
            } else {
                buffer.reserve(1);  
            }

            let capacity = buffer.capacity();
            buffer.set_len(capacity);

            let read = match self.read(&mut buffer.as_mut_slice()[at..]) {
                // A reader only ever says this for a buffer it was given room in, and this
                // one always has room left, so it is the peer having hung up. What arrived
                // before it stays in the buffer and is counted here all the same.
                Ok(0) => {
                    buffer.set_len(at);
                    return Ok((total_len, true));
                }
                Ok(read) => read,
                // Nothing was read and nothing went wrong, so the very same room is
                // offered again rather than taken for a read that came up short.
                Err(err) if err.kind() == io::ErrorKind::Interrupted => {
                    buffer.set_len(at);
                    continue;
                }
                Err(err) if err.kind() == io::ErrorKind::WouldBlock => {
                    buffer.set_len(at);
                    return Ok((total_len, false));
                }
                Err(err) => {
                    buffer.set_len(at);
                    return Err(err);
                }
            };

            total_len += read;

            // Back down to what really arrived, and not the room it was read into.
            //
            // And then round again, whatever has arrived of the message and whatever the
            // read came back with: the socket is watched edge-triggered, so it announces
            // octets arriving and not octets waiting, and anything left behind here is
            // left until the peer happens to send more. Only the reader saying it has
            // nothing left ends this.
            buffer.set_len(at + read);

        }

    }

}

pub trait WriteExt: Write {

    /// Write CDR data to this writer. This takes a [`CdrWriter`](super::cdr::CdrWriter) because it allows 
    /// changing the origin of the write without moving all the buffer to replace the
    /// part that was just written.
    /// 
    /// This is especially useful since this implementation supports non-blocking!
    /// 
    /// For example, imagine if we have filled 4 kiB of data, and the writer is just able
    /// to swallow 2 bytes, we don't want to memcpy the 4kiB-2B of data upfront. In this
    /// case we just move the origin to '2'.
    /// 
    /// We'll still need to memcpy at some point, but only after an internally defined
    /// limit for the origin!
    /// 
    /// If an error happens, it is undertermined what has been actually written!
    fn write_nb_from_cdr(&mut self, sender: &mut CdrSender) -> io::Result<()> {

        loop {
            
            let buffer = sender.buffer();
            if buffer.is_empty() {
                break;
            }

            match self.write(buffer) {
                // A socket that takes nothing at all is one there is nothing more to be done
                // with until it says otherwise.
                Ok(0) => break,
                Ok(len) => sender.ack(len),
                Err(err) if err.kind() == io::ErrorKind::Interrupted => continue,
                Err(err) if err.kind() == io::ErrorKind::WouldBlock => break,
                Err(err) => return Err(err),
            }

        }

        if sender.origin() >= SENDER_DROP_OF_THRESHOLD {
            sender.drop_off_origin();
        }

        Ok(())

    }

}

impl<R: Read> ReadExt for R {}
impl<W: Write> WriteExt for W {}

#[cfg(test)]
mod tests {

    use super::*;

    /// A reader with a given number of octets to give, which hands out as many of them as
    /// it is offered room for, and then either says it is through or that it has nothing
    /// more for now: the two ways a socket ends a turn of the loop.
    struct Stream {
        left: usize,
        eof: bool,
        reads: usize,
    }

    impl Read for Stream {

        fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {

            self.reads += 1;

            if self.left == 0 {
                return match self.eof {
                    true => Ok(0),
                    false => Err(io::Error::from(io::ErrorKind::WouldBlock)),
                };
            }

            let len = usize::min(self.left, buf.len());
            buf[..len].fill(0xAA);
            self.left -= len;

            Ok(len)

        }

    }

    /// The end of the stream is given back alongside what arrived in the very read that
    /// found it, and never instead of it: a peer that writes its last message and closes
    /// right behind it is one whose octets are still taken.
    #[test]
    fn read_eof() {

        let mut buffer = Buf::from(&[0xCC; 4][..]);
        let mut reader = Stream { left: 8, eof: true, reads: 0 };

        let (len, eof) = reader.read_nb_to_cdr(&mut buffer).unwrap();

        assert_eq!(len, 8);
        assert!(eof);
        assert_eq!(buffer.as_slice(), [&[0xCC; 4][..], &[0xAA; 8]].concat());

    }

    /// And a reader with nothing more for now is not one that is through.
    #[test]
    fn read_would_block() {

        let mut buffer = Buf::new();
        let mut reader = Stream { left: 0, eof: false, reads: 0 };

        let (len, eof) = reader.read_nb_to_cdr(&mut buffer).unwrap();

        assert_eq!(len, 0);
        assert!(!eof);
        assert!(buffer.is_empty());

    }

    /// What a read is offered is the room that is already there, so a caller that has made
    /// room for the whole of a message it knows the length of takes it in a single read,
    /// and not in the run of them that growing to it a floor at a time would be.
    #[test]
    fn read_reserved() {

        let mut buffer = Buf::new();
        let mut reader = Stream { left: 100_000, eof: false, reads: 0 };

        buffer.make_mut().try_reserve(100_000).unwrap();
        let (len, eof) = reader.read_nb_to_cdr(&mut buffer).unwrap();

        assert_eq!(len, 100_000);
        assert!(!eof);
        assert_eq!(buffer.len(), 100_000);

        // The one that took it, and the one that found nothing behind it.
        assert_eq!(reader.reads, 2);

    }

}