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

Refactor out pace method.

parent db6d95b3

prrt/proto/processes/dataTransmitter.c

+1 −15
Original line number Diff line number Diff line
@@ -77,22 +77,8 @@ static bool send_packet(PrrtSocket *sock_ptr, PrrtPacket *packet) { 
    uint8_t buf[MAX_PAYLOAD_LENGTH];

    memset(buf, 0, sizeof(buf));

    prrtPacketLength_t length = PrrtPacket_size(packet);



    prrtPacketLength_t payloadLength = packet->payloadLength;

    if (sock_ptr->pacingEnabled && sock_ptr->nextSendTime != 0) {

        debug(DEBUG_DATATRANSMITTER, "About to check for pacing.");

        prrtTimeDifference_t diff = 0;

        do {

            prrtTimeDifference_t now = (prrtTimeDifference_t) PrrtClock_get_current_time_us();

            diff = ((prrtTimeDifference_t) sock_ptr->nextSendTime) - now;

            if (diff > 0) {

                debug(DEBUG_DATATRANSMITTER, "S: %u, Pacing for %d (%u).", packet->sequenceNumber, diff, now);

                usleep_nano((uint32_t) diff);

            }

        } while (diff > 0);

    } else {

        usleep_nano(1);

    }

    PrrtSocket_pace(sock_ptr);



    int64_t space = PrrtReceiver_get_space(sock_ptr->receiver);

    while (space < sock_ptr->maximum_payload_size) {

prrt/proto/socket.c

+16 −0
Original line number Diff line number Diff line
@@ -252,6 +252,22 @@ bool PrrtSocket_closing(PrrtSocket *s) { 
    return atomic_load_explicit(&s->closing, memory_order_acquire);

}



void inline PrrtSocket_pace(PrrtSocket *s) {

    if (s->pacingEnabled && s->nextSendTime != 0) {

        debug(DEBUG_DATATRANSMITTER, "About to check for pacing.");

        prrtTimeDifference_t diff = 0;

        do {

            prrtTimeDifference_t now = (prrtTimeDifference_t) PrrtClock_get_current_time_us();

            diff = ((prrtTimeDifference_t) s->nextSendTime) - now;

            if (diff > 0) {

                usleep_nano((uint32_t) diff);

            }

        } while (diff > 0);

    } else {

        usleep_nano(1);

    }

}



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

    PrrtPacket *packet;

    packet = PrrtPacketDeliveryStore_get_packet(s->packetDeliveryStore, 0, MAX_TIMESTAMP);

prrt/proto/socket.h

+1 −0
Original line number Diff line number Diff line
@@ -125,6 +125,7 @@ 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 inline PrrtSocket_pace(PrrtSocket *s);



bool PrrtSocket_cleanup(PrrtSocket *s);