「实验记录」CS144 Lab1 StreamReassembler

发布时间:2024年01月01日

一、Motivation

我们都知道 TCP 是基于字节流的传输方式,即 Receiver 收到的数据应该和 Sender 发送的数据是一样的。这里所说的 “一样” 指的是字节的顺序

比如,Sender 发送字符串 “abc”,在不出意外的情况下,Receiver 应该收到 “abc”,而不是 “acb” 或者 “cab” 之类的乱序字符串。事与愿违,网络环境通常是不太稳定的,经常会发生乱序的情况

我们无力稳定网络中的环境,即无法阻止数据乱序到达的情况;但是,我们可以将这些乱序到达的数据,按照每个字节的 index 将其有序排列,从而实现基于字节流传输的稳定方式。这就是 Lab1: StreamReassembler 的初衷,让我们自己设计一个字节排序器,理顺 Sender 发来的数据

下面我想补充的是一个关于 TCP/IP 架构的常识。为什么会发生乱序的情况呢?因为在 TCP/IP 架构中,数据都是按照分组报文的方式发送给 Receiver 的。还接着上述的例子,Sender 发出的字符串 “abc” 被切分成三个报文,报文 A 中包含了字符 a,报文 B 中包含了字符 b,报文 C 包含了字符 c。Sender 一口气将三个报文全部送上网络。之后,每个报文具体会走什么样的路线到达 Receiver,这完全取决于当时网络环境的拥堵情况和网络节点的路由算法

如果路线不同,那就不可能保证这三个报文按序到达 Receiver。可能出现的情况即是,报文 A(字符 a)先到,报文C(字符 c)其次,报文 B(字符 b)最后才到。Receiver 按照收到报文的顺序呈现出的字符串为 “acb”,这是彻头彻尾的错误(与 Sender 本意的字符串 “abc” 相悖)

二、Solutions

针对这种乱序的情况,我们来实现自己的字节流排序器,先来看一下讲义中的这张流向图,

乍一看是比较抽象的(教这门课的老师的独特审美),但是也能够理清我们 TCP 连接中字节流的几个重要节点

第一个,first unread 节点之前的数据是已经写入 ByteStream 中的,且被用户读取了;随后的 first unread 到 first unassembled 之间的数据是 Receiver 已收到且排好序写入到 ByteStream 中的内容;first unassembled 到 first unacceptable 区间的内容是 Recevier 断断续续收到的字节。可能不是连续的,比如,已经收到了字符串 “bc”,但还没有收到字符 ‘a’,此时构不成有序的 “abc”。只有在收到字符 ‘a’ 之后,才能将 “abc” 写入 ByteStream

ByteStream 讲究的是有序字节序列,即写入 ByteStream 的最后一个字节,它之前的字节都应该是接着前一个的 index 而来的,即字符 ‘c’ 之前就应该是字符 ‘b’,字符 ‘b’ 之前的就应该是字符 ‘a’,缺一个都是不连续的

S1 - StreamReassembler的对外接口

我们先来看看 Lab1: StreamReassembler 已经声明好的接口及注解,

//! \brief A class that assembles a series of excerpts from a byte stream (possibly out of order,
//! possibly overlapping) into an in-order byte stream.
class StreamReassembler {
  private:
    // Your code here -- add private members as necessary.
    ByteStream _output;  //!< The reassembled in-order byte stream
    size_t _capacity;    //!< The maximum number of bytes
  	/* 自添加的变量 */
    std::map<size_t, char> buf_;
    bool eof_flag_;
    size_t eof_idx_;
    
  private:
  	/* 自定义的关键节点 get 方法 */
    size_t first_unread() const { return _output.bytes_read(); }
    size_t first_unassembled() const { return _output.bytes_written(); }
    size_t first_unacceptable() const { return first_unread()+_capacity; }

  public:
    //! \brief Construct a `StreamReassembler` that will store up to `capacity` bytes.
    //! \note This capacity limits both the bytes that have been reassembled,
    //! and those that have not yet been reassembled.
    StreamReassembler(const size_t capacity);

    //! \brief Receives a substring and writes any newly contiguous bytes into the stream.
    //!
    //! If accepting all the data would overflow the `capacity` of this
    //! `StreamReassembler`, then only the part of the data that fits will be
    //! accepted. If the substring is only partially accepted, then the `eof`
    //! will be disregarded.
    //!
    //! \param data the string being added
    //! \param index the index of the first byte in `data`
    //! \param eof whether or not this segment ends with the end of the stream
    void push_substring(const std::string &data, const uint64_t index, const bool eof);

    //! \name Access the reassembled byte stream
    //!@{
    const ByteStream &stream_out() const { return _output; }
    ByteStream &stream_out() { return _output; }
    //!@}

    //! The number of bytes in the substrings stored but not yet reassembled
    //!
    //! \note If the byte at a particular index has been submitted twice, it
    //! should only be counted once for the purpose of this function.
    size_t unassembled_bytes() const;

    //! \brief Is the internal state empty (other than the output stream)?
    //! \returns `true` if no substrings are waiting to be assembled
    bool empty() const;
};

不难理解,类的初始化方法是设定 StreamReassember’s ByteStream 容量的,即 _output 的大小,其定义如下,

StreamReassembler::StreamReassembler(const size_t capacity) : _output(capacity), _capacity(capacity), 
  buf_({}), eof_flag_(false), eof_idx_(0) {}

其中的 ByteStream 在 Lab0: networking warmup 中有定义,详情请移步。至于 eof_flag 和 eof_idx_ 会在之后的 push_substring() 中用到

接下来,第一个也是最重要的对外接口是 push_substring(data, index, eof),它会在 Receiver 收到新的字节时被调用。大致的功能是理顺刚收到的(可能会乱序)字节序列,并将其写入到 _output 中,以供上层读取。关于实现这个接口的具体细节我将在之后展开

之后便是上层获取 StreamReassember’s ByteStream 的方法 stream_out()。再强调一次!ByteStream 里面存放的是已有序的字节序列。未被排序的字节是没有资格写入其中的,这个是由 push_substring() 控制的

对外提供的最后两个关于容量的接口就较为简单了,unassembled_bytes() 返回的是已被接收但还未被排序的字节数,定义如下,

size_t StreamReassembler::unassembled_bytes() const { return buf_.size()-_output.bytes_written(); }

empty() 判断 StreamReassember 接收的字节序列是否已全部写入其 ByteStream 中,定义如下,

bool StreamReassembler::empty() const { return buf_.size()==_output.bytes_written(); }

其中,我们自定义的缓冲区 buf_ 的类型是 std::map<size_t, char>,作用是缓存 StreamReassember 接收的字节。利用 std::map 自带排序且去重的特性,根据每个字节的 index 对其进行排序。举个例子来说明,StreamReassember 现接收了三个字节,分别是 < 2, ‘b’ >,< 3, ‘c’ > 和 < 4, ‘d’ >,缓存进 buf_ 后应该是这样的,

假设 buf_ 的容量无穷大,目前已经收到了 3 个字符,我们根据字符的 index 将其归位,最直观的效果如上图左半边所示,和右半边我在程序中采用了 std::map 来实现缓冲排序的功能是一样的效果

我们可以看到目前字符 “bcd” 是连续的,但依然不可以写入 ByteStream;因为在此之前还缺少 index 1 的字符 ‘a’,只有等其到来,才能将字符串 “abcd” 写入 ByteStream

从这个例子中就可以理清 unassembled_bytes()empty() 的逻辑了。前者等于已收到的字节数为 3(已写入 buf_ 的字节) - 0(已写入 ByteStream 的字节数) = 3;后者判断缓冲区是否为空,就看 buf_ 中的字节是否已全部写入 ByteStream 中。很明显,例子中并未全部写入

S2 - push_substring序列写入ByteStream

理清 StreamReassember 的缓冲区 buf_ 的结构之后,我们再来考虑接收到新的字节应该如何处理?无非是先暂存新字节序列并对其进行排序,然后再将有序的部分写入到 ByteStream 中。且看定义,

//! \details This function accepts a substring (aka a segment) of bytes,
//! possibly out-of-order, from the logical stream, and assembles any newly
//! contiguous substrings and writes them into the output stream in order.
void StreamReassembler::push_substring(const string &data, const size_t index, const bool eof) {
  // DUMMY_CODE(data, index, eof);

  /* 检查若为 eof,则记录下结尾的位置 */
  if(eof) {
    eof_flag_ = true;
    eof_idx_ = index+data.length();
  }
  
  /* 暂存 up2date 的字符串 */
  for(size_t i=0; i<data.length(); i++) {
    /* 已写入 ByteStream 的字符不需要进 buf */
    if(index+i >= first_unassembled()) 
      buf_.emplace(index+i, data[i]);
  }
    
  /* 将已 in-order 的字符串写入 Bytestream 中 */
  size_t idx = first_unassembled();
  string res = "";

  for(auto ite=buf_.begin(); ite!=buf_.end(); ite++) {
    /* 找到下标 index 的字符 */
    if(ite->first < idx)  
      continue;
    
    /* 已不连续的情况下,无需继续往下寻找可以连接的字符 */
    if(ite->first>idx || idx>=first_unacceptable())
      break;
    
    if(ite->first==idx && idx<first_unacceptable()) {
      res.push_back(ite->second);
      idx++;
    }
  }
 
  /* 在 ByteStream 容量范围之内,将已 in-order 的字符串写入其中 */
  if(!res.empty()) 
    _output.write(res);  

  /* 得知已 eof 且所有已 in-order 的字符已写入 ByteStream 中,则关闭之 */
  if(eof_flag_ && first_unassembled()==eof_idx_) 
    _output.end_input();
}

当然,在流程的开始环节就需要判断 Receiver 这次接收到的字节是否是 Sender 发来的最后一个报文(有 eof 标志)

如果是最后一个报文,则需要提前做好结束标记。为什么要提前做好结束标记呢?是因为 TCP/IP 网络采用分组报文的方式发送数据,也就是说传输过程中并不能保证到达的先后顺序,Sender 发送的包含 eof 的结束报文可能因为路由选择反而第一个到达 Receiver,真正有数据的报文在之后才送达。这种情况在 TCP/IP 网络中应该是很常见的

在判断过 eof 之后,我们先一股脑地将收到的字节缓存到 buf_ 中,这里我做了一个小优化,即加上 if 判断,无视那些已经写入 ByteStream 中的字节,不将其写入 buf_ 中

下面就进入核心环节,即检出 buf_ 中已有序的字节序列。请注意,我这里说的已有序还有一个附加条件,即是从 buf_ 中的第一个字节开始就已经有序!!!

如果遍历到一个字符,它与前一个不成连续关系(index 是否加 1),那么我们就能够认定此时已经不连续了。可以跳出检出环节,将有序序列写入 ByteStream 中

最后,需要判断一下含有 eof 的字符是否已经被写入 ByteStream。如果 eof 字符都已经写入 ByteStream 了,那么就意味着在它之前的所有字符也都已经写入 ByteStream。此时,我们就可以关闭 ByteStream 了,告诉它 Sender 不会再发来数据了

三、Result

CS144 Lab 是一环套一环的结构,也就是说 Lab1: StreamReassembler 是基于 Lab0 networking warmup code 的,所以,我们在开始之前还需要透过 git fetchgit merge 拉取合并最新的代码,

$ git fetch
$ git merge origin/lab1-startercode

以上命令在 Lab1: StreamReassembler 的根目录中下发即可。编写好 code 之后,进入 bulid 目录先透过 make 命令进行编译,随后再下发 make check_lab1 检查是否通过了每个测试点,

$ make check_lab1
[100%] Testing the stream reassembler...
Test project /home/jeffrey/Documents/lab1-stitching-substrings-into-a-byte-stream/build
      Start 15: t_strm_reassem_single
 1/16 Test #15: t_strm_reassem_single ............   Passed    0.01 sec
      Start 16: t_strm_reassem_seq
 2/16 Test #16: t_strm_reassem_seq ...............   Passed    0.01 sec
      Start 17: t_strm_reassem_dup
 3/16 Test #17: t_strm_reassem_dup ...............   Passed    0.01 sec
      Start 18: t_strm_reassem_holes
 4/16 Test #18: t_strm_reassem_holes .............   Passed    0.00 sec
      Start 19: t_strm_reassem_many
 5/16 Test #19: t_strm_reassem_many ..............   Passed    1.77 sec
      Start 20: t_strm_reassem_overlapping
 6/16 Test #20: t_strm_reassem_overlapping .......   Passed    0.00 sec
      Start 21: t_strm_reassem_win
 7/16 Test #21: t_strm_reassem_win ...............   Passed    2.13 sec
      Start 22: t_byte_stream_construction
 8/16 Test #22: t_byte_stream_construction .......   Passed    0.01 sec
      Start 23: t_byte_stream_one_write
 9/16 Test #23: t_byte_stream_one_write ..........   Passed    0.01 sec
      Start 24: t_byte_stream_two_writes
10/16 Test #24: t_byte_stream_two_writes .........   Passed    0.00 sec
      Start 25: t_byte_stream_capacity
11/16 Test #25: t_byte_stream_capacity ...........   Passed    0.00 sec
      Start 26: t_byte_stream_many_writes
12/16 Test #26: t_byte_stream_many_writes ........   Passed    0.01 sec
      Start 27: t_webget
13/16 Test #27: t_webget .........................   Passed    0.46 sec
      Start 47: t_address_dt
14/16 Test #47: t_address_dt .....................   Passed    0.01 sec
      Start 48: t_parser_dt
15/16 Test #48: t_parser_dt ......................   Passed    0.00 sec
      Start 49: t_socket_dt
16/16 Test #49: t_socket_dt ......................   Passed    0.01 sec

100% tests passed, 0 tests failed out of 16

Total Test time (real) =   4.46 sec
[100%] Built target check_lab1

四、My Code

  1. CS144 Labs code 总入口

  2. CS144 Lab1 入口

五、Reference

  1. CSDN - CS144 lab1
文章来源:https://blog.csdn.net/qq_34872231/article/details/135321013
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。