spill 0.5.0

A single-user BitTorrent daemon for Linux and macOS
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

use std::{
    collections::HashMap,
    io,
    net::SocketAddr,
    sync::{Arc, Mutex},
};
use tokio::{
    net::UdpSocket,
    sync::{mpsc, oneshot, watch},
};

#[derive(Clone)]
pub(in crate::engine) struct Transport {
    pub(in crate::engine) socket: Arc<UdpSocket>,
    pub(super) routes: Arc<Mutex<Routes>>,
    pub(super) utp: Option<Arc<Mutex<super::utp::Mux>>>,
}
#[derive(Default)]
pub(super) struct Routes {
    waiting: HashMap<(u32, SocketAddr), Waiting>,
    pub(super) dht: Option<mpsc::Sender<(Vec<u8>, SocketAddr)>>,
}
struct Waiting {
    token: Arc<()>,
    sender: oneshot::Sender<Vec<u8>>,
}
pub(in crate::engine) struct Registration {
    transport: Transport,
    key: (u32, SocketAddr),
    token: Arc<()>,
}
impl Drop for Registration {
    fn drop(&mut self) {
        let mut routes = self.transport.routes.lock().unwrap();
        let waiting = &mut routes.waiting;
        // A delivered transaction's ID may already have been reused.
        if waiting
            .get(&self.key)
            .is_some_and(|entry| Arc::ptr_eq(&entry.token, &self.token))
        {
            waiting.remove(&self.key);
        }
    }
}
impl Transport {
    pub(super) fn bind(address: SocketAddr) -> io::Result<Self> {
        let socket = socket2::Socket::new(
            socket2::Domain::for_address(address),
            socket2::Type::DGRAM,
            Some(socket2::Protocol::UDP),
        )?;
        if address.is_ipv6() {
            socket.set_only_v6(!address.ip().is_unspecified())?;
        }
        // The default kernel buffer can be smaller than one uTP receive window
        // once per-datagram kernel overhead is included.
        socket.set_recv_buffer_size(2 * 1024 * 1024)?;
        socket.set_nonblocking(true)?;
        socket.bind(&address.into())?;
        Ok(Self {
            socket: Arc::new(UdpSocket::from_std(socket.into())?),
            routes: Arc::default(),
            utp: None,
        })
    }
    pub(in crate::engine) fn register(
        &self,
        endpoint: SocketAddr,
    ) -> io::Result<(u32, Registration, oneshot::Receiver<Vec<u8>>)> {
        let mut routes = self.routes.lock().unwrap();
        let waiting = &mut routes.waiting;
        loop {
            let mut bytes = [0; 4];
            getrandom::getrandom(&mut bytes).map_err(io::Error::other)?;
            let id = u32::from_ne_bytes(bytes);
            let key = (id, endpoint);
            if let std::collections::hash_map::Entry::Vacant(entry) = waiting.entry(key) {
                let (sender, receiver) = oneshot::channel();
                let token = Arc::new(());
                entry.insert(Waiting {
                    token: token.clone(),
                    sender,
                });
                return Ok((
                    id,
                    Registration {
                        transport: self.clone(),
                        key,
                        token,
                    },
                    receiver,
                ));
            }
        }
    }
    fn route(&self, bytes: &[u8], source: SocketAddr) {
        let source = SocketAddr::new(source.ip().to_canonical(), source.port());
        if bytes.first() == Some(&b'd') {
            if bytes.len() <= crate::engine::dht::DATAGRAM
                && let Some(sender) = self.routes.lock().unwrap().dht.as_ref()
            {
                let _ = sender.try_send((bytes.to_vec(), source));
            }
        } else if bytes.len() >= 20 && bytes[0] & 0x0f == 1 && bytes[0] >> 4 <= 4 {
            if let Some(mux) = &self.utp {
                super::utp::Mux::receive(mux, bytes, source, std::time::Instant::now());
            }
        } else if bytes.len() >= 8 {
            let id = u32::from_be_bytes(bytes[4..8].try_into().unwrap());
            if let Some(entry) = self.routes.lock().unwrap().waiting.remove(&(id, source)) {
                let _ = entry.sender.send(bytes.to_vec());
            }
        }
    }
    pub(super) async fn run(self, mut stop: watch::Receiver<bool>) -> io::Result<()> {
        let mut bytes = vec![0; 65_536];
        loop {
            tokio::select! {
                biased;
                _ = stop.changed() => return Ok(()),
                received = self.socket.recv_from(&mut bytes) => {
                    match received {
                        Ok((length, source)) => self.route(&bytes[..length], source),
                        Err(error) => tracing::warn!(%error, "session UDP receive failed"),
                    }
                }
            }
        }
    }
}