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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
use std::time::Duration;
use tokio::time::Instant;
use crate::{avg::SlidingDurationAvg, counter::ThruputCounters};
/// At any given time, a connection with a peer is in one of the below states.
#[derive(Clone, Default, Copy, Debug, PartialEq)]
pub enum ConnectionState {
/// The peer connection has not yet been connected or it had been connected
/// before but has been stopped.
#[default]
Disconnected,
/// The state during which the TCP connection is established.
Connecting,
/// The state after establishing the TCP connection and exchanging the
/// initial BitTorrent handshake.
Handshaking,
// This state is optional, it is used to verify that the bitfield exchange
// occurrs after the handshake and not later. It is set once the
// handshakes are exchanged and changed as soon as we receive the
// bitfield or the the first message that is not a bitfield. Any
// subsequent bitfield messages are rejected and the connection is
// dropped, as per the standard. AvailabilityExchange,
/// This is the normal state of a peer session, in which any messages,
/// apart from the 'handshake' and 'bitfield', may be exchanged.
Connected,
/// This state is set when the program is gracefully shutting down,
/// In this state, we don't send the outgoing blocks to the tracker on
/// shutdown.
Quitting,
}
/// Contains the state of both sides of the connection.
#[derive(Clone, Copy, Debug)]
pub struct State {
/// The current state of the connection.
pub connection: ConnectionState,
/// If we're choked, peer doesn't allow us to download pieces from them.
pub am_choking: bool,
/// If we're interested, peer has pieces that we don't have.
pub am_interested: bool,
/// If peer is choked, we don't allow them to download pieces from us.
pub peer_choking: bool,
/// If peer is interested in us, they mean to download pieces that we have.
pub peer_interested: bool,
// when the torrent is paused, those values will be set, so we can
// assign them again when the torrent is resumed.
// peer interested will be calculated by parsing the peers pieces
pub prev_peer_choking: bool,
}
impl Default for State {
/// By default, both sides of the connection start off as choked and not
/// interested in the other.
fn default() -> Self {
Self {
connection: Default::default(),
am_choking: true,
am_interested: false,
peer_choking: true,
peer_interested: false,
prev_peer_choking: true,
}
}
}
/// Holds and provides facilities to modify the state of a peer session.
#[derive(Debug)]
pub struct Session {
/// The session state.
pub state: State,
/// Measures various transfer statistics.
pub counters: ThruputCounters,
/// Whether we're in endgame mode.
pub in_endgame: bool,
/// The target request queue size is the number of block requests we keep
/// outstanding
pub target_request_queue_len: u16,
/// The last time some requests were sent to the peer.
pub last_outgoing_request_time: Option<Instant>,
/// Updated with the time of receipt of the most recently received
/// requested block.
pub last_incoming_block_time: Option<Instant>,
/// Updated with the time of receipt of the most recently uploaded block.
pub last_outgoing_block_time: Option<Instant>,
/// This is the average network round-trip-time between the last issued
/// a request and receiving the next block.
///
/// Note that it doesn't have to be the same block since peers are not
/// required to serve our requests in order, so this is more of a general
/// approximation.
pub avg_request_rtt: SlidingDurationAvg,
pub request_timed_out: bool,
pub timed_out_request_count: usize,
/// The time the BitTorrent connection was established (i.e. after
/// handshaking)
pub connected_time: Option<Instant>,
/// If the torrent was fully downloaded, all peers will become seed only.
/// They will only seed but not download anything anymore.
pub seed_only: bool,
}
impl Default for Session {
fn default() -> Self {
Self {
state: State::default(),
counters: ThruputCounters::default(),
in_endgame: false,
target_request_queue_len: Session::DEFAULT_REQUEST_QUEUE_LEN,
connected_time: None,
avg_request_rtt: SlidingDurationAvg::default(),
request_timed_out: false,
timed_out_request_count: 0,
last_incoming_block_time: None,
last_outgoing_block_time: None,
last_outgoing_request_time: None,
seed_only: false,
}
}
}
impl Session {
/// The value of outstanding blocks for a peer.
///
/// Before we do an extended handshake,
/// we do not have access to `reqq`.
/// And so this value is initialized with a sane default,
/// most clients support 250+ inflight requests.
///
/// After the extended handshake, this value is not used
/// in favour of the `reqq`, if the peer has it.
pub const DEFAULT_REQUEST_QUEUE_LEN: u16 = 150;
/// The smallest timeout value we can give a peer. Very fast peers will have
/// an average round-trip-times, so a slight deviation would punish them
/// unnecessarily. Therefore we use a somewhat larger minimum threshold for
/// timeouts.
const MIN_TIMEOUT: Duration = Duration::from_secs(2);
/// Returns the current request timeout value, based on the running average
/// of past request round trip times.
pub fn request_timeout(&self) -> Duration {
// we allow up to four times the average deviation from the mean
// let t = self.avg_request_rtt.mean() + 4 *
// self.avg_request_rtt.deviation(); t.max(Self::MIN_TIMEOUT)
Self::MIN_TIMEOUT
}
/// Updates state to reflect that peer was timed out.
pub fn register_request_timeout(&mut self) {
// peer has timed out, only allow a single outstanding request
// from now until peer hasn't timed out
// self.target_request_queue_len -= 1;
self.timed_out_request_count += 1;
self.request_timed_out = true;
}
/// Updates various statistics around a block download.
/// This should be called every time a block is received.
pub fn update_download_stats(&mut self, block_len: u32) {
let now = Instant::now();
// update request time
if let Some(last_outgoing_request_time) =
&mut self.last_outgoing_request_time
{
// Due to what is presumed to be inconsistencies with the
// `Instant::now()` API, it happens in rare circumstances that using
// the regular `duration_since` here panics (#48). I suspect this
// happens when requests are made a very short interval before this
// function is called, which is likely in very fast downloads.
// Either way, we guard against this by defaulting to 0.
let elapsed_since_last_request =
now.saturating_duration_since(*last_outgoing_request_time);
// If we timed out before, check if this request arrived within the
// timeout window, or outside of it. If it arrived within the
// window, we can mark peer as having recovered from the timeout.
if self.request_timed_out
&& elapsed_since_last_request <= self.request_timeout()
{
self.request_timed_out = false;
}
let request_rtt = elapsed_since_last_request;
self.avg_request_rtt.update(request_rtt);
}
self.counters.payload.down += block_len as u64;
self.last_incoming_block_time = Some(now);
}
pub fn record_waste(&mut self, block_len: u32) {
self.counters.waste += block_len as u64;
}
pub fn update_upload_stats(&mut self, block_len: u32) {
self.last_outgoing_block_time = Some(Instant::now());
self.counters.payload.up += block_len as u64;
}
}