Adding congestion control and ARQ.

.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

@@ -5,6 +5,7 @@
 ## Features
 
 * Hybrid error control (FEC + ARQ) using systematic Vandermonde codes
+* Congestion control using a variant of [BBR](https://groups.google.com/forum/#!forum/bbr-dev)
 * Clock synchronization between sending stack and receiving stack
 * Applications can specify packet-level expiration times
 * Different receive modes for ASAP and time-synchronized operation
@@ -29,9 +30,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 +45,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:
@@ -84,13 +87,13 @@ Packet 9
 If you find PRRT useful and incorporate it in your works, we are very happy to hear about it. Please also consider to cite us like this:
 
 ```bibtex
-@misc{sic2018prrt, 
-      author = {Schmidt, Andreas}, 
-      title = {PRRT: Predictably Reliable Real-time Transport}, 
-      howpublished={Web page}, 
-      url = {http://prrt.larn.systems}, 
-      year = {2018} 
-} 
+@misc{sic2018prrt,
+      author = {Schmidt, Andreas},
+      title = {PRRT: Predictably Reliable Real-time Transport},
+      howpublished={Web page},
+      url = {http://prrt.larn.systems},
+      year = {2018}
+}
 ```
 
 ## License

dissect/prrt.lua

@@ -89,9 +89,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
 
@@ -109,7 +116,7 @@ local function dissect_data(buffer, pinfo, root)
     tree:add(pf_data_groupRTprop, buffer:range(8,4))
     tree:add(pf_data_packettimeout, buffer:range(12,4))
 
-    local label = "DATA Len=" .. getDataLength()
+    local label = "[D] Idx=" .. getIndex() .. " Len=" .. getDataLength()
     tree:set_text(label)
     pinfo.cols.info:set(label)
 end
@@ -121,7 +128,7 @@ local function dissect_redundancy(buffer, pinfo, root)
     tree:add(pf_red_n, buffer:range(6,1))
     tree:add(pf_red_k, buffer:range(7,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
@@ -139,7 +146,7 @@ local function dissect_feedback(buffer, pinfo, root)
     tree:add(pf_fb_acktype, buffer:range(20,1))
     tree:add(pf_fb_ackSeqN, buffer:range(21, 2))
 
-    local label = "FEEDBACK"
+    local label = "[F]"
     tree:set_text(label)
     pinfo.cols.info:set(label)
 end

docker/Dockerfile

@@ -2,9 +2,13 @@ FROM gcc:5
 
 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 \
+    traceroute \
+    tshark
 
 COPY CMakeLists.txt /prrt/
 COPY prrt /prrt/prrt
@@ -15,8 +19,7 @@ WORKDIR /prrt
 
 RUN cmake . \
     && make
-ENV PATH /prrt/bin:$PATH
-WORKDIR /prrt/bin
+ENV PATH /prrt:$PATH
 
 VOLUME /output
 

docker/Dockerfile_tcp

+FROM gcc:5
+
+MAINTAINER Andreas Schmidt <schmidt@nt.uni-saarland.de>
+
+ENV DEBIAN_FRONTEND noninteractive
+
+RUN apt-get update && apt-get install --yes --force-yes \
+    bc \
+    cmake \
+    traceroute \
+    tshark
+
+COPY CMakeLists.txt /prrt/
+COPY prrt /prrt/prrt
+COPY tests /prrt/tests
+COPY docker/entrypoint.sh /
+
+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)
+    PRRT+=("$1")
+    shift
+    ;;
+    -p|--port|-r|--rounds|-s|--size|-R|--rcvbuf|-S|--sndbuf|-o|--output|-a|--appdelay)
     PRRT+=("$1 $2")
     shift
     shift
@@ -34,8 +53,31 @@ done
 PRRT_PARAMS="${PRRT[@]}"
 NETEM_PARAMS="${NETEM[@]}"
 
+echo "Starting Wireshark."
+tshark -i eth0 -w $OUTPUT.pcap &
+TSHARK_PID=$!
+sleep 5
+
+start=$(date +%s.%N);
+echo "Checking reachability of $TARGET."
+until ping -c1 $TARGET &>/dev/null; sleep 1; do :; done
+dur=$(echo "$(date +%s.%N) - $start" | bc);
+printf "Reachable after %.6f seconds\n" $dur
+
+traceroute $TARGET > $OUTPUT.tr
+echo "Traceroute done."
+
+if [[ "$command" == "sender" || "$command" == "time-sender" ]]; then
+    echo "Delaying sender start."
+else
+    echo "Delaying receiver start."
+fi
+
 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
+/prrt/$command $PRRT_PARAMS
+echo "Done."
 tc qdisc del dev $dev root
+
+kill $TSHARK_PID

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,13 @@ 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 +146,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/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
@@ -15,10 +13,14 @@ set (PRRT_SOURCES ../defines.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.25
+#define DRAIN_GAIN 0.75
+#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.c

@@ -2,12 +2,15 @@
 #include <netdb.h>
 #include <string.h>
 #include "../../defines.h"
+#include "../timer.h"
 #include "../receiver.h"
 #include "../socket.h"
 #include "../types/block.h"
 #include "../../util/dbg.h"
 #include "../../util/common.h"
+#include "../../util/time.h"
 #include "dataTransmitter.h"
+#include <math.h>
 
 bool send_to_socket(PrrtSocket* sock_ptr, uint8_t* buf, prrtPacketLength_t length,  struct timespec *packet_timestamp, uint64_t* packet_clockstamp) {
     PrrtReceiver *recv = sock_ptr->receiver;
@@ -77,6 +80,41 @@ static bool send_packet(PrrtSocket *sock_ptr, PrrtPacket *packet) {
     uint8_t buf[MAX_PAYLOAD_LENGTH];
     memset(buf, 0, sizeof(buf));
     prrtPacketLength_t length = PrrtPacket_size(packet);
+    prrtPacketLength_t payloadLength = packet->payloadLength;
+
+    bool paceSuccessful = PrrtSocket_pace(sock_ptr);
+    if (!paceSuccessful) {
+        debug(DEBUG_DATATRANSMITTER, "Pacing interrupted.");
+        PrrtPacket_destroy(packet);
+        return false;
+    }
+    debug(DEBUG_DATATRANSMITTER, "Pacing interval passed.");
+
+    bool waitSuccessful = PrrtReceiver_wait_for_space(sock_ptr->receiver, sock_ptr->maximum_payload_size,
+                                                      sock_ptr->applicationConstraints);
+    if(!waitSuccessful) {
+        debug(DEBUG_DATATRANSMITTER, "Wait for space interrupted.");
+        PrrtPacket_destroy(packet);
+        return false;
+    }
+    debug(DEBUG_DATATRANSMITTER, "Space available.");
+
+    prrtTimestamp_t now = PrrtClock_get_current_time_us();
+    if (sock_ptr->pacingEnabled) {
+        double pacing_rate = PrrtReceiver_getBBRPacingRate(sock_ptr->receiver);
+        if(pacing_rate != 0) {
+            prrtTimedelta_t pacingTime = (prrtTimedelta_t) round(((1000 * 1000 * ( (double)payloadLength)) / pacing_rate));
+            debug(DEBUG_DATATRANSMITTER, "Payload: %u, PacingRate: %f, Pacing Time: %u", packet->payloadLength, pacing_rate, pacingTime);
+            sock_ptr->nextSendTime = now + pacingTime;
+        }
+    }
+    // Update timestamp
+    if(PrrtPacket_type(packet) == PACKET_TYPE_DATA) {
+        ((PrrtPacketDataPayload*) (packet->payload))->timestamp = PrrtClock_get_current_time_us();
+        ((PrrtPacketDataPayload*) (packet->payload))->btlbw = PrrtReceiver_get_BBR_btlDr(sock_ptr->receiver);
+    } else if(PrrtPacket_type(packet) == PACKET_TYPE_REDUNDANCY) {
+        ((PrrtPacketRedundancyPayload*) (packet->payload))->timestamp = PrrtClock_get_current_time_us();
+    }
 
     check(PrrtPacket_encode(buf, MAX_PAYLOAD_LENGTH, packet), "Buffer too small.");
 
@@ -125,57 +163,110 @@ static bool send_packet(PrrtSocket *sock_ptr, PrrtPacket *packet) {
     return false;
 }
 
-void *send_data_loop(void *ptr) {
-    PrrtSocket *sock_ptr = ptr;
-    PrrtBlock *block = NULL;
+typedef struct timer_arg {
+    PrrtSocket* socket;
+    PrrtBlock* block;
+} RetransmissionTimerArgs;
 
-    while (1) {
-        ListNode *job;
-        do {
-            job = Pipe_pull(sock_ptr->sendDataQueue);
-            if (PrrtSocket_closing(sock_ptr)) {
-                if (block != NULL) {
-                    PrrtBlock_destroy(block);
-                }
-                return NULL;
-            }
-        } while (!job);
-        PrrtPacket *packet = PrrtPacket_byListNode(job);
-        XlapTimeStampClock(sock_ptr, ts_data_packet, packet->sequenceNumber, PrrtTransmitStart);
-        XlapTimeStampCycle(sock_ptr, ts_data_packet, packet->sequenceNumber, PrrtTransmitStart);
-        if (block == NULL) {
-            block = PrrtBlock_create(PrrtCodingConfiguration_copy(sock_ptr->codingParameters), PrrtCoder_copy(sock_ptr->coder), packet->sequenceNumber);
+void retransmission_round_handler(void *arg) {
+    uint8_t j;
+    RetransmissionTimerArgs* args = (RetransmissionTimerArgs*) arg;
+
+    PrrtBlock *block = args->block;
+    PrrtSocket *socket = args->socket;
+
+    if(block->inRound > 0) {
+        PrrtReceiver_rto_check(socket->receiver, socket->applicationConstraints);
+    }
+
+    if (PrrtSocket_closing(socket) || block->inRound >= block->codingParams->c) {
+        PrrtBlock_destroy(block);
+        free(arg);
+        return;
+    }
+
+    uint32_t redundancyPackets = block->codingParams->n_cycle[block->inRound];
+    for (j = 0; j < redundancyPackets; j++) {
+        PrrtPacket *red_pkt = PrrtBlock_get_first_red_data(block);
+        bool sendResult = send_packet(socket, red_pkt);
+        if(!sendResult) {
+            debug(DEBUG_DATATRANSMITTER, "Sending redundancy data failed.");
+            PrrtBlock_destroy(block);
+            free(arg);
+            return;
         }
+    }
 
-        packet->index = (uint8_t) (packet->sequenceNumber - block->baseSequenceNumber);
+    block->inRound++;
+    PrrtTimerTask task = {
+            .arg = arg,
+            .fun = retransmission_round_handler
+    };
 
-        PrrtPacketDataPayload *payload = packet->payload;
-        payload->groupRTprop_us = PrrtSocket_get_rtprop_fwd(sock_ptr);
+    uint32_t waittime_us = PrrtReceiver_get_retransmission_delay(socket->receiver, socket->applicationConstraints);
+    prrtTimerDate deadline = abstime_from_now(waittime_us);
+    debug(DEBUG_DATATRANSMITTER, "Set timer to expire in: %dus", waittime_us);
+    PrrtTimer_submit(socket->retransmissionTimer, &deadline, &task);
+}
+
+
+void PrrtDataTransmitter_transmit(PrrtSocket *sock_ptr, PrrtPacket *packet) {
+    XlapTimeStampClock(sock_ptr, ts_data_packet, packet->sequenceNumber, PrrtTransmitStart);
+    XlapTimeStampCycle(sock_ptr, ts_data_packet, packet->sequenceNumber, PrrtTransmitStart);
+    if (sock_ptr->receiveBlock == NULL) {
+        sock_ptr->receiveBlock = PrrtBlock_create(PrrtCodingConfiguration_copy(sock_ptr->codingParameters), PrrtCoder_copy(sock_ptr->coder), packet->sequenceNumber);
+        sock_ptr->receiveBlock->senderBlock = true;
+    }
+
+    packet->index = (uint8_t) (packet->sequenceNumber - sock_ptr->receiveBlock->baseSequenceNumber);
+
+    PrrtPacketDataPayload *payload = packet->payload;
+    payload->groupRTprop_us = PrrtSocket_get_rtprop_fwd(sock_ptr);
 
-        PrrtPacket *packetToSend = PrrtPacket_copy(packet);
-        debug(DEBUG_DATATRANSMITTER, "Send: %i", packet->sequenceNumber);
-        send_packet(sock_ptr, packetToSend);
+    PrrtPacket *packetToSend = PrrtPacket_copy(packet);
+    debug(DEBUG_DATATRANSMITTER, "Send: %i", packet->sequenceNumber);
+    int sendResult = send_packet(sock_ptr, packetToSend);
+    if (sendResult) {
         XlapTimeStampClock(sock_ptr, ts_data_packet, packet->sequenceNumber, PrrtTransmitEnd);
         XlapTimeStampCycle(sock_ptr, ts_data_packet, packet->sequenceNumber, PrrtTransmitEnd);
 
-        PrrtBlock_insert_data_packet(block, packet);
+        if(PrrtBlock_insert_data_packet(sock_ptr->receiveBlock, packet) == false) {
+            PERROR("Failed to insert packet: %d", packet->sequenceNumber);
+        }
 
-        // TODO: redundancy should only be sent when necessary
-        if (PrrtBlock_encode_ready(block)) {
-            uint32_t j = 0;
+        if (PrrtBlock_encode_ready(sock_ptr->receiveBlock)) {
             unsigned int redundancy_seqno = sock_ptr->sequenceNumberRedundancy;
             XlapTimeStampCycle(sock_ptr, ts_redundancy_packet, redundancy_seqno, PrrtEncodeStart);
-            PrrtBlock_encode(block, &sock_ptr->sequenceNumberRedundancy);
+            PrrtBlock_encode(sock_ptr->receiveBlock, &sock_ptr->sequenceNumberRedundancy);
             XlapTimeStampCycle(sock_ptr, ts_redundancy_packet, redundancy_seqno, PrrtEncodeEnd);
 
-            uint32_t redundancyPackets = List_count(block->redundancyPackets);
-            for (j = 0; j < redundancyPackets; j++) {
-                PrrtPacket *red_pkt = PrrtBlock_get_first_red_data(block);
-                send_packet(sock_ptr, red_pkt);
-            }
-
-            PrrtBlock_destroy(block);
-            block = NULL;
+            RetransmissionTimerArgs* args = (RetransmissionTimerArgs*) calloc(1, sizeof(RetransmissionTimerArgs));
+            args->block = sock_ptr->receiveBlock;
+            sock_ptr->receiveBlock = NULL;
+            args->socket = sock_ptr;
+            retransmission_round_handler(args);
         }
+    } else {
+        PrrtPacket_destroy(packet);
+    }
+}
+
+void *PrrtDataTransmitter_send_data_loop(void *ptr) {
+    PrrtSocket *s = ptr;
+
+    while (1) {
+        ListNode *job;
+        do {
+            job = Pipe_pull(s->sendDataQueue);
+            if (PrrtSocket_closing(s)) {
+                if (s->receiveBlock != NULL) {
+                    PrrtBlock_destroy(s->receiveBlock);
+                    s->receiveBlock = NULL;
+                }
+                return NULL;
+            }
+        } while (!job);
+        PrrtPacket *packet = PrrtPacket_byListNode(job);
+        PrrtDataTransmitter_transmit(s, packet);