qinterface 0.4.0

gm-quic's packet routing and definition of the underlying IO interface
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

use std::{
    fmt::Debug,
    future::Future,
    io,
    ops::DerefMut,
    pin::Pin,
    sync::{Arc, Weak},
    task::{Context, Poll},
};

use tokio_util::task::AbortOnDropHandle;

use crate::{
    QuicIO, QuicIoExt,
    iface::{RwInterface, physical::InterfaceEventReceiver},
    route::Router,
};

pub struct InterfaceContext {
    iface: Weak<RwInterface>,
    _task: AbortOnDropHandle<()>,
}

impl Debug for InterfaceContext {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("InterfaceContext")
            .field("iface", &"..")
            .field("task", &"..")
            .finish()
    }
}

impl InterfaceContext {
    pub fn new(rw_iface: Arc<RwInterface>, mut events: InterfaceEventReceiver) -> Self {
        let bind_uri = rw_iface.bind_uri();
        let device = bind_uri
            .as_iface_bind_uri()
            .map(|(_, device, _)| device.to_owned());
        let iface = Arc::downgrade(&rw_iface);
        let task = AbortOnDropHandle::new(tokio::spawn({
            let rw_iface = iface.clone();
            let mut receive_task =
                ReceiveTask::Running(Box::pin(receive_and_deliver(rw_iface.clone())));
            async move {
                loop {
                    tokio::select! {
                        Some(event) = events.recv() => {
                            // skip events not related to this interface
                            if Some(event.device()) != device.as_deref() {
                                continue;
                            }
                            let Some(rw_iface) = rw_iface.upgrade() else { break };
                            // If the task is stopped, or the interface is not alive: rebind it, and restart receive task
                            if matches!(receive_task, ReceiveTask::Stopped)
                                || rw_iface.is_alive().await.is_err_and(|e| {
                                    tracing::debug!(target: "interface", %bind_uri, "Interface may not alive: {e}");
                                    e.is_recoverable()
                                })
                            {
                                tracing::debug!(target: "interface", %bind_uri, "Rebinding interface");
                                _ = rw_iface.close().await;
                                rw_iface.rebind();
                                receive_task =
                                    ReceiveTask::Running(Box::pin(receive_and_deliver(Arc::downgrade(&rw_iface))));
                            }
                        }
                        result = &mut receive_task => {
                            match result {
                                Ok(()) => tracing::debug!(target: "interface", %bind_uri, "Receive task completed due to interface freed"),
                                Err(e) => tracing::debug!(target: "interface", %bind_uri, "Receive task failed with error: {e}"),
                            }
                            // Task completed (likely due to error), mark as stopped and wait for interface change
                            receive_task = ReceiveTask::Stopped;
                        }
                    }
                }
            }
        }));

        Self { iface, _task: task }
    }

    pub fn iface(&self) -> &Weak<RwInterface> {
        &self.iface
    }
}

enum ReceiveTask<F> {
    Running(F),
    Stopped,
}

impl<F: Future + Unpin> Future for ReceiveTask<F> {
    type Output = F::Output;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        if let ReceiveTask::Running(future) = self.deref_mut() {
            return Pin::new(future).poll(cx);
        }

        Poll::Pending
    }
}

async fn receive_and_deliver(iface: Weak<RwInterface>) -> io::Result<()> {
    let (mut bufs, mut hdrs) = (vec![], vec![]);
    loop {
        let pkts = match iface.upgrade() {
            Some(iface) => iface.recvmpkt(bufs.as_mut(), hdrs.as_mut()).await?,
            None => return Ok(()),
        };
        for (pkt, way) in pkts {
            Router::global().deliver(pkt, way).await;
        }
    }
}