From 9513708ddf4eedc03c41a8d07f3dcb6e48f23c0c Mon Sep 17 00:00:00 2001 From: dimitri staessens Date: Thu, 16 Jun 2016 17:35:30 +0200 Subject: irmd: clean up stale pending flows Adds a thread that cleans up stale pending flows. Compile time option IRMD_FLOW_TIMEOUT allows setting the timeout period. Default timeout is 5 seconds. The thread wakes up 20 times per period to check the flows. --- src/irmd/main.c | 66 +++++++++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 60 insertions(+), 6 deletions(-) (limited to 'src/irmd/main.c') diff --git a/src/irmd/main.c b/src/irmd/main.c index db96b6ed..c20d63db 100644 --- a/src/irmd/main.c +++ b/src/irmd/main.c @@ -37,6 +37,7 @@ #include #include #include +#include #include "utils.h" @@ -50,13 +51,19 @@ #include #ifndef IRMD_MAX_FLOWS - #define IRMD_MAX_FLOWS 4096 +#define IRMD_MAX_FLOWS 4096 #endif #ifndef IRMD_THREADPOOL_SIZE - #define IRMD_THREADPOOL_SIZE 3 +#define IRMD_THREADPOOL_SIZE 3 #endif +#ifndef IRMD_FLOW_TIMEOUT +#define IRMD_FLOW_TIMEOUT 5000 /* ms */ +#endif + +#define IRMD_CLEANUP_TIMER ((IRMD_FLOW_TIMEOUT / 20) * MILLION) /* ns */ + struct ipcp_entry { struct list_head next; instance_name_t * api; @@ -98,6 +105,7 @@ struct port_map_entry { pthread_mutex_t res_lock; enum flow_state state; + struct timespec t0; }; struct irm { @@ -116,6 +124,8 @@ struct irm { pthread_t * threadpool; int sockfd; rw_lock_t state_lock; + + pthread_t cleanup_flows; } * instance = NULL; static struct port_map_entry * port_map_entry_create() @@ -139,6 +149,9 @@ static struct port_map_entry * port_map_entry_create() return NULL; } + e->t0.tv_sec = 0; + e->t0.tv_nsec = 0; + return e; } @@ -945,6 +958,8 @@ static struct port_map_entry * flow_alloc(pid_t pid, pme->n_pid = pid; pme->state = FLOW_PENDING; + if (clock_gettime(CLOCK_MONOTONIC, &pme->t0) < 0) + LOG_WARN("Failed to set timestamp."); rw_lock_rdlock(&instance->state_lock); rw_lock_rdlock(&instance->reg_lock); @@ -1314,6 +1329,8 @@ void irmd_sig_handler(int sig, siginfo_t * info, void * c) } + pthread_cancel(instance->cleanup_flows); + rw_lock_unlock(&instance->state_lock); case SIGPIPE: @@ -1323,6 +1340,43 @@ void irmd_sig_handler(int sig, siginfo_t * info, void * c) } } +void * irm_flow_cleaner() +{ + struct timespec now; + struct list_head * pos = NULL; + struct list_head * n = NULL; + struct timespec timeout = {IRMD_CLEANUP_TIMER / BILLION, + IRMD_CLEANUP_TIMER % BILLION}; + + while (true) { + if(clock_gettime(CLOCK_MONOTONIC, &now) < 0) + LOG_WARN("Failed to get time."); + /* cleanup stale PENDING flows */ + rw_lock_rdlock(&instance->state_lock); + + list_for_each_safe(pos, n, &(instance->port_map)) { + struct port_map_entry * e = + list_entry(pos, struct port_map_entry, next); + + pthread_mutex_lock(&e->res_lock); + + if (e->state == FLOW_PENDING && + ts_diff_ms(&e->t0, &now) > IRMD_FLOW_TIMEOUT) { + LOG_DBGF("Flow time exceeded on port ID %d.", + e->port_id); + e->state = FLOW_NULL; + pthread_cond_broadcast(&e->res_signal); + } + + pthread_mutex_unlock(&e->res_lock); + } + + rw_lock_unlock(&instance->state_lock); + + nanosleep(&timeout, NULL); + } +} + void * mainloop() { uint8_t buf[IRM_MSG_BUF_SIZE]; @@ -1579,10 +1633,8 @@ int main() if (instance == NULL) return 1; - /* - * FIXME: we need a main loop that delegates messages to subthreads in a - * way that avoids all possible deadlocks for local apps - */ + + pthread_create(&instance->cleanup_flows, NULL, irm_flow_cleaner, NULL); for (t = 0; t < IRMD_THREADPOOL_SIZE; ++t) pthread_create(&instance->threadpool[t], NULL, mainloop, NULL); @@ -1591,6 +1643,8 @@ int main() for (t = 0; t < IRMD_THREADPOOL_SIZE; ++t) pthread_join(instance->threadpool[t], NULL); + pthread_join(instance->cleanup_flows, NULL); + irm_destroy(instance); return 0; -- cgit v1.2.3