CS144 lab3 tcp_sender学习笔记

文档传送门

 

实验 3：TCP 发送方

TCP 是一种协议，它可靠地传送一对流控制的字节流（每个字节流一个）方向）通过不可靠的数据报。两方参与 TCP 连接，并且每一方都充当“发送者”（它自己的输出字节流的）和“接收者”（一个传入的字节流）同时。双方被称为“端点”连接，或“同行”。

本周，您将实现 TCP 的“发送方”部分，负责从ByteStream（由某些发送方应用程序创建和写入），并转换成一系列传出的 TCP 段。在远端，一个 TCP 接收器变换那些段（那些到达的——它们可能不会全部成功）回到原始字节流，并将确认和窗口广告发送回发件人。

您的 TCPSender 有责任：

• 跟踪接收器的窗口（处理传入的确认和窗口大小）

• 尽可能填充窗口，通过从 ByteStream 读取，创建新的 TCP段（如果需要，包括 SYN 和 FIN 标志），并发送它们

• 跟踪哪些段已发送但尚未被接收方确认

• 如果自发送后经过足够长的时间，则重新发送未完成的段

⋆我为什么要这样做？基本原则是发送接收者允许的任何内容我们发送（填充窗口），并继续重传，直到接收方确认每个段。这称为“自动重复请求”(ARQ)。发件人分字节流向上分段并在接收器窗口允许的范围内发送它们。感谢您上周的工作，我们知道远程 TCP 接收器可以重建字节流，只要它至少收到每个带索引标记的字节一次——无论命令。发送方的工作是确保接收方至少获得每个字节一次。

重要的是要记住，接收方可以是有效 TCP 接收方的任何实现——它不一定是你自己的 TCPReceiver。关于 Internet 标准的有价值的事情之一是如何在它们在端点之间建立了一种通用语言，否则它们的行为可能会大不相同。

常见问题和特殊情况

• 我如何“发送”一个段？

将其推送到段输出队列。就您的 TCPSender 而言，请考虑只要你把它推到这个队列，它就会发送。很快主人就会过来弹出它（使用公共段 out() 访问器方法）并真正发送它。

• 等等，我如何既“发送”一个段又跟踪同一段作为优秀，所以我知道以后要重传什么？难道我不必复制每个segment呢？那很浪费吗？当您发送包含数据的段时，您可能希望将其推送到段出队列，并在内部保留一份它的副本在一个数据结构中,让您跟踪未完成的段以进行可能的重新传输。事实证明不要太浪费，因为段的有效负载存储为引用计数只读字符串（一个 Buffer 对象）。所以别担心——它实际上并不是复制有效载荷数据。

• 在我得到一个消息之前，我的 TCPSender 应该假设接收方的窗口大小是多少？

一个字节。

• 接收者告诉我它的窗口大小为零字节。我应该只是被卡住，永远不会再次发送任何数据？

不。如果接收器告诉您它的窗口长度为零字节，则将该信息保存为你会任何其他窗口广告，因为它对重传很重要 3.1 中描述的行为. 但是当需要填充窗口时，请表现得像窗口大小为一字节。这称为“零窗口探测”——这是一种定期探测接收器，看看它是否恰好在窗口中打开了更多空间自从我们上次听到他们的消息以来。可能发生的最糟糕的事情是接收器将忽略您的一字节段。

• 如果确认仅部分确认了一些未完成的工作，我该怎么办？我应该尝试剪掉被确认的字节吗？

TCP 发送方可以做到这一点，但就本课程而言，没有必要花哨。将每个部分视为完全杰出的，直到它被完全认可——所有的它占用的序列号小于确认号。

• 如果我发送包含“a”、“b”和“c”的三个单独的段，但它们永远不会得到确认后，我可以在包含“abc”的一个大段中重新传输它们吗？或者我必须单独重新传输每个段吗？

再说一遍：TCP 发送方可以做到这一点，但就本课程而言，无需花哨。只需单独跟踪每个未完成的段，以及何时重传定时器超时，再次发送最早的未完成段。

• 我是否应该在我的“未完成”数据结构中存储空段并重新传输它们？

不——唯一应该被跟踪为未完成的段，并且可能会重新传输的只有那些传达一些数据的段——即在序列空间中消耗一些长度。不需要记住或重传空的 ACK。

实验结果

思路:

 我们需要缓存已经发送但是没有确认的seg,在收到ack时将确认的seg缓存删除这里我用的deque,然后自定义了一个结构体,包含Segment和其ack的绝对序列号,每次发送seg就同时存入缓存,此时缓存中的segment的绝对序列号都是递增的,每次收到ack后,从头遍历缓存删除那些确认序列号小于当前收到ack的seg,

自定义了一个Timer类,用来计时,刷新以及判断超时.

window_size:还没有确认的但是已经发送的数据大小+没有发送但是可以发送的数据大小

每次从字节流中取数据,不是取window_size,而是取还没有发送的但是可以发送的数据

然后跟着文档写就是,也可以看我注解

tcp_sender.hh

class Timer{
         private:
         size_t last_flash_time{0};
         size_t now_time{0};
         public:
         void flash(){
           last_flash_time=now_time;
         }
         void tick(const size_t pass_time){
           now_time+=pass_time;
         }
         bool time_out(const size_t RTO){
           return now_time-last_flash_time>=RTO;
         }
    };
    struct seg_node{
       TCPSegment seg={};
       uint64_t ack_index{0};
    };
    //! our initial sequence number, the number for our SYN.
    WrappingInt32 _isn;

    //! outbound queue of segments that the TCPSender wants sent
    std::queue<TCPSegment> _segments_out{};

    //! retransmission timer for the connection
    unsigned int _initial_retransmission_timeout;

    //! outgoing stream of bytes that have not yet been sent
    ByteStream _stream;

    //! the (absolute) sequence number for the next byte to be sent
    uint64_t _next_seqno{0};
    uint64_t cur_ack_seqno{0};
    uint64_t fin_ack_seqno{UINT64_MAX};
    std::deque<seg_node>seg_buffer{};
    uint64_t wd_right_edge{1};
    size_t retrans_num{0};
    size_t RTO;
    Timer timer={};

tcp_sender.cc

#include "tcp_sender.hh"

#include "tcp_config.hh"

#include <random>
#include <iostream>

// Dummy implementation of a TCP sender

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


using namespace std;

//! \param[in] capacity the capacity of the outgoing byte stream
//! \param[in] retx_timeout the initial amount of time to wait before retransmitting the oldest outstanding segment
//! \param[in] fixed_isn the Initial Sequence Number to use, if set (otherwise uses a random ISN)
TCPSender::TCPSender(const size_t capacity, const uint16_t retx_timeout, const std::optional<WrappingInt32> fixed_isn)
    : _isn(fixed_isn.value_or(WrappingInt32{random_device()()}))
    , _initial_retransmission_timeout{retx_timeout}
    , _stream(capacity) ,RTO(retx_timeout){}

uint64_t TCPSender::bytes_in_flight() const { return _next_seqno-cur_ack_seqno; }

void TCPSender::fill_window() {
    //全部数据都已近发送,无数据发送
    if(_next_seqno==fin_ack_seqno){
        return ;
    }
    //syn
    if(_next_seqno==0){
      TCPSegment syn_seg;
      syn_seg.header().syn=true;
      syn_seg.header().seqno=_isn;
      _segments_out.push(syn_seg);
      _next_seqno+=syn_seg.length_in_sequence_space();
      seg_buffer.push_back({syn_seg,_next_seqno});
      return ;
    }

    uint32_t max_size=1452;
    //fill_size:可发送的字节长度
    //这里非负性判断是因为可能发送了一字节的零窗口探测,实际窗口大小没变
    uint32_t fill_size=static_cast<uint32_t>(wd_right_edge>_next_seqno?wd_right_edge-_next_seqno:0);
    string total_data=_stream.read(fill_size);
    //优先发送data,data不能填满窗口的话,才发送fin
    bool fin_flg=_stream.input_ended()&&_stream.buffer_empty()&&total_data.size()<fill_size;
    //把data按照最大字节数1452拆分为小的seg发送
    uint32_t i=0;
    while(i+max_size<total_data.size()){
        TCPSegment seg;
        seg.header().seqno=wrap(_next_seqno,_isn);
        seg.payload()=Buffer(total_data.substr(i,max_size));
        _segments_out.push(seg);
        _next_seqno+=static_cast<uint64_t>(seg.length_in_sequence_space());
        seg_buffer.push_back({seg,_next_seqno});
        i+=max_size;
    }
    //处理最后部分
    if(i<total_data.size()||fin_flg){
       TCPSegment last_seg;
       if(fin_flg){last_seg.header().fin=true;}
       last_seg.header().seqno=wrap(_next_seqno,_isn);
       if(i<total_data.size())last_seg.payload()=Buffer(total_data.substr(i));
       _segments_out.push(last_seg);
        _next_seqno+=static_cast<uint64_t>(last_seg.length_in_sequence_space());
       seg_buffer.push_back({last_seg,_next_seqno});
       if(fin_flg)fin_ack_seqno=_next_seqno;
    }
}
//! \param ackno The remote receiver's ackno (acknowledgment number)
//! \param window_size The remote receiver's advertised window size
void TCPSender::ack_received(const WrappingInt32 ackno, const uint16_t window_size) { 
    uint64_t ack=unwrap(ackno,_isn,_next_seqno);
    //错误ack
    if(ack>_next_seqno){
        return ;
    }
    //去除已近确认的seg缓存
    bool new_ack_flg=false;
    while(!seg_buffer.empty()&&seg_buffer.front().ack_index<=ack){
        new_ack_flg=true;
        cur_ack_seqno=seg_buffer.front().ack_index;
        seg_buffer.pop_front();
    }
    //如果有新的seg被确认,刷新计时器及RTO
    if(new_ack_flg){
       RTO=_initial_retransmission_timeout;
       retrans_num=0;
       timer.flash();
    }
    uint64_t now_wd_right=ack+static_cast<uint64_t>(window_size);
    if(now_wd_right>wd_right_edge){
        wd_right_edge=now_wd_right;
    }
    //window_size=0,发送一字节的窗口探测包
    if(window_size==0)wd_right_edge++;
    fill_window();
    if(window_size==0)wd_right_edge--;
 }

//! \param[in] ms_since_last_tick the number of milliseconds since the last call to this method
void TCPSender::tick(const size_t ms_since_last_tick) { 
    //=?
    timer.tick(ms_since_last_tick);
    if(!timer.time_out(RTO)){
       return ;
    }
    if(!seg_buffer.empty()){
    //超时重传第一个未确认的seg
    _segments_out.push(seg_buffer.front().seg);
    uint64_t wind_size=wd_right_edge-cur_ack_seqno;
    if(wind_size!=0ull){
        retrans_num++;
        RTO=RTO*2u;
    }
    timer.flash();
    }
 }

unsigned int TCPSender::consecutive_retransmissions() const { 
    return retrans_num;
 }

void TCPSender::send_empty_segment() {
    TCPSegment seg;
    seg.header().seqno=wrap(_next_seqno,_isn);
    _segments_out.push(seg);
}