CS144计算机网络Lab3: The TCP Sender

The arrangement of modules and dataflow in TCP implementation.

在Lab3中，需要实现一个TCPSender，主要有以下几个功能：

• 根据接收方的响应调整发送窗口；
• 发送SYN，FIN报文；
• 从ByteStream中读取、创建并发送新的TCPSegment；
• 跟踪已发送的Segment，直到被确认接收，必要时进行超时重传。

A. How does the TCPSender know if a segment was lost?

在Lab0中介绍过，TCP是在不可靠的链路上提供尽力而为的可靠传输。为了实现这一目的，在TCP中，TCPSender需要确保每一个Segment都被成功接收。

每当一个Segment被发送出去，TCPSender会记录其发送的时间并标记为未完成的，当超过retransmission timeout(RTO)毫秒后，如果依然没有确认收到，就会进行重传。

这是由一个全局的计时器来实现的，具体来说，TCPSender的重传遵循以下几个规则：

• 每隔一段时间(毫秒)，TCPSender的Tick()方法会被调用，并带有一个参数ms_since_last_tick代表自上次调用该方法以来已经过去了多少毫秒。
• 当TCPSender被构造时，会附带一个_initial_retransmission_timeout参数，代表重传时限(retransmission timeout, RTO) 的初始值。RTO是重新发送未完成的TCP Segment前要等待的毫秒数。随着通信进行，RTO可能会发生变化，但是初始值保持不变。
• 每次发送携带数据的Segment时，如果计时器(Timer)没有运行，就启动它。
• 当所有未完成的Segment都被确认后，停止Timer。
• 当Tick()函数被调用，且已经过期时：
  • 立即重传最早的未被确认的Segment。
  • 如果window size非零：(指数退避)
    • 记录连续重传次数consecutive retransmissions，并在发生重传时增加它。
    • RTO *= RTO。
  • 重启Timer。
• 当接收方向发送方发出了新的确认时：
  • 重置RTO为初始值。
  • 如果Sender中还有未被确认的Segment，就重置Timer，并使其在RTO毫秒后过期。
  • 归零重传计数器。

Time代码如下：

tcp_sender.hh&tcp_sender.cc

/*************************tcp_sender.hh*************************/

class Timer {
  private:
    bool _is_running{false};
    unsigned int _consecutive_retransmissions_times{0};
    unsigned int _retransmission_timeout{0};
    unsigned int _time_passed{0};

  public:
    Timer();

/*************************tcp_sender.cc*************************/

void Timer::start(const unsigned int _initial_retransmission_timeout) {
    _is_running = true;
    _consecutive_retransmissions_times = 0;
    _retransmission_timeout = _initial_retransmission_timeout;
    _time_passed = 0;
}

void Timer::stop() { _is_running = false; }

void Timer::reset(const unsigned int _initial_retransmission_timeout) {
    stop();
    start(_initial_retransmission_timeout);
}

void Timer::restart() { _time_passed = 0; }

void Timer::time_update(const unsigned int ms_since_last_tick) {
    if (_is_running)
        _time_passed += ms_since_last_tick;
}

void Timer::exponential_backoff() {
    _consecutive_retransmissions_times++;
    _retransmission_timeout *= 2;
}

bool Timer::check_expired(const unsigned int ms_since_last_tick) {
    if (_is_running) {
        time_update(ms_since_last_tick);
    }
    if (_time_passed >= _retransmission_timeout) {
        return true;
    }
    return false;
}

bool Timer::is_running() const { return _is_running; }

B. Implementing the TCP sender

Evolution of the TCP sender

现在我们可以来实现一下TCPSender了。首先我们需要了解一下Sender的各种状态:

Evolution of the TCP sender (as tested by the test suite).

在CS144中，各个状态能干的事情，其中在CLOSED和SYN_ACKED状态时，在发送SYN和FIN信号的同时也是可以携带数据的(不过我觉得现实中应该是不可以的)。除此之外，在SYN_SENT状态发完SYN还是可以继续发只有数据的Segment，分的不是很清楚。

状态	发送信号	发送数据	重传数据
CLOSED	1	0	0
SYN_SENT	0	1	1
SYN_ACKED	1	1	1
FIN_SENT	0	0	1
FIN_ACKED	0	0	0
ERROR	0	0	0

Functions of the TCP sender

成员变量

我添加了以下几个成员变量，其中_segments_cache用来储存还未确认收到的Segment，_ackno用来记录确认收到的最大的seqno。

Timer _timer{};
unsigned int _window_size{0};
std::queue<TCPSegment> _segments_cache{};
uint64_t _ackno{0};
enum _status { CLOSED, SYN_SENT, SYN_ACKED, FIN_SENT, FIN_ACKED, ERROR } _status{CLOSED};

bytes_in_flight()

返回未被确认的Segment数量，直接用下一个未发送seqno减去最后一个被确认的seqno即可。

uint64_t TCPSender::bytes_in_flight() const { 
  return _next_seqno - _ackno; 
}

fill_window()

先计算一下这一次调用最多能发送多少数据fill_size：用window size减去目前未确认的Segment。需要注意的是当状态为CLOSE的时候，window size是0，这个时候可以发送syn，我们可以把window size当作1。

那么显然，当状态是CLOSE或者fill_size > 0且fin未发送时，我们就可以主动发送Segment了。

当状态时CLOSE时，我们就可以开启一个TCP链接了，发送一个syn信号，并将seqno设置为_isn，状态更新成SYN SENT。

当状态是SYN SENT（CS144中SYN Segment也可能携带数据）或SYN ACKED且有需要发送的数据（发送正常的纯数据Segment）或数据流终止（关闭TCP，发送fin）时：

• 先计算能填装的数据量，fill_size和MAX_PAYLOAD_SIZE取最小。
• 填装数据。
• 如果装完之后发现数据流终止了，我们就可以设置fin信号了。（注意fin也会占用一个seqno，所以需要先确认这次发送是否能装下，否则就要等下一轮了）
• 将Segment发送出去，并更新_next_seqno。

如果在发送Segment时Timer没有运行的话，就启动Timer。

如果当前轮已发送的是一个不含数据的Segment，即指包含了信号的话，就可以直接结束这次发送了。如果含有数据，我们还需要更新一下fill_size，继续发送。

void TCPSender::fill_window() {
    size_t fill_size = (_window_size ? _window_size : 1) - bytes_in_flight();
    while (_status == CLOSED || (fill_size && _status != FIN_SENT)) {
        TCPSegment segment;
        if (_status == CLOSED) {
            segment.header().syn = true;
            segment.header().seqno = _isn;
            _status = SYN_SENT;
        }
        if (_status == SYN_SENT or (_status == SYN_ACKED && (!_stream.buffer_empty() or _stream.eof()))) {
            size_t payload_len =
                min(fill_size, TCPConfig::MAX_PAYLOAD_SIZE);
            segment.header().seqno = wrap(_next_seqno, _isn);
            segment.payload() = _stream.read(payload_len);
            if (_stream.eof() && fill_size-segment.length_in_sequence_space()) {
                segment.header().fin = true;
                _status = FIN_SENT;
            }
            _segments_out.push(segment);
            _segments_cache.push(segment);
            _next_seqno += segment.length_in_sequence_space();
        }
        if (!_timer.is_running()) {
            _timer.start(_initial_retransmission_timeout);
        }
        if (!segment.length_in_sequence_space())
            return;
        fill_size -= segment.length_in_sequence_space();
    }
}

ack_received()

先将收到的ackno转换成absoult seqno，并确保其在我们可接受地范围内（在未确认的seqno之中）。

如果当前状态是SYN SENT则更新成SYN ACKED，并清除缓存_segments_cache中对应的Segment，并重置Timer。需要注意的是，对于含有fin信号的Segment，我们需要单独处理，在这个环节就将它排除了。

若此时状态是FIN SENT且已经确认到了fin信号（注意syn和fin都会占一个seqno，且seqno是从0开始的），则更新成FIN ACKED，并清除缓存。

最后，如果缓存是空的，则停止Timer。

void TCPSender::ack_received(const WrappingInt32 ackno, const uint16_t window_size) {
    uint64_t absoult_ackno = unwrap(ackno, _isn, _ackno);
    if (absoult_ackno < _ackno || absoult_ackno > _next_seqno) {
        return;
    }

    _ackno = absoult_ackno;
    _window_size = window_size;

    if (_status == SYN_SENT)
        _status = SYN_ACKED;

    while (!_segments_cache.empty() &&
           unwrap(_segments_cache.front().header().seqno, _isn, _ackno) < unwrap(ackno, _isn, _ackno) && !_segments_cache.front().header().fin) {
        _segments_cache.pop();
        _timer.reset(_initial_retransmission_timeout);
    }

    if (_status == FIN_SENT && _ackno == _next_seqno) {
        _status = FIN_ACKED;
        _segments_cache.pop();
    }

    if (_segments_cache.empty())
        _timer.stop();
}

tick()

当Sender处于SYN SENT、SYN ACKED或是FIN SENT状态时，调用Timer检查当前是否过期。

如果已过期，则：

• 若window size非0或SYN未确认收到时，执行指数退避。
• 若重传次数小于TCP内建的限制次数时，立即重传并重启Timer。

void TCPSender::tick(const size_t ms_since_last_tick) {
    if (_status == SYN_SENT or _status == SYN_ACKED or _status == FIN_SENT)
        if (_timer.check_expired(ms_since_last_tick)) {
            if (_window_size || _status == SYN_SENT)
                _timer.exponential_backoff();
            if (consecutive_retransmissions() <= TCPConfig::MAX_RETX_ATTEMPTS) {
                _segments_out.push(_segments_cache.front());
                _timer.restart();
            } else {
                _status = ERROR;
            }
        }
}

consecutive_retransmissions()

返回Timer中记录的重传次数。

unsigned int TCPSender::consecutive_retransmissions() const { 
  return _timer.consecutive_retransmissions_times(); 
}

本文链接: https://zone.ivanz.cc/p/785ad61c.html