Commit 8bb0f31b authored by Andreas Schmidt's avatar Andreas Schmidt

Refactor.

parent e8b749f3
Pipeline #2674 passed with stages
in 1 minute and 26 seconds
......@@ -21,13 +21,13 @@ cdef extern from "proto/vdmcode/block_code.h":
ctypedef prrtCoder PrrtCoder
cdef extern from "proto/channelStateInformation.h":
cdef extern from "proto/types/channelStateInformation.h":
cdef struct prrtChannelStateInformation:
pass
ctypedef prrtChannelStateInformation PrrtChannelStateInformation
cdef extern from "proto/codingParams.h":
cdef extern from "proto/types/codingParams.h":
ctypedef struct prrtCodingConfiguration:
uint8_t k
uint8_t r
......@@ -59,7 +59,7 @@ cdef extern from "util/list.h":
void *List_remove(List *list, const ListNode *node)
cdef extern from "proto/block.h":
cdef extern from "proto/types/block.h":
cdef struct prrtBlock:
uint32_t data_count
uint32_t redundancy_count
......
set (PRRT_SOURCES ../defines.h
block.c block.h
channelStateInformation.c channelStateInformation.h
types/block.c types/block.h
types/channelStateInformation.c types/channelStateInformation.h
clock.c clock.h
codingParams.c codingParams.h
types/codingParams.c types/codingParams.h
receiver.c receiver.h
socket.c socket.h
applicationConstraints.c applicationConstraints.h
types/applicationConstraints.c types/applicationConstraints.h
processes/dataReceiver.c processes/dataReceiver.h
processes/dataTransmitter.c processes/dataTransmitter.h
stores/dataPacketStore.c stores/dataPacketStore.h
......
......@@ -5,13 +5,15 @@
#include "../../util/dbg.h"
#include "../../util/common.h"
#include "../types/lossStatistics.h"
#include "../block.h"
#include "../types/block.h"
#include "../clock.h"
#include "../socket.h"
#include "dataReceiver.h"
static void
retrieve_data_blocks(PrrtSocket *sock_ptr, prrtSequenceNumber_t base_seqno, uint8_t k, const PrrtBlock *block) {
static void retrieve_data_blocks(PrrtSocket *sock_ptr,
prrtSequenceNumber_t base_seqno,
uint8_t k,
const PrrtBlock *block) {
List *res = List_create();
PrrtDataPacketStore_remove_range(sock_ptr->dataPacketStore, res, base_seqno,
......@@ -34,7 +36,8 @@ static void decode_block(PrrtSocket *sock_ptr, PrrtBlock *block) {
while (List_count(block->dataPackets) > 0) {
PrrtPacket *pkt = List_shift(block->dataPackets);
if (PrrtDeliveredPacketTable_test_set_is_number_relevant(sock_ptr->deliveredPacketTable, pkt->sequenceNumber)) {
if (PrrtDeliveredPacketTable_test_set_is_number_relevant(sock_ptr->deliveredPacketTable,
pkt->sequenceNumber)) {
PrrtPacketDeliveryStore_insert(sock_ptr->packetDeliveryStore, pkt);
} else {
PrrtPacket_destroy(pkt);
......@@ -50,8 +53,13 @@ static void decode_block(PrrtSocket *sock_ptr, PrrtBlock *block) {
PERROR("Decoding failed.%s", "")
}
static bool send_feedback(PrrtSocket *sock_ptr, struct sockaddr_in remote, prrtSequenceNumber_t seqno, prrtTimestamp_t receiveStamp,
prrtTimestamp_t sentTimestamp, prrtPacketType_t type) {
static bool send_feedback(PrrtSocket *sock_ptr,
struct sockaddr_in remote,
prrtSequenceNumber_t seqno,
prrtTimestamp_t receiveStamp,
prrtTimestamp_t sentTimestamp,
prrtPacketType_t type) {
enum XlapTimestampPacketKind kind = ts_data_packet;
if (type == PACKET_TYPE_DATA) {
kind = ts_data_packet;
......@@ -90,7 +98,8 @@ static bool send_feedback(PrrtSocket *sock_ptr, struct sockaddr_in remote, prrtS
PrrtPacket *feedback_pkt_ptr = PrrtPacket_create_feedback_packet(0, sock_ptr->sequenceNumberFeedback++, group_RTT,
stats.gapLength, stats.gapCount, stats.burstLength,
stats.burstCount, forwardTripTime,
stats.erasureCount, stats.packetCount, feedback.seqNo,
stats.erasureCount, stats.packetCount,
feedback.seqNo,
feedback.type);
prrtPacketLength_t length = PrrtPacket_size(feedback_pkt_ptr);
void *buf = calloc(1, length);
......@@ -118,8 +127,7 @@ static bool is_timeout(prrtTimestamp_t now, prrtTimestamp_t to) {
return false;
}
static void
handle_data_packet(PrrtSocket *sock_ptr, PrrtPacket *packet) {
static void handle_data_packet(PrrtSocket *sock_ptr, PrrtPacket *packet) {
PrrtPacketDataPayload *payload = packet->payload;
prrtTimestamp_t sentTimestamp = payload->timestamp;
sock_ptr->lastSentTimestamp = sentTimestamp;
......@@ -137,9 +145,8 @@ handle_data_packet(PrrtSocket *sock_ptr, PrrtPacket *packet) {
PrrtPacket_destroy(packet);
debug(DEBUG_RECEIVER, "timeout data packet %u (%lu > %lu)", seqno, (unsigned long) now,
(unsigned long) payload->packetTimeout_us);
} else if (PrrtDeliveredPacketTable_test_set_is_number_relevant(sock_ptr->deliveredPacketTable,
packet->sequenceNumber) ==
false) {
} else if (!PrrtDeliveredPacketTable_test_set_is_number_relevant(sock_ptr->deliveredPacketTable,
packet->sequenceNumber)) {
PrrtPacket_destroy(packet);
} else {
prrtSequenceNumber_t baseSequenceNumber = packet->sequenceNumber - packet->index;
......@@ -168,7 +175,6 @@ handle_data_packet(PrrtSocket *sock_ptr, PrrtPacket *packet) {
error:
PERROR("Handling data packet failed%s.", "");
return;
}
static void handle_redundancy_packet(PrrtSocket *socket, PrrtPacket *packet) {
......@@ -184,10 +190,11 @@ static void handle_redundancy_packet(PrrtSocket *socket, PrrtPacket *packet) {
PrrtBlock *block = PrrtRepairBlockStore_get_block(socket->repairBlockStore,
redundancyPayload->baseSequenceNumber);
if (block == NULL) {
PrrtCodingConfiguration* codingParams = PrrtCodingConfiguration_create(redundancyPayload->k,
PrrtCodingConfiguration *codingParams = PrrtCodingConfiguration_create(redundancyPayload->k,
redundancyPayload->n, 0, NULL);
block = PrrtBlock_create(codingParams, PrrtSocket_get_matching_coder(socket, codingParams), redundancyPayload->baseSequenceNumber);
block = PrrtBlock_create(codingParams, PrrtSocket_get_matching_coder(socket, codingParams),
redundancyPayload->baseSequenceNumber);
PrrtRepairBlockStore_insert(socket->repairBlockStore, block);
}
......@@ -200,7 +207,6 @@ static void handle_redundancy_packet(PrrtSocket *socket, PrrtPacket *packet) {
PrrtPacket_destroy(packet);
}
}
return;
}
void handle_feedback_packet(PrrtSocket *prrtSocket, PrrtPacket *prrtPacket, prrtTimestamp_t receiveTime) {
......@@ -209,20 +215,24 @@ void handle_feedback_packet(PrrtSocket *prrtSocket, PrrtPacket *prrtPacket, prrt
PrrtPacketFeedbackPayload *feedbackPayload = (PrrtPacketFeedbackPayload *) prrtPacket->payload;
prrtTimestamp_t forwardTripTimestamp = feedbackPayload->forwardTripTimestamp_us;
bool valid_sample = PrrtReceiver_updateAndGenerateRateSample(prrtSocket->receiver, feedbackPayload->ackSequenceNumber, feedbackPayload->ackPacketType, receiveTime);
bool valid_sample = PrrtReceiver_updateAndGenerateRateSample(prrtSocket->receiver,
feedbackPayload->ackSequenceNumber,
feedbackPayload->ackPacketType, receiveTime);
debug(DEBUG_DATARECEIVER, "PrrtReceiver_updateAndGenerateRateSample ");
if(valid_sample) {
if (valid_sample) {
PrrtChannelStateInformation_update_delivery_rate(prrtSocket->receiver->csi,
prrtSocket->receiver->rateSample->delivery_rate);
}
PrrtChannelStateInformation_update_app_limited(prrtSocket->receiver->csi, prrtSocket->receiver->rateSample->is_app_limited);
PrrtChannelStateInformation_update_app_limited(prrtSocket->receiver->csi,
prrtSocket->receiver->rateSample->is_app_limited);
debug(DEBUG_DATARECEIVER, "PrrtChannelStateInformation_update_app_limited ");
PrrtChannelStateInformation_update_rtprop(prrtSocket->receiver->csi,
(prrtTimedelta_t) (receiveTime - forwardTripTimestamp));
debug(DEBUG_DATARECEIVER, "PrrtChannelStateInformation_update_rtprop ");
PrrtChannelStateInformation_update_plr(prrtSocket->receiver->csi, feedbackPayload->erasureCount, feedbackPayload->packetCount);
PrrtChannelStateInformation_update_plr(prrtSocket->receiver->csi, feedbackPayload->erasureCount,
feedbackPayload->packetCount);
debug(DEBUG_DATARECEIVER, "PrrtChannelStateInformation_update_plr ");
return;
......@@ -231,7 +241,8 @@ void handle_feedback_packet(PrrtSocket *prrtSocket, PrrtPacket *prrtPacket, prrt
}
void receive_from_socket(PrrtSocket *socket_ptr, unsigned char buffer_ptr[65528], ssize_t *received_size,
struct sockaddr_in *remote_ptr, socklen_t *remote_len_ptr, struct timespec *packet_timestamp_ptr,
struct sockaddr_in *remote_ptr, socklen_t *remote_len_ptr,
struct timespec *packet_timestamp_ptr,
uint64_t *packet_cyclestamp_ptr) {
if (socket_ptr->isHardwareTimestamping) {
struct cmsghdr *cmsg;
......@@ -338,10 +349,8 @@ void *receive_data_loop(void *ptr) {
if (packetType == PACKET_TYPE_DATA) {
handle_data_packet(s, packet);
} else if (packetType == PACKET_TYPE_REDUNDANCY) {
handle_redundancy_packet(s, packet);
} else {
goto error;
handle_redundancy_packet(s, packet);
}
send_feedback(s, remote, seqno, prrt_recv_timestamp, sentTimestamp, packetType);
XlapTimeStampCycle(s, kind, seqno, HandlePacketEnd);
......@@ -361,7 +370,6 @@ void *receive_data_loop(void *ptr) {
return NULL;
error:
PNOTIMPLEMENTED("SHOULD IMPLEMENT ERROR HANDLER HERE");
PERROR("receive_data_loop() ended unexpectedly.");
PrrtSocket_cleanup(s);
return NULL;
}
......@@ -2,9 +2,9 @@
#include <netdb.h>
#include <string.h>
#include "../../defines.h"
#include "../block.h"
#include "../receiver.h"
#include "../socket.h"
#include "../types/block.h"
#include "../../util/dbg.h"
#include "../../util/common.h"
#include "dataTransmitter.h"
......
......@@ -6,8 +6,8 @@
#include <sys/types.h>
#include <sys/socket.h>
#include <netdb.h>
#include "channelStateInformation.h"
#include "stores/inFlightPacketStore.h"
#include "types/channelStateInformation.h"
typedef struct prrtRateSample {
prrtByteCount_t prior_delivered;
......
......@@ -6,14 +6,14 @@
#include "../util/pipe.h"
#include "../util/mpsc_queue.h"
#include "../util/bptree.h"
#include "channelStateInformation.h"
#include "applicationConstraints.h"
#include "stores/dataPacketStore.h"
#include "stores/deliveredPacketTable.h"
#include "stores/packetTimeoutTable.h"
#include "stores/receptionTable.h"
#include "stores/repairBlockStore.h"
#include "stores/packetDeliveryStore.h"
#include "types/applicationConstraints.h"
#include "types/channelStateInformation.h"
#include "types/lossStatistics.h"
#include "types/packet.h"
#include "clock.h"
......
......@@ -3,7 +3,6 @@
#include "../../util/common.h"
#include "../../util/bptree.h"
#include "../../util/dbg.h"
#include "../types/packet.h"
#include "dataPacketStore.h"
......
......@@ -5,6 +5,8 @@
#ifndef PRRT_DATAPACKETSTORE_H
#define PRRT_DATAPACKETSTORE_H
#include "../types/packet.h"
typedef struct prrtDataPacketStore {
pthread_mutex_t lock;
BPTreeNode* dataStore;
......
......@@ -2,8 +2,8 @@
#define PRRT_REPAIRBLOCKSTORE_H
#include <bits/pthreadtypes.h>
#include "../types/block.h"
#include "../types/packet.h"
#include "../block.h"
#include "../../util/bptree.h"
typedef struct prrtRepairBlockStore {
......
#include <malloc.h>
#include "../util/common.h"
#include "../util/dbg.h"
#include "../../util/common.h"
#include "../../util/dbg.h"
#include "applicationConstraints.h"
PrrtApplicationConstraints *PrrtApplicationConstraints_create(prrtTimedelta_t target_delay_us)
......@@ -33,6 +33,7 @@ uint32_t PrrtApplicationConstraints_get_app_queue_size(PrrtApplicationConstraint
{
return (uint32_t) atomic_load_explicit(&applicationConstraints->queue_size, memory_order_acquire);
}
bool PrrtApplicationConstraints_set_app_queue_size(PrrtApplicationConstraints *applicationConstraints, uint32_t size)
{
atomic_store_explicit(&applicationConstraints->queue_size, size, memory_order_release);
......
#ifndef PRRT_NETWORKCONTRAINTS_H
#define PRRT_NETWORKCONTRAINTS_H
#include "types/packet.h"
#include "packet.h"
typedef struct applicationConstraints {
prrtTimedelta_t targetDelay_us;
......
#include <string.h>
#include "../defines.h"
#include "../util/list.h"
#include "../util/dbg.h"
#include "../util/common.h"
#include "types/packet.h"
#include "../../defines.h"
#include "../../util/dbg.h"
#include "../../util/common.h"
#include "block.h"
static void gather_redundancy_packets(const PrrtBlock *block_ptr, gf *const *fec, int16_t *idx_p)
......
......@@ -3,9 +3,9 @@
#include <stdbool.h>
#include "codingParams.h"
#include "types/packet.h"
#include "../util/list.h"
#include "vdmcode/block_code.h"
#include "packet.h"
#include "../vdmcode/block_code.h"
#include "../../util/list.h"
typedef struct prrtBlock {
PrrtCodingConfiguration* codingParams;
......
#include <stdlib.h>
#include "../defines.h"
#include "../util/common.h"
#include "../util/dbg.h"
#include "channelStateInformation.h"
#include "clock.h"
#include "receiver.h"
#include "../../defines.h"
#include "../../util/common.h"
#include "../../util/dbg.h"
#include "../clock.h"
#include "../receiver.h"
PrrtChannelStateInformation * PrrtChannelStateInformation_create()
{
......
......@@ -2,7 +2,7 @@
#define PRRT_CHANNELSTATEINFORMATION_H
#include <stdbool.h>
#include "types/packet.h"
#include "packet.h"
typedef struct prrtChannelStateInformation {
pthread_mutex_t lock;
......
#include <pthread.h>
#include "../util/common.h"
#include "../util/dbg.h"
#include "../defines.h"
#include "../../util/common.h"
#include "../../util/dbg.h"
#include "../../defines.h"
PrrtCodingConfiguration *PrrtCodingConfiguration_create(uint8_t k, uint8_t n, uint8_t c, uint8_t *n_cycle)
......
......@@ -3,7 +3,7 @@
#include <stdint.h>
#include "../../util/pipe.h"
#include "../codingParams.h"
#include "codingParams.h"
#include <stdbool.h>
#include <netinet/in.h>
......
......@@ -40,7 +40,7 @@
#endif
#include <linux/types.h>
#include <stdint.h>
#include "../../proto/codingParams.h"
#include "../types/codingParams.h"
/*
* You should not need to change anything beyond this point.
......
......@@ -16,9 +16,6 @@ class TimeoutException(Exception):
def __init__(self):
self.message = "The call timed out."
cdef extern from "proto/applicationConstraints.c":
pass
cdef extern from "proto/stores/dataPacketStore.c":
pass
......@@ -43,6 +40,9 @@ cdef extern from "proto/stores/repairBlockStore.c":
cdef extern from "proto/stores/packetDeliveryStore.c":
pass
cdef extern from "proto/types/applicationConstraints.c":
pass
cdef extern from "proto/types/packetTimeout.c":
pass
......@@ -55,22 +55,22 @@ cdef extern from "proto/processes/dataReceiver.c":
cdef extern from "proto/processes/dataTransmitter.c":
pass
cdef extern from "proto/block.c":
cdef extern from "proto/clock.c":
pass
cdef extern from "proto/clock.c":
cdef extern from "proto/vdmcode/block_code.c":
pass
cdef extern from "proto/channelStateInformation.c":
cdef extern from "proto/receiver.c":
pass
cdef extern from "proto/vdmcode/block_code.c":
cdef extern from "proto/types/block.c":
pass
cdef extern from "proto/codingParams.c":
cdef extern from "proto/types/channelStateInformation.c":
pass
cdef extern from "proto/receiver.c":
cdef extern from "proto/types/codingParams.c":
pass
cdef extern from "proto/types/packet.c":
......
......@@ -2,9 +2,9 @@
extern "C" {
#include "prrt/proto/block.h"
#include "prrt/proto/types/block.h"
#include "prrt/proto/types/codingParams.h"
#include "prrt/proto/types/packet.h"
#include "prrt/proto/codingParams.h"
#include "prrt/util/dbg.h"
#include "prrt/proto/vdmcode/block_code.h"
#include "prrt/util/common.h"
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment