From 6384e0f20dc67a5c5309906366f464aef2fdde9b Mon Sep 17 00:00:00 2001 From: Ian Gulliver Date: Sun, 21 Feb 2016 15:51:38 -0800 Subject: [PATCH] Wakeup actually working (probably), add delay calculator. --- adsbus/adsbus.c | 2 + adsbus/common.c | 31 ++++++++++++ adsbus/common.h | 7 +++ adsbus/incoming.c | 1 - adsbus/outgoing.c | 1 - adsbus/wakeup.c | 117 ++++++++++++++++++++++++++++++++++++++++++++-- adsbus/wakeup.h | 2 +- 7 files changed, 153 insertions(+), 8 deletions(-) diff --git a/adsbus/adsbus.c b/adsbus/adsbus.c index 78b21aa..a6634c9 100644 --- a/adsbus/adsbus.c +++ b/adsbus/adsbus.c @@ -102,6 +102,7 @@ int main(int argc, char *argv[]) { hex_init(); peer_init(); + retry_init(); wakeup_init(); send_init(); @@ -116,6 +117,7 @@ int main(int argc, char *argv[]) { peer_loop(); + retry_cleanup(); wakeup_cleanup(); send_cleanup(); diff --git a/adsbus/common.c b/adsbus/common.c index 702fbf6..0f384fe 100644 --- a/adsbus/common.c +++ b/adsbus/common.c @@ -216,3 +216,34 @@ void uuid_gen(char *out) { uuid_generate(uuid); uuid_unparse(uuid, out); } + + +static int retry_rand_fd; + +void retry_init() { + retry_rand_fd = open("/dev/urandom", O_RDONLY); + assert(retry_rand_fd >= 0); +} + +void retry_cleanup() { + assert(!close(retry_rand_fd)); +} + +#define RETRY_MIN_MS 2000 +#define RETRY_MAX_MS 64000 +#define RETRY_MULT 2 +#define RETRY_MAX_JITTER_DIV 2 +uint32_t retry_get_delay_ms(uint32_t prev_delay) { + uint32_t delay = prev_delay * RETRY_MULT; + delay = delay < RETRY_MIN_MS ? RETRY_MIN_MS : delay; + delay = delay > RETRY_MAX_MS ? RETRY_MAX_MS : delay; + + uint32_t max_jitter = delay / RETRY_MAX_JITTER_DIV; + uint32_t jitter; + assert(read(retry_rand_fd, &jitter, sizeof(jitter)) == sizeof(jitter)); + delay += jitter % max_jitter; + + delay = delay > RETRY_MAX_MS ? RETRY_MAX_MS : delay; + + return delay; +} diff --git a/adsbus/common.h b/adsbus/common.h index 6202416..7427d82 100644 --- a/adsbus/common.h +++ b/adsbus/common.h @@ -91,3 +91,10 @@ void hex_from_int(char *, uint64_t, size_t); #define UUID_LEN 37 void uuid_gen(char *); + + +///////// retry timing + +void retry_init(); +void retry_cleanup(); +uint32_t retry_get_delay_ms(uint32_t); diff --git a/adsbus/incoming.c b/adsbus/incoming.c index 4c6114b..e7eed5c 100644 --- a/adsbus/incoming.c +++ b/adsbus/incoming.c @@ -13,7 +13,6 @@ #include "common.h" #include "incoming.h" -struct incoming; struct incoming { struct peer peer; char id[UUID_LEN]; diff --git a/adsbus/outgoing.c b/adsbus/outgoing.c index d728dbb..d6b59a3 100644 --- a/adsbus/outgoing.c +++ b/adsbus/outgoing.c @@ -11,7 +11,6 @@ #include "common.h" #include "outgoing.h" -struct outgoing; struct outgoing { struct peer peer; char id[UUID_LEN]; diff --git a/adsbus/wakeup.c b/adsbus/wakeup.c index e40909e..87765eb 100644 --- a/adsbus/wakeup.c +++ b/adsbus/wakeup.c @@ -1,7 +1,13 @@ +#define _GNU_SOURCE + +#include #include #include +#include #include #include +#include +#include #include #include @@ -9,9 +15,53 @@ #include "wakeup.h" +struct wakeup_peer { + struct peer peer; + struct peer *inner_peer; +}; + +struct wakeup_request { + int fd; + uint64_t absolute_time_ms; +}; + +struct wakeup_entry { + struct wakeup_request request; + struct wakeup_entry *next; +}; + static pthread_t wakeup_thread; static int wakeup_write_fd; +static uint64_t wakeup_get_time_ms() { + struct timespec tp; + assert(!clock_gettime(CLOCK_MONOTONIC_COARSE, &tp)); +#define NS_PER_MS 1000000 + return tp.tv_sec + (tp.tv_nsec / NS_PER_MS); +} + +static void wakeup_request_add(struct wakeup_entry **head, struct wakeup_request *request) { + struct wakeup_entry *entry = malloc(sizeof(*entry)); + memcpy(&entry->request, request, sizeof(entry->request)); + + struct wakeup_entry *prev = NULL, *iter = *head; + while (iter) { + if (iter->request.absolute_time_ms > entry->request.absolute_time_ms) { + break; + } + prev = iter; + iter = iter->next; + } + + if (prev) { + entry->next = prev->next; + prev->next = entry; + } else { + entry->next = *head; + *head = entry; + } +} + static void *wakeup_main(void *arg) { int read_fd = (intptr_t) arg; @@ -20,18 +70,50 @@ static void *wakeup_main(void *arg) { struct epoll_event ev = { .events = EPOLLIN, + .data = { + .fd = read_fd, + }, }; assert(!epoll_ctl(epoll_fd, EPOLL_CTL_ADD, read_fd, &ev)); + struct wakeup_entry *head = NULL; + while (1) { -#define MAX_EVENTS 10 +#define MAX_EVENTS 1 struct epoll_event events[MAX_EVENTS]; int nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, -1); if (nfds == -1 && errno == EINTR) { continue; } - assert(nfds >= 0); - break; // XXX + + if (nfds == 1) { + assert(events[0].data.fd == read_fd); + struct wakeup_request request; + ssize_t result = read(read_fd, &request, sizeof(request)); + if (result == 0) { + // Peer closed connection, shutdown thread + break; + } + assert(result == sizeof(request)); + wakeup_request_add(&head, &request); + } else { + assert(nfds == 0); + } + + uint64_t now = wakeup_get_time_ms(); + while (head && head->request.absolute_time_ms < now) { + close(head->request.fd); + struct wakeup_entry *next = head->next; + free(head); + head = next; + } + } + + while (head) { + close(head->request.fd); + struct wakeup_entry *next = head->next; + free(head); + head = next; } assert(!close(read_fd)); @@ -39,9 +121,18 @@ static void *wakeup_main(void *arg) { return NULL; } +static void wakeup_handler(struct peer *peer) { + struct wakeup_peer *outer_peer = (struct wakeup_peer *) peer; + assert(!close(outer_peer->peer.fd)); + + struct peer *inner_peer = outer_peer->inner_peer; + free(outer_peer); + inner_peer->event_handler(inner_peer); +} + void wakeup_init() { int pipefd[2]; - assert(!pipe(pipefd)); + assert(!pipe2(pipefd, O_NONBLOCK)); assert(!pthread_create(&wakeup_thread, NULL, wakeup_main, (void *) (intptr_t) pipefd[0])); wakeup_write_fd = pipefd[1]; } @@ -51,5 +142,21 @@ void wakeup_cleanup() { assert(!pthread_join(wakeup_thread, NULL)); } -void wakeup_add(struct peer *peer, int delay_ms) { +void wakeup_add(struct peer *peer, uint32_t delay_ms) { + int pipefd[2]; + assert(!pipe2(pipefd, O_NONBLOCK)); + + struct wakeup_request request = { + .fd = pipefd[1], + .absolute_time_ms = wakeup_get_time_ms() + delay_ms, + }; + assert(write(wakeup_write_fd, &request, sizeof(request)) == sizeof(request)); + + struct wakeup_peer *outer_peer = malloc(sizeof(*outer_peer)); + assert(outer_peer); + outer_peer->peer.fd = pipefd[0]; + outer_peer->peer.event_handler = wakeup_handler; + outer_peer->inner_peer = peer; + + peer_epoll_add((struct peer *) outer_peer, EPOLLRDHUP); } diff --git a/adsbus/wakeup.h b/adsbus/wakeup.h index 793d0f4..7ac34ee 100644 --- a/adsbus/wakeup.h +++ b/adsbus/wakeup.h @@ -4,4 +4,4 @@ struct peer; void wakeup_init(); void wakeup_cleanup(); -void wakeup_add(struct peer *, int); +void wakeup_add(struct peer *, uint32_t);