CS144 计算机网络 Lab4：TCP Connection

前言

经过前面几个实验的铺垫，终于到了将他们组合起来的时候了。Lab4 将实现 TCP Connection 功能，内部含有 TCPReceiver 和 TCPSender，可以与 TCP 连接的另一个端点进行数据交换。

实验要求

简单来说，这次实验就是要在 TCPConnection 类中实现下图所示的有限状态机：

这些状态对应 TCPState 的内部枚举类 State：

复制//! \brief Official state names from the [TCP](\ref rfc::rfc793) specification
enum class State {
    LISTEN = 0,   //!< Listening for a peer to connect
    SYN_RCVD,     //!< Got the peer's SYN
    SYN_SENT,     //!< Sent a SYN to initiate a connection
    ESTABLISHED,  //!< Three-way handshake complete
    CLOSE_WAIT,   //!< Remote side has sent a FIN, connection is half-open
    LAST_ACK,     //!< Local side sent a FIN from CLOSE_WAIT, waiting for ACK
    FIN_WAIT_1,   //!< Sent a FIN to the remote side, not yet ACK'd
    FIN_WAIT_2,   //!< Received an ACK for previously-sent FIN
    CLOSING,      //!< Received a FIN just after we sent one
    TIME_WAIT,    //!< Both sides have sent FIN and ACK'd, waiting for 2 MSL
    CLOSED,       //!< A connection that has terminated normally
    RESET,        //!< A connection that terminated abnormally
};

除了三次握手和四次挥手外，我们还得处理报文段首部 RST 标志被置位的情况，这时候应该将断开连接，并将内部的输入流和输入流标记为 error，此时的 TCPState 应该是 RESET 。

代码实现

先在类声明里面加上一些成员：

复制class TCPConnection {
  private:
    TCPConfig _cfg;
    TCPReceiver _receiver{_cfg.recv_capacity};
    TCPSender _sender{_cfg.send_capacity, _cfg.rt_timeout, _cfg.fixed_isn};

    //! outbound queue of segments that the TCPConnection wants sent
    std::queue _segments_out{};

    //! Should the TCPConnection stay active (and keep ACKing)
    //! for 10 * _cfg.rt_timeout milliseconds after both streams have ended,
    //! in case the remote TCPConnection doesn't know we've received its whole stream?
    bool _linger_after_streams_finish{true};

    bool _is_active{true};

    size_t _last_segment_time{0};

    /**
     * @brief 发送报文段
     * @param fill_window 是否填满发送窗口
    */
    void send_segments(bool fill_window = false);

    // 发送 RST 报文段
    void send_rst_segment();

    // 中止连接
    void abort();

  public:
    // 省略其余成员
}

接着实现几个最简单的成员函数：

复制size_t TCPConnection::remaining_outbound_capacity() const { return _sender.stream_in().remaining_capacity(); }

size_t TCPConnection::bytes_in_flight() const { return _sender.bytes_in_flight(); }

size_t TCPConnection::unassembled_bytes() const { return _receiver.unassembled_bytes(); }

size_t TCPConnection::time_since_last_segment_received() const { return _last_segment_time; }

bool TCPConnection::active() const { return _is_active; }

主动连接

客户端可以调用 TCPConnection::connect 函数发送 SYN 报文段请求与服务端建立连接，由于 Lab3 中实现的 TCPSender::fill_window() 函数会根据发送方的状态选择要发送的报文段类型，在还没建立连接的情况下，这里直接调用 fill_window() 就会将一个 SYN 报文段放在队列中，我们只需将其取出放到 TCPConnection 的 _segments_out 队列中即可：

复制void TCPConnection::connect() {
    // 发送 SYN
    send_segments(true);
}

void TCPConnection::send_segments(bool fill_window) {
    if (fill_window)
        _sender.fill_window();

    auto &segments = _sender.segments_out();


    while (!segments.empty()) {
        auto seg = segments.front();

        // 设置 ACK、确认应答号和接收窗口大小
        if (_receiver.ackno()) {
            seg.header().ackno = _receiver.ackno().value();
            seg.header().win = _receiver.window_size();
            seg.header().ack = true;
        }

        _segments_out.push(seg);
        segments.pop();
    }
}

主动关闭

当上层程序没有更多数据需要发送时，将会调用 TCPConnection::end_input_stream() 结束输入，这时候需要发送 FIN 报文段给服务端，告诉他自己没有更多数据要发送了，但是可以继续接收服务端发来的数据。客户端的状态由 ESTABLISHED 转移到 FIN_WAIT_1，服务端收到 FIN 之后变成 CLOSE_WAIT 状态，并回复 ACK 给客户端，客户端收到之后接着转移到 FIN_WAIT_2 状态。

如果服务端数据传输完成了，会发送 FIN 报文段给客户端，转移到 LAST_ACK 状态，此时客户端会回复最后一个 ACK 给服务端并进入 TIME_WAIT 超时等待状态，如果这个等待时间内没有收到服务端重传的 FIN，就说明 ACK 顺利到达了服务端且服务端已经变成 CLOSED 状态了，此时客户端也能断开连接变成 CLOSED 了。

复制void TCPConnection::end_input_stream() {
    // 发送 FIN
    _sender.stream_in().end_input();
    send_segments(true);
}

在上述情景中，客户端是主动关闭（Active Close）的一方，服务端是被动关闭（Passive Close）的一方。

主动重置连接

有两种情况会导致发送 RST 报文段来主动重置连接：

• 当 TCPSender 超时重传的次数过多时，表明通信链路存在故障；
• TCPConnect 对象被释放但是 TCP 仍然处于连接状态的时候；

和 Lab3 中类似，TCPConnection 通过外部定期调用 tick() 函数来得知过了多长时间，在 tick() 函数里还得处理超时等待的情况：

复制//! \param[in] ms_since_last_tick number of milliseconds since the last call to this method
void TCPConnection::tick(const size_t ms_since_last_tick) {
    _sender.tick(ms_since_last_tick);

    // 重传次数太多时需要断开连接
    if (_sender.consecutive_retransmissions() > _cfg.MAX_RETX_ATTEMPTS) {
        return send_rst_segment();
    }

    // 重传数据包
    send_segments();

    _last_segment_time += ms_since_last_tick;

    //  TIME_WAIT 超时等待状态转移到 CLOSED 状态
    if (TCPState::state_summary(_receiver) == TCPReceiverStateSummary::FIN_RECV &&
        TCPState::state_summary(_sender) == TCPSenderStateSummary::FIN_ACKED &&
        _last_segment_time >= 10 * _cfg.rt_timeout) {
        _linger_after_streams_finish = false;
        _is_active = false;
    }
}

TCPConnection::~TCPConnection() {
    try {
        if (active()) {
            cerr << "Warning: Unclean shutdown of TCPConnection\n";

            // Your code here: need to send a RST segment to the peer
            send_rst_segment();
        }
    } catch (const exception &e) {
        std::cerr << "Exception destructing TCP FSM: " << e.what() << std::endl;
    }
}

void TCPConnection::send_rst_segment() {
    abort();
    TCPSegment seg;
    seg.header().rst = true;
    _segments_out.push(seg);
}

void TCPConnection::abort() {
    _is_active = false;
    _sender.stream_in().set_error();
    _receiver.stream_out().set_error();
}

接收报文段

外部通过 TCPConnection::segment_received() 将接收到的报文段传给它，在这个函数内部，需要将确认应答号和接收窗口大小告诉 TCPSender，好让他接着填满发送窗口。接着还需要把报文段传给 TCPReceiver 来重组数据，并更新确认应答号和自己的接收窗口大小。然后 TCPSender 需要根据收到的包类型进行状态转移，并决定发送含有有效数据的报文段还是空 ACK 给对方。

为什么即使没有新的数据要发送也要回复一个空 ACK 呢？因为如果不这么做，对方会以为刚刚发的包丢掉了而一直重传。

复制void TCPConnection::segment_received(const TCPSegment &seg) {
    if (!active())
        return;

    _last_segment_time = 0;

    // 是否需要发送空包回复 ACK，比如没有数据的时候收到 SYN/ACK 也要回一个 ACK
    bool need_empty_ack = seg.length_in_sequence_space();

    auto &header = seg.header();

    // 处理 RST 标志位
    if (header.rst)
        return abort();

    // 将包交给发送者
    if (header.ack) {
        need_empty_ack |= !_sender.ack_received(header.ackno, header.win);

        // 队列中已经有数据报文段了就不需要专门的空包回复 ACK
        if (!_sender.segments_out().empty())
            need_empty_ack = false;
    }

    // 将包交给接受者
    need_empty_ack |= !_receiver.segment_received(seg);

    // 被动连接
    if (TCPState::state_summary(_receiver) == TCPReceiverStateSummary::SYN_RECV &&
        TCPState::state_summary(_sender) == TCPSenderStateSummary::CLOSED)
        return connect();

    // 被动关闭
    if (TCPState::state_summary(_receiver) == TCPReceiverStateSummary::FIN_RECV &&
        TCPState::state_summary(_sender) == TCPSenderStateSummary::SYN_ACKED)
        _linger_after_streams_finish = false;

    // LAST_ACK 状态转移到 CLOSED
    if (TCPState::state_summary(_receiver) == TCPReceiverStateSummary::FIN_RECV &&
        TCPState::state_summary(_sender) == TCPSenderStateSummary::FIN_ACKED && !_linger_after_streams_finish) {
        _is_active = false;
        return;
    }

    if (need_empty_ack && TCPState::state_summary(_receiver) != TCPReceiverStateSummary::LISTEN)
        _sender.send_empty_segment();

    // 发送其余报文段
    send_segments();
}

测试

在终端中输入 make check_lab4 就能运行所有测试用例，测试结果如下：

发现有几个 txrx.sh 的测试用例失败了，但是单独运行这些测试用例却又可以通过，就很奇怪：

接着测试一下吞吐量（请确保构建类型是 Release 而不是 Debug），感觉还行， 0.71Gbit/s，超过了实验指导书要求的 0.1Gbit/s。但是实际上还可以优化一下 ByteStream 类，将内部数据类型换成 BufferList，这样在写入数据的时候就不用一个字符一个字符插入队列了，可以大大提高效率。

最后将 Lab0 中 webget 使用的 TCPSocket 换成 CS144TCPSocket，重新编译并运行 webegt，发现能够正确得到响应结果，说明我们实现的这个 CS144TCPSocket 已经能和别的操作系统实现的 Socket 进行交流了：

后记

至此，CS144 的 TCP 实验部分已全部完成，可以说是比较有挑战性的一次实验了，尤其是 Lab4 部分，各种奇奇怪怪的 bug，编码一晚上，调试时长两天半（约等于一坤天），调试的时候断点还总是失效，最后发现是优化搞的鬼，需要将 etc/cflags.cmake 第 18 行改为 set (CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -ggdb3 -O0") 才行。以上~~