前言

  • merge lab3的初始代码，出现如下变化

image.png

通读文档后一些疑问

  • 谁来调用fill_window()？
  • ack_received()需要调用fill_window()吗？
    • ack_received会更新接收方的接收窗口
  • 放入_segments_out的TCPSegment的ackno怎么填？
    • send_empty_segment()的TCPSegment的ackno怎么填？
    • 我觉得都不需要填，由TCPConnection来填，只需要填入seqno
  • 接收窗口为0的逻辑在哪里处理？怎么处理？

定时器相关疑问

  • 定时器第一次启动是什么时候？
    • 每次发送序号空间不为0的段时，就会检查定时器是否启动，如果没有启动就启动
    • 所以第一次启动，是发送SYN报文段后
  • 定时器什么时候会被启动/重新启动
    • 每次发送序号空间不为0的段时，就会检查定时器是否启动，如果没有启动就启动
    • ack_received()被调用后，接收到1个之前从来没有出现过的更大的ackno，且_segments_out未被清空，则启动或重启
      • 我觉得这里应该一定有定时器是开着的，因为_segments_out未被清空，所以一定是重启
      • 注意RTO可能被更新（变回初始值）
    • 定时器超时后，要再次启动
      • 注意RTO可能被更新
  • 定时器怎么确认启动时，当前时刻是多少？定时器什么时候超时？定时器怎么确认超时？
    • 定义一个成员变量，表示毫秒数，每次tick()被调用，更新这个毫秒数
    • 是否可以认为tick()被调用的足够频繁，以至于定时器认为此时的毫秒数变量指示的就是真实的毫秒数？
    • tick()被调用，在tick()方法中发现当前的毫秒数已经超过了定时器设置的超时毫秒数，则定时器超时
    • 定义一个成员变量_timer_ms
    • 当定时器被启动或重启时，_timer_ms置为0
    • 每次tick被调用
      • _timer_ms+=ms_since_last_tick
      • 若_timer_ms>=RTO，则定时器超时

  • 定时器什么时候会被关闭？

    • ack_received()被调用后，_segments_out被清空时，如果当前定时器开着，就关闭

  • 连续重传次数什么时候+1？连续重传次数什么时候清零？TCPSender除了对连续重传次数+1和清零，还需要做别的事情吗？

    • 如果tick()被调用，发现定时器超时，则会重传最早未被确认的报文段，此时连续重传次数+1（文档里说if window size is nonzero???）
    • ack_received()被调用，接收到1个之前从来没有出现过的更大的ackno，则连续重传次数清零
  • RTO什么时候加倍？RTO什么时候恢复初始值？
    • 如果tick()被调用，发现定时器超时，则会重传最早未被确认的报文段，此时RTO加倍（文档里说if window size is nonzero???）
    • ack_received()被调用，接收到1个之前从来没有出现过的更大的ackno，则RTO变回初始值

需要额外设置哪些成员变量？

  • _initial_retransmission_timeout存储的是RTO的初始值，我们还需要存储RTO，_retransmission_timeout
  • 最新的ackno和window_size
  • 注意_segments_out表示的是所有已经封装好，马上要被发送的报文段队列，我们还需要一个额外的数据结构，来保存那些已发送但未被确认的报文段，前者是后者的子集
    • 用std::queue来保存
  • 我们经常需要调用bytes_in_flight()来计算那些已发送但未被确认的字节数，由于std::queue不能很方便的遍历，即使要遍历的话，也很浪费时间
    • 所以直接设置一个变量_bytes_in_flight，来动态的记录这个队列中的TCP报文段的字节数之和，以提高时间效率
  • 连续重传次数，_consecutive_retransmission
  • 定时器相关
    • 计时器启动后已经经历的毫秒数，_timer_ms
    • 定时器是否启动，_timer_running

第一阶段

分析

  • 大部分的疑问和思考见前言
  • 本阶段的代码中，完全没有考虑接收方窗口大小为0的情况

    代码

    tcp_sender.hh ```cpp

    ifndef SPONGE_LIBSPONGE_TCP_SENDER_HH

    define SPONGE_LIBSPONGE_TCP_SENDER_HH

include “byte_stream.hh”

include “tcp_config.hh”

include “tcp_segment.hh”

include “wrapping_integers.hh”

include

include

//! \brief The “sender” part of a TCP implementation.

//! Accepts a ByteStream, divides it up into segments and sends the //! segments, keeps track of which segments are still in-flight, //! maintains the Retransmission Timer, and retransmits in-flight //! segments if the retransmission timer expires. class TCPSender { private: // 发送方随机选择的ISN //! our initial sequence number, the number for our SYN. WrappingInt32 _isn;

  1. // 把需要发送的报文段放到这个队列
  2. //! outbound queue of segments that the TCPSender wants sent
  3. std::queue<TCPSegment> _segments_out{};
  5. // 这个变量保存的是RTO的初始值
  6. //! retransmission timer for the connection
  7. unsigned int _initial_retransmission_timeout;
  9. // 还没有发送的数据
  10. //! outgoing stream of bytes that have not yet been sent
  11. ByteStream _stream;
  13. // 未发送数据的第一个字节的绝对序号，包括SYN和FIN
  14. //! the (absolute) sequence number for the next byte to be sent
  15. uint64_t _next_seqno{0};
  17. size_t _lately_ackno = 0;  // 从接收方返回的最大的ackno
  18. size_t _window_size = 0;  // 和_lately_ackno配套的window size
  20. size_t _bytes_in_flight = 0;  // 是_segments_outstanding中的序号空间大小
  21. std::queue<TCPSegment> _segments_outstanding{};  // 已发送但未确认的报文段
  23. size_t _retransmission_timeout = 0;  // RTO
  24. size_t _consecutive_retransmission = 0;  // 连续重传次数
  26. size_t _timer_ms = 0;  // 计时器启动后经过的毫秒数
  27. bool _timer_running = false;  // 计时器是否启动
  29. void send_segment(TCPSegment &seg);  // 发送已经填写了必要信息的报文段，这个方法内部只会再向TCPSegment中填入seqno

public: //! Initialize a TCPSender // 构造函数的参数：容量、初始超时时间(RTO的初始值)、ISN TCPSender(const size_t capacity = TCPConfig::DEFAULT_CAPACITY, const uint16_t retx_timeout = TCPConfig::TIMEOUT_DFLT, const std::optional fixed_isn = {});

  1. //! \name "Input" interface for the writer
  2. //!@{
  3. ByteStream &stream_in() { return _stream; }
  4. const ByteStream &stream_in() const { return _stream; }
  5. //!@}
  7. //! \name Methods that can cause the TCPSender to send a segment
  8. //!@{
  10. //! \brief A new acknowledgment was received
  11. void ack_received(const WrappingInt32 ackno, const uint16_t window_size);
  13. //! \brief Generate an empty-payload segment (useful for creating empty ACK segments)
  14. void send_empty_segment();
  16. //! \brief create and send segments to fill as much of the window as possible
  17. void fill_window();
  19. // 每隔几毫秒，您的TCPSender的tick方法将被调用，并带有一个参数，该参数指示自上次调用该方法以来已经过了几毫秒
  20. //! \brief Notifies the TCPSender of the passage of time
  21. void tick(const size_t ms_since_last_tick);
  22. //!@}
  24. //! \name Accessors
  25. //!@{
  27. // 已发送但未确认的字节数，SYN和FIN各占1字节
  28. //! \brief How many sequence numbers are occupied by segments sent but not yet acknowledged?
  29. //! \note count is in "sequence space," i.e. SYN and FIN each count for one byte
  30. //! (see TCPSegment::length_in_sequence_space())
  31. size_t bytes_in_flight() const;
  33. // 连续重传次数
  34. //! \brief Number of consecutive retransmissions that have occurred in a row
  35. unsigned int consecutive_retransmissions() const;
  37. //! \brief TCPSegments that the TCPSender has enqueued for transmission.
  38. //! \note These must be dequeued and sent by the TCPConnection,
  39. //! which will need to fill in the fields that are set by the TCPReceiver
  40. //! (ackno and window size) before sending.
  41. std::queue<TCPSegment> &segments_out() { return _segments_out; }
  42. //!@}
  44. //! \name What is the next sequence number? (used for testing)
  45. //!@{
  47. //! \brief absolute seqno for the next byte to be sent
  48. uint64_t next_seqno_absolute() const { return _next_seqno; }
  50. //! \brief relative seqno for the next byte to be sent
  51. WrappingInt32 next_seqno() const { return wrap(_next_seqno, _isn); }
  52. //!@}

};

endif // SPONGE_LIBSPONGE_TCP_SENDER_HH

  1. tcp_sender.cc
  2. ```cpp
  3. #include "tcp_sender.hh"
  5. #include "tcp_config.hh"
  7. #include <random>
  9. // Dummy implementation of a TCP sender
  11. // For Lab 3, please replace with a real implementation that passes the
  12. // automated checks run by `make check_lab3`.
  14. template <typename... Targs>
  15. void DUMMY_CODE(Targs &&... /* unused */) {}
  17. using namespace std;
  19. //! \param[in] capacity the capacity of the outgoing byte stream
  20. //! \param[in] retx_timeout the initial amount of time to wait before retransmitting the oldest outstanding segment
  21. //! \param[in] fixed_isn the Initial Sequence Number to use, if set (otherwise uses a random ISN)
  22. TCPSender::TCPSender(const size_t capacity, const uint16_t retx_timeout, const std::optional<WrappingInt32> fixed_isn)
  23.     : _isn(fixed_isn.value_or(WrappingInt32{random_device()()}))
  24.     , _initial_retransmission_timeout{retx_timeout}
  25.     , _stream(capacity)
  26.     ,_lately_ackno(0), _window_size(0) 
  27.     ,_bytes_in_flight(0), _segments_outstanding()
  28.     ,_retransmission_timeout{retx_timeout}, _consecutive_retransmission(0)
  29.     ,_timer_ms(0), _timer_running(false) {}
  31. // _segments_outstanding中所有报文段的和，包括SYN和FIN
  32. uint64_t TCPSender::bytes_in_flight() const { 
  33.     return _bytes_in_flight;
  34. }
  36. void TCPSender::fill_window() {
  37.     if (_next_seqno==0) {  // 状态0 CLOSED
  38.         // 发送不带数据的SYN报文
  39.         TCPSegment seg;
  40.         seg.header().syn=true;
  42.         send_segment(seg);
  43.     } else if (_next_seqno>0&&bytes_in_flight()==_next_seqno) {  // 状态1 SYN_SENT
  44.         // 在未收到任何ACK之前，假设接收方的窗口大小为1，所以这里不能再发送任何数据
  45.     } else if (!_stream.eof()&&bytes_in_flight()<_next_seqno) {  // 状态2 SYN_ACKED
  46.         /*  考察区间[_next_seqno, _lately_ackno+_window_size) 
  47.             如果区间不存在，则不能发送新的数据
  48.             如果window_size==0，则可以发送1个字节的数据
  49.         */
  50.         if (_lately_ackno+_window_size>_next_seqno) {
  51.             // 有可以发送的数据
  52.             std::string data_to_send = _stream.read(_lately_ackno+_window_size-_next_seqno);
  53.             TCPSegment seg;
  54.             seg.payload() = Buffer(std::move(data_to_send));
  55.             // 如果stream读完了，且还有余下的空间放置FIN
  56.             if (_stream.eof()&&data_to_send.size()<_lately_ackno+_window_size-_next_seqno) {
  57.                 seg.header().fin=true;
  58.             }
  59.             send_segment(seg);
  60.         }
  61.     } else if (_stream.eof()&&_next_seqno<_stream.bytes_written()+2) {  // 状态3 SYN_ACKED and FIN not Sent
  62.         // 这种状态下一定只剩下1个FIN没放进TCPSegment的队列
  63.         // 接收窗口也可能是0吧？
  64.         if (_lately_ackno+_window_size>_next_seqno) {
  65.             TCPSegment seg;
  66.             seg.header().fin=true;
  67.             send_segment(seg);
  68.         }
  69.     } else if (_stream.eof()&&_next_seqno==_stream.bytes_written()+2&&bytes_in_flight()>0) {  // 状态4 FIN_SENT
  70.         // 这里不用发送数据
  71.     } else if (_stream.eof()&&_next_seqno==_stream.bytes_written()+2&&bytes_in_flight()==0) {  // 状态5 FIN_ACKED
  72.         // 这里不用发送数据
  73.     } else {  // 应该不会到这个分支
  75.     }
  76. }
  78. //! \param ackno The remote receiver's ackno (acknowledgment number)
  79. //! \param window_size The remote receiver's advertised window size
  80. void TCPSender::ack_received(const WrappingInt32 ackno, const uint16_t window_size) { 
  81.     size_t _ackno=unwrap(ackno, _isn, _lately_ackno);
  82.     if (_ackno>=_lately_ackno) {
  83.         // if (_ackno>_lately_ackno) {  // 这样的话，部分确认也会重置RTO和清零连续重传次数
  84.         //     _consecutive_retransmission=0;
  85.         //     _retransmission_timeout=_initial_retransmission_timeout;
  86.         // }
  87.         _lately_ackno=_ackno;
  88.         _window_size=window_size;
  89.     }
  90.     bool seg_acked=false;
  91.     while (_segments_outstanding.size()) {
  92.         TCPSegment seg = _segments_outstanding.front();
  93.         size_t abs_seqno = unwrap(seg.header().seqno, _isn, _lately_ackno);
  94.         if (abs_seqno+seg.length_in_sequence_space()<=_lately_ackno) {
  95.             seg_acked=true;
  96.             _segments_outstanding.pop();
  97.         } else {
  98.             break;
  99.         }
  100.     }
  101.     // 我认为这里_timer_running一定是true
  102.     if (_segments_outstanding.empty()) {
  103.         _timer_running=false;  // 关闭定时器
  104.     } else if (seg_acked) {  // 重启定时器
  105.         _timer_running=true;
  106.         _timer_ms=0;
  107.     }
  108.     // 这样的话，部分确认不会重置RTO和清零连续重传次数吗？
  109.     if (seg_acked) {
  110.         _consecutive_retransmission=0;
  111.         _retransmission_timeout=_initial_retransmission_timeout;
  112.     }
  113.     // 如果接收窗口有了新的空间，TCPSender应该再次填充该窗口。
  114.     // todo
  115. }
  117. //! \param[in] ms_since_last_tick the number of milliseconds since the last call to this method
  118. void TCPSender::tick(const size_t ms_since_last_tick) { 
  119.     // 目前没有考虑窗口大小为0的情况
  120.     if (_timer_running) {
  121.         _timer_ms+=ms_since_last_tick;
  122.         if (_timer_ms>=_retransmission_timeout) {  // 定时器超时，需要重传最早的未被确认的报文段
  123.             _timer_ms=0;  // 重启定时器
  124.             _consecutive_retransmission++;  // 重传次数加一
  125.             _retransmission_timeout<<=1;  // RTO加倍
  126.             if (_segments_outstanding.size()) {  // 这个判断应该一定是成立的吧？
  127.                 _segments_out.push(_segments_outstanding.front());
  128.             }
  129.         }
  130.     }
  131. }
  133. unsigned int TCPSender::consecutive_retransmissions() const { 
  134.     return _consecutive_retransmission;
  135. }
  137. // 不带SYN、FIN，且不包含数据，只需要设置正确的seqno
  138. void TCPSender::send_empty_segment() {
  139.     TCPSegment seg;
  140.     seg.header().seqno=wrap(_next_seqno, _isn);
  141.     _segments_out.push(seg);
  142.     // 不用加入_segments_outstanding，也不用启动定时器
  143. }
  145. void TCPSender::send_segment(TCPSegment &seg) {
  146.     seg.header().seqno=wrap(_next_seqno, _isn);
  147.     _next_seqno+=seg.length_in_sequence_space();
  148.     _bytes_in_flight+=seg.length_in_sequence_space();
  150.     _segments_out.push(seg);
  151.     _segments_outstanding.push(seg);
  153.     // 每次发送报文段都要检查定时器，如果未启动则启动
  154.     if (!_timer_running) {
  155.         _timer_ms=0;
  156.         _timer_running=true;
  157.     }
  158. }

测试结果

image.png

When filling window, treat a ‘0’ window size as equal to ‘1’ but don’t back off RTO

Retx SYN twice at the right times, then ack

DEBUG

  • 问题一：在收到新的ack后，从_segments_outstanding中删除已经被完全确认的报文段，此时_bytes_in_flight这个变量应该更新，我忘记了
  • 问题二：接收窗口很大，且待发送的数据很多，这种情况下，调用fill_window()一次，就会产生多个TCPSegment，我实现fill_window每次只会产生1个TCPSegment
  • 问题三：在收到新的ack后，需要手动调用fill_window()，上面的代码中没有加
    • 目前，我把这个调用和RTO的初始化、连续重传次数的清零放在了一起
    • 只有在至少完全确认一个报文段后，才会调用fill_window()，如果只是部分确认，就不会调用fill_window()
    • 其它可能的写法
      • 无论是否收到新的ack，在ack_received中一定会调用fill_window()
      • 收到新的ack，部分确认了最早已发送但未确认的报文段，那么就调用fill_window()
  • 问题四：接收窗口为0的情况未处理
    • 我一开始接受窗口的初始化值为0（在接收任何接收方返回的报文段之前，应假设为1），现在修改为了1
    • 在收到新的ack，更新了_window_size=0后
      • 超时后不再增加重传次数，也不会RTO加倍
      • 在fill_window()方法中，如果读到_window_size=0，应将其当做1，所以设置了一个叫window_size的局部变量

第二阶段

代码

在修改了第一阶段的问题后，代码如下

tcp_sender.hh（无改动）

  1. #ifndef SPONGE_LIBSPONGE_TCP_SENDER_HH
  2. #define SPONGE_LIBSPONGE_TCP_SENDER_HH
  4. #include "byte_stream.hh"
  5. #include "tcp_config.hh"
  6. #include "tcp_segment.hh"
  7. #include "wrapping_integers.hh"
  9. #include <functional>
  10. #include <queue>
  12. //! \brief The "sender" part of a TCP implementation.
  14. //! Accepts a ByteStream, divides it up into segments and sends the
  15. //! segments, keeps track of which segments are still in-flight,
  16. //! maintains the Retransmission Timer, and retransmits in-flight
  17. //! segments if the retransmission timer expires.
  18. class TCPSender {
  19.   private:
  20.     // 发送方随机选择的ISN
  21.     //! our initial sequence number, the number for our SYN.
  22.     WrappingInt32 _isn;
  24.     // 把需要发送的报文段放到这个队列
  25.     //! outbound queue of segments that the TCPSender wants sent
  26.     std::queue<TCPSegment> _segments_out{};
  28.     // 这个变量保存的是RTO的初始值
  29.     //! retransmission timer for the connection
  30.     unsigned int _initial_retransmission_timeout;
  32.     // 还没有发送的数据
  33.     //! outgoing stream of bytes that have not yet been sent
  34.     ByteStream _stream;
  36.     // 未发送数据的第一个字节的绝对序号，包括SYN和FIN
  37.     //! the (absolute) sequence number for the next byte to be sent
  38.     uint64_t _next_seqno{0};
  40.     size_t _lately_ackno = 0;  // 从接收方返回的最大的ackno
  41.     size_t _window_size = 0;  // 和_lately_ackno配套的window size
  43.     size_t _bytes_in_flight = 0;  // 是_segments_outstanding中的序号空间大小
  44.     std::queue<TCPSegment> _segments_outstanding{};  // 已发送但未确认的报文段
  46.     size_t _retransmission_timeout = 0;  // RTO
  47.     size_t _consecutive_retransmission = 0;  // 连续重传次数
  49.     size_t _timer_ms = 0;  // 计时器启动后经过的毫秒数
  50.     bool _timer_running = false;  // 计时器是否启动
  52.     void send_segment(TCPSegment &seg);  // 发送已经填写了必要信息的报文段，这个方法内部只会再向TCPSegment中填入seqno
  53.   public:
  54.     //! Initialize a TCPSender
  55.     // 构造函数的参数：容量、初始超时时间(RTO的初始值)、ISN
  56.     TCPSender(const size_t capacity = TCPConfig::DEFAULT_CAPACITY,
  57.               const uint16_t retx_timeout = TCPConfig::TIMEOUT_DFLT,
  58.               const std::optional<WrappingInt32> fixed_isn = {});
  60.     //! \name "Input" interface for the writer
  61.     //!@{
  62.     ByteStream &stream_in() { return _stream; }
  63.     const ByteStream &stream_in() const { return _stream; }
  64.     //!@}
  66.     //! \name Methods that can cause the TCPSender to send a segment
  67.     //!@{
  69.     //! \brief A new acknowledgment was received
  70.     void ack_received(const WrappingInt32 ackno, const uint16_t window_size);
  72.     //! \brief Generate an empty-payload segment (useful for creating empty ACK segments)
  73.     void send_empty_segment();
  75.     //! \brief create and send segments to fill as much of the window as possible
  76.     void fill_window();
  78.     // 每隔几毫秒，您的TCPSender的tick方法将被调用，并带有一个参数，该参数指示自上次调用该方法以来已经过了几毫秒
  79.     //! \brief Notifies the TCPSender of the passage of time
  80.     void tick(const size_t ms_since_last_tick);
  81.     //!@}
  83.     //! \name Accessors
  84.     //!@{
  86.     // 已发送但未确认的字节数，SYN和FIN各占1字节
  87.     //! \brief How many sequence numbers are occupied by segments sent but not yet acknowledged?
  88.     //! \note count is in "sequence space," i.e. SYN and FIN each count for one byte
  89.     //! (see TCPSegment::length_in_sequence_space())
  90.     size_t bytes_in_flight() const;
  92.     // 连续重传次数
  93.     //! \brief Number of consecutive retransmissions that have occurred in a row
  94.     unsigned int consecutive_retransmissions() const;
  96.     //! \brief TCPSegments that the TCPSender has enqueued for transmission.
  97.     //! \note These must be dequeued and sent by the TCPConnection,
  98.     //! which will need to fill in the fields that are set by the TCPReceiver
  99.     //! (ackno and window size) before sending.
  100.     std::queue<TCPSegment> &segments_out() { return _segments_out; }
  101.     //!@}
  103.     //! \name What is the next sequence number? (used for testing)
  104.     //!@{
  106.     //! \brief absolute seqno for the next byte to be sent
  107.     uint64_t next_seqno_absolute() const { return _next_seqno; }
  109.     //! \brief relative seqno for the next byte to be sent
  110.     WrappingInt32 next_seqno() const { return wrap(_next_seqno, _isn); }
  111.     //!@}
  112. };
  114. #endif  // SPONGE_LIBSPONGE_TCP_SENDER_HH

tcp_sender.cc

  1. #include "tcp_sender.hh"
  2. #include "tcp_config.hh"
  4. #include <random>
  5. #include <iostream>
  6. #include <cmath>
  8. // Dummy implementation of a TCP sender
  10. // For Lab 3, please replace with a real implementation that passes the
  11. // automated checks run by `make check_lab3`.
  13. template <typename... Targs>
  14. void DUMMY_CODE(Targs &&... /* unused */) {}
  16. using namespace std;
  18. //! \param[in] capacity the capacity of the outgoing byte stream
  19. //! \param[in] retx_timeout the initial amount of time to wait before retransmitting the oldest outstanding segment
  20. //! \param[in] fixed_isn the Initial Sequence Number to use, if set (otherwise uses a random ISN)
  21. TCPSender::TCPSender(const size_t capacity, const uint16_t retx_timeout, const std::optional<WrappingInt32> fixed_isn)
  22.     : _isn(fixed_isn.value_or(WrappingInt32{random_device()()}))
  23.     , _initial_retransmission_timeout{retx_timeout}
  24.     , _stream(capacity)
  25.     ,_lately_ackno(0), _window_size(1)  // fixed
  26.     ,_bytes_in_flight(0), _segments_outstanding()
  27.     ,_retransmission_timeout{retx_timeout}, _consecutive_retransmission(0)
  28.     ,_timer_ms(0), _timer_running(false) {}
  30. // _segments_outstanding中所有报文段的和，包括SYN和FIN
  31. uint64_t TCPSender::bytes_in_flight() const { 
  32.     return _bytes_in_flight;
  33. }
  35. void TCPSender::fill_window() {
  36.     if (_next_seqno==0) {  // 状态0 CLOSED
  37.         // 发送不带数据的SYN报文
  38.         TCPSegment seg;
  39.         seg.header().syn=true;
  41.         send_segment(seg);
  42.         std::cout<<"~~~~~~~~~~~~~~~~~~~~~~~~  start with RTO: "<<_retransmission_timeout<<std::endl;
  43.     } else if (_next_seqno>0&&bytes_in_flight()==_next_seqno) {  // 状态1 SYN_SENT
  44.         // 在未收到任何ACK之前，假设接收方的窗口大小为1，所以这里不能再发送任何数据
  45.     } else if (!_stream.eof()&&bytes_in_flight()<_next_seqno) {  // 状态2 SYN_ACKED
  46.         /*  考察区间[_next_seqno, _lately_ackno+_window_size) 
  47.             如果区间不存在，则不能发送新的数据
  48.             如果window_size==0，则可以发送1个字节的数据
  49.         */
  50.         size_t window_size=_window_size==0?1:_window_size;  // fixed
  51.         if (_lately_ackno+window_size>_next_seqno&&!_stream.buffer_empty()) {  // fixed
  52.             // 注意MAX_PAYLOAD_SIZE=1452字节 fixed
  53.             // 有可以发送的数据
  54.             size_t size_to_send = _lately_ackno+window_size-_next_seqno;
  55.             while (size_to_send>0&&!_stream.buffer_empty()) {
  56.                 size_t size_to_read = std::min(size_to_send, TCPConfig::MAX_PAYLOAD_SIZE);
  57.                 // std::string data_to_send = _stream.read(_lately_ackno+window_size-_next_seqno);
  58.                 std::string data_to_send = _stream.read(size_to_read);
  59.                 size_to_send-=data_to_send.size();
  60.                 std::cout<<"****** size_to_read: "<<size_to_read<<" data_to_send.size(): "<<data_to_send.size();
  61.                 TCPSegment seg;
  62.                 seg.payload() = Buffer(std::move(data_to_send));
  63.                 // 如果stream读完了，且还有余下的空间放置FIN
  64.                 if (_stream.eof()&&size_to_send>0) {
  65.                     // 这样的话循环条件一定不成立，所以size_to_send也没必要再-1了
  66.                     seg.header().fin=true;
  67.                 }
  68.                 send_segment(seg);
  69.                 std::cout<<" _segments_out.size(): "<<_segments_out.size()<<std::endl;
  70.                 if (_segments_out.size()) {
  71.                     std::cout<<_segments_out.front().payload().str()<<std::endl;
  72.                 }
  73.             }
  74.         }
  75.     } else if (_stream.eof()&&_next_seqno<_stream.bytes_written()+2) {  // 状态3 SYN_ACKED and FIN not Sent
  76.         // 这种状态下一定只剩下1个FIN没放进TCPSegment的队列
  77.         // 接收窗口也可能是0吧？
  78.         size_t window_size=_window_size==0?1:_window_size;  // fixed
  79.         if (_lately_ackno+window_size>_next_seqno) {
  80.             TCPSegment seg;
  81.             seg.header().fin=true;
  82.             send_segment(seg);
  83.         }
  84.     } else if (_stream.eof()&&_next_seqno==_stream.bytes_written()+2&&bytes_in_flight()>0) {  // 状态4 FIN_SENT
  85.         // 这里不用发送数据
  86.     } else if (_stream.eof()&&_next_seqno==_stream.bytes_written()+2&&bytes_in_flight()==0) {  // 状态5 FIN_ACKED
  87.         // 这里不用发送数据
  88.     } else {  // 应该不会到这个分支
  90.     }
  91. }
  93. //! \param ackno The remote receiver's ackno (acknowledgment number)
  94. //! \param window_size The remote receiver's advertised window size
  95. void TCPSender::ack_received(const WrappingInt32 ackno, const uint16_t window_size) { 
  96.     size_t _ackno=unwrap(ackno, _isn, _lately_ackno);
  97.     std::cout<<"***** _ackno: "<<_ackno<<" _lately_ackno: "<<_lately_ackno<<std::endl;
  98.     if (_ackno>=_lately_ackno) {
  99.         // if (_ackno>_lately_ackno) {  // 这样的话，部分确认也会重置RTO和清零连续重传次数
  100.         //     _consecutive_retransmission=0;
  101.         //     _retransmission_timeout=_initial_retransmission_timeout;
  102.         // }
  103.         _lately_ackno=_ackno;
  104.         _window_size=window_size;  // 这里window_size可能是0
  105.     }
  106.     bool seg_acked=false;
  107.     while (_segments_outstanding.size()) {
  108.         TCPSegment seg = _segments_outstanding.front();
  109.         size_t abs_seqno = unwrap(seg.header().seqno, _isn, _lately_ackno);
  110.         if (abs_seqno+seg.length_in_sequence_space()<=_lately_ackno) {
  111.             seg_acked=true;
  112.             _segments_outstanding.pop();
  113.             _bytes_in_flight-=seg.length_in_sequence_space();  // fixed
  114.         } else {
  115.             break;
  116.         }
  117.     }
  118.     // 我认为这里_timer_running一定是true
  119.     if (_segments_outstanding.empty()) {
  120.         _timer_running=false;  // 关闭定时器
  121.     } else if (seg_acked) {  // 重启定时器
  122.         _timer_running=true;
  123.         _timer_ms=0;
  124.     }
  125.     // 这样的话，部分确认不会重置RTO和清零连续重传次数，也不会调用fill_window()
  126.     if (seg_acked) {
  127.         _consecutive_retransmission=0;
  128.         _retransmission_timeout=_initial_retransmission_timeout;
  129.         fill_window();  // fixed
  130.     }
  131.     // 如果接收窗口有了新的空间，TCPSender应该再次填充该窗口。
  132.     // todo
  133. }
  135. //! \param[in] ms_since_last_tick the number of milliseconds since the last call to this method
  136. void TCPSender::tick(const size_t ms_since_last_tick) { 
  137.     // 目前没有考虑窗口大小为0的情况
  138.     if (_timer_running) {
  139.         _timer_ms+=ms_since_last_tick;
  140.         if (_timer_ms>=_retransmission_timeout) {  // 定时器超时，需要重传最早的未被确认的报文段
  141.             _timer_ms=0;  // 重启定时器
  142.             // if (_window_size>0||(_next_seqno>0&&bytes_in_flight()==_next_seqno)) {  // fixed
  143.             if (_window_size>0) {  // fixed
  144.                 _consecutive_retransmission++;  // 重传次数加一
  145.                 _retransmission_timeout<<=1;  // RTO加倍
  146.             }
  147.             if (_segments_outstanding.size()) {  // 这个判断应该一定是成立的吧？
  148.                 _segments_out.push(_segments_outstanding.front());
  149.             }
  150.         }
  151.     }
  152. }
  154. unsigned int TCPSender::consecutive_retransmissions() const { 
  155.     return _consecutive_retransmission;
  156. }
  158. // 不带SYN、FIN，且不包含数据，只需要设置正确的seqno
  159. void TCPSender::send_empty_segment() {
  160.     TCPSegment seg;
  161.     seg.header().seqno=wrap(_next_seqno, _isn);
  162.     _segments_out.push(seg);
  163.     // 不用加入_segments_outstanding，也不用启动定时器
  164. }
  166. void TCPSender::send_segment(TCPSegment &seg) {
  167.     seg.header().seqno=wrap(_next_seqno, _isn);
  168.     _next_seqno+=seg.length_in_sequence_space();
  169.     _bytes_in_flight+=seg.length_in_sequence_space();
  171.     _segments_out.push(seg);
  172.     _segments_outstanding.push(seg);
  174.     // 每次发送报文段都要检查定时器，如果未启动则启动
  175.     if (!_timer_running) {
  176.         _timer_ms=0;
  177.         _timer_running=true;
  178.     }
  179. }

测试结果

image.png

第三阶段

说明

  • 引入Lab4的代码后，新增了测试，Lab3又有样例没过，经过DEBUG，发现是少考虑一种特殊情况
    • Impossible ackno (beyond next seqno) is ignored
    • 只需要在ack_received方法前加入相关判断即可

image.png

代码

  1. #ifndef SPONGE_LIBSPONGE_TCP_SENDER_HH
  2. #define SPONGE_LIBSPONGE_TCP_SENDER_HH
  4. #include "byte_stream.hh"
  5. #include "tcp_config.hh"
  6. #include "tcp_segment.hh"
  7. #include "wrapping_integers.hh"
  9. #include <functional>
  10. #include <queue>
  12. //! \brief The "sender" part of a TCP implementation.
  14. //! Accepts a ByteStream, divides it up into segments and sends the
  15. //! segments, keeps track of which segments are still in-flight,
  16. //! maintains the Retransmission Timer, and retransmits in-flight
  17. //! segments if the retransmission timer expires.
  18. class TCPSender {
  19.   private:
  20.     // 发送方随机选择的ISN
  21.     //! our initial sequence number, the number for our SYN.
  22.     WrappingInt32 _isn;
  24.     // 把需要发送的报文段放到这个队列
  25.     //! outbound queue of segments that the TCPSender wants sent
  26.     std::queue<TCPSegment> _segments_out{};
  28.     // 这个变量保存的是RTO的初始值
  29.     //! retransmission timer for the connection
  30.     unsigned int _initial_retransmission_timeout;
  32.     // 还没有发送的数据
  33.     //! outgoing stream of bytes that have not yet been sent
  34.     ByteStream _stream;
  36.     // 未发送数据的第一个字节的绝对序号，包括SYN和FIN
  37.     //! the (absolute) sequence number for the next byte to be sent
  38.     uint64_t _next_seqno{0};
  40.     size_t _lately_ackno = 0;  // 从接收方返回的最大的ackno
  41.     size_t _window_size = 0;  // 和_lately_ackno配套的window size
  43.     size_t _bytes_in_flight = 0;  // 是_segments_outstanding中的序号空间大小
  44.     std::queue<TCPSegment> _segments_outstanding{};  // 已发送但未确认的报文段
  46.     size_t _retransmission_timeout = 0;  // RTO
  47.     size_t _consecutive_retransmission = 0;  // 连续重传次数
  49.     size_t _timer_ms = 0;  // 计时器启动后经过的毫秒数
  50.     bool _timer_running = false;  // 计时器是否启动
  52.     void send_segment(TCPSegment &seg);  // 发送已经填写了必要信息的报文段，这个方法内部只会再向TCPSegment中填入seqno
  53.   public:
  54.     //! Initialize a TCPSender
  55.     // 构造函数的参数：容量、初始超时时间(RTO的初始值)、ISN
  56.     TCPSender(const size_t capacity = TCPConfig::DEFAULT_CAPACITY,
  57.               const uint16_t retx_timeout = TCPConfig::TIMEOUT_DFLT,
  58.               const std::optional<WrappingInt32> fixed_isn = {});
  60.     //! \name "Input" interface for the writer
  61.     //!@{
  62.     ByteStream &stream_in() { return _stream; }
  63.     const ByteStream &stream_in() const { return _stream; }
  64.     //!@}
  66.     //! \name Methods that can cause the TCPSender to send a segment
  67.     //!@{
  69.     //! \brief A new acknowledgment was received
  70.     void ack_received(const WrappingInt32 ackno, const uint16_t window_size);
  72.     //! \brief Generate an empty-payload segment (useful for creating empty ACK segments)
  73.     void send_empty_segment();
  75.     //! \brief create and send segments to fill as much of the window as possible
  76.     void fill_window();
  78.     // 每隔几毫秒，您的TCPSender的tick方法将被调用，并带有一个参数，该参数指示自上次调用该方法以来已经过了几毫秒
  79.     //! \brief Notifies the TCPSender of the passage of time
  80.     void tick(const size_t ms_since_last_tick);
  81.     //!@}
  83.     //! \name Accessors
  84.     //!@{
  86.     // 已发送但未确认的字节数，SYN和FIN各占1字节
  87.     //! \brief How many sequence numbers are occupied by segments sent but not yet acknowledged?
  88.     //! \note count is in "sequence space," i.e. SYN and FIN each count for one byte
  89.     //! (see TCPSegment::length_in_sequence_space())
  90.     size_t bytes_in_flight() const;
  92.     // 连续重传次数
  93.     //! \brief Number of consecutive retransmissions that have occurred in a row
  94.     unsigned int consecutive_retransmissions() const;
  96.     //! \brief TCPSegments that the TCPSender has enqueued for transmission.
  97.     //! \note These must be dequeued and sent by the TCPConnection,
  98.     //! which will need to fill in the fields that are set by the TCPReceiver
  99.     //! (ackno and window size) before sending.
  100.     std::queue<TCPSegment> &segments_out() { return _segments_out; }
  101.     //!@}
  103.     //! \name What is the next sequence number? (used for testing)
  104.     //!@{
  106.     //! \brief absolute seqno for the next byte to be sent
  107.     uint64_t next_seqno_absolute() const { return _next_seqno; }
  109.     //! \brief relative seqno for the next byte to be sent
  110.     WrappingInt32 next_seqno() const { return wrap(_next_seqno, _isn); }
  111.     //!@}
  112. };
  114. #endif  // SPONGE_LIBSPONGE_TCP_SENDER_HH
  117. #include "tcp_sender.hh"
  118. #include "tcp_config.hh"
  120. #include <random>
  121. #include <iostream>
  122. #include <cmath>
  124. // Dummy implementation of a TCP sender
  126. // For Lab 3, please replace with a real implementation that passes the
  127. // automated checks run by `make check_lab3`.
  129. template <typename... Targs>
  130. void DUMMY_CODE(Targs &&... /* unused */) {}
  132. using namespace std;
  134. //! \param[in] capacity the capacity of the outgoing byte stream
  135. //! \param[in] retx_timeout the initial amount of time to wait before retransmitting the oldest outstanding segment
  136. //! \param[in] fixed_isn the Initial Sequence Number to use, if set (otherwise uses a random ISN)
  137. TCPSender::TCPSender(const size_t capacity, const uint16_t retx_timeout, const std::optional<WrappingInt32> fixed_isn)
  138.     : _isn(fixed_isn.value_or(WrappingInt32{random_device()()}))
  139.     , _initial_retransmission_timeout{retx_timeout}
  140.     , _stream(capacity)
  141.     ,_lately_ackno(0), _window_size(1)  // fixed
  142.     ,_bytes_in_flight(0), _segments_outstanding()
  143.     ,_retransmission_timeout{retx_timeout}, _consecutive_retransmission(0)
  144.     ,_timer_ms(0), _timer_running(false) {}
  146. // _segments_outstanding中所有报文段的和，包括SYN和FIN
  147. uint64_t TCPSender::bytes_in_flight() const { 
  148.     return _bytes_in_flight;
  149. }
  151. void TCPSender::fill_window() {
  152.     if (_next_seqno==0) {  // 状态0 CLOSED
  153.         // 发送不带数据的SYN报文
  154.         TCPSegment seg;
  155.         seg.header().syn=true;
  157.         send_segment(seg);
  158.         std::cout<<"~~~~~~~~~~~~~~~~~~~~~~~~  start with RTO: "<<_retransmission_timeout<<std::endl;
  159.     } else if (_next_seqno>0&&bytes_in_flight()==_next_seqno) {  // 状态1 SYN_SENT
  160.         // 在未收到任何ACK之前，假设接收方的窗口大小为1，所以这里不能再发送任何数据
  161.     } else if (!_stream.eof()&&bytes_in_flight()<_next_seqno) {  // 状态2 SYN_ACKED
  162.         /*  考察区间[_next_seqno, _lately_ackno+_window_size) 
  163.             如果区间不存在，则不能发送新的数据
  164.             如果window_size==0，则可以发送1个字节的数据
  165.         */
  166.         size_t window_size=_window_size==0?1:_window_size;  // fixed
  167.         if (_lately_ackno+window_size>_next_seqno&&!_stream.buffer_empty()) {  // fixed
  168.             // 注意MAX_PAYLOAD_SIZE=1452字节 fixed
  169.             // 有可以发送的数据
  170.             size_t size_to_send = _lately_ackno+window_size-_next_seqno;
  171.             while (size_to_send>0&&!_stream.buffer_empty()) {
  172.                 size_t size_to_read = std::min(size_to_send, TCPConfig::MAX_PAYLOAD_SIZE);
  173.                 // std::string data_to_send = _stream.read(_lately_ackno+window_size-_next_seqno);
  174.                 std::string data_to_send = _stream.read(size_to_read);
  175.                 size_to_send-=data_to_send.size();
  176.                 std::cout<<"****** size_to_read: "<<size_to_read<<" data_to_send.size(): "<<data_to_send.size();
  177.                 TCPSegment seg;
  178.                 seg.payload() = Buffer(std::move(data_to_send));
  179.                 // 如果stream读完了，且还有余下的空间放置FIN
  180.                 if (_stream.eof()&&size_to_send>0) {
  181.                     // 这样的话循环条件一定不成立，所以size_to_send也没必要再-1了
  182.                     seg.header().fin=true;
  183.                 }
  184.                 send_segment(seg);
  185.                 std::cout<<" _segments_out.size(): "<<_segments_out.size()<<std::endl;
  186.                 if (_segments_out.size()) {
  187.                     std::cout<<_segments_out.front().payload().str()<<std::endl;
  188.                 }
  189.             }
  190.         }
  191.     } else if (_stream.eof()&&_next_seqno<_stream.bytes_written()+2) {  // 状态3 SYN_ACKED and FIN not Sent
  192.         // 这种状态下一定只剩下1个FIN没放进TCPSegment的队列
  193.         // 接收窗口也可能是0吧？
  194.         size_t window_size=_window_size==0?1:_window_size;  // fixed
  195.         if (_lately_ackno+window_size>_next_seqno) {
  196.             TCPSegment seg;
  197.             seg.header().fin=true;
  198.             send_segment(seg);
  199.         }
  200.     } else if (_stream.eof()&&_next_seqno==_stream.bytes_written()+2&&bytes_in_flight()>0) {  // 状态4 FIN_SENT
  201.         // 这里不用发送数据
  202.     } else if (_stream.eof()&&_next_seqno==_stream.bytes_written()+2&&bytes_in_flight()==0) {  // 状态5 FIN_ACKED
  203.         // 这里不用发送数据
  204.     } else {  // 应该不会到这个分支
  206.     }
  207. }
  209. //! \param ackno The remote receiver's ackno (acknowledgment number)
  210. //! \param window_size The remote receiver's advertised window size
  211. void TCPSender::ack_received(const WrappingInt32 ackno, const uint16_t window_size) { 
  212.     size_t _ackno=unwrap(ackno, _isn, _lately_ackno);
  213.     // Impossible ackno (beyond next seqno) is ignored
  214.     if (_ackno>_next_seqno) {
  215.         return;
  216.     }
  217.     std::cout<<"***** _ackno: "<<_ackno<<" _lately_ackno: "<<_lately_ackno<<std::endl;
  218.     if (_ackno>=_lately_ackno) {
  219.         // if (_ackno>_lately_ackno) {  // 这样的话，部分确认也会重置RTO和清零连续重传次数
  220.         //     _consecutive_retransmission=0;
  221.         //     _retransmission_timeout=_initial_retransmission_timeout;
  222.         // }
  223.         _lately_ackno=_ackno;
  224.         _window_size=window_size;  // 这里window_size可能是0
  225.     }
  226.     bool seg_acked=false;
  227.     while (_segments_outstanding.size()) {
  228.         TCPSegment seg = _segments_outstanding.front();
  229.         size_t abs_seqno = unwrap(seg.header().seqno, _isn, _lately_ackno);
  230.         if (abs_seqno+seg.length_in_sequence_space()<=_lately_ackno) {
  231.             seg_acked=true;
  232.             _segments_outstanding.pop();
  233.             _bytes_in_flight-=seg.length_in_sequence_space();  // fixed
  234.         } else {
  235.             break;
  236.         }
  237.     }
  238.     // 我认为这里_timer_running一定是true
  239.     if (_segments_outstanding.empty()) {
  240.         _timer_running=false;  // 关闭定时器
  241.     } else if (seg_acked) {  // 重启定时器
  242.         _timer_running=true;
  243.         _timer_ms=0;
  244.     }
  245.     // 这样的话，部分确认不会重置RTO和清零连续重传次数，也不会调用fill_window()
  246.     if (seg_acked) {
  247.         _consecutive_retransmission=0;
  248.         _retransmission_timeout=_initial_retransmission_timeout;
  249.         fill_window();  // fixed
  250.     }
  251.     // 如果接收窗口有了新的空间，TCPSender应该再次填充该窗口。
  252.     // todo
  253. }
  255. //! \param[in] ms_since_last_tick the number of milliseconds since the last call to this method
  256. void TCPSender::tick(const size_t ms_since_last_tick) { 
  257.     // 目前没有考虑窗口大小为0的情况
  258.     if (_timer_running) {
  259.         _timer_ms+=ms_since_last_tick;
  260.         if (_timer_ms>=_retransmission_timeout) {  // 定时器超时，需要重传最早的未被确认的报文段
  261.             _timer_ms=0;  // 重启定时器
  262.             // if (_window_size>0||(_next_seqno>0&&bytes_in_flight()==_next_seqno)) {  // fixed
  263.             if (_window_size>0) {  // fixed
  264.                 _consecutive_retransmission++;  // 重传次数加一
  265.                 _retransmission_timeout<<=1;  // RTO加倍
  266.             }
  267.             if (_segments_outstanding.size()) {  // 这个判断应该一定是成立的吧？
  268.                 _segments_out.push(_segments_outstanding.front());
  269.             }
  270.         }
  271.     }
  272. }
  274. unsigned int TCPSender::consecutive_retransmissions() const { 
  275.     return _consecutive_retransmission;
  276. }
  278. // 不带SYN、FIN，且不包含数据，只需要设置正确的seqno
  279. void TCPSender::send_empty_segment() {
  280.     TCPSegment seg;
  281.     seg.header().seqno=wrap(_next_seqno, _isn);
  282.     _segments_out.push(seg);
  283.     // 不用加入_segments_outstanding，也不用启动定时器
  284. }
  286. void TCPSender::send_segment(TCPSegment &seg) {
  287.     seg.header().seqno=wrap(_next_seqno, _isn);
  288.     _next_seqno+=seg.length_in_sequence_space();
  289.     _bytes_in_flight+=seg.length_in_sequence_space();
  291.     _segments_out.push(seg);
  292.     _segments_outstanding.push(seg);
  294.     // 每次发送报文段都要检查定时器，如果未启动则启动
  295.     if (!_timer_running) {
  296.         _timer_ms=0;
  297.         _timer_running=true;
  298.     }
  299. }