CS144-check3

overview

这个lab需要我们实现TCP协议中的发送方。

接收方只需要接收就行了，但发送方要考虑的就多了…

大概来说，需要实现下面的功能：

  • 根据接收方提供的信息，去发送尽可能大的报文填满接收方的window
  • 记录发送了但还没被接收方确认的报文
  • 超时重传

implement

以下为我的实现（可能并非最终版），本质是一坨面向测试的屎山，以后报错严重有可能要重构。

members

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
class TCPSender
{
...
private:
  // Variables initialized in constructor
  ByteStream input_;
  Wrap32 isn_;
  uint64_t initial_RTO_ms_;
  uint64_t RTO_ {initial_RTO_ms_};
  uint64_t next_seqno_ {};
  bool is_retry_ {false};
  uint64_t retrans_times_ {};
  bool has_send_SYN {false};
  bool has_send_FIN {false};
  uint64_t window_size_ {1};
  std::list<TCPSenderMessage> segments {};
  struct retransmission_timer {
    bool has_on {false};
    uint64_t t {};
    uint64_t expire_time;
  } alarm;
};

push

屎山中的屎山！存在以下问题：

  • too much copy-paste：像启动时钟，几乎相同的代码在同一个函数里面出现了4次，如果需要修改会带来巨大的问题。
  • 糟糕的条件判断：因为暂时还没想到一个统一的应对绝大部分情况的方法，所以写得极其丑陋
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
void TCPSender::push( const TransmitFunction& transmit )
{
  // if window_size_ < sequence_numbers_in_flight(),
  // the following sz will be a huge number(due to uint) and cost terrible situation...
  if(window_size_ < sequence_numbers_in_flight())
    return;

  // sz : the numbers of bytes sender should try to fill in 
  uint64_t sz = window_size_ - sequence_numbers_in_flight();
  uint64_t length;

  // special case
  if(sz == 0 && window_size_ == 0){
    sz = 1;
  }
    
  // special case: SYN + FIN
  if(sz > 0 && next_seqno_ == 0 && input_.reader().is_finished() && !has_send_SYN && !has_send_FIN){
    TCPSenderMessage t;
    t.FIN = true;
    t.SYN = true;
    has_send_SYN = true;
    has_send_FIN = true;
    t.seqno = Wrap32::wrap(next_seqno_, isn_);
    if(input_.has_error())
      t.RST = true;
    transmit(t);
    segments.push_back(t);
    if(alarm.has_on == false){
      alarm.has_on = true;
      alarm.t = 0;
      alarm.expire_time = RTO_;
    }
    sz -= 2;
    next_seqno_ += 2;
    return;
  }

  // special case: only FIN
  if(sz > 0 && input_.reader().bytes_buffered() == 0 && input_.reader().is_finished() && !has_send_FIN){
    TCPSenderMessage t;
    t.FIN = true;
    has_send_FIN = true;
    t.seqno = Wrap32::wrap(next_seqno_, isn_);
    if(input_.has_error())
      t.RST = true;
    transmit(t);
    segments.push_back(t);
    if(alarm.has_on == false){
      alarm.has_on = true;
      alarm.t = 0;
      alarm.expire_time = RTO_;
    }
    sz -= 1;
    next_seqno_ += 1;
    return;
  }

  // normol case
  while(sz > 0 && input_.reader().bytes_buffered()){
    TCPSenderMessage t;
    uint64_t bufsz = input_.reader().bytes_buffered();
    uint64_t maxsz = TCPConfig::MAX_PAYLOAD_SIZE;

    
    if(input_.reader().bytes_popped() == 0 && next_seqno_ == 0){
      t.SYN = true;
      bufsz += 1;
      maxsz += 1;
    }
    

    if(input_.writer().is_closed()){
      t.FIN = true;
      bufsz += 1;
      maxsz += 1;
      if((bufsz > sz && maxsz > sz) || (maxsz < sz && maxsz < bufsz)){
        t.FIN = false;
        bufsz -= 1;
        maxsz -= 1;
      }
    }

    t.payload = input_.reader().peek();
    length = min({sz, maxsz, bufsz}) - t.SYN - t.FIN;
    t.payload = t.payload.substr(0, length);
    t.seqno = Wrap32::wrap(next_seqno_, isn_);
    if(input_.has_error())
      t.RST = true;

    input_.reader().pop(t.payload.size());
    sz -= t.sequence_length();

    if(t.sequence_length()){
      transmit(t);
      if(t.SYN)
        has_send_SYN = true;
      if(t.FIN)
        has_send_FIN = true;
      segments.push_back(t);
      if(window_size_ == 0)
        is_retry_ = true;
    }
      
    
    next_seqno_ += t.sequence_length();

    if(alarm.has_on == false){
      alarm.has_on = true;
      alarm.t = 0;
      alarm.expire_time = RTO_;
    }

  }

  if(sz > 0 && input_.reader().bytes_buffered() == 0 && input_.reader().bytes_popped() == 0 && next_seqno_ == 0 && !has_send_SYN){
    // send isn and syn with 0 payload
    TCPSenderMessage t;
    t.SYN = true;
    has_send_SYN = true;
    t.seqno = Wrap32::wrap(next_seqno_, isn_);
    if(input_.has_error())
      t.RST = true;
    transmit(t);
    segments.push_back(t);
    if(alarm.has_on == false){
      alarm.has_on = true;
      alarm.t = 0;
      alarm.expire_time = RTO_;
    }
    sz -= 1;
    next_seqno_ += 1;
  }
}

贴完代码，讲一下我认为如果需要再次实现时需要注意的边界/特殊情况：

  • 注意SYN和FIN都占window的位置
    • SYN相对好处理
    • FIN需要考虑，即便当前发送的报文以及到达stream的最后一个字，但加上FIN刚好无法放入window，这个报文不能加上FIN。如果sz是最小的，肯定不能放入FIN。如果payload的最大值是最小的，显然也不能加上FIN。
  • window的size如果小于outstanding的seqno数，就不要再push了。
  • window的size等于0：
    • 需要push当作它的size是1（但不要改变真正的window_size），因为可能需要这个机制来唤醒接收方，如果接收方window_size从0变大。
    • 但我们在这种情况发出去的报文，需要特别对待，下文在超时重传中会说。

receive

receive可以基本按照实验手册来写。基本的思路就是检查ackno，可能删除记录的未确认报文，修改window的size。当window的大小不为0时，需要修改is_retry，来使发送的报文的超时策略改变。（下文会提）。注意不能接受过大的ackno，过大的ackno可以看作干扰的信息，不符合协议。

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
void TCPSender::receive( const TCPReceiverMessage& msg )
{ 
  if(msg.RST)
    input_.set_error();
  
  if(msg.ackno->unwrap(isn_, input_.reader().bytes_popped()) > next_seqno_)
    return;
  
  for(auto it = segments.begin(); it != segments.end();){
    if(msg.ackno->unwrap(isn_, input_.reader().bytes_popped()) 
    >= (*it).seqno.unwrap(isn_, input_.reader().bytes_popped()) 
    + (*it).sequence_length())
    {
      it = segments.erase(it);
      RTO_ = initial_RTO_ms_;
      alarm.t = 0;
      alarm.expire_time = RTO_;
      retrans_times_ = 0;
    } else {
      it++;
    }
  }
  
  window_size_ = msg.window_size;
  if(window_size_)
    is_retry_ = false;
}

tick

tick需要对alarm（计时器）的值加上传入的时间。我们在这里讨论超时重传的策略：

  • 在第一次发送报文时（push里面），启动计时器（如果计时器还没启动）
  • 如果receive收到合法的ack，就删除seqno全部小于ackno的报文，即这些报文不会再超时重传（因为已经收到了）。
  • 在tick里面，当计时器触发超时，需要重传并设置超时时间、重置计时器
    • 如果window的size为0就不用倍增超时时间，因为需要持续问询接收方的空间
    • 如果不为0，需要倍增超时时间，因为超时代表了网络可能出现了拥塞
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
void TCPSender::tick( uint64_t ms_since_last_tick, const TransmitFunction& transmit )
{
  alarm.t += ms_since_last_tick;
  if(segments.empty())
    return;

  if(alarm.t >= alarm.expire_time){
    retrans_times_ += 1;
    transmit(segments.front());
    alarm.t = 0;
    if(!is_retry_)
      RTO_ *= 2;
    alarm.expire_time = RTO_;
  }  
}

make_empty_message

发送不需要超时重传的空信息。

1
2
3
4
5
6
7
8
TCPSenderMessage TCPSender::make_empty_message() const
{
  TCPSenderMessage t;
  t.seqno = Wrap32::wrap(next_seqno_, isn_);
  if(input_.has_error())
    t.RST = true;
  return t;
}

others

1
2
3
4
5
6
7
8
9
10
11
12
13
uint64_t TCPSender::sequence_numbers_in_flight() const
{
  uint64_t num = 0;
  for(auto it = segments.begin(); it != segments.end(); it++){
    num += (*it).sequence_length();
  }
  return num;
}

uint64_t TCPSender::consecutive_retransmissions() const
{
  return retrans_times_;
}

友情提醒

  • 即使我们在前几个lab中通过了全部的测试，check4还是可能会出问题：真实的网络可能包含测试用例无法覆盖的情况。我还没开始做check4，但已经准备好check1-3的bug在check4一起爆出的情况了。
  • 务必看熟实验手册后，有基本的想法再开始动手，至少保证不要理解错意思。在测试时，如果无法理解difftest的行为，说明你还是不理解或者理解错了。
  • 面向测试编程？大概率会发生的事情，实验手册很多地方没讲太明白（可能一行就带过了理想行为）。测试常常会出现你没有考虑到的情况，或者是你以为你代码可以处理但实际上不能的情况。这个需要先看明白正确的程序行为。
#Computer-Networking
CS144-check3
https://pactheman123.github.io/2024/09/24/CS144-check3/
作者
Xiaopac
发布于
2024年9月24日
许可协议