前言

这一个Lab主要是实现一个TCP receiver的字符串接收重组部分。

其他笔记

Lab 0: networking warmup
Lab 1: stitching substrings into a byte stream

相关链接

课程主页
lab 1

1. Getting started

CS144这个Lab本来是自上而下从头到尾代码复用的，这导致我开局顺手就把远程库给干掉了，结果没想到到Lab居然叫我merge，那就重新弄一下吧：

git remote add base  git@github.com:CS144/minnow.git
git fetch base

打开VS的分支管理器，可以看到这个远程库里面应该是check1 - 6挨个发布的，看提交记录这个应该是当时上课的时候才一步步弄的仓库，理论上来讲直接合并check6的分支就可以了

右键合并到main，接受冲突，然后提交上传，就可以继续撸码了

2. Putting substrings in sequence

2.1 需求分析

Lab1和2做的事情是写一个TCP接收器，大概工作就如同Lab0的末尾写的那样，写一个类去处理字节流，不过这个数据将不用内存传输上，而是通过网络传输。

由于网络传输的不确定性以及成本问题，在传输数据时我们都是将串切成一段一段的，比如这里提到的每个$shortsegments$不超过1460个$byte$，又是考虑到网络传输的不确定性以及TCP的性质，这些字段通常会出现乱序、丢失的情况，而我们需要保证能够重排回最初的字符串。

具体到本Lab，我们要实现一个叫$Reassembler$的东西，这是用来在接收端接受上面说的那一堆字段的，而每一个$Byte$（而非$segment$）都有一个对应的$index$。文档约束了这个类的两个必要接口，insert将一个data写入output，写入的位置自first_index起，它还用了一个bool变量去标识当前段是否为最后一个段；而bytes_pending则仅仅返回一下存在$Reassembler$中的字节，但是哪些字节存在这里面呢？我们知道单纯网络传输不保证顺序，有可能提前接收到了后面的字段，就只好暂存在$Reassembler$，等它前面的字段写完了再存进去。

然后进一步展示了这个类应当做的工作的一些细节。

首先，我们应当知道流的下一个待接收的字节（的$index$），正如上面说的那样，类内部还有一大堆字段嗷嗷待哺等着进流；

然后，我们需要处理提前到达的暂时没被推进流的串；

而对于哪些超出流接收能力的字节，应当直接扔掉；

然后这个图演示了总共存在三类byte：未进流暂存的、已进流缓存的、已被read弹出的，第三个我们这个Lab应该不用考虑。绿色内存以及类内暂存的一整块空间共同组成了capacity，可知我们的红色内存的最大值只能为capacity - buffered，超过这个的字节就得丢掉了。在实现上，这个值就是上一个Lab实现的available_capacity。

2.2 注意事项

然后是一些FAQ，我们可以提炼出这些信息：

1. 流的$index$自0始；
2. 我们会同Lab0一样有个跑分环境；
3. 每个字段都是来自字符串的准确切片，不用做异常处理；
4. 鼓励用标准库、数据结构；
5. 尽可能早地将字节推进流，免得一直存着；
6. insert接收到的data字符串是有可能与其他字符串重叠的；
7. 可以往类里面加私有成员（这不是废话吗）；
8. 对于每一个字节，类内部应当只存储它的一份副本，不要存重叠的字符串；
9. 运行./scripts/lines-of-code以计算实现代码行数，这个值一般在50-60。

2.3 代码实现

下面给出我的代码实现，里面有很多注释，就不挨着说了，不过注意我这里用到了std::ranges和std::views，因此你的编译器要在gcc13.1及以上。稍微需要探讨一下的是用什么数据结构来当做$buffer$，这个数据结构需要满足什么样的需求呢？首先它会有频繁的任意处插入，然后它需要去频繁遍历比较大小查找，给出几个常见的数据结构的复杂度：

	插入	头删	删除k个	查找
list	$O(1)$	$O(1)$	$O(k)$	$O(n)$
vector	$O(n)$	$O(n)$	$O(n)$	$O(\log n)$
deque	$O(n)$	$O(1)$	$O(n)$	$O(\log n)$
map	$O(\log n)$	$O(\log n)$	$O(\log n+k)$	$O(n)$

可以看到，综合考虑下list基本是最优秀的容器了。其中虽然map红黑树自带的查找是$O(\log n)$，但是我们的查找是要查找两个端点，如果将左右区间的pair作为key的话就不能用它内置的二分查找算法——它无法传递自定义比较谓词，而使用<algorithm>中的二分算法的话又因为它的迭代器不满足随即迭代器的条件，意味着只能$O(n)$查找。综合来看，我们维护一个有序链表是最优的。

此外在向$buffer$暂存的过程中，可能涉及到区间合并的问题，可以参考LeetCode 57. 插入区间
，给出这道题我的实现，本Lab直接套用即可：

// https://leetcode.cn/u/zi-bu-yu-mf/
class Solution {
public:
    vector<vector<int>> insert(vector<vector<int>>& intervals, vector<int>& newInterval) {
        auto beg = intervals.begin(), end = intervals.end();
        int& a = newInterval[0], & b = newInterval[1];
        auto l = lower_bound(beg, end, vector{ a, a });
        auto r = upper_bound(  l, end, vector{ b, b});
        if (l != end) a = min(a, l[ 0][0]);
        if (r != beg) b = max(b, r[-1][1]);
        intervals.insert(intervals.erase(l, r), newInterval);
        return intervals;
    }
};

/*****************************************************************//**
 * \file   reassembler.hh
 * \brief  实现一个 Reassembler 类, 用于将乱序的字符串重新组装成有序的
 *         字符串，并推入字节流.
 * 
 * \author JMC
 * \date   August 2023
 *********************************************************************/
#pragma once

#include "byte_stream.hh"

#include <string>
#include <list>
#include <tuple>

class Reassembler
{
	bool had_last_ {};	// 是否已经插入了最后一个字符串
	uint64_t next_index_ {};	// 下一个要写入的字节的索引
	uint64_t buffer_size_ {};	// buffer_中的字节数
	std::list<std::tuple<uint64_t, uint64_t, std::string>> buffer_ {};

	/**
	 * \breif 将data推入output流.
	 */
	void push_to_output(std::string data, Writer& output);

	/**
	 * \brief 将data推入buffer暂存区.
	 * \param first_index data的第一个字节的索引
	 * \param last_index  data的最后一个字节的索引
	 * \param data        待推入的字符串, 下标为[first_index, last_index]闭区间
	 */
	void buffer_push( uint64_t first_index, uint64_t last_index, std::string data );

	/**
	 * 尝试将buffer中的串推入output流.
	 */
	void buffer_pop(Writer& output);

public:
  /*
   * Insert a new substring to be reassembled into a ByteStream.
   *   `first_index`: the index of the first byte of the substring
   *   `data`: the substring itself
   *   `is_last_substring`: this substring represents the end of the stream
   *   `output`: a mutable reference to the Writer
   *
   * The Reassembler's job is to reassemble the indexed substrings (possibly out-of-order
   * and possibly overlapping) back into the original ByteStream. As soon as the Reassembler
   * learns the next byte in the stream, it should write it to the output.
   *
   * If the Reassembler learns about bytes that fit within the stream's available capacity
   * but can't yet be written (because earlier bytes remain unknown), it should store them
   * internally until the gaps are filled in.
   *
   * The Reassembler should discard any bytes that lie beyond the stream's available capacity
   * (i.e., bytes that couldn't be written even if earlier gaps get filled in).
   *
   * The Reassembler should close the stream after writing the last byte.
   */
  void insert( uint64_t first_index, std::string data, bool is_last_substring, Writer& output );

  // How many bytes are stored in the Reassembler itself?
  uint64_t bytes_pending() const;
};

/*****************************************************************//**
 * \file   reassembler.cc
 * \brief  实现一个 Reassembler 类, 用于将乱序的字符串重新组装成有序的
 *         字符串，并推入字节流.
 * \author JMC
 * \date   August 2023
 *********************************************************************/
#include "reassembler.hh"

#include <ranges>
#include <algorithm>

using namespace std;
void Reassembler::push_to_output( std::string data, Writer& output ) {
  next_index_ += data.size();
  output.push( move( data ) );
}

void Reassembler::buffer_push( uint64_t first_index, uint64_t last_index, std::string data )
{
  // 合并区间
  auto l = first_index, r = last_index;
  auto beg = buffer_.begin(), end = buffer_.end();
  auto lef = lower_bound( beg, end, l, []( auto& a, auto& b ) { return get<1>( a ) < b; } );
  auto rig = upper_bound( lef, end, r, []( auto& b, auto& a ) { return get<0>( a ) > b; } );
  if (lef != end) l = min( l, get<0>( *lef ) );
  if (rig != beg) r = max( r, get<1>( *prev( rig ) ) );
  
  // 当data已在buffer_中时，直接返回
  if ( lef != end && get<0>( *lef ) == l && get<1>( *lef ) == r ) {
    return;
  }

  buffer_size_ += 1 + r - l;
  if ( data.size() == r - l + 1 && lef == rig ) { // 当buffer_中没有data重叠的部分
	buffer_.emplace( rig, l, r, move( data ) );
	return;
  }
  string s( 1 + r - l, 0 );

  for ( auto&& it : views::iota( lef, rig ) ) {
	auto& [a, b, c] = *it;
	buffer_size_ -= c.size();
    ranges::copy(c, s.begin() + a - l);
  }
  ranges::copy(data, s.begin() + first_index - l);
  buffer_.emplace( buffer_.erase( lef, rig ), l, r, move( s ) );
}

void Reassembler::buffer_pop( Writer& output ) {
  while ( !buffer_.empty() && get<0>( buffer_.front() ) == next_index_ ) {
    auto& [a, b, c] = buffer_.front();
    buffer_size_ -= c.size();
    push_to_output( move( c ), output ); 
    buffer_.pop_front();
  }

  if ( had_last_ && buffer_.empty() ) {
    output.close();
  }
}

void Reassembler::insert( uint64_t first_index, string data, bool is_last_substring, Writer& output )
{
  if ( data.empty() ) {
    if ( is_last_substring ) {
      output.close();
    }
    return;
  }
  auto end_index = first_index + data.size();                  // data: [first_index, end_index)
  auto last_index = next_index_ + output.available_capacity(); // 可用范围: [next_index_, last_index)
  if ( end_index < next_index_ || first_index >= last_index ) {
    return; // 不在可用范围内, 直接返回
  }

  // 调整data的范围
  if ( last_index < end_index ) {
    end_index = last_index;
    data.resize( end_index - first_index );
    is_last_substring = false;
  }
  if ( first_index < next_index_ ) {
    data = data.substr( next_index_ - first_index );
    first_index = next_index_;
  }

  // 若data可以直接写入output, 则直接写入
  if ( first_index == next_index_ && ( buffer_.empty() || end_index < get<1>( buffer_.front() ) + 2 ) ) {
    if ( buffer_.size() ) { // 若重叠, 则调整data的范围
      data.resize( min( end_index, get<0>( buffer_.front() ) ) - first_index );
    }
    push_to_output( move( data ), output );
  } else { // 否则, 将data插入buffer_
    buffer_push( first_index, end_index - 1, data );
  }
  had_last_ |= is_last_substring;
  
  // 尝试将buffer_中的数据写入output
  buffer_pop(output);
}

uint64_t Reassembler::bytes_pending() const
{
  return buffer_size_;
}

3. 测试与优化

写这个Lab我跑测试遇到了挺多Bug的，心态比较麻，看一下我的git记录，大体代码从五点到六点半就写完了，然后又调了三四个小时的Bug。。。：

这个Lab最主要的优化点还是要记得使用std::move转发字符串，然后这是我不管字符串能不能直接进流，都先进buffer后进流的速度

这是我对这种情况的特殊处理后的速度，可以看到来到了9.5Gbit/s+8Gbit/s这个速度，按理说我的入buffer操作对于这种情况仅仅是多跑了一遍链表而已，没想到速度也差不少：

最后看看代码行数，运行./scripts/lines-of-code，如果报错则需要安装一个工具sudo apt-get install sloccount：

它说基础代码是22行，我们就写了77行。我看了一下这个统计行数其实就是去除了注释和空行，他说50-60行是正常的，我这里写得比较详细，想压行也不是压不了，也差不多。