Add loss detection to BBR.

prrt/cprrt.pxd

@@ -156,6 +156,7 @@ cdef extern from "proto/socket.h":
     uint32_t PrrtSocket_get_inflight(PrrtSocket *s)
     uint32_t PrrtSocket_get_pacing_rate(PrrtSocket *s)
     uint32_t PrrtSocket_get_send_quantum(PrrtSocket *s)
+    uint32_t PrrtSocket_get_pipe(PrrtSocket *s)
 
 
     bint PrrtSocket_get_app_limited(PrrtSocket *socket)

prrt/proto/bbr.c

@@ -51,13 +51,13 @@ void BBR_CheckFullPipe(BBR* bbr, PrrtRateSample* rs)
         bbr->filled_pipe = true;
 }
 
-bool BBR_IsNextCyclePhase(BBR* bbr, prrtSequenceNumber_t packets_lost, prrtByteCount_t prior_inflight)
+bool BBR_IsNextCyclePhase(BBR* bbr, prrtByteCount_t bytes_lost, prrtByteCount_t prior_inflight)
 {
     bool is_full_length = (PrrtClock_get_current_time_us() - bbr->cycle_stamp) > bbr->rtprop;
     if (bbr->pacing_gain == 1)
         return is_full_length;
     if (bbr->pacing_gain > 1)
-        return is_full_length && (packets_lost > 0 || prior_inflight >= BBR_Inflight(bbr, bbr->pacing_gain));
+        return is_full_length && (bytes_lost > 0 || prior_inflight >= BBR_Inflight(bbr, bbr->pacing_gain));
     return is_full_length || (prior_inflight <= BBR_Inflight(bbr, 1.0));
 }
 
@@ -70,8 +70,8 @@ void BBR_AdvanceCyclePhase(BBR* bbr)
     debug(DEBUG_BBR, "Advanced cycle with gain: %f", bbr->pacing_gain);
 }
 
-void BBR_CheckCyclePhase(BBR* bbr, prrtSequenceNumber_t packets_lost, prrtByteCount_t prior_inflight) {
-    if (bbr->state == PROBE_BW && BBR_IsNextCyclePhase(bbr, packets_lost, prior_inflight))
+void BBR_CheckCyclePhase(BBR* bbr, prrtByteCount_t bytes_lost, prrtByteCount_t prior_inflight) {
+    if (bbr->state == PROBE_BW && BBR_IsNextCyclePhase(bbr, bytes_lost, prior_inflight))
         BBR_AdvanceCyclePhase(bbr);
 }
 
@@ -169,7 +169,7 @@ void BBR_UpdateModelAndState(BBR *bbr, PrrtChannelStateInformation *csi, PrrtRat
                              PrrtPacketTracking *packetTracking)
 {
     BBR_UpdateBtlBw(bbr, rs, packetTracking);
-    BBR_CheckCyclePhase(bbr, packetTracking->packets_lost, packetTracking->prior_inflight);
+    BBR_CheckCyclePhase(bbr, packetTracking->bytes_lost, packetTracking->prior_inflight);
     BBR_CheckFullPipe(bbr, rs);
     BBR_CheckDrain(bbr, packetTracking->pipe);
     BBR_UpdateRTprop(bbr, PrrtChannelStateInformation_get_rtprop(csi));
@@ -187,20 +187,20 @@ void BBR_ModulateCwndForProbeRTT(BBR* bbr)
         bbr->cwnd = MIN(bbr->cwnd, BBRMinPipeCwnd);
 }
 
-void BBR_ModulateCwndForRecovery(BBR* bbr, prrtSequenceNumber_t packets_lost, prrtSequenceNumber_t packets_in_flight, prrtSequenceNumber_t packets_delivered)
+void BBR_ModulateCwndForRecovery(BBR* bbr, prrtByteCount_t bytes_lost, prrtByteCount_t pipe, prrtByteCount_t delivered)
 {
     // TODO: [cwnd] = Bytes not packets
-    if (packets_lost > 0)
-        bbr->cwnd = MAX(bbr->cwnd - packets_lost, 1);
+    if (bytes_lost > 0)
+        bbr->cwnd = MAX(bbr->cwnd - bytes_lost, SMSS);
     if (bbr->packet_conservation)
-        bbr->cwnd = MAX(bbr->cwnd, packets_in_flight + packets_delivered);
+        bbr->cwnd = MAX(bbr->cwnd, pipe + delivered);
 }
 
 
 void BBR_SetCwnd(BBR* bbr, PrrtPacketTracking* packetTracking)
 {
     BBR_UpdateTargetCwnd(bbr);
-    BBR_ModulateCwndForRecovery(bbr, packetTracking->packets_lost, packetTracking->packets_in_flight, packetTracking->packets_delivered);
+    BBR_ModulateCwndForRecovery(bbr, packetTracking->bytes_lost, packetTracking->pipe, packetTracking->delivered);
     if (!bbr->packet_conservation) {
         if (bbr->filled_pipe)
             bbr->cwnd = MIN(bbr->cwnd + packetTracking->delivered, bbr->target_cwnd);
@@ -228,9 +228,9 @@ void BBR_SetPacingRate(BBR* bbr)
 
 void BBR_SetSendQuantum(BBR* bbr) {
     if(bbr->pacing_rate < 150000) { // 1.2Mbps = 0.15 MBps = 150000 Bps
-        bbr->send_quantum = 1 * MTU;
+        bbr->send_quantum = 1 * SMSS;
     } else if (bbr->pacing_rate < 3000000) { // 24 Mbps = 20 * 1.2Mbps = 3000000
-        bbr->send_quantum = 2 * MTU;
+        bbr->send_quantum = 2 * SMSS;
     } else {
         bbr->send_quantum = MIN((prrtByteCount_t) round(((double)bbr->pacing_rate) / 1000), 64000);
     }

prrt/proto/bbr.h

@@ -10,12 +10,12 @@
 #include "types/rateSample.h"
 #include "../util/windowedFilter.h"
 
-#define MTU 1200
+#define SMSS 1200
 #define RTpropFilterLen 10000000    //10s
 #define BBRHighGain ((2885 / 1000) + 1)
-#define InitialCwnd (10 * MTU)
+#define InitialCwnd (10 * SMSS)
 #define BBRGainCycleLen 8
-#define BBRMinPipeCwnd (2 * MTU)
+#define BBRMinPipeCwnd (2 * SMSS)
 #define ProbeRTTDuration 200000 //200ms
 #define Inf UINT32_MAX
 

prrt/proto/receiver.c

@@ -173,15 +173,15 @@ bool PrrtReceiver_updateAndGenerateRateSample(PrrtReceiver *recv, prrtSequenceNu
         packet = PrrtInFlightPacketStore_get_packet(inflightPacketStore, seqnum);
     }
     recv->packetTracking->pipe -= packet->payloadLength;
-    PrrtReceiver_updateRateSample(recv->rateSample, packet, receiveTime, recv->packetTracking);
+    prrtByteCount_t lostBytes = PrrtInFlightPacketStore_remove_outstanding_packet(inflightPacketStore, seqnum);
+    recv->packetTracking->pipe -= lostBytes;
+    recv->packetTracking->bytes_lost = lostBytes;
 
+    PrrtReceiver_updateRateSample(recv->rateSample, packet, receiveTime, recv->packetTracking);
     bool result = PrrtReceiver_generateRateSample(recv->rateSample, recv->packetTracking);
-    PrrtInFlightPacketStore_remove_outstanding_packet(inflightPacketStore, seqnum);
     check(pthread_mutex_unlock(&recv->lock) == 0, "Unlock failed.");
     check(pthread_cond_broadcast(&recv->pipeNotFullCv) == 0, "Signal failed.");
 
-    recv->packetTracking->packets_delivered++;
-    recv->packetTracking->packets_in_flight = (prrtSequenceNumber_t) (PrrtInFlightPacketStore_get_queue_size(recv->dataInflightPacketStore) + PrrtInFlightPacketStore_get_queue_size(recv->redundancyInflightPacketStore));
     recv->packetTracking->prior_inflight = recv->packetTracking->pipe;
     if(recv->rateSample != NULL && result) {
         BBR_OnACK(recv->bbr, recv->csi, recv->rateSample, recv->packetTracking);

prrt/proto/socket.c

@@ -636,6 +636,9 @@ prrtByteCount_t PrrtSocket_get_send_quantum(PrrtSocket *s) {
     return BBR_getSendQuantum(s->receiver->bbr);
 };
 
+prrtByteCount_t PrrtSocket_get_pipe(PrrtSocket *s) {
+    return s->receiver->packetTracking->pipe;
+};
 
 float PrrtSocket_get_pacing_gain(PrrtSocket *s) {
     return BBR_getPacingGain(s->receiver->bbr);
@@ -645,7 +648,6 @@ uint32_t PrrtSocket_get_pacing_rate(PrrtSocket *s) {
     return BBR_getPacingRate(s->receiver->bbr);
 }
 
-
 uint32_t PrrtSocket_get_cycle_index(PrrtSocket *s) {
     return BBR_getCycleIndex(s->receiver->bbr);
 }

prrt/proto/socket.h

@@ -147,6 +147,7 @@ uint32_t PrrtSocket_get_cwnd(PrrtSocket *s);
 uint32_t PrrtSocket_get_inflight(PrrtSocket *s);
 uint32_t PrrtSocket_get_pacing_rate(PrrtSocket *s);
 prrtByteCount_t PrrtSocket_get_send_quantum(PrrtSocket *s);
+prrtByteCount_t PrrtSocket_get_pipe(PrrtSocket *s);
 
 
 #endif // PRRT_SOCKET_H

prrt/proto/stores/inFlightPacketStore.c

@@ -30,12 +30,31 @@ void* PrrtInFlightPacketStore_get_packet(PrrtInFlightPacketStore *packetStore, B
     return BPTree_get(packetStore->outstandingPackets, seqNum);
 }
 
-void PrrtInFlightPacketStore_remove_outstanding_packet(PrrtInFlightPacketStore *packetStore, BPTreeKey_t seqNum)
+prrtByteCount_t PrrtInFlightPacketStore_remove_outstanding_packet(PrrtInFlightPacketStore *packetStore, BPTreeKey_t seqNum)
 {
+    prrtByteCount_t lostBytes = 0;
     PrrtPacket* value = (PrrtPacket*) BPTree_get(packetStore->outstandingPackets, seqNum);
     PrrtPacket_destroy(value);
     packetStore->outstandingPackets = BPTree_delete(packetStore->outstandingPackets, seqNum);
     packetStore->packetQueueSize--;
+
+    // TODO: Cleanup and loss marking. Make this more approriate with a "timeout".
+    List *packetStateList = List_create();
+    if(seqNum > SEQNO_SPACE/2) {
+        BPTree_get_range(packetStore->outstandingPackets, packetStateList, seqNum - (SEQNO_SPACE / 2), seqNum - 1);
+    } else {
+        BPTree_get_range(packetStore->outstandingPackets, packetStateList, 0, seqNum - 1);
+        BPTree_get_range(packetStore->outstandingPackets, packetStateList, seqNum + (SEQNO_SPACE / 2), SEQNO_SPACE - 1);
+    }
+
+    while (List_count(packetStateList) > 0) {
+        PrrtPacket *packet = List_shift(packetStateList);
+        lostBytes += packet->payloadLength;
+        packetStore->outstandingPackets = BPTree_delete(packetStore->outstandingPackets, packet->sequenceNumber);
+        PrrtPacket_destroy(packet);
+    }
+    List_destroy(packetStateList);
+    return lostBytes;
 }
 
 bool PrrtInFlightPacketStore_destroy(PrrtInFlightPacketStore *packetStore)

prrt/proto/stores/inFlightPacketStore.h

@@ -11,7 +11,7 @@ typedef struct inFlightPackets {
 
 PrrtInFlightPacketStore* PrrtInFlightPacketStore_create(void);
 void PrrtInFlightPacketStore_add_outstanding_packet(PrrtInFlightPacketStore *packetStore, PrrtPacket *packet);
-void PrrtInFlightPacketStore_remove_outstanding_packet(PrrtInFlightPacketStore *packetStore, BPTreeKey_t seqNum);
+prrtByteCount_t PrrtInFlightPacketStore_remove_outstanding_packet(PrrtInFlightPacketStore *packetStore, BPTreeKey_t seqNum);
 bool PrrtInFlightPacketStore_destroy(PrrtInFlightPacketStore *packetStore);
 uint32_t PrrtInFlightPacketStore_get_queue_size(PrrtInFlightPacketStore *packetStore);
 void* PrrtInFlightPacketStore_get_packet(PrrtInFlightPacketStore *packetStore, BPTreeKey_t seqNum);

prrt/proto/types/packetTracking.h

@@ -9,10 +9,8 @@ typedef struct packetTracking {
     prrtTimestamp_t delivered_time;
     prrtTimestamp_t first_sent_time;
     prrtByteCount_t app_limited;
-    prrtSequenceNumber_t packets_lost;
+    prrtByteCount_t bytes_lost;
     prrtByteCount_t prior_inflight;
-    prrtSequenceNumber_t packets_in_flight;
-    prrtSequenceNumber_t packets_delivered;
 } PrrtPacketTracking;
 
 #endif //PRRT_PACKETTRACKING_H

prrt/prrt.pyx

@@ -247,6 +247,12 @@ cdef class PrrtSocket:
                 raise Exception("Not a sender.")
             return cprrt.PrrtSocket_get_send_quantum(self._c_socket)
 
+    property bbr_pipe:
+        def __get__(self):
+            if not self.isSender:
+                raise Exception("Not a sender.")
+            return cprrt.PrrtSocket_get_pipe(self._c_socket)
+
     property app_limited:
         def __get__(self):
             if not self.isSender: