packetDeliveryStore.c

#include <pthread.h>
#include "../clock.h"
#include "../types/packet.h"
#include "../../defines.h"
#include "../../util/common.h"
#include "../../util/dbg.h"
#include "../../util/time.h"
#include "packetDeliveryStore.h"

int packet_wait(PrrtPacketDeliveryStore *store, prrtTimestamp_t start, prrtTimestamp_t stop, const struct timespec *deadline) {
    store->last_start = start;
    store->last_stop = stop;
    store->in_ordered_wait = true;
    int result = pthread_cond_timedwait(&store->wait_for_data, &store->lock, deadline);
    debug(DEBUG_RECEIVER, "After wait: %d", result);
    store->in_ordered_wait = false;
    return result;
}

PrrtPacketDeliveryStore *PrrtPacketDeliveryStore_create() {
    PrrtPacketDeliveryStore *store = (PrrtPacketDeliveryStore *) calloc(1, sizeof(PrrtPacketDeliveryStore));
    store->tree = NULL;
    store->closing = false;
    store->last_start = 0;
    store->last_stop = 0;
    store->in_ordered_wait = false;

    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(&store->lock, &attr) == 0, "Mutex init failed.");

    check(pthread_cond_init(&store->wait_for_data, NULL) == EXIT_SUCCESS, "Condition init failed.");

    return store;
    error:
    PERROR("Out of memory%s.", "");
    return NULL;
}

void PrrtPacketDeliveryStore_interrupt(PrrtPacketDeliveryStore *store) {
    check(pthread_mutex_lock(&store->lock) == EXIT_SUCCESS, "Lock failed.");
    atomic_store_explicit(&store->closing, true, memory_order_release);
    pthread_cond_broadcast(&store->wait_for_data);
    check(pthread_mutex_unlock(&store->lock) == EXIT_SUCCESS, "Unlock failed.");
    return;
    error:
    PERROR("PrrtPacketDeliveryStore_interrupt() failed.");
}

bool PrrtPacketDeliveryStore_destroy(PrrtPacketDeliveryStore *store) {
    List *packetList = List_create();
    BPTree_get_range(store->tree, packetList, 0, (prrtSequenceNumber_t) (SEQNO_SPACE - 1));
    while (List_count(packetList) > 0) {
        PrrtPacket *block = List_shift(packetList);
        store->tree = BPTree_delete(store->tree, block->sequenceNumber);
        PrrtPacket_destroy(block);
    }

    List_destroy(packetList);

    if(store->tree != NULL) {
        store->tree = BPTree_destroy(store->tree);
    }

    pthread_mutex_destroy(&store->lock);
    pthread_cond_destroy(&store->wait_for_data);
    free(store);
    return true;
}

PrrtPacket * PrrtPacketDeliveryStore_get_packet_timedwait(PrrtPacketDeliveryStore *store, prrtTimestamp_t start,
                                                          prrtTimestamp_t stop, const struct timespec *deadline) {
    PrrtPacket *packet = NULL;

    check(pthread_mutex_lock(&store->lock) == EXIT_SUCCESS, "Lock failed.");
    do {
        packet = PrrtPacketDeliveryStore_get_packet(store, start, stop);
        if (packet == NULL) {
            int res = packet_wait(store, start, stop, deadline);
            packet = PrrtPacketDeliveryStore_get_packet(store, start, stop);
            if (packet == NULL && res == ETIMEDOUT) {
                errno = ETIMEDOUT;
                break;
            }
        }
        if (atomic_load_explicit(&store->closing, memory_order_acquire)) {
            break;
        }
    } while(!packet);
    check(pthread_mutex_unlock(&store->lock) == EXIT_SUCCESS, "Unlock failed.");

    return packet;

    error:
    PERROR("PrrtPacketDeliveryStore_get_packet_timedwait failed%s.", "");
    return NULL;
}

PrrtPacket *PrrtPacketDeliveryStore_get_packet_wait(PrrtPacketDeliveryStore *store) {
    PrrtPacket *packet = NULL;
    prrtTimestamp_t start = 0;
    prrtTimestamp_t stop = MAX_TIMESTAMP;

    check(pthread_mutex_lock(&store->lock) == EXIT_SUCCESS, "Lock failed.");

    packet = PrrtPacketDeliveryStore_get_packet(store, start, stop);
    if (!packet && !atomic_load_explicit(&store->closing, memory_order_acquire)) {
        pthread_cond_wait(&store->wait_for_data, &store->lock);
        packet = PrrtPacketDeliveryStore_get_packet(store, start, stop);
    }

    check(pthread_mutex_unlock(&store->lock) == EXIT_SUCCESS, "Unlock failed.");

    return packet;

    error:
    PERROR("PrrtPacketDeliveryStore_get_packet_timedwait failed%s.", "");
    return NULL;
}

PrrtPacket *PrrtPacketDeliveryStore_get_packet(PrrtPacketDeliveryStore *store, prrtTimestamp_t start, prrtTimestamp_t stop) {
    PrrtPacket *packet = NULL;

    check(pthread_mutex_lock(&store->lock) == EXIT_SUCCESS, "Lock failed.");

    if (start > stop) {
        packet = BPTree_get_first_in_range(store->tree, (BPTreeKey_t) start, TIMESTAMP_SPACE - 1);
        if (packet == NULL) {
            packet = BPTree_get_first_in_range(store->tree, 0, (BPTreeKey_t) stop);
        }
    } else {
        packet = BPTree_get_first_in_range(store->tree, (BPTreeKey_t) start, (BPTreeKey_t) stop);
    }

    if(packet != NULL) {
        store->tree = BPTree_delete(store->tree, PrrtDataPacket_packet_timeout(packet));
    }

    check(pthread_mutex_unlock(&store->lock) == EXIT_SUCCESS, "Unlock failed.");

    return packet;

    error:
    PERROR("Get failed%s.", "");
    return NULL;
}

bool PrrtPacketDeliveryStore_insert(PrrtPacketDeliveryStore *store, PrrtPacket *packet) {
    check(pthread_mutex_lock(&store->lock) == EXIT_SUCCESS, "Lock failed.");
    prrtTimestamp_t timeout = PrrtDataPacket_packet_timeout(packet);
    store->tree = BPTree_insert(store->tree, timeout, packet);

    bool packet_awaited = (PrrtTimestamp_cmp(store->last_start, timeout) <= 0) && (PrrtTimestamp_cmp(timeout, store->last_stop) <= 0);
    if (!store->in_ordered_wait || packet_awaited) {
        pthread_cond_broadcast(&store->wait_for_data);
    }

    check(pthread_mutex_unlock(&store->lock) == EXIT_SUCCESS, "Unlock failed.");
    return true;

    error:
    PERROR("Insert failed%s.", "");
    return false;
}

void PrrtPacketDeliveryStore_cleanup(PrrtPacketDeliveryStore *store, prrtTimestamp_t now) {
    check(pthread_mutex_lock(&store->lock) == EXIT_SUCCESS, "Lock failed.");
    List* packetList = List_create();
    if (now >= TIMESTAMP_SPACE/2) {
        BPTree_get_range(store->tree, packetList, now - TIMESTAMP_SPACE/2, now);
    } else {
        BPTree_get_range(store->tree, packetList, 0, now);
        BPTree_get_range(store->tree, packetList, now + TIMESTAMP_SPACE/2, TIMESTAMP_SPACE - 1);
    }
    while (List_count(packetList) > 0) {
        PrrtPacket *packet = List_shift(packetList);
        store->tree = BPTree_delete(store->tree, PrrtDataPacket_packet_timeout(packet));
        PrrtPacket_destroy(packet);
    }
    List_destroy(packetList);
    check(pthread_mutex_unlock(&store->lock) == EXIT_SUCCESS, "Unlock failed.");
    return;

    error:
    PERROR("PrrtPacketDeliveryStore_cleanup failed%s.", "");
}