.gitlab-ci.yml

@@ -38,6 +38,17 @@ build:container:
         - docker push $CI_REGISTRY_IMAGE:$DOCKER_TAG
         - docker rmi $CI_REGISTRY_IMAGE:$DOCKER_TAG
 
+build:container_tcp:
+    stage: build
+    tags:
+        - docker
+    script:
+        - export DOCKER_TAG=$(echo "$CI_BUILD_REF_NAME""_tcp" | sed 's#/#_#' | sed 's#^master$#latest#')
+        - docker build -t $CI_REGISTRY_IMAGE:$DOCKER_TAG --build-arg http_proxy=http://www-proxy.uni-saarland.de:3128 -f docker/Dockerfile_tcp .
+        - docker login -u gitlab-ci-token -p $CI_BUILD_TOKEN $CI_REGISTRY
+        - docker push $CI_REGISTRY_IMAGE:$DOCKER_TAG
+        - docker rmi $CI_REGISTRY_IMAGE:$DOCKER_TAG
+
 test:prrt_mem:
   stage: test
   dependencies:

README.md

@@ -29,9 +29,9 @@ port = int(sys.argv[1])
 s = prrt.PrrtSocket(port=port)
 
 while True:
-    d = s.recv()
+    d = s.recv().decode("utf8")
     if d != "Close":
-        print d
+        print(d)
     else:
         break
 ```
@@ -44,25 +44,27 @@ import prrt
 
 host = sys.argv[1]
 port = int(sys.argv[2])
+localport = int(sys.argv[3])
 
-s = prrt.PrrtSocket(port=port)
+s = prrt.PrrtSocket(port=localport)
 s.connect(host, port)
 
+
 for i in range(10):
-    s.send("Packet {}".format(i))
-s.send("Close")
+    s.send("Packet {}".format(i).encode("utf8"))
+s.send("Close".encode("utf8"))
 ```
 
 Start the receiver by:
 
 ```bash
-python receiver.py 5000
+python3 receiver.py 5000
 ```
 
 In a separate terminal, run:
 
 ```bash
-python sender.py 127.0.0.1 5000
+python3 sender.py 127.0.0.1 5000 6000
 ```
 
 This should generate the following output in the receiver console:

dissect/prrt.lua

@@ -25,16 +25,22 @@ local pf_data_length = ProtoField.uint32("prrt.data.length", "Length")
 local pf_data_timestamp = ProtoField.uint32("prrt.data.timestamp", "Timestamp")
 local pf_data_groupRTprop = ProtoField.uint32("prrt.data.grouprtprop", "Group RTprop")
 local pf_data_packettimeout = ProtoField.uint32("prrt.data.packettimeout", "Packet Timeout")
+local pf_data_btlbw = ProtoField.uint32("prrt.data.btlbw", "Bottleneck Bandwidth")
+local pf_data_btl_pace = ProtoField.uint32("prrt.data.btl_pace", "Bottleneck Pace")
+local pf_data_appSendTotal_pace = ProtoField.uint32("prrt.data.appSendTotal_pace", "Sender total application pace")
 
 local pf_red = ProtoField.new("Redundancy", "prrt.redundancy", ftypes.BYTES, base.NONE)
 local pf_red_baseSeqN = ProtoField.uint16("prrt.redundancy.baseSequenceNumber", "Base Sequence Number", base.DEC)
 local pf_red_timestamp = ProtoField.uint32("prrt.redundancy.timestamp", "Timestamp")
+local pf_red_btl_pace = ProtoField.uint32("prrt.redundancy.btl_pace", "Bottleneck Pace")
+local pf_red_appSendTotal_pace = ProtoField.uint32("prrt.redundancy.appSendTotal_pace", "Sender total application pace")
 local pf_red_n = ProtoField.uint8("prrt.redundancy.n", "n")
 local pf_red_k = ProtoField.uint8("prrt.redundancy.k", "k")
 
 local pf_fb = ProtoField.new("Feedback", "prrt.feedback", ftypes.BYTES, base.NONE)
 local pf_fb_groupRTT = ProtoField.uint32("prrt.feedback.groupRTT", "Group RTT")
 local pf_fb_ftt = ProtoField.uint32("prrt.feedback.FTT", "FTT")
+local pf_fb_btlPace = ProtoField.uint32("prrt.feedback.btlPace", "Bottleneck pace")
 local pf_fb_erasurecount = ProtoField.uint16("prrt.feedback.erasureCount", "Erasure count")
 local pf_fb_packetcount = ProtoField.uint16("prrt.feedback.packetCount", "Packet count")
 local pf_fb_gaplength = ProtoField.uint16("prrt.feedback.gapLength", "Gap length")
@@ -56,16 +62,22 @@ prrt_proto.fields = {
     pf_data_timestamp,
     pf_data_groupRTprop,
     pf_data_packettimeout,
+    pf_data_btlbw,
+    pf_data_btl_pace,
+    pf_data_appSendTotal_pace,
 
     pf_red,
     pf_red_baseSeqN,
     pf_red_timestamp,
+    pf_red_btl_pace,
+    pf_red_appSendTotal_pace,
     pf_red_n,
     pf_red_k,
 
     pf_fb,
     pf_fb_groupRTT,
     pf_fb_ftt,
+    pf_fb_btlPace,
     pf_fb_erasurecount,
     pf_fb_packetcount,
     pf_fb_gaplength,
@@ -89,9 +101,16 @@ local ex_type = Field.new("prrt.type")
 local function getType() return ex_type()() end
 local function getTypeName() return prrtPacketTypeNames[getType()] end
 
+local ex_index = Field.new("prrt.index")
+local function getIndex() return ex_index()() end
+
 local ex_data_length = Field.new("prrt.data.length")
 local function getDataLength() return ex_data_length()() end
 
+local ex_red_baseseqno = Field.new("prrt.redundancy.baseSequenceNumber")
+local function getRedBaseSeqNo() return ex_red_baseseqno()() end
+
+
 local ex_red_n = Field.new("prrt.redundancy.n")
 local function getRedN() return ex_red_n()() end
 
@@ -108,8 +127,11 @@ local function dissect_data(buffer, pinfo, root)
     tree:add(pf_data_timestamp, buffer:range(4,4))
     tree:add(pf_data_groupRTprop, buffer:range(8,4))
     tree:add(pf_data_packettimeout, buffer:range(12,4))
+    tree:add(pf_data_btlbw, buffer:range(16,4))
+    tree:add(pf_data_btl_pace, buffer:range(20,4))
+    tree:add(pf_data_appSendTotal_pace, buffer:range(24,4))
 
-    local label = "DATA Len=" .. getDataLength()
+    local label = "[D] Idx=" .. getIndex() .. " Len=" .. getDataLength()
     tree:set_text(label)
     pinfo.cols.info:set(label)
 end
@@ -118,10 +140,12 @@ local function dissect_redundancy(buffer, pinfo, root)
     local tree = root:add(pf_red, buffer:range(0))
     tree:add(pf_red_baseSeqN, buffer:range(0,2))
     tree:add(pf_red_timestamp, buffer:range(2,4))
-    tree:add(pf_red_n, buffer:range(6,1))
-    tree:add(pf_red_k, buffer:range(7,1))
+    tree:add(pf_red_btl_pace, buffer:range(6,4))
+    tree:add(pf_red_appSendTotal_pace, buffer:range(10,4))
+    tree:add(pf_red_n, buffer:range(14,1))
+    tree:add(pf_red_k, buffer:range(15,1))
 
-    local label = "REDUNDANCY n=" .. getRedN() .. " k=" .. getRedK()
+    local label = "[R] Idx=" .. getIndex() .. " b=" .. getRedBaseSeqNo() .. " n=" .. getRedN() .. " k=" .. getRedK()
     tree:set_text(label)
     pinfo.cols.info:set(label)
 end
@@ -130,16 +154,17 @@ local function dissect_feedback(buffer, pinfo, root)
     local tree = root:add(pf_fb, buffer:range(0))
     tree:add(pf_fb_groupRTT, buffer:range(0,4))
     tree:add(pf_fb_ftt, buffer:range(4,4))
-    tree:add(pf_fb_erasurecount, buffer:range(8,2))
-    tree:add(pf_fb_packetcount, buffer:range(10,2))
-    tree:add(pf_fb_gaplength, buffer:range(12,2))
-    tree:add(pf_fb_gapcount, buffer:range(14,2))
-    tree:add(pf_fb_burstlength, buffer:range(16,2))
-    tree:add(pf_fb_burstcount, buffer:range(18,2))
-    tree:add(pf_fb_acktype, buffer:range(20,1))
-    tree:add(pf_fb_ackSeqN, buffer:range(21, 2))
-
-    local label = "FEEDBACK"
+    tree:add(pf_fb_btlPace, buffer:range(8,4))
+    tree:add(pf_fb_erasurecount, buffer:range(12,2))
+    tree:add(pf_fb_packetcount, buffer:range(14,2))
+    tree:add(pf_fb_gaplength, buffer:range(16,2))
+    tree:add(pf_fb_gapcount, buffer:range(18,2))
+    tree:add(pf_fb_burstlength, buffer:range(20,2))
+    tree:add(pf_fb_burstcount, buffer:range(22,2))
+    tree:add(pf_fb_acktype, buffer:range(24,1))
+    tree:add(pf_fb_ackSeqN, buffer:range(25, 2))
+
+    local label = "[F]"
     tree:set_text(label)
     pinfo.cols.info:set(label)
 end

docker/Dockerfile

-FROM gcc:5
+FROM gcc:8
 
 MAINTAINER Andreas Schmidt <schmidt@nt.uni-saarland.de>
 
-RUN apt-get update \
-    && apt-get upgrade -y \
-    && apt-get install -y cmake
+ENV DEBIAN_FRONTEND noninteractive
+
+RUN apt-get update && apt-get install --yes --force-yes \
+    bc \
+    cmake \
+    iperf3 \
+    psmisc \
+    traceroute \
+    tshark
 
 COPY CMakeLists.txt /prrt/
 COPY prrt /prrt/prrt
 COPY tests /prrt/tests
 COPY docker/entrypoint.sh /
+COPY docker/sysctl.conf /etc/sysctl.d/01-disable-ipv6.conf
 
 WORKDIR /prrt
 
 RUN cmake . \
     && make
-ENV PATH /prrt/bin:$PATH
-WORKDIR /prrt/bin
+ENV PATH /prrt:$PATH
 
 VOLUME /output
 

docker/Dockerfile_tcp

+FROM gcc:8
+
+MAINTAINER Andreas Schmidt <schmidt@nt.uni-saarland.de>
+
+ENV DEBIAN_FRONTEND noninteractive
+
+RUN apt-get update && apt-get install --yes --force-yes \
+    bc \
+    cmake \
+    iperf3 \
+    psmisc \
+    traceroute \
+    tshark
+
+COPY CMakeLists.txt /prrt/
+COPY prrt /prrt/prrt
+COPY tests /prrt/tests
+COPY docker/entrypoint.sh /
+COPY docker/sysctl.conf /etc/sysctl.d/01-disable-ipv6.conf
+
+WORKDIR /prrt
+
+RUN cmake -DTCP=1 . \
+    && make
+ENV PATH /prrt:$PATH
+
+VOLUME /output
+
+ENTRYPOINT ["/entrypoint.sh"]

docker/entrypoint.sh

@@ -5,13 +5,15 @@ dev=eth0
 command=$1
 shift
 
-if [[ "$command" == "sender" || "$command" == "receiver" ]]; then
+if [[ "$command" == "sender" || "$command" == "receiver" || "$command" == "time-sender" || "$command" == "time-receiver" ]]; then
     :
 else
     echo "Command should be either sender or receiver."
     exit 0;
 fi
 
+TARGET="127.0.0.1"
+OUTPUT="/dev/null"
 NETEM=()
 PRRT=()
 while [[ $# -gt 0 ]]
@@ -19,7 +21,24 @@ do
 key="$1"
 
 case $key in
-    -t|--target|-p|--port|-r|--rounds)
+    -t|--target)
+    if [[  "$command" == "sender" || "$command" == "time-sender" ]]; then
+    PRRT+=("$1 $2")
+    fi
+    TARGET=("$2")
+    shift
+    shift
+    ;;
+    -w|--wireshark)
+    OUTPUT=("$2")
+    shift
+    shift
+    ;;
+    -T|--threadpinning|-U)
+    PRRT+=("$1")
+    shift
+    ;;
+    -p|--port|-r|--rounds|-s|--size|-R|--rcvbuf|-S|--sndbuf|-o|--output|-a|--appdelay|-j|--appjitter)
     PRRT+=("$1 $2")
     shift
     shift
@@ -34,8 +53,32 @@ done
 PRRT_PARAMS="${PRRT[@]}"
 NETEM_PARAMS="${NETEM[@]}"
 
+echo "Starting Wireshark."
+tshark -i eth0 -w $OUTPUT.pcap &
+TSHARK_PID=$!
+sleep 2
+
+start=$(date +%s.%N);
+echo "Checking reachability of $TARGET."
+until ping -c1 $TARGET &>/dev/null; do sleep 1; done
+dur=$(echo "$(date +%s.%N) - $start" | bc);
+printf "Reachable after %.6f seconds\n" $dur
+
+if [[ "$command" == "sender" || "$command" == "time-sender" ]]; then
+    echo "Delaying $command"
+    sleep 10
+fi
+
+start=$(date +%s.%N);
 echo "Running PRRT with command: \"$command $PRRT_PARAMS\" and link parameters: \"$NETEM_PARAMS\""
 
-tc qdisc add dev $dev root netem $NETEM_PARAMS
-/prrt/$command $PRRT_PARAMS -o /output/log.csv
-tc qdisc del dev $dev root
+#tc qdisc add dev $dev root netem $NETEM_PARAMS
+trap 'echo "Caught SIGINT."; echo "$(ps -a)"; killall -SIGINT $command' INT
+LOG=$(/prrt/$command $PRRT_PARAMS 2>&1)
+printf "$LOG\n"
+echo "Exit status: $?"
+dur=$(echo "$(date +%s.%N) - $start" | bc);
+printf "Done after %.6f seconds\n" $dur
+#tc qdisc del dev $dev root
+
+kill $TSHARK_PID

docker/sysctl.conf

+net.ipv6.conf.all.disable_ipv6 = 1
+net.ipv6.conf.default.disable_ipv6 = 1
+net.ipv6.conf.lo.disable_ipv6 = 1

prrt/CMakeLists.txt

@@ -10,14 +10,26 @@ if (XLAP)
     add_definitions(-DXLAP)
 endif()
 
+option (TCP "Set time protocol to TCP.")
+
+if (TCP)
+    add_definitions(-DTCP)
+endif()
+
 
 add_subdirectory(proto)
 add_subdirectory(util)
 
 add_executable(sender sender.c)
 add_executable(receiver receiver.c)
-add_executable(refcount refcount.c)
+add_executable(time-sender time-sender.c)
+add_executable(time-receiver time-receiver.c)
 
 target_link_libraries(sender LINK_PUBLIC PRRT UTIL ${CMAKE_THREAD_LIBS_INIT})
 target_link_libraries(receiver LINK_PUBLIC PRRT UTIL ${CMAKE_THREAD_LIBS_INIT})
+
+target_link_libraries(time-sender LINK_PUBLIC PRRT UTIL ${CMAKE_THREAD_LIBS_INIT})
+target_link_libraries(time-receiver LINK_PUBLIC PRRT UTIL ${CMAKE_THREAD_LIBS_INIT})
+
+add_executable(refcount refcount.c)
 target_link_libraries(refcount LINK_PUBLIC PRRT UTIL ${CMAKE_THREAD_LIBS_INIT})

prrt/cprrt.pxd

@@ -119,11 +119,12 @@ cdef extern from "proto/socket.h":
 
     ctypedef prrtSocket PrrtSocket
 
-    cdef PrrtSocket* PrrtSocket_create(const uint32_t mtu, const uint32_t target_delay)
+    cdef PrrtSocket* PrrtSocket_create(const uint32_t maximum_payload_size, const uint32_t target_delay)
     bint PrrtSocket_bind(PrrtSocket *sock_ptr, const_char *ipAddress, const uint16_t port)
     int PrrtSocket_close(const PrrtSocket *sock_ptr)
     int PrrtSocket_connect(PrrtSocket *sock_ptr, const_char *host, const uint16_t port)
-    int PrrtSocket_send(PrrtSocket *sock_ptr, const uint8_t *data, const size_t data_len)
+    int PrrtSocket_send_sync(PrrtSocket *sock_ptr, const uint8_t *data, const size_t data_len)
+    int PrrtSocket_send_async(PrrtSocket *sock_ptr, const uint8_t *data, const size_t data_len)
 
     int32_t PrrtSocket_recv(PrrtSocket *sock_ptr, void *buf_ptr, sockaddr* addr) nogil
     int32_t PrrtSocket_receive_asap(PrrtSocket *s, void *buf_ptr, sockaddr* addr) nogil
@@ -144,11 +145,23 @@ cdef extern from "proto/socket.h":
     uint32_t PrrtSocket_get_rtprop_fwd(PrrtSocket *socket)
     float PrrtSocket_get_plr_fwd(PrrtSocket *socket)
     uint32_t PrrtSocket_get_delivery_rate_fwd(PrrtSocket *socket)
-
     uint32_t PrrtSocket_get_btlbw_fwd(PrrtSocket *s);
     uint32_t PrrtSocket_get_btlbw_back(PrrtSocket *s);
 
-    bint PrrtSocket_get_app_limited(PrrtSocket *socket)
+    uint32_t PrrtSocket_get_bbr_state(PrrtSocket *s)
+    uint64_t PrrtSocket_get_full_bw(PrrtSocket *s)
+    bint PrrtSocket_get_filled_pipe(PrrtSocket *s)
+    uint32_t PrrtSocket_get_cycle_index(PrrtSocket *s)
+    float PrrtSocket_get_pacing_gain(PrrtSocket *s)
+    uint32_t PrrtSocket_get_cwnd(PrrtSocket *s)
+    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)
+    uint32_t PrrtSocket_get_delivered(PrrtSocket *s)
+    bint PrrtSocket_get_bbr_round_start(PrrtSocket *s)
+    uint32_t PrrtSocket_get_bbr_app_limited(PrrtSocket *socket)
+    bint PrrtSocket_get_bbr_is_app_limited(PrrtSocket *socket)
     bint PrrtSocket_enable_thread_pinning(PrrtSocket *socket)
 
 cdef extern from "proto/stores/packetDeliveryStore.h":

prrt/defines.h

@@ -30,8 +30,8 @@
 
 #define GF_BITS 8
 
-#define K_START 4
-#define N_START 7
+#define K_START 1
+#define N_START 1
 #define N_P_START 1
 
 #define RRT_ALPHA 0.125

prrt/proto/CMakeLists.txt

 set (PRRT_SOURCES ../defines.h
-        types/block.c types/block.h
-        types/channelStateInformation.c types/channelStateInformation.h
+        bbr.c bbr.h
         clock.c clock.h
-        types/codingParams.c types/codingParams.h
         receiver.c receiver.h
         socket.c socket.h
-        types/applicationConstraints.c types/applicationConstraints.h
+        timer.c timer.h
         processes/dataReceiver.c processes/dataReceiver.h
         processes/dataTransmitter.c processes/dataTransmitter.h
         stores/dataPacketStore.c stores/dataPacketStore.h
         stores/deliveredPacketTable.c stores/deliveredPacketTable.h
         stores/inFlightPacketStore.c stores/inFlightPacketStore.h
+        stores/pace.c stores/pace.h
+        stores/paceFilter.c stores/paceFilter.h
         stores/packetTimeoutTable.c stores/packetTimeoutTable.h
         stores/packetDeliveryStore.c stores/packetDeliveryStore.h
         stores/receptionTable.c stores/receptionTable.h
         stores/repairBlockStore.c stores/repairBlockStore.h
-        types/packetTimeout.c types/packetTimeout.h
+        types/applicationConstraints.c types/applicationConstraints.h
+        types/block.c types/block.h
+        types/codingParams.c types/codingParams.h
+        types/channelStateInformation.c types/channelStateInformation.h
         types/lossStatistics.c types/lossStatistics.h
         types/packet.c types/packet.h
-        vdmcode/block_code.c vdmcode/block_code.h)
+        types/packetTimeout.c types/packetTimeout.h
+        vdmcode/block_code.c vdmcode/block_code.h types/packetTracking.c types/packetTracking.h types/rateSample.c types/rateSample.h)
 
 if (XLAP)
     set(PRRT_SOURCES ${PRRT_SOURCES} ../xlap/xlap.c ../xlap/xlap.h)

prrt/proto/bbr.h

+#ifndef PRRT_BBR_H
+#define PRRT_BBR_H
+
+#include <stdint.h>
+#include "stdbool.h"
+#include "types/packet.h"
+#include "clock.h"
+#include "types/channelStateInformation.h"
+#include "types/packetTracking.h"
+#include "types/rateSample.h"
+#include "../util/windowedFilter.h"
+
+#define PROBE_GAIN 1.1
+#define DRAIN_GAIN 0.9
+#define RTpropFilterLen 10000000    //10s
+#define BBRHighGain ((((float)2885) / 1000) + 1)
+#define BBRGainCycleLen 8
+#define ProbeRTTDuration 200000 //200ms
+#define RTprop_Inf UINT32_MAX
+
+enum bbr_state {
+    STARTUP,
+    DRAIN,
+    PROBE_BW,
+    PROBE_RTT
+};
+
+typedef struct bbr {
+    pthread_mutex_t lock;
+
+    prrtByteCount_t mps;
+    prrtByteCount_t min_pipe_cwnd;
+    prrtByteCount_t initial_cwnd;
+    prrtTimedelta_t rtprop;
+    prrtTimestamp_t rtprop_stamp;
+    prrtTimestamp_t probe_rtt_done_stamp;
+    bool probe_rtt_round_done;
+    bool packet_conservation;
+    prrtByteCount_t prior_cwnd;
+    prrtByteCount_t cwnd;
+    prrtByteCount_t target_cwnd;
+    bool idle_restart;
+
+    enum bbr_state state;
+
+    double pacing_gain;
+    float cwnd_gain;
+
+    bool filled_pipe;
+    prrtByteCount_t full_bw;
+    uint32_t full_bw_count;
+
+    double pacing_rate;
+    bool has_seen_rtt;
+
+    uint32_t next_round_delivered;
+    bool round_start;
+    uint32_t round_count;
+
+    uint32_t next_rtt_delivered;
+    uint32_t rtt_count;
+    bool rtprop_expired;
+
+    bool is_loss_recovery;
+    prrtTimestamp_t loss_recovery_stamp;
+
+    prrtTimestamp_t cycle_stamp;
+    uint8_t cycle_index;
+    float* pacing_gain_cycle;
+
+    prrtByteCount_t send_quantum;
+
+    prrtDeliveryRate_t bw;
+
+    WindowedFilter* btlBwFilter;
+} BBR;
+
+BBR* BBR_Init(prrtByteCount_t maximum_payload_size);
+void BBR_OnACK(BBR* bbr, PrrtChannelStateInformation* csi, PrrtRateSample* rs, PrrtPacketTracking* packetTracking, prrtTimedelta_t rtt);
+void BBR_OnSpuriousLoss(BBR *bbr, PrrtPacketTracking *tracking);
+void BBR_OnRTOLoss(BBR *bbr);
+void BBR_OnLossExit(BBR *bbr);
+void BBR_destroy(BBR* bbr);
+
+double BBR_getPacingRate(BBR* bbr);
+prrtByteCount_t BBR_getCwnd(BBR* bbr);
+prrtDeliveryRate_t BBR_getBtlBw(BBR* bbr);
+uint32_t BBR_getState(BBR* bbr);
+prrtByteCount_t BBR_getFullBw(BBR* bbr);
+double BBR_getPacingGain(BBR* bbr);
+uint32_t BBR_getCycleIndex(BBR* bbr);
+bool BBR_getFilledPipe(BBR* bbr);
+uint32_t BBR_getRTProp(BBR* bbr);
+prrtByteCount_t BBR_getInflight(BBR* bbr);
+prrtByteCount_t BBR_getSendQuantum(BBR* bbr);
+bool BBR_getRoundStart(BBR* bbr);
+
+#endif //PRRT_BBR_H

prrt/proto/processes/dataTransmitter.h

 #ifndef PRRT_DATA_TRANSMITTER_H
 #define PRRT_DATA_TRANSMITTER_H
 
-void * send_data_loop(void *ptr);
+#include "../socket.h"
+
+void * PrrtDataTransmitter_send_data_loop(void *ptr);
+void PrrtDataTransmitter_transmit(PrrtSocket *sock_ptr, PrrtPacket *packet);
 
 #endif //PRRT_DATA_TRANSMITTER_H

prrt/proto/receiver.h

@@ -7,28 +7,10 @@
 #include <sys/socket.h>
 #include <netdb.h>
 #include "stores/inFlightPacketStore.h"
+#include "bbr.h"
+#include "types/applicationConstraints.h"
 #include "types/channelStateInformation.h"
 
-typedef struct prrtRateSample {
-    prrtByteCount_t prior_delivered;
-    prrtTimestamp_t prior_time;
-    prrtTimedelta_t send_elapsed;
-    prrtTimedelta_t ack_elapsed;
-
-    prrtTimedelta_t interval;
-    prrtByteCount_t delivered;
-    bool is_app_limited;
-
-    prrtDeliveryRate_t delivery_rate; // Bps
-} PrrtRateSample;
-
-typedef struct packetTracking {
-    prrtByteCount_t pipe;
-    prrtByteCount_t delivered;
-    prrtTimestamp_t delivered_time;
-    prrtTimestamp_t first_sent_time;
-    prrtByteCount_t app_limited;
-} PrrtPacketTracking;
 
 typedef struct prrtReceiver {
     const char *host_name;
@@ -36,23 +18,39 @@ typedef struct prrtReceiver {
     struct addrinfo *ai;
     PrrtChannelStateInformation *csi;
     pthread_mutex_t lock;
+    pthread_cond_t wait_for_space;
+    atomic_bool closing;
+
+    BBR* bbr;
 
-    PrrtInFlightPacketStore *dataPacketStates;
-    PrrtInFlightPacketStore *redundancyPacketStates;
+    PrrtInFlightPacketStore *dataInflightPacketStore;
+    PrrtInFlightPacketStore *redundancyInflightPacketStore;
 
     PrrtRateSample *rateSample;
     PrrtPacketTracking *packetTracking;
 } PrrtReceiver;
 
-PrrtReceiver *PrrtReceiver_create(const char *host, uint16_t port);
-
-bool PrrtReceiver_updateAndGenerateRateSample(PrrtReceiver *recv, prrtSequenceNumber_t seqnum, uint8_t packetType,
-                                              prrtTimestamp_t receiveTime);
+PrrtReceiver *PrrtReceiver_create(const char *host, uint16_t port, prrtByteCount_t maximum_payload_size);
 
 void PrrtReceiver_add_outstanding_packet_state(PrrtReceiver *recv, PrrtPacket *packet, prrtTimestamp_t sentTime);
+prrtByteCount_t PrrtReceiver_get_pipe(PrrtReceiver *recv);
+void PrrtReceiver_rto_check(PrrtReceiver *recv, PrrtApplicationConstraints *constraints);
+
+void PrrtReceiver_on_application_write(PrrtReceiver *receiver, uint32_t send_queue_length, prrtSequenceNumber_t sequenceNumber);
+void PrrtReceiver_on_ack(PrrtReceiver *receiver, PrrtPacketFeedbackPayload *feedbackPayload, prrtTimestamp_t receiveTime,
+                         prrtTimedelta_t rtt, PrrtApplicationConstraints *constraints);
+
+bool PrrtReceiver_wait_for_space(PrrtReceiver *receiver, prrtByteCount_t maximum_payload_size,
+                                 PrrtApplicationConstraints *pConstraints);
+uint32_t PrrtReceiver_get_retransmission_delay(PrrtReceiver *socket, PrrtApplicationConstraints *applicationConstraints);
 
-void PrrtReceiver_on_application_write(PrrtReceiver* receiver);
+prrtByteCount_t PrrtReceiver_get_space(PrrtReceiver *receiver);
+double PrrtReceiver_get_BBR_pacingRate(PrrtReceiver *receiver);
+double PrrtReceiver_get_BBR_pacingGain(PrrtReceiver *receiver);
+uint32_t PrrtReceiver_get_BBR_state(PrrtReceiver *receiver);
+prrtDeliveryRate_t PrrtReceiver_get_BBR_btlDr(PrrtReceiver *receiver);
 
+void PrrtReceiver_interrupt(PrrtReceiver *receiver);
 bool PrrtReceiver_destroy(PrrtReceiver *receiver);
 
 #endif //PRRT_RECEIVER_H

prrt/proto/socket.h

 #ifndef PRRT_SOCKET_H
 #define PRRT_SOCKET_H
 
+
 #include "../defines.h"
 #include "../util/list.h"
 #include "../util/pipe.h"
@@ -9,6 +10,8 @@
 #include "stores/dataPacketStore.h"
 #include "stores/deliveredPacketTable.h"
 #include "stores/packetTimeoutTable.h"
+#include "stores/pace.h"
+#include "stores/paceFilter.h"
 #include "stores/receptionTable.h"
 #include "stores/repairBlockStore.h"
 #include "stores/packetDeliveryStore.h"
@@ -17,6 +20,7 @@
 #include "types/lossStatistics.h"
 #include "types/packet.h"
 #include "clock.h"
+#include "timer.h"
 #include "xlap.h"
 #include "receiver.h"
 
@@ -31,8 +35,12 @@ typedef struct prrtSocket {
 
     struct sockaddr_in *address;
     bool isBound;
+    bool withTimestamp;
+
+    bool pacingEnabled;
 
     PrrtClock clock;
+    PrrtBlock* receiveBlock;
 
     pthread_t sendDataThread;
     Pipe *sendDataQueue;
@@ -50,6 +58,8 @@ typedef struct prrtSocket {
 
     atomic_bool closing;
 
+    prrtTimestamp_t nextSendTime;
+
     prrtSequenceNumber_t packetsCount;
     prrtSequenceNumber_t sequenceNumberSource;
     prrtSequenceNumber_t sequenceNumberRepetition;
@@ -69,6 +79,18 @@ typedef struct prrtSocket {
     PrrtCodingConfiguration *codingParameters;
     PrrtCoder *coder;
 
+    prrtTimedelta_t rtt;
+
+    // Pacing
+    PrrtPace* appSendPace;
+    PrrtPace* prrtTransmitPace;
+    PrrtPace* prrtReceivePace;
+    PrrtPace* appDeliverPace;
+
+    prrtTimedelta_t send_peer_btl_pace;
+    prrtTimedelta_t recv_peer_btl_pace;
+    prrtTimedelta_t send_peer_app_total_pace;
+
     _Atomic (XlapTimestampTable *) tstable[2];
 
     pthread_attr_t *sendDataThreadAttr;
@@ -77,8 +99,12 @@ typedef struct prrtSocket {
     atomic_bool isHardwareTimestamping;
     char *interfaceName;
 
+    PrrtChannelStateInformation* senderChannelStateInformation;
+
     atomic_bool isThreadPinning;
     prrtByteCount_t maximum_payload_size;
+
+    PrrtTimer *retransmissionTimer;
 } PrrtSocket;
 
 
@@ -106,7 +132,9 @@ int PrrtSocket_close(PrrtSocket *s);
 
 bool PrrtSocket_connect(PrrtSocket *s, const char *host, const uint16_t port);
 
-int PrrtSocket_send(PrrtSocket *s, const uint8_t *data, size_t data_len);
+int PrrtSocket_send_async(PrrtSocket *s, const uint8_t *data, const size_t data_len);
+int PrrtSocket_send_sync(PrrtSocket *s, const uint8_t *data, size_t data_len);
+bool PrrtSocket_pace(PrrtSocket *sock_ptr, bool prepace);
 
 int32_t PrrtSocket_recv(PrrtSocket *s, void *buf_ptr, struct sockaddr* addr);
 
@@ -118,7 +146,6 @@ int32_t PrrtSocket_receive_ordered(PrrtSocket *s, void *buf_ptr, struct sockaddr
 int32_t PrrtSocket_receive_ordered_wait(PrrtSocket *s, void *buf_ptr, struct sockaddr* addr, prrtTimedelta_t time_window_us);
 int32_t PrrtSocket_receive_ordered_timedwait(PrrtSocket *s, void *buf_ptr, struct sockaddr* addr, prrtTimedelta_t time_window_us, struct timespec* deadline);
 
-
 bool PrrtSocket_cleanup(PrrtSocket *s);
 
 bool PrrtSocket_closing(PrrtSocket *s);
@@ -126,11 +153,23 @@ bool PrrtSocket_closing(PrrtSocket *s);
 bool PrrtSocket_uses_thread_pinning(PrrtSocket *s);
 
 uint32_t PrrtSocket_get_rtprop_fwd(PrrtSocket *s);
-
 prrtPacketLossRate_t PrrtSocket_get_plr_fwd(PrrtSocket *s);
 prrtDeliveryRate_t PrrtSocket_get_delivery_rate_fwd(PrrtSocket *s);
 prrtDeliveryRate_t PrrtSocket_get_btlbw_fwd(PrrtSocket *s);
 prrtDeliveryRate_t PrrtSocket_get_btlbw_back(PrrtSocket *s);
-bool PrrtSocket_get_app_limited(PrrtSocket *s);
+uint64_t PrrtSocket_get_full_bw(PrrtSocket *s);
+bool PrrtSocket_get_filled_pipe(PrrtSocket *s);
+uint32_t PrrtSocket_get_bbr_state(PrrtSocket *s);
+uint32_t PrrtSocket_get_bbr_app_limited(PrrtSocket *s);
+bool PrrtSocket_get_bbr_is_app_limited(PrrtSocket *s);
+uint32_t PrrtSocket_get_cycle_index(PrrtSocket *s);
+float PrrtSocket_get_pacing_gain(PrrtSocket *s);
+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);
+prrtByteCount_t PrrtSocket_get_delivered(PrrtSocket *s);
+bool PrrtSocket_get_bbr_round_start(PrrtSocket *s);
 
 #endif // PRRT_SOCKET_H

prrt/proto/stores/dataPacketStore.c

@@ -80,9 +80,11 @@ void PrrtDataPacketStore_remove_range(PrrtDataPacketStore *store, List *res, prr
 bool PrrtDataStore_insert(PrrtDataPacketStore *store, PrrtPacket *packet)
 {
     check(pthread_mutex_lock(&store->lock) == EXIT_SUCCESS, "Lock failed.");
+    debug(DEBUG_BLOCK, "PrrtDataStore_insert: %d", packet->sequenceNumber);
     if(BPTree_get(store->dataStore, packet->sequenceNumber) == NULL) {
         store->dataStore = BPTree_insert(store->dataStore, packet->sequenceNumber, packet);
         check(pthread_mutex_unlock(&store->lock) == EXIT_SUCCESS, "Unlock failed.");
+        debug(DEBUG_BLOCK, "PrrtDataStore_size: %d", PrrtDataPacketStore_size(store));
         return true;
     }
     check(pthread_mutex_unlock(&store->lock) == EXIT_SUCCESS, "Unlock failed.");

prrt/proto/stores/inFlightPacketStore.c

 #include <stdint.h>
-#include "inFlightPacketStore.h"
+#include "../clock.h"
 #include "../../util/common.h"
 #include "../../util/dbg.h"
+#include "inFlightPacketStore.h"
 
 
 PrrtInFlightPacketStore* PrrtInFlightPacketStore_create(void)
@@ -9,7 +10,8 @@ PrrtInFlightPacketStore* PrrtInFlightPacketStore_create(void)
     PrrtInFlightPacketStore* packetStore = calloc(1, sizeof(PrrtInFlightPacketStore));
     check_mem(packetStore);
 
-    packetStore->outstandingPackets = NULL;
+    packetStore->outstandingPackets_by_seqno = NULL;
+    packetStore->outstandingPackets_by_senttime = NULL;
 
     return packetStore;
 
@@ -20,34 +22,63 @@ PrrtInFlightPacketStore* PrrtInFlightPacketStore_create(void)
 
 void PrrtInFlightPacketStore_add_outstanding_packet(PrrtInFlightPacketStore *packetStore, PrrtPacket *packet)
 {
-    packetStore->outstandingPackets = BPTree_insert(packetStore->outstandingPackets, packet->sequenceNumber,
+    packetStore->outstandingPackets_by_seqno = BPTree_insert(packetStore->outstandingPackets_by_seqno, packet->sequenceNumber,
                                                     packet);
-    packetStore->packetQueueSize++;
+    packetStore->outstandingPackets_by_senttime = BPTree_insert(packetStore->outstandingPackets_by_senttime, packet->sent_time,
+                                                             packet);
+}
+
+void* PrrtInFlightPacketStore_get_packet_by_seqno(PrrtInFlightPacketStore *packetStore, BPTreeKey_t seqNum)
+{
+    return BPTree_get(packetStore->outstandingPackets_by_seqno, seqNum);
 }
 
-void* PrrtInFlightPacketStore_get_packet(PrrtInFlightPacketStore *packetStore, BPTreeKey_t seqNum)
+void* PrrtInFlightPacketStore_get_packet_by_senttime(PrrtInFlightPacketStore *packetStore, prrtTimestamp_t senttime)
 {
-    return BPTree_get(packetStore->outstandingPackets, seqNum);
+    return BPTree_get(packetStore->outstandingPackets_by_senttime, senttime);
 }
 
-void PrrtInFlightPacketStore_remove_outstanding_packet(PrrtInFlightPacketStore *packetStore, BPTreeKey_t seqNum)
+void PrrtInFlightPacketStore_remove_outstanding_packet_by_seqno(PrrtInFlightPacketStore *packetStore,
+                                                                           BPTreeKey_t seqNum)
 {
-    PrrtPacket* packet = (PrrtPacket*) BPTree_get(packetStore->outstandingPackets, seqNum);
-    if (packet != NULL) {
-        packetStore->outstandingPackets = BPTree_delete(packetStore->outstandingPackets, seqNum);
-        packetStore->packetQueueSize--;
+    PrrtPacket* outstanding_packet = (PrrtPacket*) BPTree_get(packetStore->outstandingPackets_by_seqno, seqNum);
+    if(outstanding_packet != NULL) {
+        packetStore->outstandingPackets_by_seqno = BPTree_delete(packetStore->outstandingPackets_by_seqno, seqNum);
+        packetStore->outstandingPackets_by_senttime = BPTree_delete(packetStore->outstandingPackets_by_senttime, outstanding_packet->sent_time);
+        PrrtPacket_destroy(outstanding_packet);
+    }
+}
+
+prrtByteCount_t PrrtInFlightPacketStore_clear_before(PrrtInFlightPacketStore *packetStore,
+                                                     prrtTimestamp_t time) {
+    prrtByteCount_t lostBytes = 0;
+    List *packetStateList = List_create();
+    if(time >= HALF_TIMESTAMP) {
+        BPTree_get_range(packetStore->outstandingPackets_by_senttime, packetStateList, time - HALF_TIMESTAMP, time - 1);
+    } else {
+        BPTree_get_range(packetStore->outstandingPackets_by_senttime, packetStateList, 0, time - 1);
+        BPTree_get_range(packetStore->outstandingPackets_by_senttime, packetStateList, time + HALF_TIMESTAMP, MAX_TIMESTAMP);
+    }
+    while (List_count(packetStateList) > 0) {
+        PrrtPacket *packet = List_shift(packetStateList);
+        lostBytes += packet->payloadLength;
+        packetStore->outstandingPackets_by_seqno = BPTree_delete(packetStore->outstandingPackets_by_seqno, packet->sequenceNumber);
+        packetStore->outstandingPackets_by_senttime = BPTree_delete(packetStore->outstandingPackets_by_senttime, packet->sent_time);
         PrrtPacket_destroy(packet);
     }
+    List_destroy(packetStateList);
+    return lostBytes;
 }
 
 bool PrrtInFlightPacketStore_destroy(PrrtInFlightPacketStore *packetStore)
 {
-    if (packetStore->outstandingPackets != NULL) {
+    if (packetStore->outstandingPackets_by_seqno != NULL) {
         List *packetStateList = List_create();
-        BPTree_get_range(packetStore->outstandingPackets, packetStateList, 0, SEQNO_SPACE - 1);
+        BPTree_get_range(packetStore->outstandingPackets_by_seqno, packetStateList, 0, SEQNO_SPACE - 1);
         while (List_count(packetStateList) > 0) {
             PrrtPacket *packet = List_shift(packetStateList);
-            packetStore->outstandingPackets = BPTree_delete(packetStore->outstandingPackets, packet->sequenceNumber);
+            packetStore->outstandingPackets_by_seqno = BPTree_delete(packetStore->outstandingPackets_by_seqno, packet->sequenceNumber);
+            packetStore->outstandingPackets_by_senttime = BPTree_delete(packetStore->outstandingPackets_by_senttime, packet->sent_time);
             PrrtPacket_destroy(packet);
         }
 
@@ -56,8 +87,3 @@ bool PrrtInFlightPacketStore_destroy(PrrtInFlightPacketStore *packetStore)
     free(packetStore);
     return true;
 }
-
-uint32_t PrrtInFlightPacketStore_get_queue_size(PrrtInFlightPacketStore *packetStore)
-{
-    return packetStore->packetQueueSize;
-}
\ No newline at end of file

prrt/proto/stores/inFlightPacketStore.h

@@ -5,16 +5,19 @@
 #include "../types/packet.h"
 
 typedef struct inFlightPackets {
-    BPTreeNode* outstandingPackets;
-    uint32_t packetQueueSize;
+    BPTreeNode* outstandingPackets_by_seqno;
+    BPTreeNode* outstandingPackets_by_senttime;
 } PrrtInFlightPacketStore;
 
 PrrtInFlightPacketStore* PrrtInFlightPacketStore_create(void);
 void PrrtInFlightPacketStore_add_outstanding_packet(PrrtInFlightPacketStore *packetStore, PrrtPacket *packet);
-void PrrtInFlightPacketStore_remove_outstanding_packet(PrrtInFlightPacketStore *packetStore, BPTreeKey_t seqNum);
+void PrrtInFlightPacketStore_remove_outstanding_packet_by_seqno(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);
+void* PrrtInFlightPacketStore_get_packet_by_seqno(PrrtInFlightPacketStore *packetStore, BPTreeKey_t seqNum);
+PrrtPacket * PrrtInFlightPacketStore_get_first_packet_before_seqno(PrrtInFlightPacketStore *packetStore,
+                                                                   prrtSequenceNumber_t seqNum);
+prrtByteCount_t PrrtInFlightPacketStore_clear_before(PrrtInFlightPacketStore *packetStore, prrtTimestamp_t time);
 
 
 

prrt/proto/stores/pace.c

+#include "pace.h"
+#include "../clock.h"
+#include "../../util/common.h"
+#include "../../util/dbg.h"
+#include "../../util/time.h"
+#include <math.h>
+
+PrrtPace* PrrtPace_create(void) {
+    PrrtPace* pace = (PrrtPace*) calloc(1, sizeof(PrrtPace));
+    check_mem(pace);
+    prrtTimedelta_t filterLength_us = 2 * 1000 * 1000; // 2 seconds
+    pace->internalPace = PrrtPaceFilter_create(filterLength_us, FILTER_TYPE_MAX);
+    pace->dependentPace = PrrtPaceFilter_create(filterLength_us, FILTER_TYPE_MAX);
+    pace->externalPace = PrrtPaceFilter_create(filterLength_us, FILTER_TYPE_MAX);
+
+    pace->totalPauseDuration_ns = 0;
+    pace->initialized = false;
+    pace->firstRoundDone = false;
+    clock_gettime(CLOCK_REALTIME, &pace->lastStartTimestamp);
+    clock_gettime(CLOCK_REALTIME, &pace->lastEndTimestamp);
+    clock_gettime(CLOCK_REALTIME, &pace->lastPauseTimestamp);
+    return pace;
+
+    error:
+    PERROR("Out of memory%s.", "");
+    return NULL;
+
+}
+bool PrrtPace_destroy(PrrtPace* pace) {
+    if(pace->internalPace != NULL) {
+        check(PrrtPaceFilter_destroy(pace->internalPace), "Cannot destroy internalPace.")
+    }
+
+    if(pace->dependentPace != NULL) {
+        check(PrrtPaceFilter_destroy(pace->dependentPace), "Cannot destroy dependentPace.")
+    }
+
+    if(pace->externalPace != NULL) {
+        check(PrrtPaceFilter_destroy(pace->externalPace), "Cannot destroy externalPace.")
+    }
+
+    free(pace);
+    return true;
+    error:
+    return false;
+}
+
+prrtTimedelta_t PrrtPace_get_internal(PrrtPace* pace) {
+    return PrrtPaceFilter_get(pace->internalPace);
+}
+
+prrtTimedelta_t PrrtPace_get_external(PrrtPace* pace) {
+    return PrrtPaceFilter_get(pace->externalPace);
+}
+
+prrtTimedelta_t PrrtPace_get_dependent(PrrtPace* pace) {
+    return PrrtPaceFilter_get(pace->dependentPace);
+}
+
+prrtTimedelta_t PrrtPace_get_total(PrrtPace* pace) {
+    return PrrtPaceFilter_get(pace->internalPace) + PrrtPaceFilter_get(pace->externalPace);
+}
+
+prrtTimedelta_t PrrtPace_get_effective(PrrtPace* pace) {
+    return (prrtTimedelta_t) MAX(0, ((int64_t) PrrtPaceFilter_get(pace->externalPace)) + ((int64_t)PrrtPaceFilter_get(pace->internalPace)) - ((int64_t)PrrtPaceFilter_get(pace->dependentPace)));
+}
+
+void PrrtPace_track_start(PrrtPace* pace) {
+    struct timespec now;
+    clock_gettime(CLOCK_REALTIME, &now);
+    if (pace->initialized) {
+        long long int internalDelta_ns = timedelta(&pace->lastEndTimestamp, &pace->lastStartTimestamp);
+        long long int externalDelta_ns = timedelta(&now, &pace->lastEndTimestamp);
+
+        if (pace->firstRoundDone) {
+            // make sure internal >= dependent
+            PrrtPaceFilter_update(pace->internalPace, (prrtTimedelta_t) MAX(0, round(((double) MAX(internalDelta_ns, pace->totalPauseDuration_ns)) / 1000)));
+            PrrtPaceFilter_update(pace->externalPace, (prrtTimedelta_t) MAX(0, round(((double) externalDelta_ns) / 1000)));
+            PrrtPaceFilter_update(pace->dependentPace, (prrtTimedelta_t) MAX(0, round(((double) pace->totalPauseDuration_ns) / 1000)));
+        }
+        pace->firstRoundDone = true;
+    }
+    pace->totalPauseDuration_ns = 0;
+    pace->lastStartTimestamp = now;
+    pace->initialized = true;
+}
+void PrrtPace_track_end(PrrtPace* pace) {
+    clock_gettime(CLOCK_REALTIME, &pace->lastEndTimestamp);
+}
+
+void PrrtPace_track_pause(PrrtPace* pace) {
+    clock_gettime(CLOCK_REALTIME, &pace->lastPauseTimestamp);
+}
+
+void PrrtPace_track_resume(PrrtPace* pace) {
+    struct timespec now;
+    clock_gettime(CLOCK_REALTIME, &now);
+    long long int delta = timedelta(&now, &pace->lastPauseTimestamp);
+    clock_gettime(CLOCK_REALTIME, &pace->lastPauseTimestamp);
+    pace->totalPauseDuration_ns += delta;
+}

prrt/proto/stores/pace.h

+#ifndef PRRT_PACE_H
+#define PRRT_PACE_H
+
+#include "paceFilter.h"
+
+typedef struct prrtPace {
+    PrrtPaceFilter* internalPace;
+    PrrtPaceFilter* externalPace;
+    PrrtPaceFilter* dependentPace;
+
+    struct timespec lastStartTimestamp;
+    struct timespec lastEndTimestamp;
+    struct timespec lastPauseTimestamp;
+    prrtTimedelta_t totalPauseDuration_ns;
+    bool initialized;
+    bool firstRoundDone;
+} PrrtPace;
+
+PrrtPace* PrrtPace_create(void);
+bool PrrtPace_destroy(PrrtPace* pace);
+
+prrtTimedelta_t PrrtPace_get_internal(PrrtPace* pace);
+prrtTimedelta_t PrrtPace_get_external(PrrtPace* pace);
+prrtTimedelta_t PrrtPace_get_dependent(PrrtPace* pace);
+prrtTimedelta_t PrrtPace_get_total(PrrtPace* pace);
+prrtTimedelta_t PrrtPace_get_effective(PrrtPace* pace);
+
+void PrrtPace_track_start(PrrtPace* pace);
+void PrrtPace_track_end(PrrtPace* pace);
+void PrrtPace_track_pause(PrrtPace* pace);
+void PrrtPace_track_resume(PrrtPace* pace);
+
+#endif //PRRT_PACE_H

prrt/proto/stores/paceFilter.c

+#include "paceFilter.h"
+#include "../clock.h"
+#include "../../util/common.h"
+#include "../../util/dbg.h"
+#include <math.h>
+#include <assert.h>
+
+void invalidate(PrrtPaceFilter* filter) {
+    prrtTimestamp_t now = PrrtClock_get_current_time_us();
+    filter->valid = false;
+    filter->updated = now;
+    //filter->value = 0;
+}
+
+
+PrrtPaceFilter* PrrtPaceFilter_create(prrtTimedelta_t window_us, prrtPaceFilterType type) {
+    PrrtPaceFilter* filter = (PrrtPaceFilter*) calloc(1, sizeof(PrrtPaceFilter));
+    check_mem(filter);
+
+	assert(type < FILTER_TYPE__COUNT);
+
+	filter->type = type;
+    filter->updated = PrrtClock_get_current_time_us();
+    filter->valid = false;
+    filter->value = 0;
+    filter->window_us = window_us;
+
+    return filter;
+    error:
+    PERROR("Out of memory%s.", "");
+    return NULL;
+
+}
+
+bool PrrtPaceFilter_destroy(PrrtPaceFilter* filter) {
+    free(filter);
+    return true;
+}
+
+prrtTimedelta_t PrrtPaceFilter_get(PrrtPaceFilter* filter) {
+    prrtTimestamp_t now = PrrtClock_get_current_time_us();
+    if (filter->updated + filter->window_us < now) {
+        invalidate(filter);
+    }
+    return filter->value;
+}
+
+static void PrrtPaceFilter_max_update(PrrtPaceFilter* filter, prrtTimedelta_t value) {
+    prrtTimestamp_t now = PrrtClock_get_current_time_us();
+    if (filter->valid == false) {
+        filter->value = value;
+        filter->valid = true;
+        filter->updated = now;
+    } else {
+        if (value > filter->value) {
+            filter->value = value;
+            filter->updated = now;
+        } else {
+            // ignore sample
+        }
+    }
+
+    if (filter->updated + filter->window_us < now) {
+        invalidate(filter);
+    }
+
+}
+
+static void PrrtPaceFilter_min_update(PrrtPaceFilter* filter, prrtTimedelta_t value) {
+    prrtTimestamp_t now = PrrtClock_get_current_time_us();
+    if (filter->valid == false) {
+        filter->value = value;
+        filter->valid = true;
+        filter->updated = now;
+    } else {
+        if (value < filter->value) {
+            filter->value = value;
+            filter->updated = now;
+        } else {
+            // ignore sample
+        }
+    }
+
+    if (filter->updated + filter->window_us < now) {
+        invalidate(filter);
+    }
+
+}
+
+typedef void (*PrrtPaceUpdateFunction)(PrrtPaceFilter *, prrtTimedelta_t);
+
+static PrrtPaceUpdateFunction updates[] = {
+ [FILTER_TYPE_MIN] = PrrtPaceFilter_min_update,
+ [FILTER_TYPE_MAX] = PrrtPaceFilter_max_update,
+ NULL
+};
+
+void PrrtPaceFilter_update(PrrtPaceFilter *filter, prrtTimedelta_t value)
+{
+	updates[filter->type](filter, value);
+}
+

prrt/proto/stores/paceFilter.h

+#ifndef PRRT_PACEFILTER_H
+#define PRRT_PACEFILTER_H
+
+#include"../types/packet.h"
+
+typedef enum prrtPaceFilterType {
+  FILTER_TYPE_MIN,   // maximum filter
+  FILTER_TYPE_MAX,   // minimum filter
+  FILTER_TYPE__COUNT // only used for bounds checking
+} prrtPaceFilterType;
+
+typedef struct prrtPaceFilter {
+	prrtPaceFilterType type;
+    prrtTimedelta_t window_us;
+    bool valid;
+    prrtTimestamp_t updated;
+    prrtTimedelta_t value;
+} PrrtPaceFilter;
+
+PrrtPaceFilter* PrrtPaceFilter_create(prrtTimedelta_t window_us, prrtPaceFilterType type);
+bool PrrtPaceFilter_destroy(PrrtPaceFilter* filter);
+prrtTimedelta_t PrrtPaceFilter_get(PrrtPaceFilter* filter);
+void PrrtPaceFilter_update(PrrtPaceFilter* filter, prrtTimedelta_t value);
+
+
+#endif //PRRT_PACEFILTER_H

prrt/proto/stores/packetDeliveryStore.c

@@ -6,16 +6,7 @@
 #include "../../util/dbg.h"
 #include "../../util/time.h"
 #include "packetDeliveryStore.h"
-
-int packet_wait(PrrtPacketDeliveryStore *store, prrtTimestamp_t start, prrtTimestamp_t stop, const struct timespec *deadline) {
-    store->last_start = start;
-    store->last_stop = stop;
-    store->in_ordered_wait = true;
-    int result = pthread_cond_timedwait(&store->wait_for_data, &store->lock, deadline);
-    debug(DEBUG_RECEIVER, "After wait: %d", result);
-    store->in_ordered_wait = false;
-    return result;
-}
+#include "pace.h"
 
 PrrtPacketDeliveryStore *PrrtPacketDeliveryStore_create() {
     PrrtPacketDeliveryStore *store = (PrrtPacketDeliveryStore *) calloc(1, sizeof(PrrtPacketDeliveryStore));
@@ -28,7 +19,7 @@ PrrtPacketDeliveryStore *PrrtPacketDeliveryStore_create() {
     pthread_mutexattr_t attr;
     check(pthread_mutexattr_init(&attr) == EXIT_SUCCESS, "Mutex attr init failed.");
     check(pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE) == EXIT_SUCCESS, "Setting type failed.");
-    check(pthread_mutex_init(&store->lock, &attr) == 0, "Mutex init failed.");
+    check(pthread_mutex_init(&store->lock, &attr) == EXIT_SUCCESS, "Mutex init failed.");
 
     check(pthread_cond_init(&store->wait_for_data, NULL) == EXIT_SUCCESS, "Condition init failed.");
 
@@ -69,15 +60,23 @@ bool PrrtPacketDeliveryStore_destroy(PrrtPacketDeliveryStore *store) {
     return true;
 }
 
-PrrtPacket * PrrtPacketDeliveryStore_get_packet_timedwait(PrrtPacketDeliveryStore *store, prrtTimestamp_t start,
-                                                          prrtTimestamp_t stop, const struct timespec *deadline) {
+PrrtPacket *PrrtPacketDeliveryStore_get_packet_timedwait(PrrtPacketDeliveryStore *store, prrtTimestamp_t start,
+                                                         prrtTimestamp_t stop, const struct timespec *deadline,
+                                                         PrrtPace *pace) {
     PrrtPacket *packet = NULL;
 
     check(pthread_mutex_lock(&store->lock) == EXIT_SUCCESS, "Lock failed.");
     do {
         packet = PrrtPacketDeliveryStore_get_packet(store, start, stop);
         if (packet == NULL) {
-            int res = packet_wait(store, start, stop, deadline);
+            store->last_start = start;
+            store->last_stop = stop;
+            store->in_ordered_wait = true;
+            PrrtPace_track_pause(pace);
+            int res = pthread_cond_timedwait(&store->wait_for_data, &store->lock, deadline);
+            PrrtPace_track_resume(pace);
+            debug(DEBUG_RECEIVER, "After wait: %d", res);
+            store->in_ordered_wait = false;
             packet = PrrtPacketDeliveryStore_get_packet(store, start, stop);
             if (packet == NULL && res == ETIMEDOUT) {
                 errno = ETIMEDOUT;
@@ -85,6 +84,7 @@ PrrtPacket * PrrtPacketDeliveryStore_get_packet_timedwait(PrrtPacketDeliveryStor
             }
         }
         if (atomic_load_explicit(&store->closing, memory_order_acquire)) {
+            errno = ETIMEDOUT;
             break;
         }
     } while(!packet);
@@ -97,28 +97,6 @@ PrrtPacket * PrrtPacketDeliveryStore_get_packet_timedwait(PrrtPacketDeliveryStor
     return NULL;
 }
 
-PrrtPacket *PrrtPacketDeliveryStore_get_packet_wait(PrrtPacketDeliveryStore *store) {
-    PrrtPacket *packet = NULL;
-    prrtTimestamp_t start = 0;
-    prrtTimestamp_t stop = MAX_TIMESTAMP;
-
-    check(pthread_mutex_lock(&store->lock) == EXIT_SUCCESS, "Lock failed.");
-
-    packet = PrrtPacketDeliveryStore_get_packet(store, start, stop);
-    if (!packet && !atomic_load_explicit(&store->closing, memory_order_acquire)) {
-        pthread_cond_wait(&store->wait_for_data, &store->lock);
-        packet = PrrtPacketDeliveryStore_get_packet(store, start, stop);
-    }
-
-    check(pthread_mutex_unlock(&store->lock) == EXIT_SUCCESS, "Unlock failed.");
-
-    return packet;
-
-    error:
-    PERROR("PrrtPacketDeliveryStore_get_packet_timedwait failed%s.", "");
-    return NULL;
-}
-
 PrrtPacket *PrrtPacketDeliveryStore_get_packet(PrrtPacketDeliveryStore *store, prrtTimestamp_t start, prrtTimestamp_t stop) {
     PrrtPacket *packet = NULL;
 
@@ -153,10 +131,7 @@ bool PrrtPacketDeliveryStore_insert(PrrtPacketDeliveryStore *store, PrrtPacket *
     prrtTimestamp_t timeout = PrrtDataPacket_packet_timeout(packet);
     store->tree = BPTree_insert(store->tree, timeout, packet);
 
-    bool packet_awaited = (PrrtTimestamp_cmp(store->last_start, timeout) <= 0) && (PrrtTimestamp_cmp(timeout, store->last_stop) <= 0);
-    if (!store->in_ordered_wait || packet_awaited) {
-        pthread_cond_broadcast(&store->wait_for_data);
-    }
+    pthread_cond_broadcast(&store->wait_for_data);
 
     check(pthread_mutex_unlock(&store->lock) == EXIT_SUCCESS, "Unlock failed.");
     return true;

prrt/proto/stores/packetDeliveryStore.h

@@ -2,6 +2,7 @@
 #define PRRT_RECEIVEDATAQUEUE_H
 
 #include <sys/types.h>
+#include "pace.h"
 #include "../../util/bptree.h"
 
 typedef struct prrtReceiveDataQueue {
@@ -21,7 +22,8 @@ PrrtPacket *PrrtPacketDeliveryStore_get_packet(PrrtPacketDeliveryStore *store, p
 PrrtPacket *PrrtPacketDeliveryStore_get_packet_wait(PrrtPacketDeliveryStore *store);
 
 PrrtPacket *PrrtPacketDeliveryStore_get_packet_timedwait(PrrtPacketDeliveryStore *store, prrtTimestamp_t start,
-                                                         prrtTimestamp_t stop, const struct timespec *deadline);
+                                                         prrtTimestamp_t stop, const struct timespec *deadline,
+                                                         PrrtPace *pace);
 
 bool PrrtPacketDeliveryStore_insert(PrrtPacketDeliveryStore *store, PrrtPacket *packet);
 

prrt/proto/stores/receptionTable.h

@@ -6,20 +6,27 @@
 #include "../types/packet.h"
 #include "../../util/bitmap.h"
 #include "../types/lossStatistics.h"
+#include "../stores/pace.h"
 
 
 typedef struct {
-    prrtSequenceNumber_t start;
     Bitmap* bitmap;
-    prrtSequenceNumber_t nextNumberAfterWindow;
-    prrtSequenceNumber_t windowSize;
+    PrrtLossStatistics stats;
+    prrtSequenceNumber_t windowStart;
+    prrtSequenceNumber_t prevStart;
+    prrtSequenceNumber_t prevEnd;
+
+    prrtSequenceNumber_t latestNumber;
+    prrtTimestamp_t latestStamp;
+
     pthread_mutex_t lock;
 } PrrtReceptionTable;
 
-PrrtReceptionTable * PrrtReceptionTable_create(void);
-bool PrrtReceptionTable_mark_received(PrrtReceptionTable* t, prrtSequenceNumber_t markedNumber);
-PrrtLossStatistics PrrtReceptionTable_calculate_statistics(PrrtReceptionTable *lossGatherer);
+PrrtReceptionTable * PrrtReceptionTable_create();
+bool PrrtReceptionTable_mark_received(PrrtReceptionTable* t, prrtSequenceNumber_t markedNumber, prrtTimestamp_t timestamp);
+PrrtLossStatistics PrrtReceptionTable_calculate_statistics(PrrtReceptionTable *lossGatherer, prrtTimedelta_t ipt, prrtTimestamp_t start, prrtTimestamp_t end);
 bool PrrtReceptionTable_destroy(PrrtReceptionTable * t);
 
+#define PRRT_LOSS_STATISTICS_WINDOW_SIZE ((prrtSequenceNumber_t) (SEQNO_SPACE / 2))
 
 #endif //PRRT_RECEPTIONTABLE_H

prrt/proto/stores/repairBlockStore.c

@@ -20,6 +20,7 @@ PrrtRepairBlockStore *PrrtRepairBlockStore_create(void)
 
 bool PrrtRepairBlockStore_destroy(PrrtRepairBlockStore *store)
 {
+    check(pthread_mutex_lock(&store->lock) == EXIT_SUCCESS, "Lock failed.");
     List *blockList = List_create();
     BPTree_get_range(store->blockTree, blockList, 0, (prrtSequenceNumber_t) (SEQNO_SPACE - 1));
     while (List_count(blockList) > 0) {
@@ -27,16 +28,19 @@ bool PrrtRepairBlockStore_destroy(PrrtRepairBlockStore *store)
         store->blockTree = BPTree_delete(store->blockTree, block->baseSequenceNumber);
         PrrtBlock_destroy(block);
     }
-
     List_destroy(blockList);
 
     if(store->blockTree != NULL) {
         store->blockTree = BPTree_destroy(store->blockTree);
     }
 
-    pthread_mutex_destroy(&store->lock);
+    check(pthread_mutex_unlock(&store->lock) == EXIT_SUCCESS, "Unlock failed.");
+    check(pthread_mutex_destroy(&store->lock) == EXIT_SUCCESS, "Destroy lock failed.");
     free(store);
     return true;
+    error:
+    PERROR("PrrtRepairBlockStore_destroy() failed.");
+    return false;
 }
 
 bool PrrtRepairBlockStore_delete(PrrtRepairBlockStore *store, prrtSequenceNumber_t sequenceNumber)
@@ -68,7 +72,7 @@ bool PrrtRepairBlockStore_insert(PrrtRepairBlockStore *store, PrrtBlock *block)
     check(pthread_mutex_unlock(&store->lock) == EXIT_SUCCESS, "Unlock failed.");
     return true;
     error:
-    PERROR("Could not insert%s.","");
+    PERROR("Could not insert block with base sequence number %d.",block->baseSequenceNumber);
     return false;
 }
 

prrt/proto/timer.h

+#ifndef PRRT_TIMER_H
+#define PRRT_TIMER_H
+
+#include <pthread.h>
+#include <stdatomic.h>
+#include "../util/futex.h"
+#include "../util/list.h"
+
+typedef void *prrtTimerTaskArg;
+typedef void (*prrtTimerTaskFun)(prrtTimerTaskArg);
+
+typedef struct timespec prrtTimerDate;
+
+typedef struct prrtTimerTask {
+	prrtTimerTaskFun fun;
+	prrtTimerTaskArg arg;
+} PrrtTimerTask;
+
+typedef struct prrtTimer PrrtTimer;
+
+PrrtTimer *PrrtTimer_create(unsigned int core);
+
+int PrrtTimer_submit(PrrtTimer *timer, const prrtTimerDate *when, const PrrtTimerTask *what);
+
+void PrrtTimer_end(PrrtTimer *timer);
+
+#endif // PRRT_TIMER_H
\ No newline at end of file

prrt/proto/types/block.c

@@ -41,36 +41,47 @@ static void clear_list(gf *const *src, uint8_t k)
     }
 }