Commit 02a43b57 authored by Andreas Schmidt's avatar Andreas Schmidt
Browse files

Add sync/async receive modes.

parent db7bcf84

prrt/cprrt.pxd

+3 −1
Original line number Diff line number Diff line
@@ -123,7 +123,9 @@ cdef extern from "proto/socket.h": 
    bint PrrtSocket_bind(PrrtSocket *sock_ptr, const_char *ipAddress, const uint16_t port)

    int PrrtSocket_close(const PrrtSocket *sock_ptr)

    int PrrtSocket_connect(PrrtSocket *sock_ptr, const_char *host, const uint16_t port)

    int PrrtSocket_send(PrrtSocket *sock_ptr, const uint8_t *data, const size_t data_len)

    int PrrtSocket_send_sync(PrrtSocket *sock_ptr, const uint8_t *data, const size_t data_len)

    int PrrtSocket_send_async(PrrtSocket *sock_ptr, const uint8_t *data, const size_t data_len)





    int32_t PrrtSocket_recv(PrrtSocket *sock_ptr, void *buf_ptr, sockaddr* addr) nogil

    int32_t PrrtSocket_receive_asap(PrrtSocket *s, void *buf_ptr, sockaddr* addr) nogil

prrt/proto/processes/dataTransmitter.c

+45 −41
Original line number Diff line number Diff line
@@ -151,29 +151,14 @@ static bool send_packet(PrrtSocket *sock_ptr, PrrtPacket *packet) { 
    return false;

}



void *send_data_loop(void *ptr) {

    PrrtSocket *sock_ptr = ptr;

    PrrtBlock *block = NULL;



    while (1) {

        ListNode *job;

        do {

            job = Pipe_pull(sock_ptr->sendDataQueue);

            if (PrrtSocket_closing(sock_ptr)) {

                if (block != NULL) {

                    PrrtBlock_destroy(block);

                }

                return NULL;

            }

        } while (!job);

        PrrtPacket *packet = PrrtPacket_byListNode(job);

void PrrtDataTransmitter_transmit(PrrtSocket *sock_ptr, PrrtPacket *packet) {

    XlapTimeStampClock(sock_ptr, ts_data_packet, packet->sequenceNumber, PrrtTransmitStart);

    XlapTimeStampCycle(sock_ptr, ts_data_packet, packet->sequenceNumber, PrrtTransmitStart);

        if (block == NULL) {

            block = PrrtBlock_create(PrrtCodingConfiguration_copy(sock_ptr->codingParameters), PrrtCoder_copy(sock_ptr->coder), packet->sequenceNumber);

    if (sock_ptr->receiveBlock == NULL) {

        sock_ptr->receiveBlock = PrrtBlock_create(PrrtCodingConfiguration_copy(sock_ptr->codingParameters), PrrtCoder_copy(sock_ptr->coder), packet->sequenceNumber);

    }



        packet->index = (uint8_t) (packet->sequenceNumber - block->baseSequenceNumber);

    packet->index = (uint8_t) (packet->sequenceNumber - sock_ptr->receiveBlock->baseSequenceNumber);



    PrrtPacketDataPayload *payload = packet->payload;

    payload->groupRTprop_us = PrrtSocket_get_rtprop_fwd(sock_ptr);
@@ -184,24 +169,43 @@ void *send_data_loop(void *ptr) { 
    XlapTimeStampClock(sock_ptr, ts_data_packet, packet->sequenceNumber, PrrtTransmitEnd);

    XlapTimeStampCycle(sock_ptr, ts_data_packet, packet->sequenceNumber, PrrtTransmitEnd);



        PrrtBlock_insert_data_packet(block, packet);

    PrrtBlock_insert_data_packet(sock_ptr->receiveBlock, packet);



    // TODO: redundancy should only be sent when necessary

        if (PrrtBlock_encode_ready(block)) {

    if (PrrtBlock_encode_ready(sock_ptr->receiveBlock)) {

        uint32_t j = 0;

        unsigned int redundancy_seqno = sock_ptr->sequenceNumberRedundancy;

        XlapTimeStampCycle(sock_ptr, ts_redundancy_packet, redundancy_seqno, PrrtEncodeStart);

            PrrtBlock_encode(block, &sock_ptr->sequenceNumberRedundancy);

        PrrtBlock_encode(sock_ptr->receiveBlock, &sock_ptr->sequenceNumberRedundancy);

        XlapTimeStampCycle(sock_ptr, ts_redundancy_packet, redundancy_seqno, PrrtEncodeEnd);



            uint32_t redundancyPackets = List_count(block->redundancyPackets);

        uint32_t redundancyPackets = List_count(sock_ptr->receiveBlock->redundancyPackets);

        for (j = 0; j < redundancyPackets; j++) {

                PrrtPacket *red_pkt = PrrtBlock_get_first_red_data(block);

            PrrtPacket *red_pkt = PrrtBlock_get_first_red_data(sock_ptr->receiveBlock);

            send_packet(sock_ptr, red_pkt);

        }



            PrrtBlock_destroy(block);

            block = NULL;

        PrrtBlock_destroy(sock_ptr->receiveBlock);

        sock_ptr->receiveBlock = NULL;

    }

}



void *PrrtDataTransmitter_send_data_loop(void *ptr) {

    PrrtSocket *sock_ptr = ptr;



    while (1) {

        ListNode *job;

        do {

            job = Pipe_pull(sock_ptr->sendDataQueue);

            if (PrrtSocket_closing(sock_ptr)) {

                if (sock_ptr->receiveBlock != NULL) {

                    PrrtBlock_destroy(sock_ptr->receiveBlock);

                    sock_ptr->receiveBlock = NULL;

                }

                return NULL;

            }

        } while (!job);

        PrrtPacket *packet = PrrtPacket_byListNode(job);

        PrrtDataTransmitter_transmit(sock_ptr, packet);

    }

}

prrt/proto/processes/dataTransmitter.h

+4 −1
Original line number Diff line number Diff line 
#ifndef PRRT_DATA_TRANSMITTER_H

#define PRRT_DATA_TRANSMITTER_H



void * send_data_loop(void *ptr);

#include "../socket.h"



void * PrrtDataTransmitter_send_data_loop(void *ptr);

void PrrtDataTransmitter_transmit(PrrtSocket *sock_ptr, PrrtPacket *packet);



#endif //PRRT_DATA_TRANSMITTER_H

prrt/proto/socket.c

+18 −4
Original line number Diff line number Diff line
@@ -196,7 +196,7 @@ bool PrrtSocket_bind(PrrtSocket *s, const char *ipAddress, const uint16_t port) 
        pin_thread_to_core(s->sendDataThreadAttr, 1);

    }



    check(pthread_create(&s->sendDataThread, s->sendDataThreadAttr, send_data_loop,

    check(pthread_create(&s->sendDataThread, s->sendDataThreadAttr, PrrtDataTransmitter_send_data_loop,

                         (void *) s) == EXIT_SUCCESS, "Cannot create send thread.");



    check(pthread_create(&s->receiveDataThread, s->receiveDataThreadAttr, receive_data_loop,
@@ -218,7 +218,7 @@ int PrrtSocket_connect(PrrtSocket *s, const char *host, const uint16_t port) { 
    return 0;

}



int PrrtSocket_send(PrrtSocket *s, const uint8_t *data, const size_t data_len) {

int sendpacket(PrrtSocket *s, const uint8_t *data, const size_t data_len, bool sync) {

    if (data_len > s->maximum_payload_size) {

        PERROR("Data to be sent (%ld bytes) is too long, as MTU is %d.\n", data_len, s->maximum_payload_size);

        return -1;
@@ -240,7 +240,13 @@ int PrrtSocket_send(PrrtSocket *s, const uint8_t *data, const size_t data_len) { 
    XlapTimestampPlaceholderUse(s, ts_data_packet, packet->sequenceNumber, &tsph);



    XlapTimeStampCycle(s, ts_data_packet, packet->sequenceNumber, PrrtSubmitPackage);

    if (sync) {

        PrrtDataTransmitter_transmit(s, packet);

        PrrtSocket_pace(s);

    } else {

        Pipe_push(s->sendDataQueue, &packet->asListNode);

    }



    PrrtReceiver_on_application_write(s->receiver, Pipe_get_size(s->sendDataQueue), s->sequenceNumberSource);



    XlapTimeStampClock(s, ts_data_packet, packet->sequenceNumber, PrrtSendEnd);
@@ -248,6 +254,14 @@ int PrrtSocket_send(PrrtSocket *s, const uint8_t *data, const size_t data_len) { 
    return 0;

}



int PrrtSocket_send_sync(PrrtSocket *s, const uint8_t *data, const size_t data_len) {

    return sendpacket(s, data, data_len, true);

}



int PrrtSocket_send_async(PrrtSocket *s, const uint8_t *data, const size_t data_len) {

    return sendpacket(s, data, data_len, false);

}



bool PrrtSocket_closing(PrrtSocket *s) {

    return atomic_load_explicit(&s->closing, memory_order_acquire);

}

prrt/proto/socket.h

+5 −3
Original line number Diff line number Diff line
@@ -36,6 +36,7 @@ typedef struct prrtSocket { 
    bool pacingEnabled;



    PrrtClock clock;

    PrrtBlock* receiveBlock;



    pthread_t sendDataThread;

    Pipe *sendDataQueue;
@@ -113,7 +114,10 @@ int PrrtSocket_close(PrrtSocket *s); 


int PrrtSocket_connect(PrrtSocket *s, const char *host, uint16_t port);



int PrrtSocket_send(PrrtSocket *s, const uint8_t *data, size_t data_len);

int PrrtSocket_send_async(PrrtSocket *s, const uint8_t *data, const size_t data_len);

int PrrtSocket_send_sync(PrrtSocket *s, const uint8_t *data, size_t data_len);

void PrrtSocket_pace(PrrtSocket *s);





int32_t PrrtSocket_recv(PrrtSocket *s, void *buf_ptr, struct sockaddr* addr);
@@ -125,8 +129,6 @@ int32_t PrrtSocket_receive_ordered(PrrtSocket *s, void *buf_ptr, struct sockaddr 
int32_t PrrtSocket_receive_ordered_wait(PrrtSocket *s, void *buf_ptr, struct sockaddr* addr, prrtTimedelta_t time_window_us);

int32_t PrrtSocket_receive_ordered_timedwait(PrrtSocket *s, void *buf_ptr, struct sockaddr* addr, prrtTimedelta_t time_window_us, struct timespec* deadline);



void PrrtSocket_pace(PrrtSocket *s);



bool PrrtSocket_cleanup(PrrtSocket *s);



bool PrrtSocket_closing(PrrtSocket *s);