Add support for delivery rate estimation.

prrt/cprrt.pxd

@@ -71,7 +71,7 @@ cdef extern from "proto/block.h":
 
     ctypedef prrtBlock PrrtBlock
 
-cdef extern from "proto/packet.h":
+cdef extern from "proto/types/packet.h":
     cdef struct prrtPacket:
         uint8_t type_priority;
         uint8_t index;
@@ -145,6 +145,7 @@ cdef extern from "proto/socket.h":
     bint PrrtSocket_uses_thread_pinning(PrrtSocket *socket)
     uint32_t PrrtSocket_get_rtt(PrrtSocket *socket)
     float PrrtSocket_get_plr(PrrtSocket *socket)
+    uint32_t PrrtSocket_get_delivery_rate(PrrtSocket *socket)
     bint PrrtSocket_enable_thread_pinning(PrrtSocket *socket)
 
 cdef extern from "proto/stores/packetDeliveryStore.h":

prrt/proto/CMakeLists.txt

@@ -3,22 +3,23 @@ add_library(PRRT ../defines.h
         channelStateInformation.c channelStateInformation.h
         clock.c clock.h
         codingParams.c codingParams.h
-        packet.c packet.h
         receiver.c receiver.h
         socket.c socket.h
         ../xlap/xlap.c ../xlap/xlap.h
         applicationConstraints.c applicationConstraints.h
-        vdmcode/block_code.c vdmcode/block_code.h
-        stores/deliveredPacketTable.c stores/deliveredPacketTable.h
-        types/lossStatistics.c types/lossStatistics.h
         processes/dataReceiver.c processes/dataReceiver.h
         processes/feedbackReceiver.c processes/feedbackReceiver.h
         processes/dataTransmitter.c processes/dataTransmitter.h
-        stores/repairBlockStore.c stores/repairBlockStore.h
-        stores/packetTimeoutTable.c stores/packetTimeoutTable.h
         stores/dataPacketStore.c stores/dataPacketStore.h
-        types/packetTimeout.c types/packetTimeout.h
+        stores/deliveredPacketTable.c stores/deliveredPacketTable.h
+        stores/inFlightPacketStore.c stores/inFlightPacketStore.h
+        stores/packetTimeoutTable.c stores/packetTimeoutTable.h
         stores/packetDeliveryStore.c stores/packetDeliveryStore.h
-        stores/receptionTable.c stores/receptionTable.h)
+        stores/receptionTable.c stores/receptionTable.h
+        stores/repairBlockStore.c stores/repairBlockStore.h
+        types/packetTimeout.c types/packetTimeout.h
+        types/lossStatistics.c types/lossStatistics.h
+        types/packet.c types/packet.h
+        vdmcode/block_code.c vdmcode/block_code.h)
 
 target_link_libraries(PRRT rt)

prrt/proto/applicationConstraints.h

 #ifndef PRRT_NETWORKCONTRAINTS_H
 #define PRRT_NETWORKCONTRAINTS_H
 
-#include "packet.h"
+#include "types/packet.h"
 
 typedef struct applicationConstraints {
     prrtTimedelta_t targetDelay_us;

prrt/proto/block.c

@@ -3,7 +3,7 @@
 #include "../util/list.h"
 #include "../util/dbg.h"
 #include "../util/common.h"
-#include "packet.h"
+#include "types/packet.h"
 #include "block.h"
 
 static void gather_redundancy_packets(const PrrtBlock *block_ptr, gf *const *fec, int16_t *idx_p)

prrt/proto/block.h

@@ -3,7 +3,7 @@
 
 #include <stdbool.h>
 #include "codingParams.h"
-#include "packet.h"
+#include "types/packet.h"
 #include "../util/list.h"
 #include "vdmcode/block_code.h"
 

prrt/proto/channelStateInformation.c

@@ -14,6 +14,7 @@ PrrtChannelStateInformation * PrrtChannelStateInformation_create()
 
     csi->rtprop = TIMESTAMP_SPACE;
     csi->rtprop_filter_length_us = 2 * 1000 * 1000; // 2 seconds
+    csi->deliveryRate = 0;
     return csi;
 
     error:
@@ -41,6 +42,12 @@ void PrrtChannelStateInformation_update_plr(PrrtChannelStateInformation *csi, pr
 }
 
 
+void PrrtChannelStateInformation_update_delivery_rate(PrrtChannelStateInformation *csi, prrtDeliveryRate_t rate) {
+    pthread_mutex_lock(&csi->lock);
+    csi->deliveryRate = rate;
+    pthread_mutex_unlock(&csi->lock);
+}
+
 prrtTimedelta_t PrrtChannelStateInformation_get_rtprop(PrrtChannelStateInformation *csi)
 {
     pthread_mutex_lock(&csi->lock);
@@ -61,3 +68,7 @@ bool PrrtChannelStateInformation_destroy(PrrtChannelStateInformation *csi)
 prrtPacketLossRate_t PrrtChannelStateInformation_get_plr(PrrtChannelStateInformation *csi) {
     return csi->plr;
 }
+
+prrtDeliveryRate_t PrrtChannelStateInformation_get_delivery_rate(PrrtChannelStateInformation *csi) {
+    return csi->deliveryRate;
+}
\ No newline at end of file

prrt/proto/channelStateInformation.h

@@ -2,7 +2,7 @@
 #define PRRT_CHANNELSTATEINFORMATION_H
 
 #include <stdbool.h>
-#include "packet.h"
+#include "types/packet.h"
 
 typedef struct prrtChannelStateInformation {
     pthread_mutex_t lock;
@@ -11,14 +11,17 @@ typedef struct prrtChannelStateInformation {
     prrtTimedelta_t rtprop_filter_length_us;
     bool rtprop_expired;
     prrtPacketLossRate_t plr;
+    prrtDeliveryRate_t deliveryRate;
 } PrrtChannelStateInformation;
 
 PrrtChannelStateInformation* PrrtChannelStateInformation_create(void);
 void PrrtChannelStateInformation_update_rtprop(PrrtChannelStateInformation *csi, prrtTimedelta_t rtprop);
 prrtTimedelta_t PrrtChannelStateInformation_get_rtprop(PrrtChannelStateInformation *csi);
+prrtDeliveryRate_t PrrtChannelStateInformation_get_delivery_rate(PrrtChannelStateInformation *csi);
 prrtPacketLossRate_t PrrtChannelStateInformation_get_plr(PrrtChannelStateInformation* csi);
 void PrrtChannelStateInformation_update_plr(PrrtChannelStateInformation *csi, prrtSequenceNumber_t erasures,
                                                    prrtSequenceNumber_t packets);
+void PrrtChannelStateInformation_update_delivery_rate(PrrtChannelStateInformation *csi, prrtDeliveryRate_t rate);
 bool PrrtChannelStateInformation_destroy(PrrtChannelStateInformation* csi);
 
 void PrrtChannelStateInformation_print(PrrtChannelStateInformation *csi);

prrt/proto/clock.h

 #ifndef PRRT_CLOCK_H
 #define PRRT_CLOCK_H
 
-#include "packet.h"
+#include "types/packet.h"
 #include <stdint.h>
 #include <stdbool.h>
 

prrt/proto/processes/dataReceiver.c

@@ -50,7 +50,25 @@ static void decode_block(PrrtSocket *sock_ptr, PrrtBlock *block) {
     PERROR("Decoding failed.%s", "")
 }
 
-static bool send_feedback(PrrtSocket *sock_ptr, struct sockaddr_in remote) {
+static bool send_feedback(PrrtSocket *sock_ptr, struct sockaddr_in remote, prrtSequenceNumber_t seqno, prrtTimestamp_t receiveStamp,
+                          prrtTimestamp_t sentTimestamp, prrtPacketType_t type) {
+    enum XlapTimestampPacketKind kind = ts_data_packet;
+    if (type == PACKET_TYPE_DATA) {
+        kind = ts_data_packet;
+    } else if (type == PACKET_TYPE_REDUNDANCY) {
+        kind = ts_redundancy_packet;
+    }
+
+    debug(DEBUG_FEEDBACK, "Send feedback %d %d", type, seqno);
+    XlapTimeStampCycle(sock_ptr, kind, seqno, SendFeedbackStart);
+
+    prrtFeedback_t feedback = {
+            .seqNo = seqno,
+            .type = type,
+            .receivedTime = receiveStamp,
+            .sentTime = sentTimestamp
+    };
+
     uint16_t remote_port = ntohs(remote.sin_port);
     char *remote_host = inet_ntoa(remote.sin_addr);
 
@@ -72,7 +90,8 @@ static bool send_feedback(PrrtSocket *sock_ptr, struct sockaddr_in remote) {
     PrrtPacket *feedback_pkt_ptr = PrrtPacket_create_feedback_packet(0, sock_ptr->sequenceNumberFeedback++, group_RTT,
                                                                      stats.gapLength, stats.gapCount, stats.burstLength,
                                                                      stats.burstCount, forwardTripTime,
-                                                                     stats.erasureCount, stats.packetCount);
+                                                                     stats.erasureCount, stats.packetCount, feedback.seqNo,
+                                                                     feedback.type);
     prrtPacketLength_t length = PrrtPacket_size(feedback_pkt_ptr);
     void *buf = calloc(1, length);
     check_mem(buf);
@@ -84,6 +103,7 @@ static bool send_feedback(PrrtSocket *sock_ptr, struct sockaddr_in remote) {
     free(buf);
 
     PrrtPacket_destroy(feedback_pkt_ptr);
+    XlapTimeStampCycle(sock_ptr, kind, seqno, SendFeedbackEnd);
 
     return true;
 
@@ -98,11 +118,12 @@ static bool is_timeout(prrtTimestamp_t now, prrtTimestamp_t to) {
     return false;
 }
 
-static void handle_data_packet(PrrtSocket *sock_ptr, PrrtPacket *packet, struct sockaddr_in remote) {
+static void
+handle_data_packet(PrrtSocket *sock_ptr, PrrtPacket *packet) {
     PrrtPacketDataPayload *payload = packet->payload;
-    prrtTimestamp_t dataTimestamp = payload->timestamp;
-    sock_ptr->lastSentTimestamp = dataTimestamp;
-    PrrtClock_update(&sock_ptr->clock, dataTimestamp, payload->groupRTprop_us);
+    prrtTimestamp_t sentTimestamp = payload->timestamp;
+    sock_ptr->lastSentTimestamp = sentTimestamp;
+    PrrtClock_update(&sock_ptr->clock, sentTimestamp, payload->groupRTprop_us);
 
     PrrtPacketTimeout *packetTimeout = PrrtPacketTimeout_create(packet);
     check(PrrtPacketTimeoutTable_insert(sock_ptr->packetTimeoutTable, packetTimeout), "Could not insert data packet.");
@@ -110,14 +131,9 @@ static void handle_data_packet(PrrtSocket *sock_ptr, PrrtPacket *packet, struct
     prrtSequenceNumber_t seqno = packet->sequenceNumber;
     PrrtReceptionTable_mark_received(sock_ptr->dataReceptionTable, seqno);
 
-    XlapTimeStampCycle(sock_ptr, ts_data_packet, seqno, SendFeedbackStart);
-    check(send_feedback(sock_ptr, remote), "Sending feedback failed.");
-    XlapTimeStampCycle(sock_ptr, ts_data_packet, seqno, SendFeedbackEnd);
-
     prrtTimestamp_t now = PrrtClock_get_prrt_time_us(&sock_ptr->clock);
     if (is_timeout(now, payload->packetTimeout_us)) {
         PrrtDeliveredPacketTable_test_set_is_number_relevant(sock_ptr->deliveredPacketTable, packet->sequenceNumber);
-        // TODO: note this as loss
         PrrtPacket_destroy(packet);
         debug(DEBUG_RECEIVER, "timeout data packet %u (%lu > %lu)", seqno, (unsigned long) now,
               (unsigned long) payload->packetTimeout_us);
@@ -247,7 +263,8 @@ void *receive_data_loop(void *ptr) {
         receive_from_socket(sock_ptr, buffer, &n, &remote, &addrlen, &packet_recv_stamp);
         debug(DEBUG_HARDSTAMPING, "Packet TS:\t%ld.%09ld; Who? %s", (long) packet_recv_stamp.tv_sec,
               packet_recv_stamp.tv_nsec, inet_ntoa(remote.sin_addr));
-        sock_ptr->lastReceivedTimestamp = PrrtClock_TimespecToPrrtTimestamp(packet_recv_stamp);
+        prrtTimestamp_t prrt_recv_timestamp = PrrtClock_TimespecToPrrtTimestamp(packet_recv_stamp);
+        sock_ptr->lastReceivedTimestamp = prrt_recv_timestamp;
 
         XlapTimeStampClock(&tsph1, ts_any_packet, 0, LinkReceive);
         XlapTimeStampCycle(&tsph1, ts_any_packet, 0, LinkReceive);
@@ -263,11 +280,15 @@ void *receive_data_loop(void *ptr) {
 
         prrtPacketType_t packetType = PrrtPacket_type(packet);
         debug(DEBUG_DATARECEIVER, "received packet %d:%u", (int) packetType, seqno);
+
         enum XlapTimestampPacketKind kind = ts_any_packet;
+        prrtTimestamp_t sentTimestamp;
         if (packetType == PACKET_TYPE_DATA) {
             kind = ts_data_packet;
+            sentTimestamp = PrrtPacket_get_data_timestamp(packet);
         } else if (packetType == PACKET_TYPE_REDUNDANCY) {
             kind = ts_redundancy_packet;
+            sentTimestamp = PrrtPacket_get_redundancy_timestamp(packet);
         }
         if (packetType == PACKET_TYPE_DATA || packetType == PACKET_TYPE_REDUNDANCY) {
             XlapTimeStampValue(sock_ptr, kind, seqno, ChannelReceive, packet_recv_stamp);
@@ -277,8 +298,11 @@ void *receive_data_loop(void *ptr) {
             XlapTimestampPlaceholderUse(sock_ptr, kind, seqno, &tsph3);
 
             XlapTimeStampCycle(sock_ptr, kind, seqno, HandlePacketStart);
+
+            send_feedback(sock_ptr, remote, seqno, prrt_recv_timestamp, sentTimestamp, packetType);
+
             if (packetType == PACKET_TYPE_DATA) {
-                handle_data_packet(sock_ptr, packet, remote);
+                handle_data_packet(sock_ptr, packet);
             } else if (packetType == PACKET_TYPE_REDUNDANCY) {
                 handle_redundancy_packet(sock_ptr, packet);
             } else {

prrt/proto/processes/dataReceiver.h

 #ifndef PRRT_DATA_RECEIVER_H
 #define PRRT_DATA_RECEIVER_H
 
+#include "../types/packet.h"
+
+typedef struct prrtFeedback {
+    prrtSequenceNumber_t seqNo;
+    prrtPacketType_t type;
+    prrtTimestamp_t sentTime;
+    prrtTimestamp_t receivedTime;
+} prrtFeedback_t;
+
 void * receive_data_loop(void *ptr);
 
 #endif //PRRT_DATA_RECEIVER_H

prrt/proto/processes/dataTransmitter.c

@@ -95,6 +95,8 @@ static bool send_packet(PrrtSocket *sock_ptr, PrrtPacket *packet) {
     send_to_socket(sock_ptr, sock_ptr->receiver, buf, length, &timestamp);
     XlapTimeStampValue(sock_ptr, ts_data_packet, packet->sequenceNumber, ChannelTransmit, timestamp);
 
+    PrrtReceiver_add_outstanding_packet_state(sock_ptr->receiver, packet, PrrtClock_TimespecToPrrtTimestamp(timestamp));
+
     switch (PrrtPacket_type(packet)) {
         case PACKET_TYPE_DATA:
             XlapTimeStampClock(sock_ptr, ts_data_packet, packet->sequenceNumber, LinkTransmitEnd);
@@ -111,8 +113,6 @@ static bool send_packet(PrrtSocket *sock_ptr, PrrtPacket *packet) {
         default:;
     }
 
-    PrrtPacket_destroy(packet);
-
     return true;
 
     error:

prrt/proto/processes/feedbackReceiver.c

@@ -40,6 +40,10 @@ static void handle_feedback(PrrtSocket *prrtSocket, const size_t length)
     PrrtPacketFeedbackPayload *feedbackPayload = (PrrtPacketFeedbackPayload *) prrtPacket->payload;
     prrtTimestamp_t forwardTripTimestamp = feedbackPayload->forwardTripTimestamp_us;
 
+    PrrtReceiver_updateAndGenerateRateSample(prrtSocket->receiver, feedbackPayload->ackSequenceNumber, feedbackPayload->ackPacketType, receiveTime);
+
+    PrrtChannelStateInformation_update_delivery_rate(prrtSocket->receiver->csi, prrtSocket->receiver->rateSample->delivery_rate);
+
     PrrtChannelStateInformation_update_rtprop(prrtSocket->receiver->csi,
                                               (prrtTimedelta_t) (receiveTime - forwardTripTimestamp));
     PrrtChannelStateInformation_update_plr(prrtSocket->receiver->csi, feedbackPayload->erasureCount, feedbackPayload->packetCount);

prrt/proto/receiver.c

 #include <malloc.h>
+#include "../xlap/xlap.h"
 #include "../util/common.h"
 #include "../util/dbg.h"
+#include "stores/inFlightPacketStore.h"
 #include "receiver.h"
 
-PrrtReceiver *PrrtReceiver_create(const char *host, uint16_t port)
-{
+PrrtReceiver *PrrtReceiver_create(const char *host, uint16_t port) {
     PrrtReceiver *recv = calloc(1, sizeof(PrrtReceiver));
     check_mem(recv);
     recv->host_name = strdup(host);
-	check_mem(recv->host_name);
+    check_mem(recv->host_name);
     recv->port = port;
 
-	struct addrinfo *info;
-	struct addrinfo hints;
-	char portstr[sizeof(port)*8+1];
-	snprintf(portstr, sizeof(portstr), "%u", (unsigned int) port);
-	memset(&hints, 0x0, sizeof(hints));
-	hints.ai_family = AF_INET;
-	hints.ai_socktype = SOCK_DGRAM;
-	check(0 == getaddrinfo(host, portstr, &hints, &info), "getaddrinfo");
+    PrrtRateSample *rateSample = calloc(1, sizeof(PrrtRateSample));
+    check_mem(rateSample);
+    recv->rateSample = rateSample;
 
-	recv->ai = info;
 
-	recv->csi = PrrtChannelStateInformation_create();
+    PrrtPacketTracking *packetTracking = calloc(1, sizeof(PrrtPacketTracking));
+    check_mem(packetTracking);
+    recv->packetTracking = packetTracking;
+
+    recv->csi = PrrtChannelStateInformation_create();
+
+    recv->dataPacketStates = PrrtInFlightPacketStore_create();
+    recv->redundancyPacketStates = PrrtInFlightPacketStore_create();
+
+
+    struct addrinfo *info;
+    struct addrinfo hints;
+    char portstr[sizeof(port) * 8 + 1];
+    snprintf(portstr, sizeof(portstr), "%u", (unsigned int) port);
+    memset(&hints, 0x0, sizeof(hints));
+    hints.ai_family = AF_INET;
+    hints.ai_socktype = SOCK_DGRAM;
+    check(0 == getaddrinfo(host, portstr, &hints, &info), "getaddrinfo");
+
+    recv->ai = info;
+
+    check(pthread_mutex_init(&recv->lock, NULL) == 0, "lock init failed.");
+    check(pthread_cond_init(&recv->pipeNotFullCv, NULL) == 0, "pipeNotFullCv init failed.");
+    check(pthread_cond_init(&recv->recordNotFoundCv, NULL) == 0, "recordNotFound init failed.");
 
     return recv;
 
     error:
     if (recv != NULL) {
-		free((void *) recv->host_name);
-		free(recv);
-	}
+        free((void *) recv->host_name);
+        free(recv);
+    }
     //PERROR("Memory issue.%s","");
     return NULL;
 }
 
 bool PrrtReceiver_destroy(PrrtReceiver *receiver) {
-	if (receiver->csi != NULL) {
-		PrrtChannelStateInformation_destroy(receiver->csi);
-	}
-
-	freeaddrinfo(receiver->ai);
-	free((void *) receiver->host_name);
-	free(receiver);
-	return true;
+    if (receiver->csi != NULL) {
+        PrrtChannelStateInformation_destroy(receiver->csi);
+    }
+
+    if (receiver->dataPacketStates != NULL) {
+        PrrtInFlightPacketStore_destroy(receiver->dataPacketStates);
+    }
+
+    if (receiver->redundancyPacketStates != NULL) {
+        PrrtInFlightPacketStore_destroy(receiver->redundancyPacketStates);
+    }
+
+    if (receiver->packetTracking != NULL) {
+        free(receiver->packetTracking);
+    }
+
+    if (receiver->rateSample != NULL) {
+        free(receiver->rateSample);
+    }
+
+
+    freeaddrinfo(receiver->ai);
+    free((void *) receiver->host_name);
+    check(pthread_mutex_destroy(&receiver->lock) == 0, "lock destroy failed.");
+    check(pthread_cond_destroy(&receiver->pipeNotFullCv) == 0, "pipeNotFullCv destroy failed.");
+    check(pthread_cond_destroy(&receiver->recordNotFoundCv) == 0, "recordNotFoundCv destroy failed.");
+    free(receiver);
+    return true;
+
+    error:
+    PERROR("PrrtReceiver_destroy failed.")
+    return false;
+}
+
+void PrrtReceiver_updateRateSample(PrrtRateSample *rateSample, PrrtPacket *packet, prrtTimestamp_t receiveTime,
+                                   PrrtPacketTracking *packetTracking) {
+    if (packet->delivered_time == 0)
+        return;
+
+    packetTracking->delivered += packet->payloadLength;
+    packetTracking->delivered_time = receiveTime;
+
+    if (packet->delivered > rateSample->prior_delivered) {
+        rateSample->prior_delivered = packet->delivered;
+        rateSample->prior_time = packet->delivered_time;
+        rateSample->is_app_limited = packet->is_app_limited;
+        rateSample->send_elapsed = packet->sent_time - packet->first_sent_time;
+        rateSample->ack_elapsed = packetTracking->delivered_time - packet->delivered_time;
+        packetTracking->first_sent_time = packet->sent_time;
+    }
+    packet->delivered_time = 0;
+}
+
+void PrrtReceiver_generateRateSample(PrrtRateSample *rateSample, PrrtPacketTracking *packetTracking) {
+    /* Clear app-limited field */
+    if (packetTracking->app_limited && packetTracking->delivered > packetTracking->app_limited)
+        packetTracking->app_limited = 0;
+
+    if (rateSample->prior_time == 0) {
+        //printf("Prior Time is 0, Cancelling Rate sample calculation\n");
+        return;
+    }
+
+    rateSample->interval = MAX(rateSample->send_elapsed, rateSample->ack_elapsed);
+
+    rateSample->delivered = packetTracking->delivered - rateSample->prior_delivered;
+
+    if (rateSample->interval != 0) {
+        // delivered: bytes; interval: us; convert to bps
+        rateSample->delivery_rate =
+                rateSample->delivered * 1000 * 1000 * 8 / rateSample->interval;
+    }