mtorrent 0.3.9

Fast and lightweight BitTorrent client in Rust
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

use super::super::PeerReporter;
use local_async_utils::prelude::*;
use mtorrent_core::pwp;
use mtorrent_utils::peer_id::PeerId;
use std::io;
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
use tokio::net::{TcpSocket, TcpStream};
use tokio::runtime;
use tokio_util::sync::CancellationToken;

/// Re-register socket so that it will be polled on PWP runtime
macro_rules! marshal_stream {
    ($stream:expr, $rt_handle:expr) => {{
        let std_stream = $stream.into_std()?;
        // note: EnterGuard must NEVER live across a suspension point
        let _g = $rt_handle.enter();
        TcpStream::from_std(std_stream)?
    }};
}

fn bound_pwp_socket(local_addr: SocketAddr) -> io::Result<TcpSocket> {
    let socket = match local_addr {
        SocketAddr::V4(_) => TcpSocket::new_v4()?,
        SocketAddr::V6(_) => TcpSocket::new_v6()?,
    };

    // To use the same local addr and port for outgoing PWP connections and for TCP listener,
    // (in order to deal with endpoint-independent NAT mappings, https://www.rfc-editor.org/rfc/rfc5128#section-2.3)
    // we need to set SO_REUSEADDR on Windows, and SO_REUSEADDR and SO_REUSEPORT on Linux.
    // See https://stackoverflow.com/a/14388707/4432988 for details.
    socket.set_reuseaddr(true)?;
    #[cfg(not(windows))]
    socket.set_reuseport(true)?;
    // To avoid putting socket into TIME_WAIT when disconnecting someone, enable SO_LINGER with 0 timeout
    // See https://stackoverflow.com/a/71975993
    #[expect(deprecated)]
    socket.set_linger(Some(sec!(0)))?;
    socket.set_nodelay(true)?;
    socket.bind(local_addr)?;
    Ok(socket)
}

pub async fn new_outbound_connection(
    local_peer_id: &PeerId,
    info_hash: &[u8; 20],
    extension_protocol_enabled: bool,
    peer_addr: SocketAddr,
    local_port: u16,
    pwp_runtime: &runtime::Handle,
) -> io::Result<(pwp::DownloadChannels, pwp::UploadChannels, Option<pwp::ExtendedChannels>)> {
    let local_addr = match &peer_addr {
        SocketAddr::V4(_) => Ipv4Addr::UNSPECIFIED.into(),
        SocketAddr::V6(_) => Ipv6Addr::UNSPECIFIED.into(),
    };

    let socket = bound_pwp_socket(SocketAddr::new(local_addr, local_port))?;
    let stream = socket.connect(peer_addr).await?;
    let stream = marshal_stream!(stream, pwp_runtime);

    let (download_chans, upload_chans, extended_chans, runner) =
        pwp::channels_for_outbound_connection(
            local_peer_id,
            info_hash,
            extension_protocol_enabled,
            peer_addr,
            stream,
            None,
        )
        .await?;

    pwp_runtime.spawn(async move {
        if let Err(e) = runner.await
            && e.kind() != io::ErrorKind::BrokenPipe
            && e.kind() != io::ErrorKind::UnexpectedEof
        {
            log::warn!("Peer runner for {peer_addr} exited: {e}");
        }
    });

    Ok((download_chans, upload_chans, extended_chans))
}

pub async fn new_inbound_connection(
    local_peer_id: &[u8; 20],
    info_hash: &[u8; 20],
    extension_protocol_enabled: bool,
    remote_ip: SocketAddr,
    stream: TcpStream,
    pwp_runtime: &runtime::Handle,
) -> io::Result<(pwp::DownloadChannels, pwp::UploadChannels, Option<pwp::ExtendedChannels>)> {
    let (download_chans, upload_chans, extended_chans, runner) =
        pwp::channels_for_inbound_connection(
            local_peer_id,
            Some(info_hash),
            extension_protocol_enabled,
            remote_ip,
            stream,
        )
        .await?;

    pwp_runtime.spawn(async move {
        if let Err(e) = runner.await
            && e.kind() != io::ErrorKind::BrokenPipe
            && e.kind() != io::ErrorKind::UnexpectedEof
        {
            log::warn!("Peer runner for {remote_ip} exited: {e}");
        }
    });

    Ok((download_chans, upload_chans, extended_chans))
}

pub async fn run_listener(
    local_addr: SocketAddr,
    peer_reporter: PeerReporter,
    canceller: CancellationToken,
) -> io::Result<()> {
    let task = async move {
        let socket = bound_pwp_socket(local_addr)?;
        let listener = socket.listen(1024)?;
        log::info!("TCP listener started on {}", listener.local_addr()?);
        loop {
            let (stream, addr) = listener.accept().await?;
            peer_reporter.report_accepted(addr, stream).await;
        }
    };
    canceller.run_until_cancelled_owned(task).await.unwrap_or(Ok(()))
}