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

use mnemos_kernel::{
    comms::{
        bbq::{new_bidi_channel, BidiHandle},
        kchannel::KChannel,
    },
    drivers::simple_serial::{Request, Response, SimpleSerialError, SimpleSerialService},
    registry::Message,
    Kernel,
};
use std::{net::SocketAddr, sync::Arc};
use tokio::{
    io::{self, AsyncWriteExt},
    net::{TcpListener, TcpStream},
    sync::Notify,
};
use tracing::{info_span, trace, warn, Instrument};

pub struct TcpSerial {
    _inner: (),
}

impl TcpSerial {
    pub async fn register(
        kernel: &'static Kernel,
        ip: SocketAddr,
        incoming_size: usize,
        outgoing_size: usize,
        irq: Arc<Notify>,
    ) -> Result<(), ()> {
        let (a_ring, b_ring) = new_bidi_channel(kernel.heap(), incoming_size, outgoing_size).await;
        let (prod, cons) = KChannel::<Message<SimpleSerialService>>::new_async(kernel, 2)
            .await
            .split();

        let listener = TcpListener::bind(ip).await.unwrap();
        tracing::info!("TCP serial port driver listening on {ip}");

        kernel
            .spawn(async move {
                let handle = b_ring;

                // Reply to the first request, giving away the serial port
                let req = cons.dequeue_async().await.map_err(drop).unwrap();
                let Request::GetPort = req.msg.body;
                let resp = req.msg.reply_with(Ok(Response::PortHandle { handle }));

                req.reply.reply_konly(resp).await.map_err(drop).unwrap();

                // And deny all further requests after the first
                loop {
                    let req = cons.dequeue_async().await.map_err(drop).unwrap();
                    let Request::GetPort = req.msg.body;
                    let resp = req
                        .msg
                        .reply_with(Err(SimpleSerialError::AlreadyAssignedPort));
                    req.reply.reply_konly(resp).await.map_err(drop).unwrap();
                }
            })
            .await;

        let _ = tokio::spawn(
            async move {
                let mut handle = a_ring;
                loop {
                    match listener.accept().await {
                        Ok((stream, addr)) => {
                            irq.notify_one();
                            process_stream(&mut handle, stream, irq.clone())
                                .instrument(info_span!("process_stream", client.addr = %addr))
                                .await
                        }
                        Err(error) => {
                            warn!(%error, "Error accepting incoming TCP connection");
                            return;
                        }
                    };
                }
            }
            .instrument(info_span!("TCP Serial", ?ip)),
        );

        kernel
            .with_registry(|reg| reg.register_konly::<SimpleSerialService>(&prod))
            .await
            .map_err(drop)
    }
}

pub(crate) fn default_addr() -> SocketAddr {
    SocketAddr::from(([127, 0, 0, 1], 9999))
}

async fn process_stream(handle: &mut BidiHandle, mut stream: TcpStream, irq: Arc<Notify>) {
    loop {
        // Wait until either the socket has data to read, or the other end of
        // the BBQueue has data to write.
        tokio::select! {
            // The kernel wants to write something.
            outmsg = handle.consumer().read_grant() => {
                trace!(len = outmsg.len(), "Got outgoing message",);
                let wall = stream.write_all(&outmsg);
                wall.await.unwrap();
                let len = outmsg.len();
                outmsg.release(len);
            }
            // The socket has more bytes to read.
            _ = stream.readable() => {
                // Simulate an "interrupt", waking the kernel if it's waiting
                // an IRQ.
                irq.notify_one();
                let mut in_grant = handle.producer().send_grant_max(256).await;

                // Try to read data, this may still fail with `WouldBlock`
                // if the readiness event is a false positive.
                match stream.try_read(&mut in_grant) {
                    Ok(used) if used == 0 => {
                        warn!("Empty read, socket probably closed.");
                        return;
                    },
                    Ok(used) => {
                        trace!(len = used, "Got incoming message",);
                        in_grant.commit(used);
                    },
                    // WouldBlock here indicates that the `readable()` event was
                    // spurious. That's fine, just continue waiting for the
                    // sender to become ready or the socket to be readable again.
                    Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                        continue;
                    },
                    // Other errors indicate something is actually wrong.
                    Err(error) => {
                        warn!(%error, "Error reading from TCP stream");
                        return;
                    },
                }
            }
        }
    }
}