data_receiver.c

#include <netdb.h>
#include <stdio.h>
#include <arpa/inet.h>
#include <src/defines.h>
#include <src/util/dbg.h>
#include <src/prrt/socket.h>
#include <src/prrt/block.h>
#include <src/prrt/packet.h>
#include "data_receiver.h"

void decode_block(const PrrtSocket *sock_ptr, uint16_t base_seqno);

void retrieve_data_blocks(PrrtSocket *sock_ptr, uint16_t base_seqno, uint8_t k, const PrrtBlock *block);

int send_feedback(const PrrtSocket *sock_ptr, const char *remote_host, const int remote_port) {
    struct sockaddr_in targetaddr;
    memset((char *) &targetaddr, 0, sizeof(targetaddr));
    targetaddr.sin_family = AF_INET;
    targetaddr.sin_port = htons((uint16_t) (remote_port + 1));

    struct hostent *hp;
    hp = gethostbyname(remote_host);
    memcpy((void *) &targetaddr.sin_addr, hp->h_addr_list[0], (size_t) hp->h_length);

    PrrtPacket *feedback_pkt_ptr = PrrtPacket_create_feedback_packet(0, 19, 4715, 7, 3, 50, 4, 6, 8, 9, 5, 1);
    uint8_t buf[MAX_PAYLOAD_LENGTH];
    uint32_t length = PrrtPacket_size(feedback_pkt_ptr);

    check(PrrtPacket_encode(buf, MAX_PAYLOAD_LENGTH, feedback_pkt_ptr) == EXIT_SUCCESS,
          "Buffer for encoding feedback is too small");
    check(sendto(sock_ptr->feedbackSocketFd, buf, length, 0, (struct sockaddr *) &targetaddr, sizeof(targetaddr)) ==
          length, "Sending feedback failed.");

    return EXIT_SUCCESS;

    error:
    return EXIT_FAILURE;
}

void *receive_data_loop(void *ptr) {
    ssize_t n;
    uint16_t remote_port;
    char *remote_host;
    struct sockaddr_in remote;
    socklen_t addrlen = sizeof(remote);
    unsigned char buffer[MAX_PAYLOAD_LENGTH];
    PrrtSocket *sock_ptr = ptr;
    PrrtPacket *packet;
    PrrtPacketRedundancyPayload *redundancyPayload;
    PrrtBlock *block;

    while (1) {
        memset(buffer, 0, MAX_PAYLOAD_LENGTH);
        n = recvfrom(sock_ptr->dataSocketFd, buffer, MAX_PAYLOAD_LENGTH, 0, (struct sockaddr *) &remote, &addrlen);
        remote_port = ntohs(remote.sin_port);
        remote_host = inet_ntoa(remote.sin_addr);

        packet = calloc(1, sizeof(PrrtPacket));
        check_mem(packet);
        PrrtPacket_decode(buffer, (uint16_t) n, packet);

        check(send_feedback(sock_ptr, remote_host, remote_port) == EXIT_SUCCESS, "Sending feedback failed.");

        switch (PrrtPacket_type(packet)) {
            case PACKET_TYPE_DATA:
                // packet.timestamp + packet.timeout < now: break

                if (PrrtForwardPacketTable_test_set_is_number_relevant(sock_ptr->forwardPacketTable, packet->seqno) ==
                    FALSE) {
                    break;
                }

                // check incomplete_prrt_blocks for this seqno: insert if found
                // else: insert in data_packet_store

                sock_ptr->dataStore = BPTree_insert(sock_ptr->dataStore, packet->seqno, packet);

                // forward to application layer

                pthread_mutex_lock(&sock_ptr->inQueueFilledMutex);
                List_push(sock_ptr->inQueue, packet);
                pthread_cond_signal(&sock_ptr->inQueueFilledMutexCv);
                pthread_mutex_unlock(&sock_ptr->inQueueFilledMutex);

                decode_block(sock_ptr, packet->seqno - packet->index);
                break;
            case PACKET_TYPE_REDUNDANCY:
                redundancyPayload = packet->payload;

                if (!PrrtForwardPacketTable_test_is_block_relevant(sock_ptr->forwardPacketTable,
                                                                   redundancyPayload->base_seqno,
                                                                   redundancyPayload->n)) {
                    //printf("-------- IRRELEVANT -----------------------\n");
                    //PrrtPacket_print(packet);
                    //printf("-------- IRRELEVANT -----------------------\n");
                    PrrtPacket_destroy(packet);
                    break;
                }

                block = BPTree_get(sock_ptr->blockStore, redundancyPayload->base_seqno);
                if (block == NULL) {
                    PrrtCodingParams *cpar = calloc(1, sizeof(PrrtCodingParams));
                    PrrtCodingParams_init(cpar);
                    cpar->k = redundancyPayload->k;
                    cpar->n = redundancyPayload->n;
                    PrrtBlock_alloc(&block, cpar, redundancyPayload->base_seqno);
                    sock_ptr->blockStore = BPTree_insert(sock_ptr->blockStore, redundancyPayload->base_seqno, block);
                }

                PrrtBlock_insert_redundancy_packet(block, packet);
                decode_block(sock_ptr, redundancyPayload->base_seqno);
                break;
            default:
                //PrrtPacket_print(packet);
                break;
        }


    }

    error:
    free(packet);
}

void decode_block(const PrrtSocket *sock_ptr, uint16_t base_seqno) {
    PrrtBlock *block = BPTree_get(sock_ptr->blockStore, base_seqno);

    if (block != NULL && PrrtBlock_decode_ready(block)) {
        retrieve_data_blocks(sock_ptr, base_seqno, block->coding_params.k, block);

        PrrtBlock_decode(block);

        LIST_FOREACH(block->data_blocks, first, next, cur) {
            PrrtPacket *pkt = cur->value;
            if (PrrtForwardPacketTable_test_set_is_number_relevant(sock_ptr->forwardPacketTable, pkt->seqno)) {
                printf("RECOVERED %d via coding.\n", pkt->seqno);
                pthread_mutex_lock(&sock_ptr->inQueueFilledMutex);
                List_push(sock_ptr->inQueue, pkt);
                pthread_cond_signal(&sock_ptr->inQueueFilledMutexCv);
                pthread_mutex_unlock(&sock_ptr->inQueueFilledMutex);
            }
        }
    }
}

void retrieve_data_blocks(PrrtSocket *sock_ptr, uint16_t base_seqno, uint8_t k, const PrrtBlock *block) {
    List *res = List_create();
    BPTree_get_range(sock_ptr->dataStore, res, base_seqno, base_seqno + k - 1);

    LIST_FOREACH(res, first, next, cur) {
            PrrtPacket *packetPtr = cur->value;
            PrrtBlock_insert_data_packet(block, packetPtr);
            sock_ptr->blockStore = BPTree_delete(sock_ptr->blockStore, packetPtr->seqno);
        }
}