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
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
use std::cmp::{max, Ordering};
use std::num::Wrapping;
use std::ops::{Add, Div, Mul, Neg, Sub};
use std::time::{Duration, Instant};
use std::u32;

pub mod connection;
pub mod handshake;
pub mod receiver;
pub mod sender;
pub mod stats;

/// Timestamp in us after creation
/// These wrap every 2^32 microseconds
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub struct TimeStamp(Wrapping<u32>);

/// Signed duration in us, e.g. RTT
#[derive(Debug, Copy, Clone, PartialEq, Eq, Ord, PartialOrd)]
pub struct TimeSpan(i32);

const TIMESTAMP_MASK: u128 = u32::MAX as u128;

#[derive(Copy, Clone, Debug)]
pub struct TimeBase(Instant);

impl TimeSpan {
    pub fn from_micros(us: i32) -> Self {
        Self(us)
    }

    pub fn as_micros(self) -> i32 {
        self.0
    }

    pub fn abs(self) -> Self {
        Self(self.0.abs())
    }

    pub fn as_secs_f64(self) -> f64 {
        self.0 as f64 / 1e6
    }
}

impl TimeStamp {
    pub fn from_micros(us: u32) -> Self {
        Self(Wrapping(us))
    }

    pub fn as_micros(self) -> u32 {
        (self.0).0
    }

    pub fn as_secs_f64(self) -> f64 {
        (self.0).0 as f64 / 1e6
    }

    pub fn as_duration(self) -> Duration {
        Duration::from_micros(u64::from(self.as_micros()))
    }
}

impl PartialOrd<TimeStamp> for TimeStamp {
    fn partial_cmp(&self, other: &TimeStamp) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for TimeStamp {
    fn cmp(&self, other: &Self) -> Ordering {
        // this is a "best effort" implementation, and goes for close
        // if timestamps are very far apart, this will not work (and cannot)
        (*self - *other).as_micros().cmp(&0)
    }
}

impl Add<TimeSpan> for TimeStamp {
    type Output = TimeStamp;

    #[allow(clippy::suspicious_arithmetic_impl)]
    fn add(self, rhs: TimeSpan) -> Self::Output {
        TimeStamp(if rhs.0 > 0 {
            self.0 + Wrapping(rhs.0 as u32)
        } else {
            self.0 - Wrapping(rhs.0.abs() as u32)
        })
    }
}

impl Sub<TimeSpan> for TimeStamp {
    type Output = TimeStamp;

    fn sub(self, rhs: TimeSpan) -> Self::Output {
        self + -rhs
    }
}

impl Sub<TimeStamp> for TimeStamp {
    type Output = TimeSpan;

    fn sub(self, rhs: TimeStamp) -> TimeSpan {
        // This is also a "best effort" implementation, and cannot be precise
        let pos_sub = self.0 - rhs.0;
        let neg_sub = rhs.0 - self.0;
        if pos_sub < neg_sub {
            TimeSpan(pos_sub.0 as i32)
        } else {
            -TimeSpan(neg_sub.0 as i32)
        }
    }
}

impl Neg for TimeSpan {
    type Output = TimeSpan;

    fn neg(self) -> Self::Output {
        Self(-self.0)
    }
}

impl Mul<i32> for TimeSpan {
    type Output = TimeSpan;

    fn mul(self, rhs: i32) -> Self::Output {
        Self(self.0 * rhs)
    }
}

impl Add<TimeSpan> for TimeSpan {
    type Output = TimeSpan;

    fn add(self, rhs: TimeSpan) -> Self::Output {
        Self(self.0 + rhs.0)
    }
}

impl Div<i32> for TimeSpan {
    type Output = TimeSpan;

    fn div(self, rhs: i32) -> Self::Output {
        Self(self.0 / rhs)
    }
}

impl Sub<TimeSpan> for TimeSpan {
    type Output = TimeSpan;

    fn sub(self, rhs: TimeSpan) -> Self::Output {
        Self(self.0 - rhs.0)
    }
}

impl TimeBase {
    pub fn new(start_time: Instant) -> Self {
        Self(start_time)
    }

    #[allow(clippy::trivially_copy_pass_by_ref)]
    pub fn timestamp_from(&self, instant: Instant) -> TimeStamp {
        assert!(
            self.0 <= instant,
            "Timestamps are only valid after the timebase start time"
        );
        TimeStamp(Wrapping(
            ((instant - self.0).as_micros() & TIMESTAMP_MASK) as u32,
        ))
    }

    // Get Instant closest to `now` that is consistent with `timestamp`
    #[allow(clippy::trivially_copy_pass_by_ref)]
    pub fn instant_from(&self, now: Instant, timestamp: TimeStamp) -> Instant {
        let wraps = ((now - self.0).as_micros() >> 32) as u64;
        self.0
            + Duration::from_micros(wraps * u64::from(std::u32::MAX) + timestamp.as_micros() as u64)
    }

    pub fn adjust(&mut self, delta: TimeSpan) {
        if delta.0 > 0 {
            self.0 += Duration::from_micros(delta.0 as u64);
        } else {
            self.0 -= Duration::from_micros(delta.0.abs() as u64);
        }
    }

    pub fn origin_time(&self) -> Instant {
        self.0
    }
}

#[cfg(test)]
mod timebase {
    use super::*;
    use proptest::prelude::*;

    proptest! {
        #[test]
        fn timestamp_roundtrip(expected_ts: u32) {
            let timebase = TimeBase::new(Instant::now());
            let expected_ts = TimeStamp::from_micros(expected_ts);

            let ts = timebase.timestamp_from(timebase.instant_from(Instant::now(), expected_ts));
            assert_eq!(ts, expected_ts);
        }

        #[test]
        fn timestamp_from(expected_ts: u32, n in 0u64..10) {
            let now = Instant::now();
            let timebase = TimeBase::new(now);
            let delta = ((std::u32::MAX as u64 + 1)* n) + expected_ts as u64;
            let instant =  now + Duration::from_micros(delta as u64);
            let ts = timebase.timestamp_from(instant);
            assert_eq!(ts, TimeStamp::from_micros(expected_ts));
        }

        #[test]
        fn adjust(drift: i16) {
            let now = Instant::now();
            let mut timebase = TimeBase::new(now);
            let drift = TimeSpan::from_micros(i32::from(drift));

            let original_ts = timebase.timestamp_from(now);
            timebase.adjust(drift);
            let ts = timebase.timestamp_from(now + Duration::from_micros(1_000_000));

            assert_eq!(ts, original_ts - drift + TimeSpan::from_micros(1_000_000));
        }

    }
}

#[cfg(test)]
mod timestamp {
    use super::*;

    #[test]
    #[allow(clippy::eq_op)]
    fn subtract_timestamp() {
        let a = TimeStamp::from_micros(10);
        let max = a - TimeSpan(11);
        let b = TimeStamp::from_micros(11);

        assert_eq!(a - a, TimeSpan::from_micros(0));
        assert_eq!(b - a, TimeSpan::from_micros(1));
        assert_eq!(a - b, TimeSpan::from_micros(-1));
        assert!(max < a);
        assert!(b > a);
        assert!(b > max);
        assert_eq!(max.as_micros(), u32::MAX);
    }
}

//4. Timers
//
//   UDT uses four timers to trigger different periodical events. Each
//   event has its own period and they are all independent. They use the
//   system time as origins and should process wrapping if the system time
//   wraps.
//
//   For a certain periodical event E in UDT, suppose the time variable is
//   ET and its period is p. If E is set or reset at system time t0 (ET =
//   t0), then at any time t1, (t1 - ET >= p) is the condition to check if
//   E should be triggered.
//
//   The four timers are ACK, NAK, EXP and SND. SND is used in the sender
//   only for rate-based packet sending (see Section 6.1), whereas the
//   other three are used in the receiver only.
//
//   ACK is used to trigger an acknowledgement (ACK). Its period is set by
//   the congestion control module. However, UDT will send an ACK no
//   longer than every 0.01 second, even though the congestion control
//   does not need timer-based ACK. Here, 0.01 second is defined as the
//   SYN time, or synchronization time, and it affects many of the other
//   timers used in UDT.
//
//   NAK is used to trigger a negative acknowledgement (NAK). Its period
//   is dynamically updated to 4 * RTT_+ RTTVar + SYN, where RTTVar is the
//   variance of RTT samples.
//
//   EXP is used to trigger data packets retransmission and maintain
//   connection status. Its period is dynamically updated to 4 * RTT +
//   RTTVar + SYN.
//
//   The recommended granularity of their periods is microseconds. The
//   system time is queried after each time bounded UDP receiving (there
//   will be additional necessary data processing time if a UDP packet is
//   received) to check if any of the ACK, NAK, or EXP event should be
//   triggered. The timeout value of UDP receiving should be at least SYN.
//
//   In the rest of this document, a name of a time variable will be used
//   to represent the associated event, the variable itself, or the value
//   of its period, depending on the context. For example, ACK can mean
//   either the ACK event or the value of ACK period.

pub struct Timer {
    period: Duration,
    last: Instant,
}

impl Timer {
    const MIN_PERIOD: Duration = Duration::from_micros(1);

    pub fn new(period: Duration, now: Instant) -> Timer {
        Timer {
            period: max(period, Self::MIN_PERIOD),
            last: now,
        }
    }

    pub fn period(&mut self) -> Duration {
        self.period
    }

    pub fn next_instant(&self) -> Instant {
        self.last + self.period
    }

    pub fn reset(&mut self, now: Instant) {
        self.last = now;
    }

    pub fn set_period(&mut self, period: Duration) {
        self.period = period;
    }

    pub fn check_expired(&mut self, now: Instant) -> Option<Instant> {
        if self.period.as_nanos() == 0 {
            return Some(now);
        }
        if now >= self.next_instant() {
            let elapsed = now - self.last;
            let elapsed_periods = elapsed.as_nanos() / self.period.as_nanos();
            self.last += self.period * elapsed_periods as u32;
            Some(self.last)
        } else {
            None
        }
    }
}