CS144 Lab Checkpoint 2: the TCP receiver

Overview

TCPReceiver 从对等的sender接收消息，使用 receive() 方法，然后调用 Reassembler() 方法，后者写入 ByteStream 中 然后应用程序从 ByteSteam 中读取。
同时，TCPReceiver 还会通过 send() 方法给sender发送消息，告诉sender：

1. 确认号ackno ，这是接收方需要从发送方接受的第一个字节；
2. ByteStream的可用容量，也就是窗口大小 window size。

本实验最难的部分在于如何思考 TCP 将如何表示每个字节在流中的位置，也称为“序列号”。

Translating between 64-bit indexes and 32-bit seqnos

TCP协议中包含三种序号分别是：seqno、absolute seqno、stream index，其转换关系如图所:示：


可以看到，absolute Seqno和stream index只是在开头和结尾的SYN和FIN有差异，且都是64位，所以第一个问题是如何实现seqno和absolute seqno之间的转换。

absolute seqno 到 seqno

首先，seqno有一个随机数ISN，作为起始序列编码，而absolute seqno从0开始，可以得到两者之间的转换关系为：
$$seqno\equiv absoluteseqno+ISN(mod{2}^{32})$$
所以wrap函数实现如下：

Wrap32 Wrap32::wrap( uint64_t n, Wrap32 zero_point )
{
  // Your code here.
  return Wrap32 { zero_point.raw_value_ + static_cast<uint32_t>(n)};
}

seqno 到 absolute seqno

和上面的类似，现在反向推倒absolute seqno
$$absoluteseqno\equiv seqno-ISN(mod{2}^{32})$$

这样就可以得到absolute seqno的低32位，然后在函数中给出了checkpoint，现在的目标是找出离checkpoint最近的absolute seqno的高32位，这个值一定满足：

$$checkpoint-2^{31}\le absoluteseqno\le checkpoint+2^{31}$$
之所以满足这个条件，是因为如果不在这个范围内，则可以通过+或者- $2^{32}$来更接近checkpoint。

所以高32位的计算方法是通过给checkpoint + 或者 - $2^{31}$再取出其中的高32位，然后找出两个中接近checkpoint的值，即为所求。

uint64_t Wrap32::unwrap( Wrap32 zero_point, uint64_t checkpoint ) const
{
  // Your code here.
  uint32_t low32_bits = this->raw_value_ - zero_point.raw_value_; 
  uint64_t high32_bits1 = (checkpoint + (1 << 31)) & 0xFFFFFFFF00000000;
  uint64_t high32_bits2 = (checkpoint - (1 << 31)) & 0xFFFFFFFF00000000;
  uint64_t aseqno1 = low32_bits | high32_bits1;
  uint64_t aseqno2 = low32_bits | high32_bits2;
  if(max(aseqno1,checkpoint) - min(aseqno1,checkpoint) < max(aseqno2,checkpoint) - min(aseqno2,checkpoint))
    return aseqno1;
  return aseqno2;
}

Implementing the TCP receiver

实验的剩余部分，要求我们实现TCPReceiver，实现：

1. 从sender接收消息，并调用Reassembler对ByteStream重组；
2. 将包含确认号ackno和窗口大小发送给sender.

首先添加几个成员变量和修改构造函数

class TCPReceiver
{
public:
  // Construct with given Reassembler
  explicit TCPReceiver( Reassembler&& reassembler ) : 
  reassembler_( std::move( reassembler ) ) ,
  isn(-1), 
  open(false),
  _capacity (std::min((size_t)reassembler_.writer().available_capacity(),(size_t)UINT16_MAX))
  {}

  /*
   * The TCPReceiver receives TCPSenderMessages, inserting their payload into the Reassembler
   * at the correct stream index.
   */
  void receive( TCPSenderMessage message );

  // The TCPReceiver sends TCPReceiverMessages to the peer's TCPSender.
  TCPReceiverMessage send() const;

  // Access the output (only Reader is accessible non-const)
  const Reassembler& reassembler() const { return reassembler_; }
  Reader& reader() { return reassembler_.reader(); }
  const Reader& reader() const { return reassembler_.reader(); }
  const Writer& writer() const { return reassembler_.writer(); }

private:
  Reassembler reassembler_;
  Wrap32 isn;                   // zero_point
  bool open;                    // ISN
  size_t _capacity;             // 容量，最大为UINT16_MAX
};

然后修改成员函数，其中有三个特殊的标志位，分别是：
SYN：接收到SYN才开始传输报文，此前的全部丢弃
FIN：接收到FIN就结束传输报文
RST：接收到RST就将ByteSteam置错，并且停止传输。

所以在receive函数里：

1. 先判断RST信号，如果出现RST，就给reader()置错；
2. 判断是否有SYN，之前没有就丢弃，否则就记录ISN；
3. 根据已经push进重组器的字节数为checkpoint，以ISN和checkpoint调用前面写的unwarp函数计算绝对序列号；
4. 如上一问所述，在stream中索引并不是完全和64位的绝对索引相同的，stream中的索引不包括SYN和FIN，所以要判断一下。

值得注意的是，我第一次在实现这个转换的时候以为SYN始终都不会进入stream中，实际上不是这样的，第一个SYN的序列号作为ISN是会进入stream中，而其后的各报文的SYN端不会进入stream，FIN在后文也需要判断，所以我们需要根据报文的SYN标志对stream index和absolute seqno之间进行转换。

接下来在send函数，其中RST根据reader()是否出现error，而windows size则根据总大小减去重组器中已经存储的大小，接下来主要介绍ackno的计算：

1. 如果还没有建立连接，即还没有SYN，就直接回复空(nullopt)；
2. 否则，就根据已经push进重组器的字节 + 1(确认号是下一个需要的序列号，所以加1)，这里又需要判断是否是FIN，如果是，因为FIN也要占一个绝对序列号，所以还要再加1；
3. 调用wrap转换成seqno，并返回message。

void TCPReceiver::receive( TCPSenderMessage message )
{
  // Your code here.
  if ( message.RST ) {
    reassembler_.reader().set_error();
    return;
  } 
  if ( open == false ) {
    if ( !message.SYN )
      return;
    else {
      open = true;
      isn = message.seqno;
    }
  }
  Wrap32 seqno = message.seqno;
  uint64_t checkpoint = reassembler_.writer().bytes_pushed();
  uint64_t ab_seqno = seqno.unwrap( isn, checkpoint );
  uint64_t index = message.SYN ? ab_seqno : ab_seqno - 1;
  reassembler_.insert( index, message.payload, message.FIN );
}

TCPReceiverMessage TCPReceiver::send() const
{
  // Your code here.
  TCPReceiverMessage message;
  if ( !open ) {
    message.ackno = nullopt;
  } else {
    uint64_t ab_seqno = reassembler_.writer().bytes_pushed() + 1 + ( reassembler_.writer().is_closed() ? 1 : 0 );
    message.ackno = Wrap32::wrap( ab_seqno, isn );
  }
  message.RST = reassembler_.reader().has_error();
  message.window_size = _capacity - reassembler_.reader().bytes_buffered();
  return message;
}

实验结果