Skip to content

GitLab

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

Refactoring.

parent edf0c135
Hide whitespace changes
Inline Side-by-side
Showing with 185 additions and 195 deletions
src/prrt/block.c
View file @ ba9d6c5b
......@@ -43,7 +43,7 @@ void clear_list(gf *const *src, uint8_t k)
}
}
void PrrtBlock_free(PrrtBlock *mblock)
void PrrtBlock_destroy(PrrtBlock *mblock)
{
while(List_count(mblock->dataPackets) > 0) {
PrrtPacket *pkt = List_shift(mblock->dataPackets);
......@@ -60,7 +60,7 @@ void PrrtBlock_free(PrrtBlock *mblock)
free(mblock);
}
int PrrtBlock_alloc(PrrtBlock *mblock, const PrrtCodingParams *cpar, uint16_t base_seqno)
int PrrtBlock_create(PrrtBlock *mblock, const PrrtCodingParams *cpar, uint16_t base_seqno)
{
mblock->codingParams = *cpar;
mblock->dataPackets = List_create();
......@@ -186,9 +186,7 @@ void PrrtBlock_decode(PrrtBlock *block_ptr)
PrrtCoder_decode(coder, fec, idx_p, length);
for(j = 0; j < k; j++) {
PrrtPacket *packet = calloc(1, sizeof(PrrtPacket));
check_mem(packet);
PrrtPacket_create_data_packet(packet, 0, fec[j], length, (uint16_t) (block_ptr->baseSequenceNumber + j));
PrrtPacket *packet = PrrtPacket_create_data_packet(0, fec[j], length, (uint16_t) (block_ptr->baseSequenceNumber + j));
PrrtBlock_insert_data_packet(block_ptr, packet);
}
......
src/prrt/block.h
View file @ ba9d6c5b
......@@ -18,12 +18,12 @@ typedef struct {
/**
* Allocate space for a block.
*/
int PrrtBlock_alloc(PrrtBlock *mblock, const PrrtCodingParams *cpar, uint16_t base_seqno);
int PrrtBlock_create(PrrtBlock *mblock, const PrrtCodingParams *cpar, uint16_t base_seqno);
/**
* Frees the PrrtBlock data structure.
*/
void PrrtBlock_free(PrrtBlock *mblock);
void PrrtBlock_destroy(PrrtBlock *mblock);
int PrrtBlock_insert_data_packet(PrrtBlock *prrtBlock, const PrrtPacket *prrtPacket);
int PrrtBlock_insert_redundancy_packet(PrrtBlock *prrtBlock, const PrrtPacket *prrtPacket);
......
src/prrt/packet.c
View file @ ba9d6c5b
......@@ -15,11 +15,11 @@ void *encode_redundancy_header(void *buf_ptr, const void *payload);
void *encode_feedback_header(void *buf_ptr, const void *payload);
void *decode_data_header(void *buf_ptr, const void *payload_ptr);
void *decode_data_header(void *dstBuffer, const void *srcBuffer);
void *decode_redundancy_header(void *buf_ptr, const void *payload_ptr);
void *decode_redundancy_header(void *dstBuffer, const void *srcBuffer);
void *decode_feedback_header(void *buf_ptr, const void *payload_ptr);
void *decode_feedback_header(void *dstBuffer, const void *srcBuffer);
uint8_t PrrtPacket_type(PrrtPacket *packet_ptr)
{
......@@ -106,6 +106,23 @@ PrrtPacket *PrrtPacket_copy(PrrtPacket *original)
PERROR("Not enough memory for packet copies.");
}
PrrtPacket *create_header(uint8_t priority, uint16_t seqno, uint32_t size, uint8_t type, uint8_t index)
{
PrrtPacket *packet = calloc(1, sizeof(PrrtPacket));
check_mem(packet);
packet->type_priority = type << 4;
packet->type_priority |= priority & 0x0F;
packet->index = index;
packet->seqno = seqno;
packet->payload_len = size;
return packet;
error:
PERROR("Could not create packet.%s", "");
}
int PrrtPacket_encode(void *buf_ptr, uint16_t buf_size, PrrtPacket *packet_ptr)
{
void *payload = packet_ptr->payload;
......@@ -243,33 +260,33 @@ void *encode_general_header(void *buf_ptr, const PrrtPacket *packet)
return buf_ptr;
}
int PrrtPacket_decode(void *buf_ptr, uint16_t buf_size, PrrtPacket *packet_ptr)
int PrrtPacket_decode(void *srcBuffer, uint16_t srcBufferSize, PrrtPacket *targetPacket)
{
uint32_t payload_len = (uint32_t) (buf_size - PRRT_PACKET_GENERAL_HEADER_SIZE);
packet_ptr->type_priority = *(uint8_t *) buf_ptr;
buf_ptr += 1;
uint8_t *index_ptr = (uint8_t *) buf_ptr;
packet_ptr->index = *index_ptr;
buf_ptr += 1;
uint16_t *seqno_prt = (uint16_t *) buf_ptr;
packet_ptr->seqno = ntohs(*seqno_prt);
buf_ptr += 2;
uint32_t payload_len = (uint32_t) (srcBufferSize - PRRT_PACKET_GENERAL_HEADER_SIZE);
targetPacket->type_priority = *(uint8_t *) srcBuffer;
srcBuffer += 1;
uint8_t *index_ptr = (uint8_t *) srcBuffer;
targetPacket->index = *index_ptr;
srcBuffer += 1;
uint16_t *seqno_prt = (uint16_t *) srcBuffer;
targetPacket->seqno = ntohs(*seqno_prt);
srcBuffer += 2;
void *payload_buffer = calloc(1, payload_len);
check_mem(payload_buffer);
packet_ptr->payload = payload_buffer;
packet_ptr->payload_len = payload_len;
if(PrrtPacket_type(packet_ptr) == PACKET_TYPE_DATA) {
buf_ptr = decode_data_header(buf_ptr, payload_buffer);
PrrtPacket_copy_buffer_to_payload(packet_ptr, buf_ptr, PRRT_PACKET_DATA_HEADER_SIZE);
} else if(PrrtPacket_type(packet_ptr)) {
buf_ptr = decode_redundancy_header(buf_ptr, payload_buffer);
PrrtPacket_copy_buffer_to_payload(packet_ptr, buf_ptr, PRRT_PACKET_REDUNDANCY_HEADER_SIZE);
} else if(PrrtPacket_type(packet_ptr) == PACKET_TYPE_FEEDBACK) {
buf_ptr = decode_feedback_header(buf_ptr, payload_buffer);
PrrtPacket_copy_buffer_to_payload(packet_ptr, buf_ptr, PRRT_PACKET_FEEDBACK_HEADER_SIZE);
targetPacket->payload = payload_buffer;
targetPacket->payload_len = payload_len;
if(PrrtPacket_type(targetPacket) == PACKET_TYPE_DATA) {
srcBuffer = decode_data_header(srcBuffer, payload_buffer);
PrrtPacket_copy_buffer_to_payload(targetPacket, srcBuffer, PRRT_PACKET_DATA_HEADER_SIZE);
} else if(PrrtPacket_type(targetPacket)) {
srcBuffer = decode_redundancy_header(srcBuffer, payload_buffer);
PrrtPacket_copy_buffer_to_payload(targetPacket, srcBuffer, PRRT_PACKET_REDUNDANCY_HEADER_SIZE);
} else if(PrrtPacket_type(targetPacket) == PACKET_TYPE_FEEDBACK) {
srcBuffer = decode_feedback_header(srcBuffer, payload_buffer);
PrrtPacket_copy_buffer_to_payload(targetPacket, srcBuffer, PRRT_PACKET_FEEDBACK_HEADER_SIZE);
} else {
printf("NOT IMPLEMENTED\n");
}
......@@ -279,203 +296,181 @@ int PrrtPacket_decode(void *buf_ptr, uint16_t buf_size, PrrtPacket *packet_ptr)
return EXIT_FAILURE;
}
void *decode_redundancy_header(void *buf_ptr, const void *payload_ptr)
void *decode_redundancy_header(void *dstBuffer, const void *srcBuffer)
{
PrrtPacketRedundancyPayload *redundancyPayload = (PrrtPacketRedundancyPayload *) payload_ptr;
PrrtPacketRedundancyPayload *redundancyPayload = (PrrtPacketRedundancyPayload *) srcBuffer;
uint16_t *baseSeqNo = (uint16_t *) buf_ptr;
uint16_t *baseSeqNo = (uint16_t *) dstBuffer;
redundancyPayload->base_seqno = ntohs(*baseSeqNo);
buf_ptr += 2;
dstBuffer += 2;
uint8_t *n = (uint8_t *) buf_ptr;
uint8_t *n = (uint8_t *) dstBuffer;
redundancyPayload->n = *n;
buf_ptr += 1;
dstBuffer += 1;
uint8_t *k = (uint8_t *) buf_ptr;
uint8_t *k = (uint8_t *) dstBuffer;
redundancyPayload->k = *k;
buf_ptr += 1;
dstBuffer += 1;
return buf_ptr;
return dstBuffer;
}
void *decode_feedback_header(void *buf_ptr, const void *payload_ptr)
void *decode_feedback_header(void *dstBuffer, const void *srcBuffer)
{
PrrtPacketFeedbackPayload *feedback_payload = (PrrtPacketFeedbackPayload *) payload_ptr;
PrrtPacketFeedbackPayload *feedback_payload = (PrrtPacketFeedbackPayload *) srcBuffer;
uint32_t *receiver_addr = (uint32_t *) buf_ptr;
feedback_payload->receiver_addr = ntohl(*receiver_addr);
buf_ptr += 4;
uint32_t *receiverAddr = (uint32_t *) dstBuffer;
feedback_payload->receiver_addr = ntohl(*receiverAddr);
dstBuffer += 4;
uint32_t *group_round_trip_time = (uint32_t *) buf_ptr;
feedback_payload->group_round_trip_time = ntohl(*group_round_trip_time);
buf_ptr += 4;
uint32_t *groupRoundTripTime = (uint32_t *) dstBuffer;
feedback_payload->group_round_trip_time = ntohl(*groupRoundTripTime);
dstBuffer += 4;
uint32_t *forward_trip_time = (uint32_t *) buf_ptr;
feedback_payload->forward_trip_time = ntohl(*forward_trip_time);
buf_ptr += 4;
uint32_t *forwardTripTime = (uint32_t *) dstBuffer;
feedback_payload->forward_trip_time = ntohl(*forwardTripTime);
dstBuffer += 4;
uint32_t *packet_loss_rate = (uint32_t *) buf_ptr;
feedback_payload->packet_loss_rate = ntohl(*packet_loss_rate);
buf_ptr += 4;
uint32_t *packetLossRate = (uint32_t *) dstBuffer;
feedback_payload->packet_loss_rate = ntohl(*packetLossRate);
dstBuffer += 4;
uint16_t *gap = (uint16_t *) buf_ptr;
uint16_t *gap = (uint16_t *) dstBuffer;
feedback_payload->gap = ntohs(*gap);
buf_ptr += 2;
dstBuffer += 2;
uint16_t *ngap = (uint16_t *) buf_ptr;
uint16_t *ngap = (uint16_t *) dstBuffer;
feedback_payload->ngap = ntohs(*ngap);
buf_ptr += 2;
dstBuffer += 2;
uint16_t *burst = (uint16_t *) buf_ptr;
uint16_t *burst = (uint16_t *) dstBuffer;
feedback_payload->burst = ntohs(*burst);
buf_ptr += 2;
dstBuffer += 2;
uint16_t *nburst = (uint16_t *) buf_ptr;
uint16_t *nburst = (uint16_t *) dstBuffer;
feedback_payload->nburst = ntohs(*nburst);
buf_ptr += 2;
dstBuffer += 2;
uint32_t *bandwidth_estimate = (uint32_t *) buf_ptr;
feedback_payload->bandwidth_estimate = ntohl(*bandwidth_estimate);
buf_ptr += 4;
uint32_t *bandwidthEstimate = (uint32_t *) dstBuffer;
feedback_payload->bandwidth_estimate = ntohl(*bandwidthEstimate);
dstBuffer += 4;
uint32_t *buffer_feedback = (uint32_t *) buf_ptr;
feedback_payload->buffer_feedback = ntohl(*buffer_feedback);
buf_ptr += 4;
uint32_t *bufferFeedback = (uint32_t *) dstBuffer;
feedback_payload->buffer_feedback = ntohl(*bufferFeedback);
dstBuffer += 4;
return buf_ptr;
return dstBuffer;
}
void *decode_data_header(void *buf_ptr, const void *payload_ptr)
void *decode_data_header(void *dstBuffer, const void *srcBuffer)
{
PrrtPacketDataPayload *data_payload = (PrrtPacketDataPayload *) payload_ptr;
PrrtPacketDataPayload *data_payload = (PrrtPacketDataPayload *) srcBuffer;
uint32_t *timestamp = (uint32_t *) buf_ptr;
uint32_t *timestamp = (uint32_t *) dstBuffer;
data_payload->timestamp = ntohl(*timestamp);
buf_ptr += 4;
dstBuffer += 4;
uint16_t *group_round_trip_time = (uint16_t *) buf_ptr;
uint16_t *group_round_trip_time = (uint16_t *) dstBuffer;
data_payload->group_round_trip_time = ntohs(*group_round_trip_time);
buf_ptr += 2;
dstBuffer += 2;
uint16_t *packet_timeout = (uint16_t *) buf_ptr;
uint16_t *packet_timeout = (uint16_t *) dstBuffer;
data_payload->packet_timeout = ntohs(*packet_timeout);
buf_ptr += 2;
dstBuffer += 2;
uint16_t *decoding_timeout = (uint16_t *) buf_ptr;
uint16_t *decoding_timeout = (uint16_t *) dstBuffer;
data_payload->decoding_timeout = ntohs(*decoding_timeout);
buf_ptr += 2;
dstBuffer += 2;
uint16_t *feedback_timeout = (uint16_t *) buf_ptr;
uint16_t *feedback_timeout = (uint16_t *) dstBuffer;
data_payload->feedback_timeout = ntohs(*feedback_timeout);
buf_ptr += 2;
return buf_ptr;
dstBuffer += 2;
return dstBuffer;
}
int PrrtPacket_destroy(PrrtPacket *packet_ptr)
int PrrtPacket_destroy(PrrtPacket *packet)
{
if(packet_ptr->payload != NULL) {
free(packet_ptr->payload);
if(packet->payload != NULL) {
free(packet->payload);
}
free(packet_ptr);
free(packet);
return 0;
}
int PrrtPacket_create_data_packet(PrrtPacket *packet_ptr, uint8_t priority, const void *data_ptr,
uint32_t data_len, uint16_t seqno)
PrrtPacket *PrrtPacket_create_data_packet(uint8_t priority, const void *payloadPointer, uint32_t payloadLength,
uint16_t sequenceNumber)
{
uint32_t payload_length = (uint32_t) (data_len + PRRT_PACKET_DATA_HEADER_SIZE);
packet_ptr->type_priority = PACKET_TYPE_DATA << 4;
packet_ptr->type_priority |= priority & 0x0F;
packet_ptr->index = 0;
packet_ptr->seqno = seqno;
PrrtPacket *packet = create_header(priority, sequenceNumber,
(uint32_t) (payloadLength + PRRT_PACKET_DATA_HEADER_SIZE), PACKET_TYPE_DATA, 0);
void *content_buf = calloc(1, payload_length);
check_mem(content_buf);
PrrtPacketDataPayload *source_payload = content_buf;
PrrtPacketDataPayload *dataPayload = calloc(1, packet->payload_len);
check_mem(dataPayload);
packet->payload = dataPayload;
struct timeval tv;
gettimeofday(&tv, NULL);
unsigned long time_in_micros = (unsigned long) (1000000 * tv.tv_sec + tv.tv_usec);
source_payload->timestamp = (uint32_t) time_in_micros;
source_payload->group_round_trip_time = 95; // TODO: payload->rtt = CURRENT ESTIMATE
source_payload->packet_timeout = 110; // TODO: payload->packet_timeout = NOW + maximum delay
source_payload->decoding_timeout = 150; // TODO: payload->decoding_timeout
source_payload->feedback_timeout = 170; // TODO: payload->feedback_timer
dataPayload->timestamp = (uint32_t) time_in_micros;
dataPayload->group_round_trip_time = 95; // TODO: payload->rtt = CURRENT ESTIMATE
dataPayload->packet_timeout = 110; // TODO: payload->packet_timeout = NOW + maximum delay
dataPayload->decoding_timeout = 150; // TODO: payload->decoding_timeout
dataPayload->feedback_timeout = 170; // TODO: payload->feedback_timer
PrrtPacket_copy_buffer_to_payload(packet, payloadPointer, PRRT_PACKET_DATA_HEADER_SIZE)
packet_ptr->payload = content_buf;
packet_ptr->payload_len = payload_length;
PrrtPacket_copy_buffer_to_payload(packet_ptr, data_ptr, PRRT_PACKET_DATA_HEADER_SIZE)
return packet;
error:
return -1;
PERROR("Could not create packet.");
}
PrrtPacket *PrrtPacket_create_redundancy_packet(uint8_t priority, void *data_ptr, uint32_t data_len,
uint16_t seqno, uint8_t index, uint16_t base_seqno,
PrrtPacket *PrrtPacket_create_redundancy_packet(uint8_t priority, void *payloadPointer, uint32_t payloadLength,
uint16_t sequenceNumber, uint8_t index, uint16_t baseSequenceNumber,
PrrtCodingParams codingParams)
{
uint32_t payload_length = (uint32_t) (data_len + PRRT_PACKET_REDUNDANCY_HEADER_SIZE);
PrrtPacket *packet_ptr = calloc(1, sizeof(PrrtPacket));
check_mem(packet_ptr);
packet_ptr->type_priority = PACKET_TYPE_REDUNDANCY << 4;
packet_ptr->type_priority |= priority & 0x0F;
packet_ptr->index = index;
packet_ptr->seqno = seqno;
PrrtPacket *packet = create_header(priority, sequenceNumber,
(uint32_t) (payloadLength + PRRT_PACKET_REDUNDANCY_HEADER_SIZE),
PACKET_TYPE_REDUNDANCY, index);
void *content_buf = calloc(1, PRRT_PACKET_REDUNDANCY_HEADER_SIZE + data_len);
check_mem(content_buf);
PrrtPacketRedundancyPayload *redundancyPayload = calloc(1, packet->payload_len);
check_mem(redundancyPayload);
packet->payload = redundancyPayload;
PrrtPacketRedundancyPayload *redundancy_payload = content_buf;
redundancyPayload->base_seqno = baseSequenceNumber;
redundancyPayload->k = codingParams.k;
redundancyPayload->n = codingParams.n;
PrrtPacket_copy_buffer_to_payload(packet, payloadPointer, PRRT_PACKET_REDUNDANCY_HEADER_SIZE);
redundancy_payload->base_seqno = base_seqno;
redundancy_payload->k = codingParams.k;
redundancy_payload->n = codingParams.n;
packet_ptr->payload = content_buf;
packet_ptr->payload_len = payload_length;
PrrtPacket_copy_buffer_to_payload(packet_ptr, data_ptr, PRRT_PACKET_REDUNDANCY_HEADER_SIZE);
return packet_ptr;
return packet;
error:
return NULL;
}
PrrtPacket *PrrtPacket_create_feedback_packet(uint8_t priority, uint8_t index, uint16_t seqno, uint32_t round_trip_time,
uint32_t forward_trip_time, uint32_t packet_loss_rate, uint16_t gap,
uint16_t ngap,
uint16_t burst, uint16_t nburst, uint32_t bandwidth,
uint32_t buffer_feedback)
PrrtPacket *PrrtPacket_create_feedback_packet(uint8_t priority, uint8_t index, uint16_t sequenceNumber,
uint32_t roundTripTime, uint32_t forwardTripTime, uint32_t packetLossRate,
uint16_t gap, uint16_t ngap, uint16_t burst, uint16_t nburst,
uint32_t bandwidth, uint32_t bufferFeedback)
{
PrrtPacket *packet_ptr = calloc(1, sizeof(PrrtPacket));
check_mem(packet_ptr);
packet_ptr->type_priority = PACKET_TYPE_FEEDBACK << 4;
packet_ptr->type_priority |= priority & 0x0F;
packet_ptr->index = index;
packet_ptr->seqno = seqno;
PrrtPacketFeedbackPayload *payload = calloc(1, PRRT_PACKET_FEEDBACK_HEADER_SIZE);
PrrtPacket *packet = create_header(priority, sequenceNumber, PRRT_PACKET_FEEDBACK_HEADER_SIZE, PACKET_TYPE_FEEDBACK,
index);
PrrtPacketFeedbackPayload *payload = calloc(1, packet->payload_len);
check_mem(payload);
packet->payload = payload;
payload->group_round_trip_time = round_trip_time;
payload->forward_trip_time = forward_trip_time;
payload->packet_loss_rate = packet_loss_rate;
payload->group_round_trip_time = roundTripTime;
payload->forward_trip_time = forwardTripTime;
payload->packet_loss_rate = packetLossRate;
payload->gap = gap;
payload->ngap = ngap;
payload->burst = burst;
payload->nburst = nburst;
payload->bandwidth_estimate = bandwidth;
payload->buffer_feedback = buffer_feedback;
payload->buffer_feedback = bufferFeedback;
packet_ptr->payload = (void *) payload;
packet_ptr->payload_len = (uint32_t) (PRRT_PACKET_FEEDBACK_HEADER_SIZE);
return packet_ptr;
return packet;
error:
return NULL;
......
src/prrt/packet.h
View file @ ba9d6c5b
......@@ -63,20 +63,19 @@ uint16_t PrrtPacket_size(PrrtPacket *packet_ptr);
int PrrtPacket_print(PrrtPacket *packet_ptr);
PrrtPacket *PrrtPacket_copy(PrrtPacket *original);
int PrrtPacket_create_data_packet(PrrtPacket *packet_ptr, uint8_t priority, const void *data_ptr,
uint32_t data_len, uint16_t seqno);
PrrtPacket * PrrtPacket_create_data_packet(uint8_t priority, const void *payloadPointer, uint32_t payloadLength, uint16_t sequenceNumber);
PrrtPacket *PrrtPacket_create_feedback_packet(uint8_t priority, uint8_t index, uint16_t seqno, uint32_t round_trip_time,
uint32_t forward_trip_time, uint32_t packet_loss_rate, uint16_t gap,
PrrtPacket *PrrtPacket_create_feedback_packet(uint8_t priority, uint8_t index, uint16_t sequenceNumber, uint32_t roundTripTime,
uint32_t forwardTripTime, uint32_t packetLossRate, uint16_t gap,
uint16_t ngap, uint16_t burst, uint16_t nburst, uint32_t bandwidth,
uint32_t buffer_feedback);
uint32_t bufferFeedback);
PrrtPacket *PrrtPacket_create_redundancy_packet(uint8_t priority, void *data_ptr, uint32_t data_len,
uint16_t seqno, uint8_t index, uint16_t base_seqno, PrrtCodingParams codingParams);
PrrtPacket *PrrtPacket_create_redundancy_packet(uint8_t priority, void *payloadPointer, uint32_t payloadLength,
uint16_t sequenceNumber, uint8_t index, uint16_t baseSequenceNumber, PrrtCodingParams codingParams);
int PrrtPacket_decode(void *buf_ptr, uint16_t buf_size, PrrtPacket *packet_ptr);
int PrrtPacket_decode(void *srcBuffer, uint16_t srcBufferSize, PrrtPacket *targetPacket);
int PrrtPacket_encode(void *buf_ptr, uint16_t buf_size, PrrtPacket *packet_ptr);
int PrrtPacket_destroy(PrrtPacket *packet_ptr);
int PrrtPacket_destroy(PrrtPacket *packet);
#define PrrtPacket_copy_payload_to_buffer(dst, packet, header_size) memcpy(dst, packet->payload + header_size, packet->payload_len - header_size);
#define PrrtPacket_copy_buffer_to_payload(packet, src, header_size) memcpy(packet->payload + header_size, src, packet->payload_len - header_size);
......
src/prrt/processes/data_receiver.c
View file @ ba9d6c5b
......@@ -43,7 +43,7 @@ void decode_block(PrrtSocket *sock_ptr, uint16_t base_seqno)
}
}
PrrtBlock_free(block);
PrrtBlock_destroy(block);
sock_ptr->blockStore = BPTree_delete(sock_ptr->blockStore, base_seqno);
}
......@@ -142,7 +142,7 @@ void *receive_data_loop(void *ptr)
block = calloc(1, sizeof(PrrtBlock));
check_mem(block);
PrrtBlock_alloc(block, cpar, redundancyPayload->base_seqno);
PrrtBlock_create(block, cpar, redundancyPayload->base_seqno);
free(cpar);
sock_ptr->blockStore = BPTree_insert(sock_ptr->blockStore, redundancyPayload->base_seqno,
block);
......
src/prrt/processes/data_transmitter.c
View file @ ba9d6c5b
......@@ -57,7 +57,9 @@ void * send_data_loop(void *ptr) {
pthread_mutex_lock(&sock_ptr->closingMutex);
if(sock_ptr->closing) {
free(cpar);
PrrtBlock_free(block);
if(block != NULL) {
PrrtBlock_destroy(block);
}
block = NULL;
pthread_mutex_unlock(&sock_ptr->closingMutex);
......@@ -72,7 +74,7 @@ void * send_data_loop(void *ptr) {
block = calloc(1, sizeof(PrrtBlock));
check_mem(block);
PrrtBlock_alloc(block, cpar, sock_ptr->sequenceNumberRedundancy);
PrrtBlock_create(block, cpar, sock_ptr->sequenceNumberRedundancy);
}
PrrtPacket *packet = List_shift(sock_ptr->outQueue);
......@@ -95,7 +97,7 @@ void * send_data_loop(void *ptr) {
send_packet(sock_ptr, data_pkt);
}
PrrtBlock_free(block);
PrrtBlock_destroy(block);
block = NULL;
}
......
src/prrt/socket.c
View file @ ba9d6c5b
......@@ -95,18 +95,13 @@ int PrrtSocket_connect(PrrtSocket *sock_ptr, const char *host, const uint16_t po
int PrrtSocket_send(PrrtSocket *sock_ptr, const void *data, const size_t data_len) {
pthread_mutex_lock(&sock_ptr->outQueueFilledMutex);
PrrtPacket *packet = calloc(1, sizeof(PrrtPacket));
check_mem(packet);
PrrtPacket_create_data_packet(packet, 5, data, (uint32_t) data_len, sock_ptr->sequenceNumberSource++);
PrrtPacket *packet = PrrtPacket_create_data_packet(5, data, (uint32_t) data_len, sock_ptr->sequenceNumberSource++);
List_push(sock_ptr->outQueue, packet);
pthread_cond_signal(&sock_ptr->outQueueFilledCv);
pthread_mutex_unlock(&sock_ptr->outQueueFilledMutex);
return 0;
error:
return -1;
}
int32_t PrrtSocket_recv(PrrtSocket *sock_ptr, void *buf_ptr) {
......@@ -199,7 +194,7 @@ int PrrtSocket_close(PrrtSocket *sock_ptr) {
BPTree_get_range(sock_ptr->blockStore, blockList, 0, SEQNO_SPACE - 1);
while (List_count(blockList) > 0) {
PrrtBlock *block = List_shift(blockList);
PrrtBlock_free(block);
PrrtBlock_destroy(block);
}
List_destroy(blockList);
......
tests/PrrtBlock_tests.cpp
View file @ ba9d6c5b
#include <gtest/gtest.h>
extern "C" {
#include "src/prrt/block.h"
#include "src/prrt/packet.h"
#include "src/prrt/coding_params.h"
#include "src/prrt/block.h"
#include "src/prrt/packet.h"
#include "src/prrt/coding_params.h"
}
class PrrtBlockTest : public ::testing::Test {
protected:
virtual void SetUp() {
virtual void SetUp()
{
encBlock = (PrrtBlock *) calloc(1, sizeof(PrrtBlock));
decBlock = (PrrtBlock *) calloc(1, sizeof(PrrtBlock));
PrrtCodingParams *cpar = (PrrtCodingParams *) calloc(1, sizeof(PrrtCodingParams));
PrrtCodingParams_init(cpar);
PrrtBlock_alloc(encBlock, cpar, 0);
PrrtBlock_alloc(decBlock, cpar, 0);
PrrtBlock_create(encBlock, cpar, 0);
PrrtBlock_create(decBlock, cpar, 0);
}
PrrtBlock *encBlock;
PrrtBlock *decBlock;
};
TEST_F(PrrtBlockTest, Encode) {
TEST_F(PrrtBlockTest, Encode)
{
uint16_t base = 0;
PrrtPacket* packets = (PrrtPacket *) calloc(2, sizeof(PrrtPacket));
PrrtPacket refPackets[2];
memset(refPackets, 0, 2*sizeof(PrrtPacket));
PrrtPacket *packets[2];
PrrtPacket *refPackets[2];
PrrtPacket* redPackets = (PrrtPacket *) calloc(2, sizeof(PrrtPacket));
PrrtPacket *redPackets = (PrrtPacket *) calloc(2, sizeof(PrrtPacket));
for (int i = 1; i < 3; i++) {
for(int i = 1; i < 3; i++) {
char data[3];
sprintf(data, "%d", i);
PrrtPacket *ptr = &packets[i - 1];
PrrtPacket_create_data_packet(ptr, 0, data, (uint32_t) strlen(data), i);
PrrtPacket_create_data_packet(refPackets + (i-1), 0, data, (uint32_t) strlen(data), i);
PrrtBlock_insert_data_packet(encBlock, ptr);