對應課程視頻: 【計算機網(wǎng)絡】 斯坦福大學CS144課程

Lab 1 對應的PDF: Lab Checkpoint 1: stitching substrings into a byte stream

實驗結(jié)構(gòu)

這幅圖完整的說明了CS144 這門實驗的結(jié)構(gòu)：

其中， ByteStream 是我們已經(jīng)在 Lab0 中實現(xiàn)完成的。

我們將在接下來的實驗中分別實現(xiàn)：

• Lab1 StreamReassembler：實現(xiàn)一個流重組器，一個將字節(jié)流的字串或者小段按照正確順序來拼接回連續(xù)字節(jié)流的模塊
• Lab2 TCPReceiver：實現(xiàn)入站字節(jié)流的TCP部分。
• Lab3 TCPSender：實現(xiàn)出站字節(jié)流的TCP部分。
• Lab4 TCPConnection: 結(jié)合之前的工作來創(chuàng)建一個有效的 TCP 實現(xiàn)。最后我們可以使用這個 TCP 實現(xiàn)來和真實世界的服務器進行通信。

該實驗引導我們以模塊化的方式構(gòu)建一個 TCP 實現(xiàn)。

流重組器在 TCP 起到了相當重要的作用。迫于網(wǎng)絡環(huán)境的限制，TCP 發(fā)送者會將數(shù)據(jù)切割成一個個小段的數(shù)據(jù)分批發(fā)送。但這就可能帶來一些新的問題：數(shù)據(jù)在網(wǎng)絡中傳輸時可能丟失、重排、多次重傳等等。而TCP接收者就必須通過流重組器，將接收到的這些重排重傳等等的數(shù)據(jù)包重新組裝成新的連續(xù)字節(jié)流。

如何調(diào)試

先 cmake && make 一個 Debug 版本的程序。

所有的評測程序位于build/tests/中，先一個個手動執(zhí)行過去。

若輸出了錯誤信息，則使用 gdb 調(diào)試一下。

StreamReassembler 實現(xiàn)

在我們所實現(xiàn)的流重組器中，有以下幾種特性：

• 接收子字符串。這些子字符串中包含了一串字節(jié)，以及該字符串在總的數(shù)據(jù)流中的第一個字節(jié)的索引。

• 流的每個字節(jié)都有自己唯一的索引，從零開始向上計數(shù)。

• StreamReassembler 中存在一個 ByteStream 用于輸出，當重組器知道了流的下一個字節(jié)，它就會將其寫入至 ByteStream中。

需要注意的是，傳入的子串中：

• 子串之間可能相互重復，存在重疊部分

  • 但假設重疊部分數(shù)據(jù)完全重復。
  • 不存在某些 index 下的數(shù)據(jù)在某個子串中是一種數(shù)據(jù)，在另一個子串里又是另一種數(shù)據(jù)。
  • 重疊部分的處理最為麻煩。

• 可能會傳一些已經(jīng)被裝配了的數(shù)據(jù)

• 如果 ByteStream 已滿，則必須暫停裝配，將未裝配數(shù)據(jù)暫時保存起來

除了上面的要求以外，容量 Capacity 需要嚴格限制：

為了便于說明，將圖中的綠色區(qū)域稱為 ByteStream，將圖中存放紅色區(qū)域的內(nèi)存范圍（即 first unassembled - first unacceptable）稱為 Unassembled_strs。

CS144 要求將 ByteStream + Unassembled_strs 的內(nèi)存占用總和限制在 Reassember 中構(gòu)造函數(shù)傳入的 capacity 大小。因此我們在構(gòu)造 Reassembler 時，需要既將傳入的 capacity 參數(shù)設置為 ByteStream的緩沖區(qū)大小上限，也將其設置為first unassembled - first unacceptable的范圍大小，以避免極端情況下的內(nèi)存使用。

注意：first unassembled - first unacceptable的范圍大小，并不等同于存放尚未裝配子串的結(jié)構(gòu)體內(nèi)存大小上限，別混淆了。

Capacity 這個概念很重要，因為它不僅用于限制高內(nèi)存占用，而且它還會起到流量控制的作用（見 lab2）。

代碼實現(xiàn)如下:

• stream_reassembler.hh

//! \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.
    struct Datum {
        char ch = 0;
        bool valid = false;
    };
    // 用于存放未按序達到的字節(jié)流
    std::vector<Datum> buffer_;  // buffer to store unassembled data.
    size_t buffer_header_;       // pointer to the begining of the unssembled bytes in
                                 // buffer.
    size_t unassembled_bytes_;
    size_t eof_byte_;
    bool is_eof_set_;    // true if read the eof
    // 存放按序到達的字節(jié)流,但是這部分字節(jié)流還沒有被read
    ByteStream _output;  //!< The reassembled in-order byte stream
    size_t _capacity;    //!< The maximum number of bytes

  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 Receive a substring and write any newly contiguous bytes
    //! into the stream.
    //!
    //! The StreamReassembler will stay within the memory limits of the
    //! `capacity`. Bytes that would exceed the capacity are silently
    //! discarded.
    //!
    //! \param data the substring
    //! \param index indicates the index (place in sequence) of the first
    //! byte in `data` \param eof the last byte of `data` will be the last
    //! byte in the entire 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 pushed more than
    //! once, 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;
};

• stream_reassembler.cc

#include "stream_reassembler.hh"

#include <cassert>

// For Lab 1, please replace with a real implementation that passes the
// automated checks run by `make check_lab1`.

// You will need to add private members to the class declaration in
// `stream_reassembler.hh`

using namespace std;

StreamReassembler::StreamReassembler(const size_t capacity)
    : buffer_(capacity)
    , buffer_header_(0)
    , unassembled_bytes_(0)
    , eof_byte_(0)
    , is_eof_set_(false)
    , _output(capacity)
    , _capacity(capacity) {}

//! \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 (_output.input_ended())
        return;
    const size_t max_byte = _output.bytes_read() + _capacity;
    const size_t index_start = std::max(_output.bytes_written(), index);
    size_t index_end = std::min(max_byte, index + data.length());
    // check for eof
    if (eof) {
        is_eof_set_ = true;
        eof_byte_ = index + data.length();
    }
    if (is_eof_set_)
        index_end = std::min(index_end, eof_byte_);

    // buffer the data
    for (size_t write_index = index_start; write_index < index_end; write_index++) {
        size_t cache_index = (buffer_header_ + write_index - _output.bytes_written()) % _capacity;
        assert(!(buffer_[cache_index].valid && buffer_[cache_index].ch != data[write_index - index]));
        if (!buffer_[cache_index].valid)
            unassembled_bytes_++;
        buffer_[cache_index].valid = true;
        buffer_[cache_index].ch = data[write_index - index];
    }

    // write the data.
    while (buffer_[buffer_header_].valid) {
        buffer_[buffer_header_].valid = false;
        _output.write_char(buffer_[buffer_header_].ch);
        unassembled_bytes_--;
        buffer_header_ = (buffer_header_ + 1) % _capacity;
    }
    if (is_eof_set_ && _output.bytes_written() >= eof_byte_)
        _output.end_input();
}

size_t StreamReassembler::unassembled_bytes() const { return unassembled_bytes_; }

bool StreamReassembler::empty() const { return unassembled_bytes_ == 0; }

代碼可能不太容易理解，但是大家對照下圖，把幾種可能出現(xiàn)的情況看明白，再回去看代碼，相信就不難了:

核心一點: buffer用于暫存未按序到達的這部分不連續(xù)的字節(jié)流，而output用于存放按序到達的這部分字節(jié)流,但是這段字節(jié)流還沒有被read。

測試代碼正確性:
文章來源地址http://www.zghlxwxcb.cn/news/detail-595432.html

到了這里，關于CS 144 Lab One -- 流重組器的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關文章，希望大家以后多多支持TOY模板網(wǎng)！