From 7efe2871fcd0e3bebcebd96618ff7fcbedd46a2c Mon Sep 17 00:00:00 2001 From: Tom Smeding Date: Thu, 2 Aug 2018 19:02:19 +0200 Subject: Keepalive pings in icmpd --- clientd.c | 4 +++- icmpd.c | 67 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++------- util.c | 7 +++++++ util.h | 3 ++- 4 files changed, 72 insertions(+), 9 deletions(-) diff --git a/clientd.c b/clientd.c index 2462669..0f8bde1 100644 --- a/clientd.c +++ b/clientd.c @@ -14,7 +14,9 @@ #include "util.h" int main(void) { - struct icmpd *d = icmpd_create_client(inet_addr("127.0.0.1")); + const char *remote_addr = "127.0.0.1"; + // const char *remote_addr = "198.211.118.67"; + struct icmpd *d = icmpd_create_client(inet_addr(remote_addr)); int d_fd = icmpd_get_select_fd(d); while (true) { diff --git a/icmpd.c b/icmpd.c index 1cfc051..fc47daa 100644 --- a/icmpd.c +++ b/icmpd.c @@ -2,6 +2,7 @@ #include #include #include +#include #include #include #include @@ -13,6 +14,9 @@ #include "icmp_client.h" +#define KEEPALIVE_DELAY 5000 // maximum time to send nothing (milliseconds) + + struct icmpd { // CONSTANTS @@ -31,6 +35,10 @@ struct icmpd { // server: seqnum is the sequence number of the last-received ECHO // value is -1 if not set int seqnum; + + // client: timestamp of when last message was sent + // value is -1 if no message sent yet + int64_t last_send_stamp; }; // Global state @@ -137,7 +145,7 @@ static void* thread_entry(void *arg) { size_t recvqu_cap = 8, recvqu_len = 0; struct recvqu_item *recvqu = malloc(recvqu_cap * sizeof(struct recvqu_item)); - // Server messages submitted for transmission while no non-ponged client + // Server messages submitted for transmission while no outstanding client // ping is available. size_t sendqu_cap = 8, sendqu_len = 0; struct sendqu_item *sendqu = malloc(sendqu_cap * sizeof(struct sendqu_item)); @@ -149,17 +157,33 @@ static void* thread_entry(void *arg) { if (sock_server >= 0) FD_SET(sock_server, &inset); if (sock_client >= 0) FD_SET(sock_client, &inset); - int nfds = thread_in; - if (sock_server > nfds) nfds = sock_server; - if (sock_client > nfds) nfds = sock_client; + int nfds = maxi(maxi(thread_in, sock_server), sock_client) + 1; + + int64_t now = gettimestamp(); + int64_t wait_interval = INT64_MAX; // microseconds + + for (size_t i = 0; i < conns_len; i++) { + if (conns[i]->isserver) continue; + + mt_mutex_lock(&conns[i]->mut); + int64_t stamp = conns[i]->last_send_stamp; + mt_mutex_unlock(&conns[i]->mut); - int ret = select(nfds + 1, &inset, NULL, NULL, NULL); + if (stamp + KEEPALIVE_DELAY * 1000 - now < wait_interval) { + wait_interval = stamp + KEEPALIVE_DELAY * 1000 - now; + } + } + + struct timeval timeout_tv; + timeout_tv.tv_sec = wait_interval / 1000000; + timeout_tv.tv_usec = wait_interval % 1000000; + + int ret = select(nfds, &inset, NULL, NULL, wait_interval == INT64_MAX ? NULL : &timeout_tv); if (ret < 0) { if (errno == EINTR) continue; perror("select"); assert(false); } - assert(ret > 0); if (FD_ISSET(thread_in, &inset)) { struct msg_in msg = recv_message(thread_in); @@ -293,6 +317,7 @@ static void* thread_entry(void *arg) { mt_mutex_lock(&d->mut); client_increment_seqnum(d); int seqnum = d->seqnum; + d->last_send_stamp = gettimestamp(); mt_mutex_unlock(&d->mut); int ret = icmp_client_send( @@ -403,6 +428,7 @@ static void* thread_entry(void *arg) { mt_mutex_lock(&d->mut); client_increment_seqnum(d); int seqnum = d->seqnum; + d->last_send_stamp = gettimestamp(); mt_mutex_unlock(&d->mut); int ret = icmp_client_send(sock_client, d->other_addr, seqnum, NULL, 0); @@ -411,6 +437,29 @@ static void* thread_entry(void *arg) { } } } + + now = gettimestamp(); + + for (size_t i = 0; i < conns_len; i++) { + if (conns[i]->isserver) continue; + + mt_mutex_lock(&conns[i]->mut); + int64_t stamp = conns[i]->last_send_stamp; + mt_mutex_unlock(&conns[i]->mut); + + if (now - stamp >= KEEPALIVE_DELAY) { + mt_mutex_lock(&conns[i]->mut); + client_increment_seqnum(conns[i]); + int seqnum = conns[i]->seqnum; + conns[i]->last_send_stamp = gettimestamp(); + mt_mutex_unlock(&conns[i]->mut); + + int ret = icmp_client_send(sock_client, conns[i]->other_addr, seqnum, NULL, 0); + if (ret < 0) { + perror("icmp_client_send"); + } + } + } } return NULL; @@ -449,6 +498,7 @@ static struct icmpd* icmpd_create_base(int id, bool isserver, uint32_t other_add d->id = id; d->seqnum = -1; d->outstanding = false; + d->last_send_stamp = gettimestamp(); int pp[2]; assert(pipe(pp) == 0); @@ -483,7 +533,7 @@ void icmpd_server_set_outstanding(struct icmpd *d, int seqnum) { } bool icmpd_peek(struct icmpd *d) { -#if 0 +#if 1 send_message(host_out, MSG_PEEK, &d, sizeof d); struct msg_in msg = recv_message(host_in); assert(msg.hdr.type == MSG_PEEK_ANS); @@ -492,6 +542,9 @@ bool icmpd_peek(struct icmpd *d) { free(msg.data); return ret; #else + // This version has the same fallacy as just selecting on icmpd_get_select_fd() directly, + // i.e. that select(2) can return when there's nothing to read + fd_set inset; FD_ZERO(&inset); FD_SET(d->signal_out, &inset); diff --git a/util.c b/util.c index 0f3f6bc..9c7e1a3 100644 --- a/util.c +++ b/util.c @@ -4,6 +4,7 @@ #include #include #include +#include #include "util.h" @@ -69,3 +70,9 @@ ssize_t writeall(int fd, const void *data, size_t length) { int maxi(int a, int b) { return a > b ? a : b; } + +int64_t gettimestamp(void) { + struct timeval tv; + gettimeofday(&tv, NULL); + return tv.tv_sec * 1000000LL + tv.tv_usec; +} diff --git a/util.h b/util.h index 3e228f9..88ad996 100644 --- a/util.h +++ b/util.h @@ -1,6 +1,7 @@ #pragma once #include +#include #include @@ -8,5 +9,5 @@ int uniqid(void); void xxd(const void *buf, size_t length); ssize_t readall(int fd, void *data, size_t length); ssize_t writeall(int fd, const void *data, size_t length); - int maxi(int a, int b); +int64_t gettimestamp(void); -- cgit v1.2.3