Switch out custom wakeup system for timerfd.

This commit is contained in:
Ian Gulliver
2016-03-08 21:53:57 -08:00
parent 3985182304
commit b915f38dca
4 changed files with 41 additions and 60 deletions

View File

@@ -111,7 +111,7 @@ static void outgoing_connect_result(struct outgoing *outgoing, int result) {
default: default:
LOG(outgoing->id, "Can't connect to %s/%s: %s", hbuf, sbuf, strerror(result)); LOG(outgoing->id, "Can't connect to %s/%s: %s", hbuf, sbuf, strerror(result));
peer_close(&outgoing->peer); assert(!close(outgoing->peer.fd));
outgoing->addr = outgoing->addr->ai_next; outgoing->addr = outgoing->addr->ai_next;
// Tail recursion :/ // Tail recursion :/
outgoing_connect_next(outgoing); outgoing_connect_next(outgoing);

View File

@@ -10,7 +10,6 @@
#include "log.h" #include "log.h"
#include "server.h" #include "server.h"
#include "wakeup.h"
#include "peer.h" #include "peer.h"
@@ -118,7 +117,7 @@ void peer_loop() {
} }
#define MAX_EVENTS 10 #define MAX_EVENTS 10
struct epoll_event events[MAX_EVENTS]; struct epoll_event events[MAX_EVENTS];
int delay = list_is_empty(&peer_always_trigger_head) ? wakeup_get_delay() : 0; int delay = list_is_empty(&peer_always_trigger_head) ? -1 : 0;
int nfds = epoll_wait(peer_epoll_fd, events, MAX_EVENTS, delay); int nfds = epoll_wait(peer_epoll_fd, events, MAX_EVENTS, delay);
if (nfds == -1 && errno == EINTR) { if (nfds == -1 && errno == EINTR) {
continue; continue;
@@ -136,7 +135,5 @@ void peer_loop() {
peer_call(iter); peer_call(iter);
} }
} }
wakeup_dispatch();
} }
} }

View File

@@ -1,83 +1,69 @@
#include <assert.h> #include <assert.h>
#include <limits.h> #include <limits.h>
#include <stdlib.h> #include <stdlib.h>
#include <time.h> #include <sys/timerfd.h>
#include <unistd.h>
#include "peer.h" #include "peer.h"
#include "rand.h" #include "rand.h"
#include "wakeup.h" #include "wakeup.h"
struct wakeup_entry { struct wakeup {
int fd; struct peer peer;
uint64_t absolute_time_ms; struct peer *inner_peer;
struct peer *peer;
struct list_head wakeup_list; struct list_head wakeup_list;
}; };
static struct list_head wakeup_head = LIST_HEAD_INIT(wakeup_head); static struct list_head wakeup_head = LIST_HEAD_INIT(wakeup_head);
static uint64_t wakeup_get_time_ms() { static void wakeup_handler(struct peer *peer) {
struct timespec tp; struct wakeup *wakeup = container_of(peer, struct wakeup, peer);
assert(!clock_gettime(CLOCK_MONOTONIC_COARSE, &tp));
#define MS_PER_S UINT64_C(1000) uint64_t events;
#define NS_PER_MS UINT64_C(1000000) assert(read(wakeup->peer.fd, &events, sizeof(events)) == sizeof(events));
assert(tp.tv_sec >= 0); assert(events == 1);
assert(tp.tv_nsec >= 0);
return ((uint64_t) tp.tv_sec * MS_PER_S) + ((uint64_t) tp.tv_nsec / NS_PER_MS); peer_close(&wakeup->peer);
peer_call(wakeup->inner_peer);
list_del(&wakeup->wakeup_list);
free(wakeup);
} }
void wakeup_init() { void wakeup_init() {
} }
void wakeup_cleanup() { void wakeup_cleanup() {
struct wakeup_entry *iter, *next; struct wakeup *iter, *next;
list_for_each_entry_safe(iter, next, &wakeup_head, wakeup_list) { list_for_each_entry_safe(iter, next, &wakeup_head, wakeup_list) {
free(iter); free(iter);
} }
} }
int wakeup_get_delay() {
if (list_is_empty(&wakeup_head)) {
return -1;
}
uint64_t now = wakeup_get_time_ms();
struct wakeup_entry *next_to_fire = list_entry(wakeup_head.next, struct wakeup_entry, wakeup_list);
if (next_to_fire->absolute_time_ms > now) {
uint64_t delta = next_to_fire->absolute_time_ms - now;
assert(delta < INT_MAX);
return (int) delta;
} else {
return 0;
}
}
void wakeup_dispatch() {
uint64_t now = wakeup_get_time_ms();
struct wakeup_entry *iter, *next;
list_for_each_entry_safe(iter, next, &wakeup_head, wakeup_list) {
if (iter->absolute_time_ms > now) {
break;
}
peer_call(iter->peer);
list_del(&iter->wakeup_list);
free(iter);
}
}
void wakeup_add(struct peer *peer, uint32_t delay_ms) { void wakeup_add(struct peer *peer, uint32_t delay_ms) {
struct wakeup_entry *entry = malloc(sizeof(*entry)); struct wakeup *wakeup = malloc(sizeof(*wakeup));
entry->absolute_time_ms = wakeup_get_time_ms() + delay_ms; wakeup->inner_peer = peer;
entry->peer = peer; list_add(&wakeup->wakeup_list, &wakeup_head);
struct wakeup_entry *iter, *next; wakeup->peer.fd = timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
list_for_each_entry_safe(iter, next, &wakeup_head, wakeup_list) { assert(wakeup->peer.fd >= 0);
if (iter->absolute_time_ms > entry->absolute_time_ms) {
list_add(&entry->wakeup_list, &iter->wakeup_list); #define MS_PER_S UINT64_C(1000)
return; #define NS_PER_MS UINT64_C(1000000)
} const struct itimerspec ts = {
} .it_interval = {
list_add(&entry->wakeup_list, &wakeup_head); .tv_sec = 0,
.tv_nsec = 0,
},
.it_value = {
.tv_sec = delay_ms / MS_PER_S,
.tv_nsec = (delay_ms % MS_PER_S) * NS_PER_MS,
},
};
assert(!timerfd_settime(wakeup->peer.fd, 0, &ts, NULL));
wakeup->peer.event_handler = wakeup_handler;
peer_epoll_add(&wakeup->peer, EPOLLIN);
} }
#define RETRY_MIN_MS 2000 #define RETRY_MIN_MS 2000

View File

@@ -6,7 +6,5 @@ struct peer;
void wakeup_init(void); void wakeup_init(void);
void wakeup_cleanup(void); void wakeup_cleanup(void);
int wakeup_get_delay(void);
void wakeup_dispatch(void);
void wakeup_add(struct peer *, uint32_t); void wakeup_add(struct peer *, uint32_t);
uint32_t wakeup_get_retry_delay_ms(uint32_t); uint32_t wakeup_get_retry_delay_ms(uint32_t);