Commit ddf7d8de authored by Andreas Schmidt's avatar Andreas Schmidt
Browse files

Add packet loss statistics.

parent 349b320d
Pipeline #1718 failed with stages
in 1 minute and 16 seconds
...@@ -143,10 +143,10 @@ cdef extern from "proto/socket.h": ...@@ -143,10 +143,10 @@ cdef extern from "proto/socket.h":
uint32_t PrrtSocket_get_rtprop(PrrtSocket *socket) uint32_t PrrtSocket_get_rtprop(PrrtSocket *socket)
bint PrrtSocket_uses_thread_pinning(PrrtSocket *socket) bint PrrtSocket_uses_thread_pinning(PrrtSocket *socket)
uint32_t PrrtSocket_get_rtt(PrrtSocket *socket)
float PrrtSocket_get_plr(PrrtSocket *socket)
bint PrrtSocket_enable_thread_pinning(PrrtSocket *socket) bint PrrtSocket_enable_thread_pinning(PrrtSocket *socket)
cdef extern from "proto/stores/packetDeliveryStore.h": cdef extern from "proto/stores/packetDeliveryStore.h":
ctypedef struct PrrtPacketDeliveryStore: ctypedef struct PrrtPacketDeliveryStore:
pass pass
......
...@@ -10,7 +10,7 @@ add_library(PRRT ../defines.h ...@@ -10,7 +10,7 @@ add_library(PRRT ../defines.h
applicationConstraints.c applicationConstraints.h applicationConstraints.c applicationConstraints.h
vdmcode/block_code.c vdmcode/block_code.h vdmcode/block_code.c vdmcode/block_code.h
stores/deliveredPacketTable.c stores/deliveredPacketTable.h stores/deliveredPacketTable.c stores/deliveredPacketTable.h
stores/lossGatherer.c stores/lossGatherer.h types/lossStatistics.c types/lossStatistics.h
processes/dataReceiver.c processes/dataReceiver.h processes/dataReceiver.c processes/dataReceiver.h
processes/feedbackReceiver.c processes/feedbackReceiver.h processes/feedbackReceiver.c processes/feedbackReceiver.h
processes/dataTransmitter.c processes/dataTransmitter.h processes/dataTransmitter.c processes/dataTransmitter.h
...@@ -18,6 +18,7 @@ add_library(PRRT ../defines.h ...@@ -18,6 +18,7 @@ add_library(PRRT ../defines.h
stores/packetTimeoutTable.c stores/packetTimeoutTable.h stores/packetTimeoutTable.c stores/packetTimeoutTable.h
stores/dataPacketStore.c stores/dataPacketStore.h stores/dataPacketStore.c stores/dataPacketStore.h
types/packetTimeout.c types/packetTimeout.h types/packetTimeout.c types/packetTimeout.h
stores/packetDeliveryStore.c stores/packetDeliveryStore.h) stores/packetDeliveryStore.c stores/packetDeliveryStore.h
stores/receptionTable.c stores/receptionTable.h)
target_link_libraries(PRRT rt) target_link_libraries(PRRT rt)
...@@ -33,6 +33,13 @@ void PrrtChannelStateInformation_update_rtprop(PrrtChannelStateInformation *csi, ...@@ -33,6 +33,13 @@ void PrrtChannelStateInformation_update_rtprop(PrrtChannelStateInformation *csi,
pthread_mutex_unlock(&csi->lock); pthread_mutex_unlock(&csi->lock);
} }
void PrrtChannelStateInformation_update_plr(PrrtChannelStateInformation *csi, prrtSequenceNumber_t erasures,
prrtSequenceNumber_t packets) {
pthread_mutex_lock(&csi->lock);
csi->plr = ((float) erasures) / packets;
pthread_mutex_unlock(&csi->lock);
}
prrtTimedelta_t PrrtChannelStateInformation_get_rtprop(PrrtChannelStateInformation *csi) prrtTimedelta_t PrrtChannelStateInformation_get_rtprop(PrrtChannelStateInformation *csi)
{ {
...@@ -50,3 +57,7 @@ bool PrrtChannelStateInformation_destroy(PrrtChannelStateInformation *csi) ...@@ -50,3 +57,7 @@ bool PrrtChannelStateInformation_destroy(PrrtChannelStateInformation *csi)
error: error:
return false; return false;
} }
prrtPacketLossRate_t PrrtChannelStateInformation_get_plr(PrrtChannelStateInformation *csi) {
return csi->plr;
}
...@@ -10,11 +10,15 @@ typedef struct prrtChannelStateInformation { ...@@ -10,11 +10,15 @@ typedef struct prrtChannelStateInformation {
prrtTimestamp_t rtprop_stamp; prrtTimestamp_t rtprop_stamp;
prrtTimedelta_t rtprop_filter_length_us; prrtTimedelta_t rtprop_filter_length_us;
bool rtprop_expired; bool rtprop_expired;
prrtPacketLossRate_t plr;
} PrrtChannelStateInformation; } PrrtChannelStateInformation;
PrrtChannelStateInformation* PrrtChannelStateInformation_create(void); PrrtChannelStateInformation* PrrtChannelStateInformation_create(void);
void PrrtChannelStateInformation_update_rtprop(PrrtChannelStateInformation *csi, prrtTimedelta_t rtprop); void PrrtChannelStateInformation_update_rtprop(PrrtChannelStateInformation *csi, prrtTimedelta_t rtprop);
prrtTimedelta_t PrrtChannelStateInformation_get_rtprop(PrrtChannelStateInformation *csi); prrtTimedelta_t PrrtChannelStateInformation_get_rtprop(PrrtChannelStateInformation *csi);
prrtPacketLossRate_t PrrtChannelStateInformation_get_plr(PrrtChannelStateInformation* csi);
void PrrtChannelStateInformation_update_plr(PrrtChannelStateInformation *csi, prrtSequenceNumber_t erasures,
prrtSequenceNumber_t packets);
bool PrrtChannelStateInformation_destroy(PrrtChannelStateInformation* csi); bool PrrtChannelStateInformation_destroy(PrrtChannelStateInformation* csi);
void PrrtChannelStateInformation_print(PrrtChannelStateInformation *csi); void PrrtChannelStateInformation_print(PrrtChannelStateInformation *csi);
......
...@@ -409,7 +409,6 @@ void *decode_data_header(void *dstBuffer, const void *srcBuffer) { ...@@ -409,7 +409,6 @@ void *decode_data_header(void *dstBuffer, const void *srcBuffer) {
return dstBuffer; return dstBuffer;
} }
int PrrtPacket_destroy(PrrtPacket *packet) { int PrrtPacket_destroy(PrrtPacket *packet) {
if (packet->payload != NULL) { if (packet->payload != NULL) {
free(packet->payload); free(packet->payload);
...@@ -494,7 +493,8 @@ PrrtPacket *PrrtPacket_create_feedback_packet(uint8_t priority, uint8_t index, p ...@@ -494,7 +493,8 @@ PrrtPacket *PrrtPacket_create_feedback_packet(uint8_t priority, uint8_t index, p
prrtTimedelta_t groupRTT, prrtSequenceNumber_t gapLength, prrtTimedelta_t groupRTT, prrtSequenceNumber_t gapLength,
prrtSequenceNumber_t gapCount, prrtSequenceNumber_t burstLength, prrtSequenceNumber_t gapCount, prrtSequenceNumber_t burstLength,
prrtSequenceNumber_t burstCount, uint32_t bandwidth, prrtSequenceNumber_t burstCount, uint32_t bandwidth,
uint32_t receiverAddr, prrtTimestamp_t forwardTripTime) { uint32_t receiverAddr, prrtTimestamp_t forwardTripTime,
prrtSequenceNumber_t erasureCount, prrtSequenceNumber_t packetCount) {
PrrtPacket *packet = create_header(priority, sequenceNumber, PRRT_PACKET_FEEDBACK_HEADER_SIZE, PACKET_TYPE_FEEDBACK, PrrtPacket *packet = create_header(priority, sequenceNumber, PRRT_PACKET_FEEDBACK_HEADER_SIZE, PACKET_TYPE_FEEDBACK,
index); index);
...@@ -505,8 +505,8 @@ PrrtPacket *PrrtPacket_create_feedback_packet(uint8_t priority, uint8_t index, p ...@@ -505,8 +505,8 @@ PrrtPacket *PrrtPacket_create_feedback_packet(uint8_t priority, uint8_t index, p
payload->receiverAddress = receiverAddr; payload->receiverAddress = receiverAddr;
payload->groupRTT_us = groupRTT; payload->groupRTT_us = groupRTT;
payload->forwardTripTimestamp_us = forwardTripTime; payload->forwardTripTimestamp_us = forwardTripTime;
payload->erasureCount = 0; payload->erasureCount = erasureCount;
payload->packetCount = 0; payload->packetCount = packetCount;
payload->gapLength = gapLength; payload->gapLength = gapLength;
payload->gapCount = gapCount; payload->gapCount = gapCount;
payload->burstLength = burstLength; payload->burstLength = burstLength;
......
...@@ -22,6 +22,7 @@ typedef uint32_t prrtTimestamp_t; ...@@ -22,6 +22,7 @@ typedef uint32_t prrtTimestamp_t;
typedef uint32_t prrtTimedelta_t; typedef uint32_t prrtTimedelta_t;
typedef int32_t prrtTimeDifference_t; typedef int32_t prrtTimeDifference_t;
typedef uint32_t prrtPacketLength_t; typedef uint32_t prrtPacketLength_t;
typedef float prrtPacketLossRate_t;
typedef struct prrtIncompleteBlock { typedef struct prrtIncompleteBlock {
prrtSequenceNumber_t sequenceNumberBase; prrtSequenceNumber_t sequenceNumberBase;
...@@ -29,7 +30,7 @@ typedef struct prrtIncompleteBlock { ...@@ -29,7 +30,7 @@ typedef struct prrtIncompleteBlock {
} PrrtIncompleteBlock; } PrrtIncompleteBlock;
typedef struct prrtPacket { typedef struct prrtPacket {
ListNode asListNode; ListNode asListNode;
uint8_t type_priority; uint8_t type_priority;
prrtIndex_t index; prrtIndex_t index;
prrtSequenceNumber_t sequenceNumber; prrtSequenceNumber_t sequenceNumber;
...@@ -90,12 +91,14 @@ PrrtPacket *PrrtPacket_create_feedback_packet(uint8_t priority, uint8_t index, p ...@@ -90,12 +91,14 @@ PrrtPacket *PrrtPacket_create_feedback_packet(uint8_t priority, uint8_t index, p
prrtTimedelta_t groupRTT, prrtSequenceNumber_t gapLength, prrtTimedelta_t groupRTT, prrtSequenceNumber_t gapLength,
prrtSequenceNumber_t gapCount, prrtSequenceNumber_t burstLength, prrtSequenceNumber_t gapCount, prrtSequenceNumber_t burstLength,
prrtSequenceNumber_t burstCount, uint32_t bandwidth, prrtSequenceNumber_t burstCount, uint32_t bandwidth,
uint32_t receiverAddr, prrtTimestamp_t forwardTripTime); uint32_t receiverAddr, prrtTimestamp_t forwardTripTime,
prrtSequenceNumber_t erasureCount, prrtSequenceNumber_t packetCount);
PrrtPacket *PrrtPacket_create_redundancy_packet(uint8_t priority, void *payloadPointer, PrrtPacket *PrrtPacket_create_redundancy_packet(uint8_t priority, void *payloadPointer,
prrtPacketLength_t payloadLength, prrtPacketLength_t payloadLength,
prrtSequenceNumber_t sequenceNumber, uint8_t index, prrtSequenceNumber_t sequenceNumber, uint8_t index,
prrtSequenceNumber_t baseSequenceNumber, PrrtCodingParams* codingParams); prrtSequenceNumber_t baseSequenceNumber,
PrrtCodingParams *codingParams);
bool PrrtPacket_decode(void *srcBuffer, uint16_t srcBufferSize, PrrtPacket *targetPacket); bool PrrtPacket_decode(void *srcBuffer, uint16_t srcBufferSize, PrrtPacket *targetPacket);
......
#include <netdb.h> #include <netdb.h>
#include <stdio.h> #include <stdio.h>
#include <arpa/inet.h> #include <arpa/inet.h>
#include <sys/time.h>
#include "../../defines.h" #include "../../defines.h"
#include "../../util/dbg.h" #include "../../util/dbg.h"
#include "../../util/common.h" #include "../../util/common.h"
#include "../types/lossStatistics.h"
#include "../block.h" #include "../block.h"
#include "../clock.h" #include "../clock.h"
#include "../socket.h" #include "../socket.h"
...@@ -66,9 +66,11 @@ static bool send_feedback(const PrrtSocket *sock_ptr, struct sockaddr_in remote) ...@@ -66,9 +66,11 @@ static bool send_feedback(const PrrtSocket *sock_ptr, struct sockaddr_in remote)
prrtTimestamp_t forwardTripTime = (uint32_t) (((int64_t) PrrtClock_get_current_time_us()) + prrtTimestamp_t forwardTripTime = (uint32_t) (((int64_t) PrrtClock_get_current_time_us()) +
(sock_ptr->lastSentTimestamp - sock_ptr->lastReceivedTimestamp)); (sock_ptr->lastSentTimestamp - sock_ptr->lastReceivedTimestamp));
PrrtLossStatistics stats = sock_ptr->lossStatistics;
PrrtPacket *feedback_pkt_ptr = PrrtPacket_create_feedback_packet(0, 19, 4715, 0, 4, 6, 8, 9, 5, PrrtPacket *feedback_pkt_ptr = PrrtPacket_create_feedback_packet(0, 19, 4715, 0, 4, 6, 8, 9, 5,
sock_ptr->address->sin_addr.s_addr, sock_ptr->address->sin_addr.s_addr,
forwardTripTime); forwardTripTime, stats.erasureCount, stats.packetCount);
prrtPacketLength_t length = PrrtPacket_size(feedback_pkt_ptr); prrtPacketLength_t length = PrrtPacket_size(feedback_pkt_ptr);
void *buf = calloc(1, length); void *buf = calloc(1, length);
check_mem(buf); check_mem(buf);
...@@ -104,6 +106,7 @@ static void handle_data_packet(PrrtSocket *sock_ptr, PrrtPacket *packet, struct ...@@ -104,6 +106,7 @@ static void handle_data_packet(PrrtSocket *sock_ptr, PrrtPacket *packet, struct
check(PrrtPacketTimeoutTable_insert(sock_ptr->packetTimeoutTable, packetTimeout), "Could not insert data packet."); check(PrrtPacketTimeoutTable_insert(sock_ptr->packetTimeoutTable, packetTimeout), "Could not insert data packet.");
prrtSequenceNumber_t seqno = packet->sequenceNumber; prrtSequenceNumber_t seqno = packet->sequenceNumber;
PrrtReceptionTable_mark_received(sock_ptr->dataReceptionTable, seqno);
XlapTimeStampCycle(sock_ptr, ts_data_packet, seqno, SendFeedbackStart); XlapTimeStampCycle(sock_ptr, ts_data_packet, seqno, SendFeedbackStart);
check(send_feedback(sock_ptr, remote), "Sending feedback failed."); check(send_feedback(sock_ptr, remote), "Sending feedback failed.");
...@@ -153,6 +156,8 @@ static void handle_data_packet(PrrtSocket *sock_ptr, PrrtPacket *packet, struct ...@@ -153,6 +156,8 @@ static void handle_data_packet(PrrtSocket *sock_ptr, PrrtPacket *packet, struct
static void handle_redundancy_packet(PrrtSocket *socket, PrrtPacket *packet) { static void handle_redundancy_packet(PrrtSocket *socket, PrrtPacket *packet) {
PrrtPacketRedundancyPayload *redundancyPayload = packet->payload; PrrtPacketRedundancyPayload *redundancyPayload = packet->payload;
PrrtReceptionTable_mark_received(socket->redundancyReceptionTable, packet->sequenceNumber);
if (!PrrtDeliveredPacketTable_test_is_block_relevant(socket->deliveredPacketTable, if (!PrrtDeliveredPacketTable_test_is_block_relevant(socket->deliveredPacketTable,
redundancyPayload->baseSequenceNumber, redundancyPayload->baseSequenceNumber,
redundancyPayload->n)) { redundancyPayload->n)) {
......
...@@ -37,10 +37,12 @@ static void handle_feedback(PrrtSocket *prrtSocket, const size_t length) ...@@ -37,10 +37,12 @@ static void handle_feedback(PrrtSocket *prrtSocket, const size_t length)
check_mem(prrtPacket); check_mem(prrtPacket);
PrrtPacket_decode(bufin, (uint16_t) n, prrtPacket); PrrtPacket_decode(bufin, (uint16_t) n, prrtPacket);
prrtTimestamp_t forwardTripTimestamp = ((PrrtPacketFeedbackPayload *) prrtPacket->payload)->forwardTripTimestamp_us; PrrtPacketFeedbackPayload *feedbackPayload = (PrrtPacketFeedbackPayload *) prrtPacket->payload;
prrtTimestamp_t forwardTripTimestamp = feedbackPayload->forwardTripTimestamp_us;
PrrtChannelStateInformation_update_rtprop(prrtSocket->receiver->csi, PrrtChannelStateInformation_update_rtprop(prrtSocket->receiver->csi,
(prrtTimedelta_t) (receiveTime - forwardTripTimestamp)); (prrtTimedelta_t) (receiveTime - forwardTripTimestamp));
PrrtChannelStateInformation_update_plr(prrtSocket->receiver->csi, feedbackPayload->erasureCount, feedbackPayload->packetCount);
error: error:
if(prrtPacket != NULL) { PrrtPacket_destroy(prrtPacket); } if(prrtPacket != NULL) { PrrtPacket_destroy(prrtPacket); }
......
...@@ -104,6 +104,9 @@ PrrtSocket *PrrtSocket_create(const bool is_sender, prrtTimedelta_t target_delay ...@@ -104,6 +104,9 @@ PrrtSocket *PrrtSocket_create(const bool is_sender, prrtTimedelta_t target_delay
s->repairBlockStore = PrrtRepairBlockStore_create(); s->repairBlockStore = PrrtRepairBlockStore_create();
s->packetDeliveryStore = PrrtPacketDeliveryStore_create(); s->packetDeliveryStore = PrrtPacketDeliveryStore_create();
s->dataReceptionTable = PrrtReceptionTable_create();
s->redundancyReceptionTable = PrrtReceptionTable_create();
} }
return s; return s;
...@@ -423,6 +426,16 @@ int PrrtSocket_close(PrrtSocket *s) { ...@@ -423,6 +426,16 @@ int PrrtSocket_close(PrrtSocket *s) {
s->sendDataQueue = NULL; s->sendDataQueue = NULL;
} }
if (s->dataReceptionTable != NULL) {
PrrtReceptionTable_destroy(s->dataReceptionTable);
s->dataReceptionTable = NULL;
}
if (s->redundancyReceptionTable != NULL) {
PrrtReceptionTable_destroy(s->redundancyReceptionTable);
s->redundancyReceptionTable = NULL;
}
if (s->packetDeliveryStore != NULL) { if (s->packetDeliveryStore != NULL) {
PrrtPacketDeliveryStore_destroy(s->packetDeliveryStore); PrrtPacketDeliveryStore_destroy(s->packetDeliveryStore);
s->packetDeliveryStore = NULL; s->packetDeliveryStore = NULL;
...@@ -549,6 +562,8 @@ bool PrrtSocket_cleanup(PrrtSocket *s) { ...@@ -549,6 +562,8 @@ bool PrrtSocket_cleanup(PrrtSocket *s) {
PrrtRepairBlockStore_expire_block_range(s->repairBlockStore, (prrtSequenceNumber_t) (current_start - SEQNO_SPACE/2), PrrtRepairBlockStore_expire_block_range(s->repairBlockStore, (prrtSequenceNumber_t) (current_start - SEQNO_SPACE/2),
(prrtSequenceNumber_t) (current_start - 1)); (prrtSequenceNumber_t) (current_start - 1));
} }
s->lossStatistics = PrrtLossStatistics_add(PrrtReceptionTable_calculate_statistics(s->dataReceptionTable), PrrtReceptionTable_calculate_statistics(s->redundancyReceptionTable));
} }
return true; return true;
} }
...@@ -560,3 +575,7 @@ bool PrrtSocket_uses_thread_pinning(PrrtSocket *s) { ...@@ -560,3 +575,7 @@ bool PrrtSocket_uses_thread_pinning(PrrtSocket *s) {
uint32_t PrrtSocket_get_rtprop(PrrtSocket *s) { uint32_t PrrtSocket_get_rtprop(PrrtSocket *s) {
return PrrtChannelStateInformation_get_rtprop(s->receiver->csi); return PrrtChannelStateInformation_get_rtprop(s->receiver->csi);
} }
prrtPacketLossRate_t PrrtSocket_get_plr(PrrtSocket *socket) {
return PrrtChannelStateInformation_get_plr(socket->receiver->csi);
}
...@@ -13,8 +13,10 @@ ...@@ -13,8 +13,10 @@
#include "stores/dataPacketStore.h" #include "stores/dataPacketStore.h"
#include "stores/deliveredPacketTable.h" #include "stores/deliveredPacketTable.h"
#include "stores/packetTimeoutTable.h" #include "stores/packetTimeoutTable.h"
#include "stores/receptionTable.h"
#include "stores/repairBlockStore.h" #include "stores/repairBlockStore.h"
#include "stores/packetDeliveryStore.h" #include "stores/packetDeliveryStore.h"
#include "types/lossStatistics.h"
#include "clock.h" #include "clock.h"
#include "../xlap/xlap.h" #include "../xlap/xlap.h"
#include "receiver.h" #include "receiver.h"
...@@ -62,6 +64,11 @@ typedef struct prrtSocket { ...@@ -62,6 +64,11 @@ typedef struct prrtSocket {
prrtTimestamp_t lastSentTimestamp; prrtTimestamp_t lastSentTimestamp;
prrtTimestamp_t lastReceivedTimestamp; prrtTimestamp_t lastReceivedTimestamp;
PrrtLossStatistics lossStatistics;
PrrtReceptionTable* dataReceptionTable;
PrrtReceptionTable* redundancyReceptionTable;
PrrtApplicationConstraints *applicationConstraints; PrrtApplicationConstraints *applicationConstraints;
PrrtCodingParams *codingParameters; PrrtCodingParams *codingParameters;
...@@ -122,4 +129,6 @@ bool PrrtSocket_uses_thread_pinning(PrrtSocket *s); ...@@ -122,4 +129,6 @@ bool PrrtSocket_uses_thread_pinning(PrrtSocket *s);
uint32_t PrrtSocket_get_rtprop(PrrtSocket *s); uint32_t PrrtSocket_get_rtprop(PrrtSocket *s);
prrtPacketLossRate_t PrrtSocket_get_plr(PrrtSocket *socket);
#endif // PRRT_SOCKET_H #endif // PRRT_SOCKET_H
...@@ -5,6 +5,7 @@ ...@@ -5,6 +5,7 @@
#include <stdbool.h> #include <stdbool.h>
#include "../packet.h" #include "../packet.h"
#include "../../util/bitmap.h" #include "../../util/bitmap.h"
#include "../types/lossStatistics.h"
typedef struct { typedef struct {
prrtSequenceNumber_t start; prrtSequenceNumber_t start;
...@@ -22,4 +23,5 @@ bool PrrtDeliveredPacketTable_test_is_block_relevant(PrrtDeliveredPacketTable *t ...@@ -22,4 +23,5 @@ bool PrrtDeliveredPacketTable_test_is_block_relevant(PrrtDeliveredPacketTable *t
void PrrtDeliveredPacketTable_forward_start(PrrtDeliveredPacketTable *fpt_ptr, prrtSequenceNumber_t new_start); void PrrtDeliveredPacketTable_forward_start(PrrtDeliveredPacketTable *fpt_ptr, prrtSequenceNumber_t new_start);
#endif //PRRT_DELIVERED_PACKET_TABLE_H #endif //PRRT_DELIVERED_PACKET_TABLE_H
#include <stddef.h>
#include <malloc.h>
#include "../../defines.h"
#include "../../util/common.h"
#include "../../util/dbg.h"
#include "lossGatherer.h"
LossGatherer *LossGatherer_create()
{
LossGatherer *lg = calloc(1, sizeof(LossGatherer));
check_mem(lg);
lg->bitmap = Bitmap_create(false, SEQNO_SPACE);
return lg;
error:
PERROR("Could not create loss gatherer.%s", "");
return NULL;
}
bool LossGatherer_destroy(LossGatherer *lossGatherer)
{
Bitmap_destroy(lossGatherer->bitmap);
free(lossGatherer);
return true;
}
LossStatistics LossGatherer_calculate_statistics(LossGatherer *lossGatherer, uint32_t start, uint16_t length)
{
check(length != 0, "Cannot calculate stats over empty sequence.");
bool x0 = LossGatherer_check_loss(lossGatherer, start);
LossStatistics statistics = {
.errorCount = (uint16_t) x0,
.packetCount = length,
.gapCount = (uint16_t) x0,
.burstCount = (uint16_t) (1 - ((uint16_t) x0)),
};
statistics.gapLength = statistics.gapCount;
statistics.burstLength = statistics.burstCount;
bool isInGap = x0;
uint16_t i = 0;
for(i = 1; i < length; i++) {
bool currentPosition = LossGatherer_check_loss(lossGatherer, start + i);
statistics.errorCount += currentPosition;
if(isInGap && currentPosition) {
statistics.gapLength++;
} else if(isInGap && (currentPosition == false)) {
isInGap = false;
statistics.burstCount++;
statistics.burstLength++;
} else if((isInGap == false) && currentPosition) {
isInGap = true;
statistics.gapCount++;
statistics.gapLength++;
} else {
statistics.burstLength++;
}
}
return statistics;
error:
PERROR("Stats error.%s","");
LossStatistics ls;
return ls;
}
bool LossGatherer_check_loss(LossGatherer *lossGatherer, uint32_t sequenceNumber)
{
return Bitmap_get(lossGatherer->bitmap, sequenceNumber);
}
bool LossGatherer_mark_loss(LossGatherer *lossGatherer, uint32_t sequenceNumber)
{
return Bitmap_set(lossGatherer->bitmap, sequenceNumber, true);
}
#include "../../defines.h"
#include "../../util/common.h"
#include "../../util/dbg.h"
#include "receptionTable.h"
PrrtReceptionTable *PrrtReceptionTable_create(void) {
PrrtReceptionTable* t = calloc(1, sizeof(PrrtReceptionTable));
check_mem(t);
t->start = 0;
t->maxNumberInWindow = 0;
t->windowSize = 1000;
t->bitmap = Bitmap_create(false, SEQNO_SPACE);
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(&t->lock, &attr) == 0, "Mutex init failed.");
return t;
error:
PERROR("Out of memory.%s", "");
return NULL;
}
bool PrrtReceptionTable_destroy(PrrtReceptionTable *t) {
pthread_mutex_destroy(&t->lock);
Bitmap_destroy(t->bitmap);
free(t);
return true;
}
bool PrrtReceptionTable_mark_received(PrrtReceptionTable *t, prrtSequenceNumber_t markedNumber) {
check(pthread_mutex_lock(&t->lock) == EXIT_SUCCESS, "Lock failed.");
prrtSequenceNumber_t markedNumber_relative = (prrtSequenceNumber_t) (markedNumber - t->start);
prrtSequenceNumber_t maxNumberInWindow_relative = (prrtSequenceNumber_t) (t->maxNumberInWindow - t->start);
if (markedNumber_relative < t->windowSize) {
Bitmap_set_1(t->bitmap, markedNumber);
if(markedNumber_relative > maxNumberInWindow_relative) {
t->maxNumberInWindow = markedNumber;
}
}
if(markedNumber_relative > (t->windowSize/2)) {
prrtSequenceNumber_t new_start = (prrtSequenceNumber_t) (markedNumber - (t->windowSize / 2));
Bitmap_set_range_0(t->bitmap, t->start, (prrtSequenceNumber_t ) (new_start - t->start));
t->start = new_start;
}
check(pthread_mutex_unlock(&t->lock) == EXIT_SUCCESS, "Lock failed.");
return 0;
error:
PERROR("PrrtReceptionTable_mark_received failed%s.", "");
return false;
}
PrrtLossStatistics PrrtReceptionTable_calculate_statistics(PrrtReceptionTable *t) {
check(pthread_mutex_lock(&t->lock) == EXIT_SUCCESS, "Lock failed.");
prrtSequenceNumber_t length = (prrtSequenceNumber_t) ((t->maxNumberInWindow - t->start) + 1);
bool x0 = Bitmap_get(t->bitmap, t->start) == false;
PrrtLossStatistics statistics = {
.erasureCount = (uint16_t) x0,
.packetCount = length,
.gapCount = (uint16_t) x0,
.burstCount = (uint16_t) (1 - ((uint16_t) x0)),
};
statistics.gapLength = statistics.gapCount;
statistics.burstLength = statistics.burstCount;
bool isInGap = x0;
uint16_t i = 0;
for(i = 1; i < length; i++) {
bool currentPosition = Bitmap_get(t->bitmap, t->start + i) == false;
statistics.erasureCount += currentPosition;
if(isInGap && currentPosition) {
statistics.gapLength++;
} else if(isInGap && (currentPosition == false)) {
isInGap = false;
statistics.burstCount++;
statistics.burstLength++;
} else if((isInGap == false) && currentPosition) {
isInGap = true;
statistics.gapCount++;
statistics.gapLength++;
} else {
statistics.burstLength++;
}
}
check(pthread_mutex_unlock(&t->lock) == EXIT_SUCCESS, "Lock failed.");
return statistics;
error:
PERROR("PrrtReceptionTable_calculate_statistics failed%s.", "");
return statistics;
}
#ifndef PRRT_RECEPTIONTABLE_H
#define PRRT_RECEPTIONTABLE_H
#include <stdint.h>
#include <stdbool.h>
#include "../packet.h"
#include "../../util/bitmap.h"
#include "../types/lossStatistics.h"
typedef struct {
prrtSequenceNumber_t start;
Bitmap* bitmap;
prrtSequenceNumber_t maxNumberInWindow;
prrtSequenceNumber_t windowSize;
pthread_mutex_t lock;
} PrrtReceptionTable;
PrrtReceptionTable * PrrtReceptionTable_create(void);