Merge branch 'develop'

.gitignore

@@ -9,3 +9,4 @@ tests/__pycache__/
 MANIFEST
 prrt.cpython*.so
 prrt.so
+.ipynb_checkpoints/

.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:
@@ -78,4 +89,20 @@ deploy:pypi:
     - echo "username=$PYPI_USER" >> ~/.pypirc
     - echo "password=$PYPI_PASSWORD" >> ~/.pypirc
     - python3 setup.py check sdist bdist upload -r on
-    - rm -vf ~/.pypirc
\ No newline at end of file
+    - rm -vf ~/.pypirc
+
+deploy:profile:
+  stage: deploy
+  tags:
+    - gprof
+  script:
+    - ls -lahv
+    - rm CMakeCache.txt
+    - CC=gcc-5 CXX=g++-5 cmake . -DGPROF=1
+    - make
+    - bash profiling/profile.sh
+  artifacts:
+    paths:
+    - gprof-send.txt
+    - gprof-recv.txt
+    expire_in: 30 days
\ No newline at end of file

CMakeLists.txt

@@ -2,16 +2,22 @@ cmake_minimum_required (VERSION 2.8.11)
 project (PRRT)
 
 option(PRRT_TESTS "Build tests" OFF)
+option(GPROF "Compile with profiler" OFF)
 
 set(CMAKE_ARCHIVE_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR})
 set(CMAKE_LIBRARY_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR})
 set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR})
 
 set(CMAKE_C_FLAGS "-O2 -Wall -std=gnu11 -D_GNU_SOURCE -fPIC" )
+set(CMAKE_C_FLAGS_DEBUG "-O0 -fsanitize=undefined -fsanitize=address -g3" )
 set(CMAKE_CXX_FLAGS "-fstack-protector -fstack-protector-all -Wall -std=gnu++11 -D_GNU_SOURCE" )
-set(CMAKE_CXX_FLAGS_DEBUG "-O2 -Wall -ggdb" )
+set(CMAKE_CXX_FLAGS_DEBUG "-O0 -Wall -ggdb -fsanitize=undefined -fsanitize=address -g3" )
 set(CMAKE_CXX_FLAGS_RELEASE "-Os -Wall" )
 
+if(GPROF)
+    set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -pg")
+    set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -pg")
+endif()
 
 find_package (Threads)
 find_library(M_LIB m)

README.md

@@ -4,12 +4,14 @@
 
 ## Features
 
-* Forward Error Correction (FEC) using systematic Vandermonde codes
+* Hybrid error control (FEC + ARQ) using systematic Vandermonde codes
+* Congestion control and pacing 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
 * Passive measurement of propagation delay, bottleneck data rate and packet loss rate
 * Packet-level timing analysis using [X-Lap](http://xlap.larn.systems)
+* Wireshark dissector written in Lua
 * [Hardware timestamping support](https://git.nt.uni-saarland.de/LARN/PRRT/wikis/hardware-timestamping)
 
 ## Installation
@@ -29,9 +31,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 +46,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:
@@ -79,6 +83,20 @@ Packet 9
 * [PRRT Wiki](https://git.nt.uni-saarland.de/LARN/PRRT/wikis)
 * [LARN Project Website](http://larn.systems)
 
+## Citing Us
+
+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}
+}
+```
+
 ## License
 
 [MIT Licence](LICENSE)

dissect/prrt.lua

+-- For an excellent example script, see
+-- https://wiki.wireshark.org/Lua/Examples?action=AttachFile&do=get&target=dissector.lua
+
+
+-- declare our protocol
+local prrt_proto = Proto("prrt","Predictably Reliable Real-time Transport")
+
+local prrtPacketTypeNames = {
+    [0] = "Data",
+    [1] = "Repetition",
+    [2] = "Redundancy",
+    [3] = "Feedback",
+    [4] = "Pre-sent Redundancy",
+    [5] = "Channel Feedback"
+}
+
+-- create the fields
+local pf_type = ProtoField.uint8("prrt.type", "Type", base.DEC, prrtPacketTypeNames, 240)
+local pf_prio = ProtoField.uint8("prrt.priority", "Priority", base.DEC, nil, 15)
+local pf_idx = ProtoField.uint8("prrt.index", "Index")
+local pf_seqN = ProtoField.uint16("prrt.sequenceNumber", "Sequence Number")
+
+local pf_data = ProtoField.new("Data", "prrt.data", ftypes.BYTES, base.NONE)
+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_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_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_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")
+local pf_fb_gapcount = ProtoField.uint16("prrt.feedback.gapCount", "Gap count")
+local pf_fb_burstlength = ProtoField.uint16("prrt.feedback.burstLength", "Burst length")
+local pf_fb_burstcount = ProtoField.uint16("prrt.feedback.burstCount", "Burst count")
+local pf_fb_acktype = ProtoField.uint8("prrt.feedback.ackPacketType", "Ack Packet Type", base.DEC, prrtPacketTypeNames)
+local pf_fb_ackSeqN = ProtoField.uint16("prrt.feedback.ackSequenceNumber", "Ack Sequence Number")
+
+-- add the fields to the protocol
+prrt_proto.fields = {
+    pf_type,
+    pf_prio,
+    pf_idx,
+    pf_seqN,
+
+    pf_data,
+    pf_data_length,
+    pf_data_timestamp,
+    pf_data_groupRTprop,
+    pf_data_packettimeout,
+
+    pf_red,
+    pf_red_baseSeqN,
+    pf_red_timestamp,
+    pf_red_n,
+    pf_red_k,
+
+    pf_fb,
+    pf_fb_groupRTT,
+    pf_fb_ftt,
+    pf_fb_erasurecount,
+    pf_fb_packetcount,
+    pf_fb_gaplength,
+    pf_fb_gapcount,
+    pf_fb_burstlength,
+    pf_fb_burstcount,
+    pf_fb_acktype,
+    pf_fb_ackSeqN,
+}
+
+-- create expert info fields
+local ef_too_short = ProtoExpert.new("prrt.too_short.expert", "PRRT Packet too short",
+                                     expert.group.MALFORMED, expert.severity.ERROR)
+
+prrt_proto.experts = {
+    ef_too_short
+}
+
+-- Create extractor fields (for some reason this can't be done with the fields that already exist)
+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
+
+local ex_red_k = Field.new("prrt.redundancy.k")
+local function getRedK() return ex_red_k()() end
+
+-- some constants
+local PRRT_MIN_SIZE = 8
+
+-- create sub-dissectors for different types
+local function dissect_data(buffer, pinfo, root)
+    local tree = root:add(pf_data, buffer:range(0))
+    tree:add(pf_data_length, buffer:range(0,4))
+    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))
+
+    local label = "[D] Idx=" .. getIndex() .. " Len=" .. getDataLength()
+    tree:set_text(label)
+    pinfo.cols.info:set(label)
+end
+
+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))
+
+    local label = "[R] Idx=" .. getIndex() .. " b=" .. getRedBaseSeqNo() .. " n=" .. getRedN() .. " k=" .. getRedK()
+    tree:set_text(label)
+    pinfo.cols.info:set(label)
+end
+
+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 = "[F]"
+    tree:set_text(label)
+    pinfo.cols.info:set(label)
+end
+
+local subdissectors = {
+    [0] = dissect_data,
+    [2] = dissect_redundancy,
+    [3] = dissect_feedback
+}
+
+-- create a function to dissect each frame
+function prrt_proto.dissector(buffer,pinfo,root)
+    -- set the protocol column
+    pinfo.cols.protocol:set("PRRT")
+
+    local pktlen = buffer:reported_length_remaining()
+
+    --[[ TODO
+         parse whether the packet has a payload first
+         if it has, do not use the entire buffer:range(0)
+         in the tree label, but only up to the payload
+         and add another "Data" subtree containing only
+         the payload
+    --]]
+    local tree = root:add(prrt_proto, buffer:range(0))
+
+    if pktlen < PRRT_MIN_SIZE then
+        tree:add_proto_expert_info(ef_too_short)
+        return
+    end
+
+    tree:add(pf_type, buffer:range(0,1))
+    tree:add(pf_prio, buffer:range(0,1))
+    tree:add(pf_idx,  buffer:range(1,1))
+    tree:add(pf_seqN, buffer:range(2,2))
+
+    if subdissectors[getType()] then
+        subdissectors[getType()](buffer:range(4,nil):tvb(), pinfo, tree)
+    end
+
+end
+
+local current_port = 6000
+DissectorTable.get("udp.port"):add(current_port, prrt_proto)
+
+prrt_proto.prefs.port = Pref.uint("Port number", current_port, "The UDP Port number for PRRT")
+
+function prrt_proto.prefs_changed()
+    if current_port ~= prrt_proto.prefs.port then
+        if curent_port ~= 0 then
+            DissectorTable.get("udp.port"):remove(current_port, prrt_proto)
+        end
+        current_port = prrt_proto.prefs.port
+        if current_port ~= 0 then
+            DissectorTable.get("udp.port"):add(current_port, prrt_proto)
+        end
+    end
+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

eval.ipynb

+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import pandas as pd"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "pd.read_csv(\"eval.csv\")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": []
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.5.2"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}

eval.py

+import sys
+sys.path.insert(0, "./build")
+
+import tests.perf as perf
+
+def run_setup_and_report(setup):
+    results = setup.run()
+    print("Setup:\n   ", setup)
+    print("Results:\n    " + str(results).replace("\n","\n    "))
+    results.export()
+
+if __name__ == "__main__":
+    # Packet Count Works: 2^15; Segfault at: 2^16
+
+    # TODO: support multiple tests via proper socket termination
+    setups = [
+        perf.TestSetup(packets=2**17,delay=1,loss=0,reorder=0,duplicate=0)
+    ]
+
+    for setup in setups:
+        run_setup_and_report(setup)

examples/sender.py

@@ -5,7 +5,7 @@ host = sys.argv[1]
 port = int(sys.argv[2])
 localport = int(sys.argv[3])
 
-s = prrt.PrrtSocket(("127.0.1.1", localport))
+s = prrt.PrrtSocket(("127.0.1.1", localport), maximum_payload_size=150)
 s.connect((host, port))
 
 for i in range(10):

profiling/profile.sh

+#!/usr/bin/env bash
+
+which gprof
+
+run_in() {
+    (cd $1; ${@:2})
+}
+
+to="timeout -s INT 30 "
+
+mkdir -p gprof_send gprof_recv
+
+run_in gprof_recv $to ../receiver -p 5000 -r 4095 -o receiver.csv &
+run_in gprof_send $to ../sender -t 127.0.0.1 -p 5000 -r 4095 -o sender.csv
+
+wait
+
+gprof ./receiver gprof_recv/gmon.out > gprof-recv.txt
+gprof ./sender gprof_send/gmon.out > gprof-send.txt
+
+rm -r gprof_send gprof_recv
\ No newline at end of file

prrt/CMakeLists.txt

@@ -10,12 +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 ../tests/common.h)
+add_executable(receiver receiver.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/__init__.py

-