【斯坦福计网CS144项目】Lab2: TCPReceiver

本节在 Lab1 实现的 StreamReassembler 的基础上进行 TCP 协议中接收端 TCPReceiver 的实现。

序号与索引

Lab 1 中实现的核心函数 push_substring 认为输入的数据索引是一个 64 位表示（size_t/uint64_t），从 0 开始依序增大的数字。然而 TCP 的序号并非如此：

1. TCP 包头中的表示序号的长度为 32 位，相当于 4GB，可能发生溢出，溢出后则重新从 0 开始增长。
2. TCP 的序号并非从 0，而是从一个随机的 32 位数开始，该数字称为 Initial Sequence Number (ISN)。后面的数据序号依次增长。
3. 整个数据流的开头和结尾各会占据一个序号（虽然它们不表示实际的数据）。分别对应 TCP 规定的 SYN 和 FIN 标识。

为此，讲义中给出了 seqno, absoulute seqno 和 stream index 的概念，下面这张图可以很好解释各自的含义：

后两者的区别仅为相差 1，主要的问题就是如何实现从 seqno（TCP 包携带）到 stream index（StreamReassembler 接受）的转换。为此，写一个前者的包装类 WrappingInt32，然后实现两个全局函数 wrap 和 unwrap 完成 WrappingInt32 和 uint64_t 的转换。注意这里 wrap 和 unwrap 进行的是 seqno 和 absolute seqno 间的转换，因为头尾标识的判断属于接收端应处理的逻辑，放到这两个辅助函数中会不必要的增加耦合度。

wrap 函数

给定 absolute seqno 和 isn，生成 seqno。实现逻辑为先将 64 位的 absolute seqno 对 32 位数的最大值取余，然后加上 isn。利用向更窄数据类型做转换发生溢出的效果与取余相同的特性，实现如下：

WrappingInt32 wrap(uint64_t n, WrappingInt32 isn) {
    // overflowing n by casting it to uint32_t is equivalent to n % (UINT32_MAX+1)
    return WrappingInt32(isn + static_cast<uint32_t>(n));
}
1
2
3
4

unwrap 函数

只给定 isn 和较短的 seqno 时无法判断 absolute seqno 该值是否经历了若干倍的溢出，因此还需要一个 64 位参数 checkpoint，表示 seqno 是所有可能的取值中与 checkpoint 距离最小的那个。这里我的实现不是最简单的，逻辑应该还能简化，仅提供一种原始思路：

（随手画的见谅ww）上图中蓝色表示无数个整数倍节点，用 seqno 减去 isn 得到所求 abs seqno 相对于它的偏移值，称为 n_mod，对应无数个可能的 abs seqno取值（绿点），红色表示 checkpoint 位置，余数称为 cp_mod。对于图中这种情况，即 n_mod 大于 cp_mod，真实的 abs seqno 只可能在图中左右两个绿点取到。首先如果 cp 除的倍数是0，也就是 cp 左边的蓝点已经是 0 了，自然左边的绿点也不存在，必然是右边。否则，用算式比较两边的距离，取更近的。

如果 n_mod 小于 cp_mod，则可能的两个绿点位置如上图所示，同样可以直观写出两边的距离并比较。

该部分代码完成后可以运行 ctest -R wrap 测试。

TCPReceiver

Receiver 接收的是一个 TCPSegment，结构如下图所示。本节中我们需要关注的是 Header 中的 seqno 和 SYN、FIN 标识位。

另外作为接收端还应该负责汇报 ackno，即期待对方发送的下一个数据的 seqno，该信息会由 Lab4 实现的 TCPConnection 类负责从 Receiver 读取并放入待发送的 TCPSegment 中。处理对方数据包的 ackno 是发送端负责的任务，这里无需关注。

主要实现的函数有两个：segment_received 和 ackno。

segment_received

函数形式为：void segment_received(const TCPSegment &seg);

Receiver 的生命周期如下图所示：

（本节无需考虑 error 情况，因为 Receiver 不带错误状态，会在 TCPConnection 层处理，错误状态直接设置到 ByteStream 上）

第一个带 SYN 标识的数据包意味着有效连接的开始，而带有 FIN 的数据包意味着连接即将结束，这两个状态分别用 _syn_set 和 _fin_set 记录。在 _syn_set 之前，不带 SYN 标识的数据包应该被视为无效而丢弃。第一个带 SYN 的数据包到来时使 _syn_set 变为 true，其 seqno 就是 isn。利用刚才实现的 unwrap 函数可以将 seqno 转换为 stream index，其中 checkpoint 可以使用 isn，同时注意对于非第一个（带 SYN 标识）的数据包，应该将 seqno 前移一位。然后就可以将数据和转换后的索引交给 Reassembler 进行重组。带 FIN 标识的数据包到来时使 _fin_set 变为 true，但此时还不能立刻结束数据流，因为传输乱序可能有数据包在 FIN 包之后才到达，只有同时检测到 Reassembler 中没有待重组的数据时才说明所有数据已到位，可以 end_input。实现如下：

void TCPReceiver::segment_received(const TCPSegment &seg) {
    const TCPHeader &header = seg.header();
    bool syn = header.syn;
    bool fin = header.fin;
    // before SYN is set in receiver, segments with no SYN flag should be disposed.
    if (!syn && !_syn_set)
        return;
    if (!_syn_set) {
        _syn_set = true;
        _init_seqno = header.seqno;
    }
    string data = seg.payload().copy();
    if (!data.empty()) {
        // there's a special case in t_ack_rst that a segment with data whose seqno belongs to SYN,
        // that data should be ignored
        if (syn || header.seqno != _init_seqno) {
            // we treat _init_seqno as the index of the first valid byte (though it's actually for SYN)
            // so for segments without SYN, the index should be shifted back by 1
            size_t index = unwrap(header.seqno - (!syn), _init_seqno, _reassembler.wait_index());
            _reassembler.push_substring(data, index, fin);
        }
    }
    // set FIN flag if FIN arrives, and from then on keep checking
    // if the reassembler is clear so that we can close the output stream
    if (fin || _fin_set) {
        _fin_set = true;
        if (_reassembler.unassembled_bytes() == 0)
            _reassembler.stream_out().end_input();
    }
}
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

ackno

函数形式为：std::optional<WrappingInt32> ackno() const;

由于只有当已经收到含 SYN 标识的数据包后才知道对方的 isn，ackno 也才能存在，所以返回类型使用了 std::optional，可以利用前面引入的 _syn_set 进行判断。这里需要用到 Reassembler 中等待 stream index 的信息，而 Lab1 讲义上没有规定这个 public 函数，因此自行定义一个 wait_index()，返回上节实现中的 _wait_index。利用 wrap 进行 stream index 到 seqno 的转换，同样要注意因为 SYN 标识占位所以要后移一位，如果数据流已经结束（所有数据 & FIN 已经真的到位了）还要再后移一位。实现如下：

optional<WrappingInt32> TCPReceiver::ackno() const {
    optional<WrappingInt32> res = nullopt;
    if (_syn_set) {
        uint64_t index = _reassembler.wait_index() + 1;
        // for ackno we should check whether the output stream has really closed
        // instead of whether FIN flag is set (there may still be unarrived bytes)
        if (_reassembler.stream_out().input_ended())
            index++;
        res.emplace(wrap(index, _init_seqno));
    }
    return res;
}
1
2
3
4
5
6
7
8
9
10
11
12

完整代码链接：
tcp_receiver.hh
tcp_receiver.cc

通关截图 😎