前言
上一篇博客中我们完成了 Lab0,使用双端队列实现了一个字节流类 ByteStream
,可以向字节流中写入数据并按写入顺序读出数据。由于网络环境的变化,发送端滑动窗口内的数据包到达接收端时可能失序,所以接收端收到数据之后不能直接写入 ByteStream
中,而是应该缓存下来并按照序号重组成正确的数据。这篇博客所介绍的 Lab1 将实现一个字节流重组器 StreamReassambler
来完成上述任务。
实验要求
接收方的数据情况如下图所示,蓝色部分表示已消费的数据,绿色表示已正确重组但是还没消费的数据,红色则是失序到达且还没重组的数据:
由于接收端缓冲区大小 capacity
有限,超出容量的数据(first unacceptable 之后的数据)将被丢弃,这些被丢弃的数据包将起到流量控制的作用,可以限制发送端滑动窗口的大小。
流重组器的接口如下所示:
复制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);
//! 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
size_t unassembled_bytes() const;
//! Is the internal state empty (other than the output stream)?
bool empty() const;
其中最重要的函数就是 StreamReassambler::push_substring()
,接收方收到数据之后就会调用此函数将数据保存起来。此函数接受三个参数:
data
: 接收到的数据index
: 数据的第一个字节的索引,由于原始数据可能很大,超过了 TCPSegment 的容量,所以会将原始数据切分成多个片段,每个片段的第一个字节的索引就是index
,最小值为 0。请注意,这个index
是流索引,不是报文段包头上的那个序列号,二者的关系将在 Lab2 中指出。eof
:是不是最后一个数据包
三个参数中,最耐人寻味的就是 index
参数,如果只是单纯的失序到达,数据之间没有发生重叠,Lab1 就比较好做了,但是实验指导书中明确指出
May substrings overlap? Yes
这就比较难搞了,因为重叠分成两种:
-
前面一部分与已重组的数据发生重叠
-
前面不与已重组的数据发生重叠
实际上由于 data
的末尾可能超出 first unacceptable
,需要对超出部分进行截断,这可能导致 eof
标志失效,但是问题不大,发送方之后会重新发送这个数据包。
代码实现
为了处理上述重叠情况,需要一个 _next_index
成员代表 first unassembled
索引,一个 _unassembles
双端队列代表 first unassembled
到 first unacceptable
之间的数据,由于里面可能只有一部分数据是有效的,所以用一个遮罩 _unassembled_mask
指出哪些数据是有效但是还没重组的。
复制class StreamReassembler {
private:
ByteStream _output; //!< The reassembled in-order byte stream
size_t _capacity; //!< The maximum number of bytes
std::deque<char> _unassembles{};
std::deque<bool> _unassemble_mask{};
size_t _unassambled_bytes{0};
uint64_t _next_index{0};
bool _is_eof{false};
/** @brief 将数据写入未重组队列中
* @param data 将被写入的字符串
* @param dstart 字符串开始写入的位置
* @param len 写入的长度
* @param astart 队列中开始写入的位置
*/
void write_unassamble(const std::string &data, size_t dstart, size_t len, size_t astart);
/** @brief 重组数据
*/
void assemble();
public:
StreamReassembler(const size_t capacity);
//! \brief Receives a substring and writes any newly contiguous bytes into 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
size_t unassembled_bytes() const;
bool empty() const;
};
收到数据时,先将不重叠的数据写入 _unassembles
队列中,之后调用 StreamReassabler::assemble()
函数重组队列中的连续数据,并更新 _next_index
:
复制StreamReassembler::StreamReassembler(const size_t capacity)
: _output(capacity), _capacity(capacity), _unassembles(capacity, '\0'), _unassemble_mask(capacity, false) {}
//! \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) {
if (index > _next_index + _capacity)
return;
if (eof)
_is_eof = true;
if (_is_eof && empty() && data.empty()) {
_output.end_input();
return;
}
auto end_index = data.size() + index;
// 新数据在后面
if (index >= _next_index) {
auto astart = index - _next_index;
auto len = min(_output.remaining_capacity() - astart, data.size());
if (len < data.size())
_is_eof = false;
write_unassamble(data, 0, len, astart);
}
// 新数据与已重组的数据部分重叠
else if (end_index > _next_index) {
auto dstart = _next_index - index;
auto len = min(_output.remaining_capacity(), data.size() - dstart);
if (len < data.size() - dstart)
_is_eof = false;
write_unassamble(data, dstart, len, 0);
}
// 最后合并数据
assemble();
if (_is_eof && empty())
_output.end_input();
}
void StreamReassembler::write_unassamble(const string &data, size_t dstart, size_t len, size_t astart) {
for (size_t i = 0; i < len; ++i) {
if (_unassemble_mask[i + astart])
continue;
_unassembles[i + astart] = data[dstart + i];
_unassemble_mask[i + astart] = true;
_unassambled_bytes++;
}
}
void StreamReassembler::assemble() {
string s;
while (_unassemble_mask.front()) {
s.push_back(_unassembles.front());
_unassembles.pop_front();
_unassemble_mask.pop_front();
_unassembles.push_back('\0');
_unassemble_mask.push_back(false);
}
if (s.empty())
return;
_output.write(s);
_next_index += s.size();
_unassambled_bytes -= s.size();
}
size_t StreamReassembler::unassembled_bytes() const { return _unassambled_bytes; }
bool StreamReassembler::empty() const { return _unassambled_bytes == 0; }
在命令行中输入:
复制cd build
make -j8
make check_lab1
可以看到测试用例也全部通过了:
调试代码
由于使用代码编辑器的是 VSCode,所以这里给出在 VSCode 中调试项目代码的方式。
tasks.json
首先在项目目录下创建 .vscode
文件夹,并新建一个 tasks.json
文件,在里面写入下述内容:
复制{
"tasks": [
{
"type": "shell",
"label": "cmake",
"command": "cd build && cmake .. -DCMAKE_BUILD_TYPE=Debug",
"detail": "CMake 生成 Makefile",
"args": [],
"problemMatcher": "$gcc"
},
{
"type": "shell",
"label": "build",
"command": "cd build && make -j8",
"detail": "编译项目",
"args": [],
"problemMatcher": "$gcc"
},
],
"version": "2.0.0"
}
这里主要配置了两个任务,一个调用 CMake 生成 Makefile,一个编译 Makefile。在 VSCode 中按下 Alt + T + R,就能在任务列表中看到这两个任务,点击之后就能执行。
launch.json
在 .vscode
文件夹中新建 launch.json
,并写入下述内容:
复制{
// Use IntelliSense to learn about possible attributes.
// Hover to view descriptions of existing attributes.
// For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
"version": "0.2.0",
"configurations": [
{
"name": "debug lab test",
"type": "cppdbg",
"request": "launch",
"program": "${workspaceFolder}/build/tests/${fileBasenameNoExtension}",
"args": [],
"stopAtEntry": false,
"cwd": "${workspaceFolder}",
"environment": [],
"externalConsole": false,
"MIMode": "gdb",
"setupCommands": [
{
"description": "Enable pretty-printing for gdb",
"text": "-enable-pretty-printing",
"ignoreFailures": true
}
],
"miDebuggerPath": "/usr/bin/gdb"
},
{
"name": "debug current file",
"type": "cppdbg",
"request": "launch",
"program": "${fileDirname}/${fileBasenameNoExtension}",
"args": [],
"stopAtEntry": false,
"cwd": "${workspaceFolder}",
"environment": [],
"externalConsole": false,
"MIMode": "gdb",
"setupCommands": [
{
"description": "Enable pretty-printing for gdb",
"text": "-enable-pretty-printing",
"ignoreFailures": true
}
],
"preLaunchTask": "C/C++: g++ build active file",
"miDebuggerPath": "/usr/bin/gdb"
}
]
}
之后打开一个测试用例,比如 tests/fsm_stream_reassembler_seq.cc
,转到 debug
标签页,在代码中打下断点, 点击绿色按钮就能开始调试了:
调试效果如下图所示:
后记
通过这次实验,可以加深对接收端数据重组和分组序号的了解,期待后面的几个实验,以上~~