links_blocking/core/
framer.rs

1//! This module contains a blocking `paired` [FrameReader] and [FrameWriter] which are designed to be used in separate threads.
2//! where each thread is only doing either reading or writing to the underlying [std::net::TcpStream].
3//!
4//! # Note
5//! The underlying [std::net::TcpStream] is cloned and therefore share a single underlying network socket.
6//!
7//! # Example
8//! ```
9//! use links_blocking::prelude::*;
10//!
11//! const FRAME_SIZE: usize = 128;
12//!
13//! let addr = "127.0.0.1:8080";
14//!
15//! let svc_listener = std::net::TcpListener::bind(addr).unwrap();
16//!
17//! let clt_stream = std::net::TcpStream::connect(addr).unwrap();
18//! let (clt_reader, clt_writer) = into_split_framer::<FixedSizeFramer<FRAME_SIZE>, FRAME_SIZE>(
19//!         ConId::clt(Some("unittest"), None, addr),
20//!         clt_stream,
21//!     );
22//!
23//! let svc_stream = svc_listener.accept().unwrap().0;
24//! let (svc_reader, svc_writer) = into_split_framer::<FixedSizeFramer<FRAME_SIZE>, FRAME_SIZE>(
25//!         ConId::svc(Some("unittest"), addr, None),
26//!         svc_stream,
27//!     );
28//!
29//! drop(clt_reader);
30//! drop(clt_writer);
31//! drop(svc_reader);
32//! drop(svc_writer);
33//! drop(svc_listener);
34//!
35//! // Note:
36//!     // paired
37//!         // clt_reader & clt_writer
38//!         // svc_reader & svc_writer
39//!     // peers
40//!         // clt_reader & svc_writer
41//!         // svc_reader & clt_writer
42//! ```
43
44use crate::prelude::{asserted_short_name, cross_os_fd, ConId, Framer};
45use bytes::{Bytes, BytesMut};
46use byteserde::utils::hex::to_hex_pretty;
47use log::{debug, log_enabled};
48use std::fmt::Display;
49use std::io::{ErrorKind, Read, Write};
50use std::mem::MaybeUninit;
51use std::net::Shutdown;
52use std::{io::Error, net::TcpStream};
53
54const EOF: usize = 0;
55
56/// Represents an abstraction for reading exactly oen frame from the [TcpStream].
57/// Each call to the [Self::read_frame] will issue a [Read::read] system call on the underlying [TcpStream].
58/// which will capture any bytes read into internal accumulator implemented as a [BytesMut]. This internal buffer will be
59/// passed to the generic impl of [Framer::get_frame] where it is user's responsibility to  inspect the buffer and split off a single frame.
60///
61/// # Generic Parameters
62///  * `F` - A type that implements the [Framer] trait. This is used to split off a single frame from the internal buffer.
63///  * `MAX_MSG_SIZE` - The maximum size of a single frame. This is used to pre-allocate the internal buffer.
64/// Set this number to the maximum size of a single frame for your protocol.
65///  
66#[derive(Debug)]
67pub struct FrameReader<F: Framer, const MAX_MSG_SIZE: usize> {
68    pub(crate) con_id: ConId,
69    pub(crate) stream_reader: TcpStream,
70    buffer: BytesMut,
71    phantom: std::marker::PhantomData<F>,
72}
73impl<F: Framer, const MAX_MSG_SIZE: usize> FrameReader<F, MAX_MSG_SIZE> {
74    /// Creates a new instance of [FrameReader]
75    /// # Arguments
76    /// * `con_id` - [ConId] a unique identifier for the connection and used for logging
77    /// * `reader` - [TcpStream] the underlying stream that will be used for reading
78    pub fn new(con_id: ConId, reader: TcpStream) -> FrameReader<F, MAX_MSG_SIZE> {
79        Self {
80            con_id,
81            stream_reader: reader,
82            buffer: BytesMut::with_capacity(MAX_MSG_SIZE),
83            phantom: std::marker::PhantomData,
84        }
85    }
86
87    /// Reads `exactly one frame` from the underlying [TcpStream] and returns it as a [Some(Bytes)] or [None] if the connection was closed.
88    ///
89    /// # Note
90    /// If the [FrameWriter] `pair` is dropped then this method will return a [Ok(None)].
91    #[inline]
92    pub fn read_frame(&mut self) -> Result<Option<Bytes>, Error> {
93        loop {
94            if let Some(bytes) = F::get_frame(&mut self.buffer) {
95                return Ok(Some(bytes));
96            } else {
97                #[allow(clippy::uninit_assumed_init)]
98                let mut buf: [u8; MAX_MSG_SIZE] = unsafe { MaybeUninit::uninit().assume_init() };
99                match self.stream_reader.read(&mut buf) {
100                    Ok(EOF) => {
101                        self.shutdown(Shutdown::Write, "read_frame EOF");
102                        if self.buffer.is_empty() {
103                            return Ok(None);
104                        } else {
105                            let msg = format!(
106                                "{} {}::read_frame connection reset by peer, residual buf:\n{}",
107                                self.con_id,
108                                asserted_short_name!("FrameReader", Self),
109                                to_hex_pretty(&self.buffer[..])
110                            );
111                            return Err(Error::new(std::io::ErrorKind::ConnectionReset, msg));
112                        }
113                    }
114                    Ok(len) => {
115                        self.buffer.extend_from_slice(&buf[..len]);
116                        continue; // more bytes added, try to get a frame again
117                    }
118                    Err(e) => {
119                        self.shutdown(Shutdown::Write, "read_frame error");
120                        let msg = format!("{} {}::read_frame caused by: [{}] residual buf:\n{}", self.con_id, asserted_short_name!("FrameReader", Self), e, to_hex_pretty(&self.buffer[..]));
121                        return Err(Error::new(e.kind(), msg));
122                    }
123                }
124            }
125        }
126    }
127    #[inline]
128    fn shutdown(&mut self, how: Shutdown, reason: &str) {
129        match self.stream_reader.shutdown(how) {
130            Ok(_) => {
131                if log_enabled!(log::Level::Debug) {
132                    debug!("{}::shutdown how: {:?}, reason: {}", self, how, reason);
133                }
134            }
135            Err(e) if e.kind() == ErrorKind::NotConnected => {
136                if log_enabled!(log::Level::Debug) {
137                    debug!("{}::shutdown while disconnected how: {:?}, reason: {}", self, how, reason);
138                }
139            }
140            Err(e) => {
141                panic!("{}::shutdown how: {:?}, reason: {}, caused by: [{}]", self, how, reason, e);
142            }
143        }
144    }
145}
146impl<F: Framer, const MAX_MSG_SIZE: usize> Drop for FrameReader<F, MAX_MSG_SIZE> {
147    /// Will shutdown the underlying [std::net::TcpStream] in both directions. This way
148    /// the `peer` connection will receive a TCP FIN flag and and once it reaches the `peer` [FrameWriter] it will
149    /// get a [ErrorKind::BrokenPipe] error which in turn shall issue a [Shutdown::Write]
150    fn drop(&mut self) {
151        self.shutdown(Shutdown::Both, "drop")
152    }
153}
154impl<F: Framer, const MAX_MSG_SIZE: usize> Display for FrameReader<F, MAX_MSG_SIZE> {
155    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
156        write!(
157            f,
158            "FrameReader<{}> {{ {}, addr: {}, peer: {}, fd: {} }}",
159            std::any::type_name::<F>().split("::").last().unwrap_or("Unknown"),
160            self.con_id,
161            match self.stream_reader.local_addr() {
162                Ok(_) => "connected",
163                Err(_) => "disconnected",
164            },
165            match self.stream_reader.peer_addr() {
166                Ok(_) => "connected",
167                Err(_) => "disconnected",
168            },
169            cross_os_fd!(&self.stream_reader),
170        )
171    }
172}
173
174/// Represents an abstraction for writing a single frame to the underlying [TcpStream].
175#[derive(Debug)]
176pub struct FrameWriter {
177    pub(crate) con_id: ConId,
178    pub(crate) stream_writer: TcpStream,
179}
180impl FrameWriter {
181    /// Creates a new instance of [FrameWriter]
182    /// # Arguments
183    /// * `con_id` - [ConId] a unique identifier for the connection and used for logging
184    /// * `stream` - [TcpStream] the underlying stream that will be used for writing
185    pub fn new(con_id: ConId, stream: TcpStream) -> Self {
186        Self { con_id, stream_writer: stream }
187    }
188    /// Writes a single frame to the underlying [TcpStream].
189    ///
190    /// # Note
191    /// If the [FrameReader] `pair` is dropped then this method will return a [Err(ErrorKind::BrokenPipe)] error.
192    #[inline]
193    pub fn write_frame(&mut self, bytes: &[u8]) -> Result<(), Error> {
194        match self.stream_writer.write_all(bytes) {
195            Ok(_) => Ok(()),
196            Err(e) => {
197                self.shutdown(Shutdown::Write, "write_frame error");
198                let msg = format!("{} FrameWriter::write_frame caused by: [{}]", self.con_id, e);
199                Err(Error::new(e.kind(), msg))
200            }
201        }
202    }
203
204    /// Shuts down the underlying [std::net::TcpStream] in the specified direction.
205    fn shutdown(&mut self, how: Shutdown, reason: &str) {
206        match self.stream_writer.shutdown(how) {
207            Ok(_) => {
208                if log_enabled!(log::Level::Debug) {
209                    debug!("{}::shutdown how: {:?}, reason: {}", self, how, reason);
210                }
211            }
212            Err(e) if e.kind() == ErrorKind::NotConnected => {
213                if log_enabled!(log::Level::Debug) {
214                    debug!("{}::shutdown while disconnected how: {:?}, reason: {}", self, how, reason);
215                }
216            }
217            Err(e) => {
218                panic!("{}::shutdown how: {:?}, reason: {}, caused by: [{}]", self, how, reason, e);
219            }
220        }
221    }
222}
223impl Drop for FrameWriter {
224    /// Will shutdown the underlying [std::net::TcpStream] in both directions.
225    fn drop(&mut self) {
226        self.shutdown(Shutdown::Both, "drop")
227    }
228}
229impl Display for FrameWriter {
230    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
231        write!(
232            f,
233            "FrameWriter {{ {}, addr: {}, peer: {}, fd: {} }}",
234            self.con_id,
235            match self.stream_writer.local_addr() {
236                Ok(_) => "connected",
237                Err(_) => "disconnected",
238            },
239            match self.stream_writer.peer_addr() {
240                Ok(_) => "connected",
241                Err(_) => "disconnected",
242            },
243            cross_os_fd!(&self.stream_writer),
244        )
245    }
246}
247
248type FrameProcessor<F, const MAX_MSG_SIZE: usize> = (FrameReader<F, MAX_MSG_SIZE>, FrameWriter);
249
250/// Creates a `paired` [FrameReader] and [FrameWriter] from a [std::net::TcpStream] by cloning
251///
252/// # Returns a tuple with
253///   * [FrameReader] - a blocking FrameReader
254///   * [FrameWriter] - a blocking FrameWriter
255///
256/// # Important
257/// If either the [FrameReader] or [FrameWriter] are dropped the underlying stream will be shutdown and all actions on the remaining `pair` will fail
258pub fn into_split_framer<F: Framer, const MAX_MSG_SIZE: usize>(mut con_id: ConId, stream: TcpStream) -> FrameProcessor<F, MAX_MSG_SIZE> {
259    con_id.set_local(stream.local_addr().unwrap());
260    con_id.set_peer(stream.peer_addr().unwrap());
261    let (reader, writer) = (stream.try_clone().expect("Failed to try_clone TcpStream for FrameReader"), stream);
262    (FrameReader::<F, MAX_MSG_SIZE>::new(con_id.clone(), reader), FrameWriter::new(con_id, writer))
263}
264
265#[cfg(test)]
266mod test {
267
268    use crate::prelude::*;
269    use byteserde::utils::hex::to_hex_pretty;
270    use links_core::{assert_error_kind_on_target_family, fmt_num, prelude::ConId, unittest::setup};
271    use log::{error, info};
272    use rand::Rng;
273    use std::{
274        net::{TcpListener, TcpStream},
275        thread::{self, sleep},
276        time::{Duration, Instant},
277    };
278
279    /// # High Level Approach
280    /// 1. Spawn Svc FrameReader in a separate thread
281    ///     1. accept connection that will be split into reader & writer
282    ///     2. only use reader and read until None or Err
283    ///     3. return frame_recv_count upon completion
284    /// 2. Create Clt FrameWriter in main thread
285    ///     1. the connection will be split into reader & writer
286    ///     2. only use writer to write N frames
287    ///     3. randomly drop either reader or writer as join FrameReader thread which should successfully exist in either case
288    ///     4. ensure number of frames sent by FrameWriter equals number of frames received by FrameReader
289    /// # Notes
290    /// * turn on LevelFilter::Debug for additional logging, it will show drop events
291    #[test]
292    fn test_reader() {
293        setup::log::configure_level(log::LevelFilter::Info);
294        const TEST_SEND_FRAME_SIZE: usize = 128;
295        const WRITE_N_TIMES: usize = 100_000;
296        pub type MsgFramer = FixedSizeFramer<TEST_SEND_FRAME_SIZE>;
297
298        let send_frame = setup::data::random_bytes(TEST_SEND_FRAME_SIZE);
299        info!("send_frame: \n{}", to_hex_pretty(send_frame));
300
301        let addr = setup::net::rand_avail_addr_port();
302
303        // CONFIGURE svc
304        let svc = thread::Builder::new()
305            .name("Thread-Svc".to_owned())
306            .spawn({
307                move || {
308                    let listener = TcpListener::bind(addr).unwrap();
309                    let (stream, _) = listener.accept().unwrap();
310                    let (mut svc_reader, _svc_writer) = into_split_framer::<MsgFramer, TEST_SEND_FRAME_SIZE>(ConId::svc(Some("unittest"), addr, None), stream);
311                    info!("svc: reader: {}", svc_reader);
312                    let mut frame_recv_count = 0_usize;
313                    loop {
314                        let res = svc_reader.read_frame();
315                        match res {
316                            Ok(frame) => {
317                                if let None = frame {
318                                    info!("svc: read_frame is None, client closed connection");
319                                    break;
320                                } else {
321                                    frame_recv_count += 1;
322                                }
323                            }
324                            Err(e) => {
325                                error!("Svc read_frame error: {}", e.to_string());
326                                break;
327                            }
328                        }
329                    }
330                    frame_recv_count
331                }
332            })
333            .unwrap();
334
335        sleep(Duration::from_millis(100)); // allow the spawned to bind
336                                           // CONFIGURE clt
337        let (mut clt_reader, mut clt_writer) = into_split_framer::<MsgFramer, TEST_SEND_FRAME_SIZE>(ConId::clt(Some("unittest"), None, addr), TcpStream::connect(addr).unwrap());
338
339        info!("clt: {}", clt_writer);
340
341        let mut frame_send_count = 0_usize;
342        let start = Instant::now();
343        for _ in 0..WRITE_N_TIMES {
344            clt_writer.write_frame(send_frame).unwrap();
345            frame_send_count += 1;
346        }
347        let elapsed = start.elapsed();
348
349        // drop either clt_reader or clt_writer and validate that the `pair` is acting correction
350        if rand::thread_rng().gen_range(1..=2) % 2 == 0 {
351            info!("dropping clt_writer");
352            drop(clt_writer);
353            let opt = clt_reader.read_frame().unwrap();
354            info!("clt_reader.read_frame() opt: {:?}", opt);
355            assert_eq!(opt, None);
356        } else {
357            info!("dropping clt_reader");
358            drop(clt_reader);
359            let err = clt_writer.write_frame(send_frame).unwrap_err();
360            info!("clt_writer.write_frame() err: {}", err);
361            assert_error_kind_on_target_family!(err, std::io::ErrorKind::BrokenPipe);
362        }
363        let frame_recv_count = svc.join().unwrap();
364
365        info!("frame_send_count: {}, frame_recv_count: {}", fmt_num!(frame_send_count), fmt_num!(frame_recv_count));
366        info!("per send elapsed: {:?}, total elapsed: {:?} ", elapsed / WRITE_N_TIMES as u32, elapsed);
367        assert_eq!(frame_send_count, frame_recv_count);
368        assert_eq!(frame_send_count, WRITE_N_TIMES);
369    }
370}
links_blocking/core/framer.rs

links_blocking/core/
framer.rs
