Skip to content
GitLab
Commit 8ec641b9 authored by Pablo Gil Pereira's avatar Pablo Gil Pereira
Browse files

Refactor block and fix gather redundancy seg fault

parent 461b903c
Pipeline #4890 failed with stages
in 1 minute and 21 seconds
Hide whitespace changes
Inline Side-by-side
CMakeLists.txt
View file @ 8ec641b9
......@@ -10,7 +10,7 @@ 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 "-fstack-protector -fstack-protector-all -Wall -std=gnu++11 -D_GNU_SOURCE -fsanitize=undefined -fsanitize=addrss" )
set(CMAKE_CXX_FLAGS_DEBUG "-O0 -Wall -ggdb -fsanitize=undefined -fsanitize=address -g3" )
set(CMAKE_CXX_FLAGS_RELEASE "-Os -Wall" )
......@@ -36,4 +36,4 @@ if(PRRT_TESTS)
endif()
add_custom_target(perftest COMMAND python3 tests/eval.py)
add_custom_target(memtest COMMAND bash ./memtest.sh DEPENDS sender receiver)
\ No newline at end of file
add_custom_target(memtest COMMAND bash ./memtest.sh DEPENDS sender receiver)
prrt/proto/processes/dataReceiver.c
View file @ 8ec641b9
......@@ -41,14 +41,14 @@ static void decode_block(PrrtSocket *sock_ptr, PrrtBlock *block) {
if (data_relevant) {
check(PrrtBlock_decode(block), "Decoding failed");
while (List_count(block->dataPackets) > 0) {
PrrtPacket *pkt = List_shift(block->dataPackets);
if (PrrtDeliveredPacketTable_test_set_is_number_relevant(sock_ptr->deliveredPacketTable,
pkt->sequenceNumber)) {
PrrtPacketDeliveryStore_insert(sock_ptr->packetDeliveryStore, pkt);
} else {
PrrtPacket_destroy(pkt);
}
while (block->nextDataPacket < block->dataPacketCount) {
PrrtPacket *pkt = PrrtBlock_get_next_data(block);
if (PrrtDeliveredPacketTable_test_set_is_number_relevant(sock_ptr->deliveredPacketTable,
pkt->sequenceNumber)) {
PrrtPacketDeliveryStore_insert(sock_ptr->packetDeliveryStore, pkt);
} else {
PrrtPacket_destroy(pkt);
}
}
}
PrrtRepairBlockStore_delete(sock_ptr->repairBlockStore, block->baseSequenceNumber);
......
prrt/proto/processes/dataTransmitter.c
View file @ 8ec641b9
......@@ -208,7 +208,7 @@ void retransmission_round_handler(void *arg) {
uint32_t redundancyPackets = block->codingParams->n_cycle[block->inRound];
for (j = 0; j < redundancyPackets; j++) {
PrrtPacket *red_pkt = PrrtBlock_get_first_red_data(block);
PrrtPacket *red_pkt = PrrtBlock_get_next_red_data(block);
bool sendResult = send_packet(socket, red_pkt);
if(!sendResult) {
debug(DEBUG_DATATRANSMITTER, "Sending redundancy data failed.");
......
prrt/proto/types/block.c
View file @ 8ec641b9
......@@ -6,18 +6,27 @@
static void gather_redundancy_packets(const PrrtBlock *block_ptr, gf *const *fec, int16_t *idx_p)
{
uint32_t i;
uint32_t m = 0;
uint32_t redundancyBlocks = List_count(block_ptr->redundancyPackets);
int k = PrrtCoder_get_k(block_ptr->coder);
int n = PrrtCoder_get_n(block_ptr->coder);
for(i = 0; i < redundancyBlocks; i++) {
PrrtPacket *packet = List_shift(block_ptr->redundancyPackets);
//for(i = 0; i < redundancyBlocks; i++) {
// PrrtPacket *packet = List_shift(block_ptr->redundancyPackets);
for(int i = k; i < n; ++i) {
PrrtPacket *packet = block_ptr->packets[i];
if(packet == NULL) {
continue;
}
while(idx_p[m] != -1) {
while(idx_p[m] != -1 && m < n) {
m++;
}
if(m == n) {
break;
}
// Take the data payload out of the redundancy payload and
// decode/deserialize it into the fec matrix
PrrtPacket_decode_payload(
......@@ -27,14 +36,18 @@ static void gather_redundancy_packets(const PrrtBlock *block_ptr, gf *const *fec
packet->payloadLength
);
idx_p[m] = packet->index;
PrrtPacket_destroy(packet);
}
}
static void gather_data_packets(PrrtBlock *block_ptr, gf *const *fec, int16_t *idx_p)
{
LIST_FOREACH(block_ptr->dataPackets, first, next, current) {
PrrtPacket *packet = current->value;
int k = PrrtCoder_get_k(block_ptr->coder);
for(int i = 0; i < k; ++i){
PrrtPacket *packet = block_ptr->packets[i];
if(packet == NULL) {
continue;
}
PrrtPacket_copy_payload_to_buffer(fec[packet->index], packet, 0);
idx_p[packet->index] = packet->index;
}
......@@ -54,19 +67,21 @@ void PrrtBlock_print(PrrtBlock *block) {
"+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+\n");
printf("| %29u | %13u | %13u |\n", block->baseSequenceNumber, block->codingParams->n, block->codingParams->k);
printf("+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+\n");
printf("| %13d | %13d |\n", List_count(block->dataPackets), List_count(block->redundancyPackets));
printf("| %13d | %13d |\n", block->dataPacketCount, block->redPacketCount);
printf("+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+\n");
}
bool PrrtBlock_destroy(PrrtBlock *block)
{
check(pthread_mutex_lock(&block->lock) == EXIT_SUCCESS, "Lock failed.");
while(List_count(block->dataPackets) > 0) {
PrrtPacket *pkt = List_shift(block->dataPackets);
PrrtPacket_destroy(pkt);
}
while(List_count(block->redundancyPackets) > 0) {
PrrtPacket *pkt = List_shift(block->redundancyPackets);
int n = PrrtCoder_get_n(block->coder);
for(int i = 0; i < n; ++i) {
PrrtPacket *pkt = block->packets[i];
if(pkt == NULL) {
continue;
}
PrrtPacket_destroy(pkt);
}
......@@ -78,8 +93,7 @@ bool PrrtBlock_destroy(PrrtBlock *block)
PrrtCodingConfiguration_destroy(block->codingParams);
}
List_destroy(block->dataPackets);
List_destroy(block->redundancyPackets);
free(block->packets);
check(pthread_mutex_unlock(&block->lock) == EXIT_SUCCESS, "Unlock failed.");
check(pthread_mutex_destroy(&block->lock) == EXIT_SUCCESS, "Mutex init failed.");
......@@ -99,13 +113,16 @@ PrrtBlock * PrrtBlock_create(PrrtCodingConfiguration *cpar, PrrtCoder* coder, pr
block_ptr->coder = coder;
block_ptr->codingParams = cpar;
block_ptr->dataPackets = List_create();
block_ptr->redundancyPackets = List_create();
block_ptr->baseSequenceNumber = baseSequenceNumber;
block_ptr->largestPayloadLength = 0;
block_ptr->nextRedundancyPacket = 0;
block_ptr->nextDataPacket = 0;
block_ptr->senderBlock = false;
block_ptr->packets = calloc(cpar->n, sizeof(PrrtPacket*));
block_ptr->dataPacketCount = 0;
block_ptr->redPacketCount = 0;
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.");
......@@ -121,15 +138,16 @@ PrrtBlock * PrrtBlock_create(PrrtCodingConfiguration *cpar, PrrtCoder* coder, pr
bool insert_data_packet(PrrtBlock * block, const PrrtPacket *packet) {
bool found = false;
debug(DEBUG_BLOCK, "S: %d, Insert [D]: %d", block->senderBlock, packet->sequenceNumber);
LIST_FOREACH(block->dataPackets, first, next, cur) {
PrrtPacket *pkt = cur->value;
if(pkt->sequenceNumber == packet->sequenceNumber) {
found = true;
}
int index = packet->index;
if(block->packets[index] != NULL) {
found = true;
}
if(found == false) {
List_push(block->dataPackets, packet);
block->packets[index] = packet;
block->dataPacketCount += 1;
}
return found;
}
......@@ -154,14 +172,15 @@ bool PrrtBlock_insert_redundancy_packet(PrrtBlock *block, const PrrtPacket *pack
bool found = false;
check(pthread_mutex_lock(&block->lock) == EXIT_SUCCESS, "Lock failed.");
debug(DEBUG_BLOCK, "S: %d, Insert [R]: %d", block->senderBlock, packet->sequenceNumber);
LIST_FOREACH(block->redundancyPackets, first, next, cur) {
PrrtPacket *pkt = cur->value;
if(pkt->sequenceNumber == packet->sequenceNumber) {
found = true;
}
int index = packet->index;
if(block->packets[index] != NULL) {
found = true;
}
if(found == false) {
List_push(block->redundancyPackets, packet);
block->packets[index] = packet;
block->redPacketCount += 1;
block->largestPayloadLength = (prrtPacketLength_t) MAX(block->largestPayloadLength,
packet->payloadLength - PRRT_PACKET_REDUNDANCY_HEADER_SIZE);
}
......@@ -176,7 +195,7 @@ bool PrrtBlock_insert_redundancy_packet(PrrtBlock *block, const PrrtPacket *pack
bool PrrtBlock_encode_ready(PrrtBlock *block_ptr)
{
check(pthread_mutex_lock(&block_ptr->lock) == EXIT_SUCCESS, "Lock failed.");
bool res = List_count(block_ptr->dataPackets) == block_ptr->codingParams->k;
bool res = block_ptr->dataPacketCount == block_ptr->codingParams->k;
check(pthread_mutex_unlock(&block_ptr->lock) == EXIT_SUCCESS, "Unlock failed.");
return res;
......@@ -188,7 +207,7 @@ bool PrrtBlock_encode_ready(PrrtBlock *block_ptr)
bool PrrtBlock_decode_ready(PrrtBlock *block_ptr)
{
check(pthread_mutex_lock(&block_ptr->lock) == EXIT_SUCCESS, "Lock failed.");
bool res = (List_count(block_ptr->dataPackets) + List_count(block_ptr->redundancyPackets) == block_ptr->codingParams->k);
bool res = (block_ptr->dataPacketCount + block_ptr->redPacketCount) == block_ptr->codingParams->k;
check(pthread_mutex_unlock(&block_ptr->lock) == EXIT_SUCCESS, "Unlock failed.");
return res;
......@@ -197,18 +216,6 @@ bool PrrtBlock_decode_ready(PrrtBlock *block_ptr)
return false;
}
PrrtPacket *PrrtBlock_get_first_data(PrrtBlock *block_ptr)
{
check(pthread_mutex_lock(&block_ptr->lock) == EXIT_SUCCESS, "Lock failed.");
PrrtPacket *res = List_shift(block_ptr->dataPackets);
check(pthread_mutex_unlock(&block_ptr->lock) == EXIT_SUCCESS, "Unlock failed.");
return res;
error:
PERROR("Get first data failed.")
return NULL;
}
void PrrtBlock_encode(PrrtBlock *block_ptr, prrtSequenceNumber_t *seqno)
{
check(pthread_mutex_lock(&block_ptr->lock) == EXIT_SUCCESS, "Lock failed.");
......@@ -221,10 +228,10 @@ void PrrtBlock_encode(PrrtBlock *block_ptr, prrtSequenceNumber_t *seqno)
gf **src = calloc(k, sizeof(gf *));
check_mem(src);
LIST_FOREACH(block_ptr->dataPackets, first, next, cur) {
for(int i = 0; i < block_ptr->codingParams->k; ++i) {
src[j] = calloc(length, sizeof(gf));
check_mem(src[j]);
PrrtPacket *pkt = cur->value;
PrrtPacket *pkt = block_ptr->packets[i];
pkt->index = (uint8_t) ((pkt->sequenceNumber - baseSequenceNumber) % SEQNO_SPACE);
PrrtPacket_encode_payload(src[j], pkt);
j++;
......@@ -246,6 +253,9 @@ void PrrtBlock_encode(PrrtBlock *block_ptr, prrtSequenceNumber_t *seqno)
PrrtBlock_insert_redundancy_packet(block_ptr, red_packet_ptr);
}
block_ptr->nextDataPacket = 0;
block_ptr->nextRedundancyPacket = k;
block_ptr->isCoded = true;
clear_list(fec, r);
......@@ -328,28 +338,43 @@ bool PrrtBlock_decode(PrrtBlock *block_ptr)
return false;
}
PrrtPacket *PrrtBlock_get_first_red_data(PrrtBlock *block_ptr)
PrrtPacket *PrrtBlock_get_next_red_data(PrrtBlock *block_ptr)
{
check(pthread_mutex_lock(&block_ptr->lock) == EXIT_SUCCESS, "Lock failed.");
// TODO: This copy should be avoided.
PrrtPacket *res = PrrtPacket_copy(PrrtBlock_get_red_data(block_ptr, block_ptr->nextRedundancyPacket));
PrrtPacket *res = PrrtBlock_get_packet(block_ptr, block_ptr->nextRedundancyPacket);
block_ptr->nextRedundancyPacket++;
check(pthread_mutex_unlock(&block_ptr->lock) == EXIT_SUCCESS, "Unlock failed.");
return res;
error:
PERROR("Get first data failed.")
PERROR("Get first redundancy data failed.")
return NULL;
}
PrrtPacket * PrrtBlock_get_next_data(PrrtBlock *block_ptr)
{
check(pthread_mutex_lock(&block_ptr->lock) == EXIT_SUCCESS, "Lock failed.");
PrrtPacket *res = PrrtBlock_get_packet(block_ptr, block_ptr->nextDataPacket);
block_ptr->nextDataPacket++;
check(pthread_mutex_unlock(&block_ptr->lock) == EXIT_SUCCESS, "Unlock failed");
return res;
error:
PERROR("Get first data failed.");
return NULL;
}
PrrtPacket *PrrtBlock_get_red_data(PrrtBlock *block_ptr, uint8_t index)
PrrtPacket *PrrtBlock_get_packet(PrrtBlock *block_ptr, int index)
{
check(pthread_mutex_lock(&block_ptr->lock) == EXIT_SUCCESS, "Lock failed.");
PrrtPacket *res = (PrrtPacket *) List_get(block_ptr->redundancyPackets, index);
PrrtPacket *res = block_ptr->packets[index];
block_ptr->packets[index] = NULL;
check(pthread_mutex_unlock(&block_ptr->lock) == EXIT_SUCCESS, "Unlock failed.");
return res;
error:
PERROR("Get first data failed.")
PERROR("Get packet failed.")
return NULL;
}
prrt/proto/types/block.h
View file @ 8ec641b9
......@@ -11,14 +11,18 @@ typedef struct prrtBlock {
PrrtCodingConfiguration* codingParams;
prrtPacketLength_t largestPayloadLength;
prrtSequenceNumber_t baseSequenceNumber;
List*dataPackets;
List*redundancyPackets;
PrrtPacket **packets;
int dataPacketCount;
int redPacketCount;
bool isCoded;
bool isDecoded;
pthread_mutex_t lock;
PrrtCoder *coder;
uint16_t inRound;
uint8_t nextRedundancyPacket;
uint8_t nextDataPacket;
bool senderBlock;
} PrrtBlock;
......@@ -39,13 +43,13 @@ bool PrrtBlock_insert_redundancy_packet(PrrtBlock *prrtBlock, const PrrtPacket *
bool PrrtBlock_encode_ready(PrrtBlock *block_ptr);
bool PrrtBlock_decode_ready(PrrtBlock *block_ptr);
PrrtPacket *PrrtBlock_get_first_data(PrrtBlock *block_ptr);
void PrrtBlock_print(PrrtBlock *block);
void PrrtBlock_encode(PrrtBlock *block_ptr, prrtSequenceNumber_t *seqno);
bool PrrtBlock_decode(PrrtBlock *block_ptr);
PrrtPacket *PrrtBlock_get_first_red_data(PrrtBlock *block_ptr);
PrrtPacket *PrrtBlock_get_red_data(PrrtBlock *block_ptr, uint8_t index);
PrrtPacket *PrrtBlock_get_next_data(PrrtBlock *block_ptr);
PrrtPacket *PrrtBlock_get_next_red_data(PrrtBlock *block_ptr);
PrrtPacket *PrrtBlock_get_packet(PrrtBlock *block_ptr, int index);
#endif //PRRT_BLOCK_H
tests/PrrtBlock_tests.cpp
View file @ 8ec641b9
......@@ -128,15 +128,15 @@ TEST_F(PrrtBlockTest, EncodeDecode)
PrrtPacket *redPackets[4];
for(uint32_t i = 0; i < 4; i++) {
char data[3];
char data[10];
sprintf(data, "%d", i);
if(i == 3) {
// Last packet has a different size.
sprintf(data, "D:%d", i);
}
packets[i] = PrrtPacket_create_data_packet(0, data, (prrtPacketLength_t) strlen(data), i + 1, 0);
packets[i] = PrrtPacket_create_data_packet(0, data, (prrtPacketLength_t) strlen(data) + 1, i + 1, 0);
packets[i]->index = (uint8_t) i;
refPackets[i] = PrrtPacket_create_data_packet(0, data, (prrtPacketLength_t) strlen(data), i + 1, 0);
refPackets[i] = PrrtPacket_create_data_packet(0, data, (prrtPacketLength_t) strlen(data) + 1, i + 1, 0);
ASSERT_TRUE(PrrtBlock_insert_data_packet(encBlock, packets[i]));
}
......@@ -145,14 +145,14 @@ TEST_F(PrrtBlockTest, EncodeDecode)
uint32_t j = 0;
PrrtBlock_encode(encBlock, &base);
uint32_t pkt_count = List_count(encBlock->dataPackets);
uint32_t pkt_count = encBlock->dataPacketCount;
for(j = 0; j < pkt_count; j++) {
PrrtPacket *data_pkt = PrrtBlock_get_first_data(encBlock);
PrrtPacket *data_pkt = PrrtBlock_get_next_data(encBlock);
ASSERT_EQ(refPackets[j]->sequenceNumber, data_pkt->sequenceNumber);
}
uint32_t red_count = List_count(encBlock->redundancyPackets);
uint32_t red_count = encBlock->redPacketCount;
for(j = 0; j < red_count; j++) {
PrrtPacket *red_pkt = PrrtBlock_get_first_red_data(encBlock);
PrrtPacket *red_pkt = PrrtBlock_get_next_red_data(encBlock);
redPackets[j] = red_pkt;
}
......@@ -169,7 +169,7 @@ TEST_F(PrrtBlockTest, EncodeDecode)
PrrtBlock_decode(decBlock);
for(int k = 0; k < 4; ++k) {
PrrtPacket *ptr = PrrtBlock_get_first_data(decBlock);
PrrtPacket *ptr = PrrtBlock_get_next_data(decBlock);
prrtIndex_t idx = static_cast<prrtIndex_t>(ptr->sequenceNumber - 1);
const char *s1 = (const char *) ((char *) refPackets[idx]->payload + PRRT_PACKET_DATA_HEADER_SIZE);
......@@ -228,7 +228,7 @@ TEST_F(PrrtBlockTest, NoCodingConfigured)
PrrtPacket *data_pkt = PrrtBlock_get_first_data(encBlock);
ASSERT_EQ(refPackets[j]->sequenceNumber, data_pkt->sequenceNumber);
}
uint32_t red_count = List_count(encBlock->redundancyPackets);
uint32_t red_count = encBlock->redPacketCount;
ASSERT_EQ(red_count, 0);
for (int i = 0; i < 1; ++i) {
......
Supports Markdown
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