timer.c and timer.h: Use new heap and queue data

structures to re-implement timers to use memory caching and
try to fix memory leak associated with timers not being freed
This commit is contained in:
tedbullock 2007-09-12 04:47:27 +00:00
parent 04caf4ff61
commit 70c255dc95
12 changed files with 187 additions and 167 deletions

View File

@ -1,3 +1,9 @@
2007-09-11 Ted Bullock <tbullock@canada.com>
* timer.c and timer.h: Use new heap and queue data
structures to re-implement timers to use memory caching and
try to fix memory leak associated with timers not being freed
2007-09-11 Ted Bullock <tbullock@canada.com> 2007-09-11 Ted Bullock <tbullock@canada.com>
* lib/heap.c, lib/heap.h: New generic priority queue data structure * lib/heap.c, lib/heap.h: New generic priority queue data structure

View File

@ -77,7 +77,7 @@ typedef struct Conn
struct Call *sendq_tail; struct Call *sendq_tail;
struct Call *recvq; /* calls waiting for a reply */ struct Call *recvq; /* calls waiting for a reply */
struct Call *recvq_tail; struct Call *recvq_tail;
Timer *watchdog; struct Timer *watchdog;
struct struct
{ {

View File

@ -324,7 +324,7 @@ conn_failure (Conn *s, int err)
} }
static void static void
conn_timeout (Timer *t, Any_Type arg) conn_timeout (struct Timer *t, Any_Type arg)
{ {
Conn *s = arg.vp; Conn *s = arg.vp;
Time now; Time now;

View File

@ -100,7 +100,7 @@ next_arrival_time_variable (Rate_Generator *rg)
} }
static void static void
tick (Timer *t, Any_Type arg) tick (struct Timer *t, Any_Type arg)
{ {
Time delay, now = timer_now (); Time delay, now = timer_now ();
Rate_Generator *rg = arg.vp; Rate_Generator *rg = arg.vp;

View File

@ -45,7 +45,7 @@ typedef struct Rate_Generator
Time start; Time start;
Time next_time; Time next_time;
Any_Type arg; Any_Type arg;
Timer *timer; struct Timer *timer;
int (*tick) (Any_Type arg); int (*tick) (Any_Type arg);
int done; int done;
Time (*next_interarrival_time) (struct Rate_Generator *rg); Time (*next_interarrival_time) (struct Rate_Generator *rg);

View File

@ -62,7 +62,7 @@ typedef struct Sess_Private_Data
u_int num_calls_in_this_burst; /* # of calls created for this burst */ u_int num_calls_in_this_burst; /* # of calls created for this burst */
u_int num_calls_target; /* total # of calls desired */ u_int num_calls_target; /* total # of calls desired */
u_int num_calls_destroyed; /* # of calls destroyed so far */ u_int num_calls_destroyed; /* # of calls destroyed so far */
Timer *timer; /* timer for session think time */ struct Timer *timer; /* timer for session think time */
} }
Sess_Private_Data; Sess_Private_Data;
@ -106,7 +106,7 @@ issue_calls (Sess *sess, Sess_Private_Data *priv)
} }
static void static void
user_think_time_expired (Timer *t, Any_Type arg) user_think_time_expired (struct Timer *t, Any_Type arg)
{ {
Sess *sess = arg.vp; Sess *sess = arg.vp;
Sess_Private_Data *priv; Sess_Private_Data *priv;

View File

@ -135,7 +135,7 @@ struct Sess_Private_Data
u_int num_calls_in_this_burst; /* # of calls created for this burst */ u_int num_calls_in_this_burst; /* # of calls created for this burst */
u_int num_calls_target; /* total # of calls desired */ u_int num_calls_target; /* total # of calls desired */
u_int num_calls_destroyed; /* # of calls destroyed so far */ u_int num_calls_destroyed; /* # of calls destroyed so far */
Timer *timer; /* timer for session think time */ struct Timer *timer; /* timer for session think time */
int total_num_reqs; /* total number of requests in this session */ int total_num_reqs; /* total number of requests in this session */
@ -250,7 +250,7 @@ issue_calls (Sess *sess, Sess_Private_Data *priv)
} }
static void static void
user_think_time_expired (Timer *t, Any_Type arg) user_think_time_expired (struct Timer *t, Any_Type arg)
{ {
Sess *sess = arg.vp; Sess *sess = arg.vp;
Sess_Private_Data *priv; Sess_Private_Data *priv;

View File

@ -92,7 +92,7 @@ typedef struct Sess_Private_Data
u_int num_created; /* # of calls created in this burst */ u_int num_created; /* # of calls created in this burst */
u_int num_destroyed; /* # of calls destroyed in this burst */ u_int num_destroyed; /* # of calls destroyed in this burst */
u_int num_reqs_completed; /* # of user reqs completed */ u_int num_reqs_completed; /* # of user reqs completed */
Timer *timer; /* timer for session think time */ struct Timer *timer; /* timer for session think time */
struct uri_list struct uri_list
{ {
struct uri_list *next; struct uri_list *next;
@ -218,7 +218,7 @@ fetch_uri (Sess *sess, Sess_Private_Data *priv, Call_Private_Data *cpriv,
} }
static void static void
user_think_time_expired (Timer *t, Any_Type arg) user_think_time_expired (struct Timer *t, Any_Type arg)
{ {
Sess *sess = arg.vp; Sess *sess = arg.vp;
Sess_Private_Data *priv; Sess_Private_Data *priv;

View File

@ -197,7 +197,7 @@ no_op (void)
} }
static void static void
perf_sample (Timer *t, Any_Type regarg) perf_sample (struct Timer *t, Any_Type regarg)
{ {
Any_Type callarg; Any_Type callarg;
@ -1034,7 +1034,7 @@ main (int argc, char **argv)
/* Update `now'. This is to keep things accurate even when some of /* Update `now'. This is to keep things accurate even when some of
the initialization routines take a long time to execute. */ the initialization routines take a long time to execute. */
timer_tick (); timer_now_forced ();
/* ensure that clients sample rates at different times: */ /* ensure that clients sample rates at different times: */
t = (param.client.id + 1.0)*RATE_INTERVAL/param.client.num_clients; t = (param.client.id + 1.0)*RATE_INTERVAL/param.client.num_clients;
@ -1060,7 +1060,6 @@ main (int argc, char **argv)
for (i = 0; i < num_stats; ++i) for (i = 0; i < num_stats; ++i)
(*stat[i]->dump)(); (*stat[i]->dump)();
timer_reset_all();
timer_free_all(); timer_free_all();
return 0; return 0;

View File

@ -32,6 +32,7 @@
#define generic_types_h #define generic_types_h
#include <sys/types.h> #include <sys/types.h>
#include <stdbool.h>
typedef union { typedef union {
char c; char c;

View File

@ -1,7 +1,6 @@
/* /*
* httperf -- a tool for measuring web server performance Copyright 2000-2007 * Copyright (C) 2000-2007 Hewlett-Packard Company
* Hewlett-Packard Company and Contributors listed in AUTHORS file. Originally * Copyright (C) 2007 Ted Bullock <tbullock@canada.com>
* contributed by David Mosberger-Tang
* *
* This file is part of httperf, a web server performance measurment tool. * This file is part of httperf, a web server performance measurment tool.
* *
@ -39,37 +38,58 @@
#include <sys/time.h> #include <sys/time.h>
#include <generic_types.h> #include <generic_types.h>
#include <heap.h>
#include <queue.h>
#include <httperf.h> #include <httperf.h>
#include <timer.h> #include <timer.h>
#define HEAP_SIZE 4096
#define WHEEL_SIZE 4096 #define WHEEL_SIZE 4096
static Time now; static Time now;
static Time next_tick; static Time next_tick;
static Timer *timer_free_list = 0;
static Timer *t_curr = 0;
typedef struct Timer_List { struct Timer {
struct Timer_List *next; u_long delta;
struct Timer *this_timer;
} Timer_List;
static Timer_List *timer_list_head = NULL;
/* /*
* What a wheel is made of, no? * Callback function called when timer expires (timeout)
*/ */
static Timer_Queue wheel[WHEEL_SIZE], *curr = 0; Timer_Callback timeout_callback;
/*
* Typically used as a void pointer to the data object being timed
*/
Any_Type timer_subject;
};
/*
* FIFO Queue of inactive timers
*/
static struct Queue *passive_timers = NULL;
/*
* Min heap of active timers
*/
static struct Heap *active_timers = NULL;
/*
* Executed once a timer has expired, enqueues the timer back into
* the passive_timers queue for later use
*/
static void static void
done(Timer * t) done(struct Timer *t)
{ {
t->q.next = timer_free_list; /*
t->q.prev = 0; * Double cast. Irritating but does the trick
timer_free_list = t; */
enqueue((Any_Type) (void *) t, passive_timers);
} }
/*
* Returns the time and calls the syscall gettimeofday. This is an expensive
* function since it requires a context switch. Use of the cache is
* preferable with the timer_now function
*/
Time Time
timer_now_forced(void) timer_now_forced(void)
{ {
@ -79,6 +99,9 @@ timer_now_forced(void)
return tv.tv_sec + tv.tv_usec * 1e-6; return tv.tv_sec + tv.tv_usec * 1e-6;
} }
/*
* Returns the current time. If timer caching is enabled then uses the cache.
*/
Time Time
timer_now(void) timer_now(void)
{ {
@ -88,128 +111,157 @@ timer_now(void)
return timer_now_forced(); return timer_now_forced();
} }
void /*
timer_init(void) * Comparison callback function used by the heap data structure to correctly
* insert the timer in the proper order (min order as in this case).
*/
_Bool
comparator(Any_Type a, Any_Type b)
{ {
now = timer_now_forced(); struct Timer *timer_a, *timer_b;
memset(wheel, 0, sizeof(wheel));
next_tick = timer_now() + TIMER_INTERVAL; timer_a = (struct Timer *) a.vp;
curr = wheel; timer_b = (struct Timer *) b.vp;
return timer_a->delta < timer_b->delta;
} }
static void
timer_free_memory(struct Timer_List *node)
{
if (node) {
if (node->next)
timer_free_memory(node->next);
memset(node->this_timer, 0, sizeof(struct Timer));
free(node->this_timer);
memset(node, 0, sizeof(struct Timer_List));
free(node);
}
}
void
timer_reset_all(void)
{
Timer *t, *t_next;
for (t = curr->next; t; t = t_next) {
(*t->func) (t, t->arg);
t_next = t->q.next;
/* /*
* Push timer into timer_free_list for later re-use * Initializes a large timer pool cache
* This is a very expensive function. Call before beginning measurements.
* Returns 0 upon a memory allocation error
*/ */
done(t); _Bool
timer_init(void)
{
passive_timers = create_queue(WHEEL_SIZE);
if (passive_timers == NULL)
goto init_failure;
active_timers = create_heap(HEAP_SIZE, &comparator);
if (active_timers == NULL)
goto init_failure;
while (!is_queue_full(passive_timers)) {
Any_Type a;
a.vp = malloc(sizeof(struct Timer));
if (a.vp == NULL)
goto init_failure;
enqueue(a, passive_timers);
} }
timer_init(); now = timer_now_forced();
}
return true;
init_failure:
fprintf(stderr, "%s.%s: %s\n", __FILE__, __func__, strerror(errno));
return false;
}
/*
* Frees all allocated timers, and timer queues
*/
void void
timer_free_all(void) timer_free_all(void)
{ {
timer_free_memory(timer_list_head); while (!is_queue_empty(passive_timers)) {
timer_free_list = NULL; Any_Type a = get_front_and_dequeue(passive_timers);
timer_list_head = NULL; free(a.vp);
}
free_queue(passive_timers);
timer_init(); while (!is_heap_empty(active_timers)) {
Any_Type a = remove_min(active_timers);
free(a.vp);
}
free_heap(active_timers);
} }
/*
* Checks for timers which have had their timeout value pass and executes their
* callback function. The timer is then removed from the active timer list
* and then enqueued back into the passive timer queue
*/
static void
expire_complete_timers(Any_Type a)
{
struct Timer *t = (struct Timer *) a.vp;
if (t->delta == 0) {
(*t->timeout_callback) (t, t->timer_subject);
Any_Type verify = remove_min(active_timers);
if (verify.vp != t)
fprintf(stderr,
"Active timer heap is out of min order!\n\t%s.%s(%d): %s\n",
__FILE__, __func__, __LINE__, strerror(errno));
/*
* Double cast. Irritating but does the trick
*/
enqueue((Any_Type) (void *) t, passive_timers);
}
}
/*
* To be used to decrement a single timer delta value with the heap_for_each
* function via a function pointer
*/
static void
decrement_timers(Any_Type a)
{
struct Timer *t = (struct Timer *) a.vp;
if (t != 0)
t->delta--;
}
/*
* Checks for timers which have had their timeout value pass and executes their
* callback function. The timer is then removed from the active timer list
* and then enqueued back into the passive timer queue
*/
void void
timer_tick(void) timer_tick(void)
{ {
Timer *t, *t_next;
assert(!t_curr);
now = timer_now_forced(); now = timer_now_forced();
while (timer_now() >= next_tick) { while (timer_now() >= next_tick) {
/* /*
* Check for timers that have timed out and expire them * Check for timers that have timed out and expire them
*/ */
for (t = curr->next; t && t->delta == 0; t = t_next) { heap_for_each(active_timers, &expire_complete_timers);
t_curr = t;
(*t->func) (t, t->arg);
t_next = t->q.next;
/* /*
* Push timer into timer_free_list for later re-use * Decrement remaining timers
*/ */
done(t); heap_for_each(active_timers, &decrement_timers);
}
t_curr = 0;
curr->next = t;
if (t) {
t->q.prev = (Timer *) curr;
--t->delta;
}
next_tick += TIMER_INTERVAL; next_tick += TIMER_INTERVAL;
if (++curr >= wheel + WHEEL_SIZE)
curr = wheel;
} }
} }
Timer * struct Timer *
timer_schedule(Timer_Callback timeout, Any_Type arg, Time delay) timer_schedule(Timer_Callback timeout, Any_Type subject, Time delay)
{ {
Timer_Queue *spoke; struct Timer *t;
Timer *t, *p;
u_long ticks; u_long ticks;
u_long delta; u_long delta;
Time behind; Time behind;
if (timer_free_list) { if (!is_queue_empty(passive_timers)) {
t = timer_free_list; Any_Type a = get_front_and_dequeue(passive_timers);
timer_free_list = t->q.next; t = (struct Timer *) a.vp;
} else { } else
struct Timer_List *node = malloc(sizeof(struct Timer_List)); return NULL;
if (!node) {
fprintf(stderr, "%s.timer_schedule: %s\n",
prog_name, strerror(errno));
return 0;
}
t = malloc(sizeof(*t)); memset(t, 0, sizeof(struct Timer));
if (!t) { t->timeout_callback = timeout;
fprintf(stderr, "%s.timer_schedule: %s\n", t->timer_subject = subject;
prog_name, strerror(errno));
return 0;
}
node->this_timer = t;
node->next = timer_list_head;
timer_list_head = node;
}
memset(t, 0, sizeof(*t));
t->func = timeout;
t->arg = arg;
behind = (timer_now() - next_tick); behind = (timer_now() - next_tick);
if (behind > 0.0) if (behind > 0.0)
@ -224,54 +276,28 @@ timer_schedule(Timer_Callback timeout, Any_Type arg, Time delay)
ticks = 1; /* minimum delay is a tick */ ticks = 1; /* minimum delay is a tick */
} }
spoke = curr + (ticks % WHEEL_SIZE);
if (spoke >= wheel + WHEEL_SIZE)
spoke -= WHEEL_SIZE;
delta = ticks / WHEEL_SIZE; delta = ticks / WHEEL_SIZE;
p = (Timer *) spoke;
while (p->q.next && delta > p->q.next->delta) {
delta -= p->q.next->delta;
p = p->q.next;
}
t->q.next = p->q.next;
t->q.prev = p;
p->q.next = t;
t->delta = delta; t->delta = delta;
if (t->q.next) {
t->q.next->q.prev = t; insert((Any_Type) (void *) t, active_timers);
t->q.next->delta -= delta;
}
if (DBG > 2) if (DBG > 2)
fprintf(stderr, "timer_schedule: t=%p, delay=%gs, arg=%lx\n", fprintf(stderr, "timer_schedule: t=%p, delay=%gs, subject=%lx\n",
t, delay, arg.l); t, delay, subject.l);
return t; return t;
} }
void void
timer_cancel(Timer * t) timer_cancel(struct Timer *t)
{ {
if (DBG > 2) if (DBG > 2)
fprintf(stderr, "timer_cancel: t=%p\n", t); fprintf(stderr, "timer_cancel: t=%p\n", t);
assert(t->q.prev);
/* /*
* A module MUST NOT call timer_cancel() for a timer that is currently * A module MUST NOT call timer_cancel() for a timer that is currently
* being processed (whose timeout has expired). * being processed (whose timeout has expired).
*/ */
if (t_curr == t) {
fprintf(stderr,
"timer_cancel() called on currently active timer!\n");
return;
}
if (t->q.next) {
t->q.next->delta += t->delta;
t->q.next->q.prev = t->q.prev;
}
t->q.prev->q.next = t->q.next;
done(t); done(t);
} }

View File

@ -37,22 +37,10 @@
struct Timer; struct Timer;
typedef void (*Timer_Callback) (struct Timer * t, Any_Type arg); typedef void (*Timer_Callback) (struct Timer * t, Any_Type arg);
typedef struct Timer_Queue {
struct Timer *next;
struct Timer *prev;
} Timer_Queue;
typedef struct Timer {
Timer_Queue q; /* must be first member! */
u_long delta;
Timer_Callback func;
Any_Type arg;
} Timer;
extern Time timer_now_forced(void); extern Time timer_now_forced(void);
extern Time timer_now(void); extern Time timer_now(void);
extern void timer_init(void); extern _Bool timer_init(void);
extern void timer_reset_all(void); extern void timer_reset_all(void);
extern void timer_free_all(void); extern void timer_free_all(void);
/* /*
@ -60,8 +48,8 @@ extern void timer_free_all(void);
*/ */
extern void timer_tick(void); extern void timer_tick(void);
extern Timer *timer_schedule(Timer_Callback timeout, Any_Type arg, extern struct Timer *timer_schedule(Timer_Callback timeout, Any_Type arg,
Time delay); Time delay);
extern void timer_cancel(Timer * t); extern void timer_cancel(struct Timer * t);
#endif /* timer_h */ #endif /* timer_h */