lab3 done right but ugly

libsponge/tcp_sender.cc

@@ -10,7 +10,7 @@
 // automated checks run by `make check_lab3`.
 
 template <typename... Targs>
-void DUMMY_CODE(Targs &&... /* unused */) {}
+void DUMMY_CODE(Targs &&.../* unused */) {}
 
 using namespace std;
 
@@ -20,19 +20,98 @@ using namespace std;
 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) {}
+    , _stream(capacity) {
+    _timer.reinit(_initial_retransmission_timeout);
+}
 
-uint64_t TCPSender::bytes_in_flight() const { return {}; }
+uint64_t TCPSender::bytes_in_flight() const { return _next_seqno - _unack_seqno; }
 
-void TCPSender::fill_window() {}
+void TCPSender::fill_window() {
+    if (_stream.eof() && _next_seqno - _stream.bytes_read() == 2)
+        return;
+    TCPHeader hdr;
+    hdr.seqno = next_seqno();
+    if (_next_seqno == 0) {
+        hdr.syn = true;
+    }
+    auto _fake_window_size = _window_size ? _window_size : 1UL;
+    auto occupied = _next_seqno - _unack_seqno;
+    auto data = _stream.read(_fake_window_size - occupied);
+    size_t sendptr = 0;
+    std::string payload = data.substr(sendptr, TCPConfig::MAX_PAYLOAD_SIZE);
+    if (data.size() > TCPConfig::MAX_PAYLOAD_SIZE)
+        while (payload.size() == TCPConfig::MAX_PAYLOAD_SIZE) {
+            TCPSegment seg;
+            seg.header() = hdr;
+            seg.payload() = Buffer(std::move(payload));
+            _segments_out.push(seg);
+            _outstandings.push(seg);
+            sendptr += TCPConfig::MAX_PAYLOAD_SIZE;
+            _next_seqno += seg.length_in_sequence_space();
+            hdr.syn = false;
+            hdr.seqno = next_seqno();
+            if (sendptr >= data.size())
+                break;
+            payload = data.substr(sendptr, TCPConfig::MAX_PAYLOAD_SIZE);
+        }
+    occupied = _next_seqno - _unack_seqno + payload.size();
+    if (_stream.eof() && _fake_window_size > occupied) {
+        hdr.fin = true;
+    }
+    TCPSegment seg;
+    seg.header() = hdr;
+    seg.payload() = Buffer(std::move(payload));
+    if (seg.length_in_sequence_space()) {
+        _segments_out.push(seg);
+        _outstandings.push(seg);
+        _next_seqno += seg.length_in_sequence_space();
+    }
+    _timer.start();
+}
 
 //! \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) { DUMMY_CODE(ackno, window_size); }
+void TCPSender::ack_received(const WrappingInt32 ackno, const uint16_t window_size) {
+    auto abs_ackno = unwrap(ackno, _isn, _next_seqno);
+    if (abs_ackno > _next_seqno)
+        return;
+    _window_size = window_size;
+    while (!_outstandings.empty()) {
+        // auto abs_seqno = unwrap(_outstandings.front().header().seqno, _isn, _next_seqno);
+        if (abs_ackno >= _unack_seqno + _outstandings.front().length_in_sequence_space()) {
+            _unack_seqno += _outstandings.front().length_in_sequence_space();
+            _outstandings.pop();
+            _timer.reinit(_initial_retransmission_timeout);
+        } else {
+            break;
+        }
+    }
+    if (_outstandings.empty()) {
+        _timer.stop();
+    } else {
+        _timer.start();
+    }
+}
 
 //! \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) { DUMMY_CODE(ms_since_last_tick); }
+void TCPSender::tick(const size_t ms_since_last_tick) {
+    if (!_timer.started())
+        return;
+    _timer.count_down(ms_since_last_tick);
+    if (!_timer.expired())
+        return;
+    _segments_out.push(_outstandings.front());
+    if (_window_size > 0)
+        _timer.consec_retrans();
+    _timer.reset();
+}
 
-unsigned int TCPSender::consecutive_retransmissions() const { return {}; }
+unsigned int TCPSender::consecutive_retransmissions() const { return _timer.consec_retrans_count(); }
 
-void TCPSender::send_empty_segment() {}
+void TCPSender::send_empty_segment() {
+    TCPHeader hdr;
+    hdr.seqno = wrap(_next_seqno - (_next_seqno ? 1 : 0), _isn);
+    TCPSegment seg;
+    seg.header() = hdr;
+    _segments_out.push(seg);
+}

libsponge/tcp_sender.hh

@@ -9,6 +9,43 @@
 #include <functional>
 #include <queue>
 
+//! \brief Timer helper class suggested in the handout
+
+//! God knows what this class does
+class TCPTimer {
+  private:
+    size_t RTO{0};
+    size_t rt_count{0};
+    size_t ti_count{0};
+    bool timer_on{false};
+
+  public:
+    TCPTimer() = default;
+    void reinit(uint32_t initial_RTO) {
+        RTO = initial_RTO;
+        ti_count = RTO;
+        rt_count = 0;
+        timer_on = false;
+    }
+    void reset() { ti_count = RTO; }
+    size_t cur_RTO() const { return RTO; }
+    void start() { timer_on = true; }
+    bool started() const { return timer_on; }
+    void stop() { timer_on = false; }
+    bool expired() const { return ti_count == 0; }
+    void count_down(size_t milli_sec) {
+        if (ti_count >= milli_sec)
+            ti_count -= milli_sec;
+        else
+            ti_count = 0;
+    }
+    void consec_retrans() {
+        rt_count++;
+        RTO <<= 1;
+    }
+    size_t consec_retrans_count() const { return rt_count; }
+};
+
 //! \brief The "sender" part of a TCP implementation.
 
 //! Accepts a ByteStream, divides it up into segments and sends the
@@ -31,6 +68,12 @@ class TCPSender {
 
     //! the (absolute) sequence number for the next byte to be sent
     uint64_t _next_seqno{0};
+    uint64_t _unack_seqno{0};
+
+    //! current window size, default (and at least) should be 1 byte according to the handout
+    uint16_t _window_size{1};
+    std::queue<TCPSegment> _outstandings{};
+    TCPTimer _timer{};
 
   public:
     //! Initialize a TCPSender