add handling for short circuit loss in SendQueue

libraries/networking/src/udt/Connection.cpp

@@ -103,6 +103,7 @@ SendQueue& Connection::getSendQueue() {
         QObject::connect(_sendQueue.get(), &SendQueue::packetRetransmitted, this, &Connection::recordRetransmission);
         QObject::connect(_sendQueue.get(), &SendQueue::queueInactive, this, &Connection::queueInactive);
         QObject::connect(_sendQueue.get(), &SendQueue::timeout, this, &Connection::queueTimeout);
+        QObject::connect(_sendQueue.get(), &SendQueue::shortCircuitLoss, this, &Connection::queueShortCircuitLoss);
         
         // set defaults on the send queue from our congestion control object and estimatedTimeout()
         _sendQueue->setPacketSendPeriod(_congestionControl->_packetSendPeriod);
@@ -140,6 +141,12 @@ void Connection::queueTimeout() {
     });
 }
 
+void Connection::queueShortCircuitLoss(quint32 sequenceNumber) {
+    updateCongestionControlAndSendQueue([this, sequenceNumber]{
+        _congestionControl->onLoss(SequenceNumber { sequenceNumber }, SequenceNumber { sequenceNumber });
+    });
+}
+
 void Connection::sendReliablePacket(std::unique_ptr<Packet> packet) {
     Q_ASSERT_X(packet->isReliable(), "Connection::send", "Trying to send an unreliable packet reliably.");
     getSendQueue().queuePacket(std::move(packet));

libraries/networking/src/udt/Connection.h

@@ -87,6 +87,7 @@ private slots:
     void recordRetransmission();
     void queueInactive();
     void queueTimeout();
+    void queueShortCircuitLoss(quint32 sequenceNumber);
     
 private:
     void sendACK(bool wasCausedBySyncTimeout = true);

libraries/networking/src/udt/SendQueue.cpp

@@ -128,8 +128,8 @@ void SendQueue::stop() {
     _emptyCondition.notify_one();
 }
     
-void SendQueue::sendPacket(const Packet& packet) {
-    _socket->writeDatagram(packet.getData(), packet.getDataSize(), _destination);
+int SendQueue::sendPacket(const Packet& packet) {
+    return _socket->writeDatagram(packet.getData(), packet.getDataSize(), _destination);
 }
     
 void SendQueue::ack(SequenceNumber ack) {
@@ -232,15 +232,16 @@ SequenceNumber SendQueue::getNextSequenceNumber() {
     return _currentSequenceNumber;
 }
 
-void SendQueue::sendNewPacketAndAddToSentList(std::unique_ptr<Packet> newPacket, SequenceNumber sequenceNumber) {
+bool SendQueue::sendNewPacketAndAddToSentList(std::unique_ptr<Packet> newPacket, SequenceNumber sequenceNumber) {
     // write the sequence number and send the packet
     newPacket->writeSequenceNumber(sequenceNumber);
-    sendPacket(*newPacket);
 
     // Save packet/payload size before we move it
     auto packetSize = newPacket->getDataSize();
     auto payloadSize = newPacket->getPayloadSize();
     
+    auto bytesWritten = sendPacket(*newPacket);
+
     {
         // Insert the packet we have just sent in the sent list
         QWriteLocker locker(&_sentLock);
@@ -251,6 +252,22 @@ void SendQueue::sendNewPacketAndAddToSentList(std::unique_ptr<Packet> newPacket,
     Q_ASSERT_X(!newPacket, "SendQueue::sendNewPacketAndAddToSentList()", "Overriden packet in sent list");
 
     emit packetSent(packetSize, payloadSize);
+
+    if (bytesWritten < 0) {
+        // this is a short-circuit loss - we failed to put this packet on the wire
+        // so immediately add it to the loss list
+
+        {
+            std::lock_guard<std::mutex> nakLocker(_naksLock);
+            _naks.append(sequenceNumber);
+        }
+
+        emit shortCircuitLoss(quint32(sequenceNumber));
+
+        return false;
+    } else {
+        return true;
+    }
 }
 
 void SendQueue::run() {
@@ -285,12 +302,12 @@ void SendQueue::run() {
     auto nextPacketTimestamp = p_high_resolution_clock::now();
 
     while (_state == State::Running) {
-        bool sentAPacket = maybeResendPacket();
+        bool attemptedToSendPacket = maybeResendPacket();
         
         // if we didn't find a packet to re-send AND we think we can fit a new packet on the wire
         // (this is according to the current flow window size) then we send out a new packet
-        if (!sentAPacket) {
-            sentAPacket = maybeSendNewPacket();
+        if (!attemptedToSendPacket) {
+            attemptedToSendPacket = maybeSendNewPacket();
         }
         
         // since we're a while loop, give the thread a chance to process events
@@ -300,7 +317,7 @@ void SendQueue::run() {
         // If the send queue has been innactive, skip the sleep for
         // Either _isRunning will have been set to false and we'll break
         // Or something happened and we'll keep going
-        if (_state != State::Running || isInactive(sentAPacket)) {
+        if (_state != State::Running || isInactive(attemptedToSendPacket)) {
             return;
         }
 
@@ -325,6 +342,9 @@ bool SendQueue::maybeSendNewPacket() {
             std::unique_ptr<Packet> firstPacket = _packets.takePacket();
             Q_ASSERT(firstPacket);
 
+
+            // attempt to send the first packet
+            if (sendNewPacketAndAddToSentList(move(firstPacket), nextNumber)) {
                 std::unique_ptr<Packet> secondPacket;
                 bool shouldSendPairTail = false;
 
@@ -336,9 +356,6 @@ bool SendQueue::maybeSendNewPacket() {
                     secondPacket = _packets.takePacket();
                 }
 
-            // definitely send the first packet
-            sendNewPacketAndAddToSentList(move(firstPacket), nextNumber);
-            
                 // do we have a second in a pair to send as well?
                 if (secondPacket) {
                     sendNewPacketAndAddToSentList(move(secondPacket), getNextSequenceNumber());
@@ -349,8 +366,9 @@ bool SendQueue::maybeSendNewPacket() {
                     static auto pairTailPacket = ControlPacket::create(ControlPacket::ProbeTail);
                     _socket->writeBasePacket(*pairTailPacket, _destination);
                 }
+            }
 
-            // We sent our packet(s), return here
+            // We attempted to send packet(s), return here
             return true;
         }
     }
@@ -377,6 +395,7 @@ bool SendQueue::maybeResendPacket() {
             auto it = _sentPackets.find(resendNumber);
 
             if (it != _sentPackets.end()) {
+
                 auto& entry = it->second;
                 // we found the packet - grab it
                 auto& resendPacket = *(entry.second);
@@ -437,7 +456,7 @@ bool SendQueue::maybeResendPacket() {
     return false;
 }
 
-bool SendQueue::isInactive(bool sentAPacket) {
+bool SendQueue::isInactive(bool attemptedToSendPacket) {
     // check for connection timeout first
 
     // that will be the case if we have had 16 timeouts since hearing back from the client, and it has been
@@ -462,7 +481,7 @@ bool SendQueue::isInactive(bool sentAPacket) {
         return true;
     }
 
-    if (!sentAPacket) {
+    if (!attemptedToSendPacket) {
         // During our processing above we didn't send any packets
         
         // If that is still the case we should use a condition_variable_any to sleep until we have data to handle.

libraries/networking/src/udt/SendQueue.h

@@ -79,6 +79,7 @@ signals:
     
     void queueInactive();
 
+    void shortCircuitLoss(quint32 sequenceNumber);
     void timeout();
     
 private slots:
@@ -91,13 +92,13 @@ private:
     
     void sendHandshake();
     
-    void sendPacket(const Packet& packet);
-    void sendNewPacketAndAddToSentList(std::unique_ptr<Packet> newPacket, SequenceNumber sequenceNumber);
+    int sendPacket(const Packet& packet);
+    bool sendNewPacketAndAddToSentList(std::unique_ptr<Packet> newPacket, SequenceNumber sequenceNumber);
     
     bool maybeSendNewPacket(); // Figures out what packet to send next
     bool maybeResendPacket(); // Determines whether to resend a packet and which one
     
-    bool isInactive(bool sentAPacket);
+    bool isInactive(bool attemptedToSendPacket);
     void deactivate(); // makes the queue inactive and cleans it up
 
     bool isFlowWindowFull() const;