爱站程序员基地Lab1-4 分别是完成一个流重组器，TCP接收端，TCP发送端，TCP连接四个部分，将四个部分组合在一起就是一个完整的TCP端了。之后经过包装就可以进行TCP的接收和发送了。
代码全部在github上了。
Lab1 流重组器
这一个实验是要实现一个流重组器，传入数据的片段以及起始位置，之后对其进行重组，并尽快将以及重组完成的数据输出。
这里我使用的是红黑树来实现，也就是C++的
std::set
来实现。将未重组完成的碎片保存在红黑树中，当新碎片到达时就尽可能地将该碎片与已有的碎片进行合并，保证红黑树中没有重叠的碎片。
这一个实验的问题就是要考虑的情况有很多，当用
lower_bound()
找到插入位置后，要对前面和后面的碎片判断能否合并，合并的情况也有很多种，包括部分重叠、正好接上、完全覆盖等情况；而且一个新的碎片可能会一次覆盖掉很多碎片。这一部分代码我写的比较混乱，因为写完之后测试发现有情况没考虑到，然后就只能打补丁，于是就越来越混乱了。
而尽快输出这个条件还是很容易的，如果当前到达碎片能够直接输出的话，就再判断一下树中第一个碎片能否输出，因为前面保证了不会有重叠碎片，所以可以只对第一个碎片进行判断。
Lab2 TCP接收端
这个实验是基于上一个的流重组器来实现一个TCP接收端，这一个还是比较简单的。就是前面流重组器的一些BUG可能会在这个实验里面被检测到，要回去改代码。
首先是实现一个
WrappingInt32
，因为TCP的序号是32位的，并且是可能发生溢出的，而在流重组器里面使用的序列号是64位，因此需要实现函数来进行转换，将64位的相对序列号根据ISN转换成32位的绝对序列号。
#include "wrapping_integers.hh"using namespace std;WrappingInt32 wrap(uint64_t n, WrappingInt32 isn) {uint64_t res = isn.raw_value() + n;return WrappingInt32{static_cast<uint32_t>(res)};}uint64_t abs(uint64_t a, uint64_t b) {if (a > b) {return a - b;} else {return b - a;}}uint64_t unwrap(WrappingInt32 n, WrappingInt32 isn, uint64_t checkpoint) {uint64_t pre = checkpoint & 0xffffffff00000000;uint64_t num;if (n.raw_value() >= isn.raw_value()) {num = n.raw_value() - isn.raw_value();} else {num = 0x0000000100000000;num += n.raw_value();num -= isn.raw_value();}uint64_t a = pre + num;uint64_t b = a + 0x0000000100000000;uint64_t c = a - 0x0000000100000000;// b a cif (abs(a, checkpoint) < abs(b, checkpoint)) {if (abs(a, checkpoint) < abs(c, checkpoint)) {return a;} else {return c;}} else {return b;}}
最后就是对流重组器进行一下包装，计算出
ackno
和
window_size
提供给后面使用，处理一下SYN和FIN标记就行了。
#include "tcp_receiver.hh"using namespace std;void TCPReceiver::segment_received(const TCPSegment &seg) {if (!_isn.has_value()) {if (seg.header().syn) {_isn = seg.header().seqno + 1;} else {std::cerr << "Error: connection not build" << std::endl;return;}}bool eof = seg.header().fin;std::string&& payload = std::string(seg.payload().str());if (seg.header().seqno == _isn.value() - 1 && !seg.header().syn && _reassembler.expect() < 0x0000ffff) {// wrong packet seqno == isnreturn;}uint64_t index = unwrap(seg.header().seqno + (seg.header().syn ? 1 : 0), _isn.value(), _reassembler.expect());_reassembler.push_substring(payload, index, eof);}optional<WrappingInt32> TCPReceiver::ackno() const {if (_isn.has_value()) {return { wrap(_reassembler.expect(), _isn.value()) + (_reassembler.stream_out().input_ended() ? 1 : 0) };} else {return std::nullopt;}}size_t TCPReceiver::window_size() const {return _capacity - _reassembler.stream_out().buffer_size();}
Lab3 TCP发送端
在这个实验里面就要考虑到TCP的一些细节了，包括
SYN
和
FIN
包的发送，
ACK
的处理，超时重传的实现了。
SYN包
之前我还在考虑客户端和服务端的
SYN
包应该是不同的，应该如何处理；而实际上两个包是相同的，都是携带
SYN
和初始序列号，不同的地方就是服务端的
SYN
要同时对客户端的
SYN
的包进行ACK。但ACK的处理是在TCP连接部分进行处理的，也就是说在TCP发送端里，只要发出一个
SYN
包就行了，剩下的不需要考虑。
而应该在什么时候发出
SYN
包呢，一开始我是在构造函数中就构造一个
SYN
包放到发送队列里面。这个做法在这个实验的测试里面是没有问题的，但是对于下一个实验就有问题了。因为服务端一开始是处于
LISTEN
状态，而这个状态下不应该有包被发出。因此，
SYN
包的发送应该放在
fill_window()
函数中，如果没有发送过
SYN
包，就先将
SYN
包发送出去。
对于一个最简单的
SYN
包，就只需要将
SYN
位置1，设置初始序列号
seqno
就行了。注意
SYN
是要占用一个序列号的。
FIN包
当发送流被用户程序结束后，就可以发送
FIN
包来关闭一个方向的连接了。这一个包的发送还是比较简单的，只要在
fill_window
中判断是否结束就行了。
FIN
包是可以和数据包一起发送的，在发送数据时发现流结束了，就将该数据包的
FIN
标志置1就行了。
FIN
也是要占用一个序列号的。当收到对方对
FIN
包的ACK后，就说明顺利关闭了。
ACK的处理
ACK数据会通过
ack_received
函数来通知发送端，当收到一个ACK后，就可以将发送窗口向右滑动，注意判断一下ACK是不是之前的ACK，避免窗口左移。最后将所有被ACK了的包从等待确认的队列中移除就行了。
重传
在这个实验里面只要求实现的是超时重传机制，但我也加入了快速重传机制。理论上当一个包超时之后就要对其进行重传，也就是每个包都要有定时器来负责重传，而这样的代价是很高的。因此，实现中是对每个TCP连接设置一个定时器，当时间超过RTO后进行重传，定时器的规则如下：
- 当发送一个包并且定时器为关闭状态：打开定时器
- 当发送一个包并且定时器为打开状态：不做任何修改
- 当收到一个ACK并且所有包都被ACK了：关闭定时器
- 当收到一个ACK并且仍有包未被ACK：重开定时器
超时重传使用的是指数退避算法，当进行了一次超时重传后，下一次超时的时间就会翻倍，也就是1RTO，2RTO，4RTO，8RTO…
当定时器超时后，就需要对包进行重传，在本实验里面只要重传第一个包就行了。这种方法的缺点就是后面丢失的包也要等到第一个被ACK了才能重发，时间会比较长。另一种选择是重传所有包，而这种的缺点就是会加大网络负担。因此，为了解决这种问题，就引入了其他的机制。
快速重传
快速重传是指当收到三个重复ACK（不包括第一次ACK）的时候，就立即进行重传，这种方法是数据驱动而不是时间驱动，可以避免超时重传的速度较慢的问题。而这仍然存在是重传一个还是所有的问题。
SACK
SACK也就是选择重传机制，接收端通过SACK来确认已收到的片段，从而对重传算法进行优化，可以不用对所有包进行重传。
而SACK存在接收方Reneging的问题，即接收方有权把已经SACK的数据给丢弃。这种丢弃是不被鼓励但还是可能发生的。因此，发送方不能完全依赖SACK，还是要依赖ACK，并维护定时器，如果后续的ACK没有增长，那么还是要对已经SACK的数据进行重传。同时，接收端也永远不能把SACK的包标记为ACK。
#include "tcp_sender.hh"#include "tcp_config.hh"#include <random>#include <cassert>using namespace std;TCPSender::TCPSender(const size_t capacity, const uint16_t retx_timeout, const std::optional<WrappingInt32> fixed_isn): _isn(fixed_isn.value_or(WrappingInt32{random_device()()})), _initial_retransmission_timeout{retx_timeout}, _stream(capacity) { }uint64_t TCPSender::bytes_in_flight() const {return _next_seqno - _expect_ack;}void TCPSender::fill_window() {if (!_syn_sent) {TCPSegment seg;seg.header().syn = true;seg.header().seqno = wrap(0, _isn);_segments_out.push(seg);_seg_not_ack.push(seg);_next_seqno = 1;_retrans_timer = _tick + _initial_retransmission_timeout;_syn_sent = true;}uint64_t remain = _window_size - bytes_in_flight();bool send = false;if (_expect_ack != 0) {// SYN receivedwhile (remain > 0 && _stream.buffer_size() > 0) {// send segmentuint64_t send_bytes = min(remain, TCPConfig::MAX_PAYLOAD_SIZE);string payload = _stream.read(send_bytes);TCPSegment seg;seg.header().seqno = wrap(_next_seqno, _isn);seg.payload() = move(payload);_next_seqno += seg.length_in_sequence_space();remain = _window_size - bytes_in_flight();if (_stream.eof() && remain > 0 && !_fin_sent) {seg.header().fin = true;_next_seqno += 1;_fin_sent = true;}_segments_out.push(seg);_seg_not_ack.push(seg);send = true;}}if (_stream.eof() && remain > 0 && !_fin_sent) {// send FINTCPSegment seg;seg.header().fin = true;seg.header().seqno = wrap(_next_seqno, _isn);_segments_out.push(seg);_seg_not_ack.push(seg);_next_seqno += 1;_fin_sent = true;send = true;}if (send && _retrans_timer == 0) {// open timer_retrans_timer = _tick + _initial_retransmission_timeout;_consecutive_retransmissions = 0;_rto_back_off = 0;}}void TCPSender::ack_received(const WrappingInt32 ackno, const uint16_t window_size) {_window_size = window_size;_do_back_off = 1;if (_window_size == 0) {_window_size = 1;_do_back_off = 0;}uint64_t ack = unwrap(ackno, _isn, _expect_ack);if (ack <= _next_seqno && ack > _expect_ack) {if (ack == _expect_ack) {_same_ack++;} else {_same_ack = 0;}_expect_ack = ack;if (bytes_in_flight() == 0) {// close timer_retrans_timer = 0;_consecutive_retransmissions = 0;_rto_back_off = 0;} else {// reopen timer_retrans_timer = _tick + _initial_retransmission_timeout;_consecutive_retransmissions = 0;_rto_back_off = 0;}}// remove all acked packetswhile (!_seg_not_ack.empty()) {TCPSegment seg = _seg_not_ack.front();if (seg.length_in_sequence_space() + unwrap(seg.header().seqno, _isn, _expect_ack) <= _expect_ack) {_seg_not_ack.pop();} else {break;}}// faster retransmitif (_same_ack == 3 && !_seg_not_ack.empty()) {// cout << "!! FASTER RETRANSMIT" << endl;_same_ack = 0;TCPSegment seg = _seg_not_ack.front();_segments_out.push(seg);_consecutive_retransmissions += 1;_rto_back_off += _do_back_off;_retrans_timer += _initial_retransmission_timeout << _rto_back_off;}}void TCPSender::tick(const size_t ms_since_last_tick) {_tick += ms_since_last_tick;if (!_seg_not_ack.empty() && _tick >= _retrans_timer) {// retransmit the first packet// cout << "retransmit" << endl;TCPSegment seg = _seg_not_ack.front();_segments_out.push(seg);_consecutive_retransmissions += 1;_rto_back_off += _do_back_off;_retrans_timer = _tick + (_initial_retransmission_timeout << _rto_back_off);}}unsigned int TCPSender::consecutive_retransmissions() const { return _consecutive_retransmissions; }void TCPSender::send_empty_segment() {TCPSegment seg;seg.header().seqno = wrap(_next_seqno, _isn);_segments_out.push(seg);}
Lab4 TCP连接
这个实验就是将之前的发送端和接收端组合起来，成为一个完整的TCP peer。
主要的工作就是将发送的数据从发送端的队列中取出，再放到发送队列中去；发送
ack
包进行确认；对
RST
进行处理；对连接的关闭和TIME_WAIT状态进行处理。
当调用
connect
函数时，就可以调用
fill_window
生成
SYN
包，然后发送出去。
当收到一个包之后，就将对于信息交给发送端和接收端进行处理，然后进行ACK，当发送队列有包时直接附带ACK就行了，如果没有就要生成一个空包进行ACK，注意当接收的包只是一个ACK包而没有任何数据的话就不要进行ACK。当接收端收到所有数据以及
FIN
包之后就会关闭接收端的输入流。
当连接的输入流关闭，就可以调用发送端的
end_input
和
fill_window
来生成
FIN
包并发送。
连接的关闭
TCP连接的关闭分为两种情况，主动关闭和被动关闭。
当发送流先结束时，就要进行主动关闭，发送
FIN
进入FIN_WAIT_1状态，收到
ACK
后进入FIN_WAIT_2状态，当收到对方的
FIN
包并进行ACK之后，就进入TIME_WAIT状态，在TIME_WAIT状态下要等待2MSL（Linux中一般为60s）才能释放连接。
TIME_WAIT状态的目的就是保证最后一个ACK包被对方接收到，因为不会对ACK进行ACK，就只能使用这种方式。如果ACK没有被对方接收到，那么对方就会重发
FIN
包，这时候就可以再次进行ACK。如果直接释放而不进行TIME_WAIT的话，那么下一个使用该端口的连接就可能会收到上一个连接重传的
FIN
包，从而导致混乱。
当对方先关闭时就是被动关闭，当收到
FIN
并ACK后，就进入CLOSE_WAIT状态。等到发送流结束后，发送
FIN
进入LAST-ACK状态，收到对方ACK后就可以关闭连接了，当被动关闭时，就不需要TIME_WAIT了。
#include "tcp_connection.hh"#include <iostream>using namespace std;void TCPConnection::send_all_segments() {if (_closed) return;while (!_sender.segments_out().empty()) {TCPSegment& seg = _sender.segments_out().front();if (_receiver.ackno().has_value()) {seg.header().ack = true;seg.header().ackno = _receiver.ackno().value();}size_t max_win = numeric_limits<uint16_t>().max();seg.header().win = min(_receiver.window_size(), max_win);_segments_out.push(seg);_sender.segments_out().pop();}if (_sender.stream_in().eof() && _sender.bytes_in_flight() == 0 && _receiver.stream_out().input_ended()) {if (_linger_after_streams_finish) {_time_wait = true;}}}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 _ticks - _last_ack_time;}void TCPConnection::segment_received(const TCPSegment &seg) {if (!_syn_sent && !seg.header().syn) return;if (seg.header().rst) {// reset connection_sender.stream_in().set_error();_receiver.stream_out().set_error();_linger_after_streams_finish = false;}_last_ack_time = _ticks;_receiver.segment_received(seg);_sender.ack_received(seg.header().ackno, seg.header().win);_sender.fill_window();_syn_sent = true;if (_receiver.stream_out().input_ended() && !_sender.stream_in().eof()) {// passive close_linger_after_streams_finish = false;}if (!_receiver.ackno().has_value()) {return; // no need for ack}if (_sender.segments_out().empty()) {// generate an empty segment to ackif (_receiver.stream_out().input_ended() && !seg.header().fin) {// no need to ack, server closed and seg not fin} else if (seg.length_in_sequence_space() == 0) {// no need to ack the empty-ack} else {_sender.send_empty_segment();}}// send with acksend_all_segments();}bool TCPConnection::active() const {if (_sender.stream_in().error() && _receiver.stream_out().error()) return false;return !(_sender.stream_in().eof() && _sender.bytes_in_flight() == 0 && _receiver.stream_out().input_ended()) || _time_wait;}size_t TCPConnection::write(const string &data) {size_t wrote = _sender.stream_in().write(data);_sender.fill_window();send_all_segments();return wrote;}void TCPConnection::tick(const size_t ms_since_last_tick) {_ticks += ms_since_last_tick;_sender.tick(ms_since_last_tick);if (_time_wait && _ticks >= _last_ack_time + _cfg.rt_timeout * 10) {// closed_time_wait = false;_closed = true;}if (_sender.consecutive_retransmissions() > _cfg.MAX_RETX_ATTEMPTS) {// RST_sender.stream_in().set_error();_receiver.stream_out().set_error();_linger_after_streams_finish = false;while (!_sender.segments_out().empty()) {// pop all segments_sender.segments_out().pop();}_sender.send_empty_segment();TCPSegment& seg = _sender.segments_out().front();seg.header().rst = true;}send_all_segments();}void TCPConnection::end_input_stream() {_sender.stream_in().end_input();_sender.fill_window();send_all_segments();}void TCPConnection::connect() {// send SYNif (!_syn_sent) {_sender.fill_window();_syn_sent = true;TCPSegment& seg = _sender.segments_out().front();size_t max_win = numeric_limits<uint16_t>().max();seg.header().win = min(_receiver.window_size(), max_win);_segments_out.push(seg);_sender.segments_out().pop();}}TCPConnection::~TCPConnection() {try {if (active()) {cerr << "Warning: Unclean shutdown of TCPConnection\\n";_sender.stream_in().set_error();_receiver.stream_out().set_error();_linger_after_streams_finish = false;while (!_sender.segments_out().empty()) {// pop all segments_sender.segments_out().pop();}_sender.send_empty_segment();TCPSegment& seg = _sender.segments_out().front();seg.header().rst = true;send_all_segments();}} catch (const exception &e) {std::cerr << "Exception destructing TCP FSM: " << e.what() << std::endl;}}
测试
至此整个简单的TCP协议就实现完了，使用这个TCP协议修改Lab0中的webget，然后就可以对网站进行访问了。使用抓包软件就可以看到完整的连接建立、数据发送、连接关闭的过程了。但这里有一个问题不知道是为什么，使用webget访问
cs144.keithw.org
，
bilibili.com
都能正常访问，但是访问
www.baidu.com
的时候，就会丢失连接建立之后的一个包，抓包看根本没收到那个包（tcp previous segment not captured），不知道是我的代码有问题还是百度的服务器使用了什么特殊的策略。
