Commit d5c268ff authored by Scott Vokes's avatar Scott Vokes
Browse files

Check remaining time during send/poll loop. Use clock_gettime if available. 

clock_gettime(CLOCK_MONOTONIC, &timespec) will be durable across time
changes. If it's not available, or the call has specified that that level
of accuracy is not needed, it will use regular gettimeofday.
parent 21cbdfb1

src/lib/bus/bus_example.c

+4 −1
Original line number Diff line number Diff line
@@ -35,6 +35,7 @@ 
#include "bus.h"

#include "atomic.h"

#include "socket99.h"

#include "util.h"



typedef struct {

    uint32_t magic_number;
@@ -444,7 +445,9 @@ static void tick_handler(example_state *s) { 


static time_t get_cur_second(void) {

    struct timeval tv;

    gettimeofday(&tv, 0);  // TODO: clock_gettime

    if (!util_timestamp(&tv, true)) {

        assert(false);

    }

    return tv.tv_sec;

}

src/lib/bus/echosrv.c

+2 −2
Original line number Diff line number Diff line
@@ -198,7 +198,7 @@ static void tick_handler(config *cfg) { 
static void listen_loop_poll(config *cfg) {

    struct timeval tv;



    gettimeofday(&tv, 0);

    if (!util_timestamp(&tv, true)) { assert(false); }

    cfg->last_second = tv.tv_sec;



    assert(cfg->client_fds[0].fd == NO_CLIENT);
@@ -206,7 +206,7 @@ static void listen_loop_poll(config *cfg) { 
    int delay = 1;



    for (;;) {

        gettimeofday(&tv, 0);  // TODO: clock_gettime

        if (!util_timestamp(&tv, true)) { assert(false); }

        if (tv.tv_sec != cfg->last_second) {

            tick_handler(cfg);

            cfg->last_second = tv.tv_sec;

src/lib/bus/listener.c

+11 −4
Original line number Diff line number Diff line
@@ -302,7 +302,11 @@ void *listener_mainloop(void *arg) { 
    struct bus *b = self->bus;

    struct timeval tv;

    

    gettimeofday(&tv, NULL);

    if (!util_timestamp(&tv, true)) {

        BUS_LOG_SNPRINTF(b, 0, LOG_LISTENER, b->udata, 64,

            "timestamp failure: %d", errno);

    }



    time_t last_sec = tv.tv_sec;



    /* The listener thread has full control over its execution -- the
@@ -313,7 +317,10 @@ void *listener_mainloop(void *arg) { 
     * need any internal locking. */



    while (self->shutdown_notify_fd == LISTENER_NO_FD) {

        gettimeofday(&tv, NULL);  // TODO: clock_gettime

        if (!util_timestamp(&tv, true)) {

            BUS_LOG_SNPRINTF(b, 0, LOG_LISTENER, b->udata, 64,

                "timestamp failure: %d", errno);

        }

        time_t cur_sec = tv.tv_sec;

        if (cur_sec != last_sec) {

            tick_handler(self);
@@ -740,7 +747,7 @@ static void tick_handler(listener *l) { 
            /* Check timeout */

            if (info->timeout_sec == 1) {

                struct timeval tv;

                if (-1 == gettimeofday(&tv, NULL)) {

                if (!util_timestamp(&tv, false)) {

                    BUS_LOG(b, 0, LOG_LISTENER,

                        "gettimeofday failure in tick_handler!", b->udata);

                    continue;
@@ -778,7 +785,7 @@ static void tick_handler(listener *l) { 
                notify_message_failure(l, info, BUS_SEND_RX_FAILURE);

            } else if (info->timeout_sec == 1) {

                struct timeval tv;

                if (-1 == gettimeofday(&tv, NULL)) {

                if (!util_timestamp(&tv, false)) {

                    BUS_LOG(b, 0, LOG_LISTENER,

                        "gettimeofday failure in tick_handler!", b->udata);

                    continue;

src/lib/bus/sender.c

+36 −11
Original line number Diff line number Diff line
@@ -57,8 +57,14 @@ bool sender_do_blocking_send(bus *b, boxed_msg *box) { 
    

    int timeout_msec = box->timeout_sec * 1000;



    struct timeval tv;

    if (0 == gettimeofday(&tv, NULL)) { box->tv_send_start = tv; }

    struct timeval start;

    if (util_timestamp(&start, true)) {

        box->tv_send_start = start;

    } else {

        BUS_LOG_SNPRINTF(b, 0, LOG_SENDER, b->udata, 128,

            "gettimeofday failure: %d", errno);

        return false;

    }



    struct pollfd fds[1];

    fds[0].fd = box->fd;
@@ -83,7 +89,25 @@ bool sender_do_blocking_send(bus *b, boxed_msg *box) { 


    int rem_msec = timeout_msec;



    for (;;) {

    while (rem_msec > 0) {

        struct timeval now;

        if (util_timestamp(&now, true)) {

            size_t usec_elapsed = (((now.tv_sec - start.tv_sec) * 1000000)

                + (now.tv_usec - start.tv_usec));

            size_t msec_elapsed = usec_elapsed / 1000;



            rem_msec = timeout_msec - msec_elapsed;

        } else {

            /* If gettimeofday fails here, the listener's hold message has

             * already been sent; it will time out later. We need to treat

             * this like a TX failure (including closing the socket) because

             * we don't know what state the connection was left in. */

            BUS_LOG_SNPRINTF(b, 0, LOG_SENDER, b->udata, 128,

                "gettimeofday failure in poll loop: %d", errno);

            handle_failure(b, box, BUS_SEND_TX_FAILURE);

            return true;

        }



        int res = poll(fds, 1, rem_msec);

        BUS_LOG_SNPRINTF(b, 3, LOG_SENDER, b->udata, 256,

            "handle_write: poll res %d", res);
@@ -109,9 +133,6 @@ bool sender_do_blocking_send(bus *b, boxed_msg *box) { 
                handle_failure(b, box, BUS_SEND_TX_FAILURE);

                return true;

            } else if (revents & POLLOUT) {

                /* TODO: use gettimeofday to figure out actual elapsed time?

                 * We're more concerned with timing out on the response than

                 * on the send, though. */

                handle_write_res hw_res = handle_write(b, box);



                BUS_LOG_SNPRINTF(b, 4, LOG_SENDER, b->udata, 256,
@@ -131,13 +152,15 @@ bool sender_do_blocking_send(bus *b, boxed_msg *box) { 
                assert(false);  /* match fail */

            }

        } else if (res == 0) {  /* timeout */

            break;

        }

    }

    BUS_LOG_SNPRINTF(b, 3, LOG_SENDER, b->udata, 256,

                "do_blocking_send on %d: TX_TIMEOUT", box->fd);

        "do_blocking_send on <fd:%d, seq_id:%lld>: TX_TIMEOUT",

        box->fd, (long long)box->out_seq_id);

    handle_failure(b, box, BUS_SEND_TX_TIMEOUT);

    return true;

}

    }

}



static bool attempt_to_enqueue_sending_request_message_to_listener(struct bus *b,

    int fd, int64_t seq_id, int16_t timeout_sec) {
@@ -236,7 +259,9 @@ static handle_write_res handle_write(bus *b, boxed_msg *box) { 


    if (rem == 0) {             /* check if whole message is written */

        struct timeval tv;

        if (0 == gettimeofday(&tv, NULL)) { box->tv_send_done = tv; }

        if (util_timestamp(&tv, true)) {

            box->tv_send_done = tv;

        }



        if (enqueue_request_sent_message_to_listener(b, box)) {

            return HW_DONE;

src/lib/bus/util.c

+20 −0
Original line number Diff line number Diff line
@@ -19,9 +19,29 @@ 
*/

#include <stdbool.h>

#include <errno.h>

#include <time.h>



#include "util.h"



bool util_is_resumable_io_error(int errno_) {

    return errno_ == EAGAIN || errno_ == EINTR || errno_ == EWOULDBLOCK;

}



bool util_timestamp(struct timeval *tv, bool relative) {

    if (tv == NULL) {

        return false;

    }



#ifdef __linux__

    if (relative) {

        struct timespec ts;

        if (0 != clock_gettime(CLOCK_MONOTONIC, &ts)) {

            return false;

        }

        tv.tv_sec = ts.tv_sec;

        tv.tv_usec = tv.tv_nsec / 1000L;

        return true;

    }

#endif

    return (0 == gettimeofday(tv, NULL));

}