From 107ab48debc87ee97d9053a112aad9a614012655 Mon Sep 17 00:00:00 2001 From: Philippe PITTOLI Date: Wed, 14 Sep 2016 00:20:04 +0200 Subject: [PATCH] pubsubd: memory leak gone, kind of a problem on pubsubc pubsubc has a problem when listening for the first time on his own message, don't know where it comes from --- lib/pubsubd.c | 2 +- pubsub/pubsubd.c | 130 +++++++++++++++++++++++++++++++++++++++++------ 2 files changed, 115 insertions(+), 17 deletions(-) diff --git a/lib/pubsubd.c b/lib/pubsubd.c index 7b09aad..013f3cc 100644 --- a/lib/pubsubd.c +++ b/lib/pubsubd.c @@ -15,7 +15,7 @@ pubsubd_channels_add (struct channels *chans, const char *chan) return NULL; } - struct channel *n = malloc (sizeof (struct channel));; + struct channel *n = malloc (sizeof (struct channel)); memset (n, 0, sizeof (struct channel)); pubsubd_channel_new (n, chan); diff --git a/pubsub/pubsubd.c b/pubsub/pubsubd.c index d18e892..b49c6cc 100644 --- a/pubsub/pubsubd.c +++ b/pubsub/pubsubd.c @@ -10,13 +10,98 @@ ohshit(int rvalue, const char* str) { exit(rvalue); } -// give this structure to the thread worker function -struct worker_params { +// head of the list +LIST_HEAD(workers, worker); + +struct workers *my_workers; + +// element of the list +// worker : process to handle (threaded) +struct worker { struct channels *chans; struct channel *chan; struct app_list_elm *ale; + LIST_ENTRY(worker) entries; }; +void pubsubd_worker_free (struct worker * w); +struct worker * pubsubd_worker_get (struct workers *wrkrs, struct worker *w); +int pubsubd_worker_eq (const struct worker *w1, const struct worker *w2); + +void pubsubd_workers_init (struct workers *wrkrs) { LIST_INIT(wrkrs); } + +struct worker * +pubsubd_workers_add (struct workers *wrkrs, const struct worker *w) +{ + if (wrkrs == NULL || w == NULL) { + printf ("pubsubd_workers_add: wrkrs == NULL or w == NULL"); + return NULL; + } + + struct worker *n = malloc (sizeof (struct worker)); + memset (n, 0, sizeof (struct worker)); + memcpy (n, w, sizeof (struct worker)); + if (w->ale != NULL) + n->ale = pubsubd_app_list_elm_copy (w->ale); + + LIST_INSERT_HEAD(wrkrs, n, entries); + + return n; +} + +void pubsubd_worker_del (struct workers *wrkrs, struct worker *w) +{ + struct worker *todel = pubsubd_worker_get (wrkrs, w); + if(todel != NULL) { + LIST_REMOVE(todel, entries); + pubsubd_worker_free (todel); + free (todel); + todel = NULL; + } +} + +void pubsubd_workers_del_all (struct workers *wrkrs) +{ + if (!wrkrs) + return; + + struct worker *w = NULL; + + while (!LIST_EMPTY(wrkrs)) { + printf ("KILL THE WORKERS : %p\n", w); + w = LIST_FIRST(wrkrs); + LIST_REMOVE(w, entries); + pubsubd_worker_free (w); + free (w); + w = NULL; + } +} + +void pubsubd_worker_free (struct worker * w) +{ + if (w == NULL) + return; + pubsubd_app_list_elm_free (w->ale); + free (w->ale); + w->ale = NULL; +} + +struct worker * pubsubd_worker_get (struct workers *wrkrs, struct worker *w) +{ + struct worker * np = NULL; + LIST_FOREACH(np, wrkrs, entries) { + if (pubsubd_worker_eq (np, w)) + return np; + } + return NULL; +} + +int pubsubd_worker_eq (const struct worker *w1, const struct worker *w2) +{ + return w1 == w2; // if it's the same pointer +} + +// a thread for each connected process void * pubsubd_worker_thread (void *params) { int s = 0; @@ -25,17 +110,15 @@ void * pubsubd_worker_thread (void *params) if (s != 0) printf ("pthread_setcancelstate: %d\n", s); - struct worker_params *wp = (struct worker_params *) params; - if (wp == NULL) { + struct worker *w = (struct worker *) params; + if (w == NULL) { fprintf (stderr, "error pubsubd_worker_thread : params NULL\n"); return NULL; } - struct channels *chans = wp->chans; - struct channel *chan = wp->chan; - struct app_list_elm *ale = wp->ale; - - free (wp); + struct channels *chans = w->chans; + struct channel *chan = w->chan; + struct app_list_elm *ale = w->ale; // main loop while (1) { @@ -68,6 +151,10 @@ void * pubsubd_worker_thread (void *params) } pubsubd_app_list_elm_free (ale); + free (w->ale); + w->ale = NULL; + + pubsubd_worker_del (my_workers, w); pthread_exit (NULL); } @@ -93,6 +180,10 @@ main(int argc, char **argv, char **env) thr = malloc (sizeof (pthread_t)); memset (thr, 0, sizeof (pthread_t)); + my_workers = malloc (sizeof (struct workers)); + memset (my_workers, 0, sizeof (struct workers)); + pubsubd_workers_init (my_workers); + int i = 0; // for (i = 0; i < NB_CLIENTS; i++) for (i = 0; ; i++) { @@ -109,12 +200,16 @@ main(int argc, char **argv, char **env) break; } - // TODO thread to handle multiple clients at a time - struct worker_params *wp = NULL; - wp = malloc (sizeof (struct worker_params)); - wp->ale = pubsubd_app_list_elm_copy (&ale); - wp->chans = &chans; - wp->chan = chan; + // thread to handle multiple clients at a time + struct worker *w = NULL; + w = malloc (sizeof (struct worker)); + w->ale = pubsubd_app_list_elm_copy (&ale); + w->chans = &chans; + w->chan = chan; + struct worker *wtmp = pubsubd_workers_add (my_workers, w); + pubsubd_worker_free (w); + free (w); + w = wtmp; // realloc memory for further workers pthread_t * tmpthr = realloc (thr, sizeof (pthread_t) * (i+1)); @@ -128,7 +223,7 @@ main(int argc, char **argv, char **env) } memset (thr + i, 0, sizeof (pthread_t)); - pthread_create (thr + i, NULL, pubsubd_worker_thread, wp); + pthread_create (thr + i, NULL, pubsubd_worker_thread, w); pthread_detach (thr[i]); pubsubd_app_list_elm_free (&ale); @@ -147,6 +242,9 @@ main(int argc, char **argv, char **env) free (thr); pubsubd_channels_del_all (&chans); + pubsubd_workers_del_all (my_workers); + + free (my_workers); // the application will shut down, and remove the service named pipe if (srv_close (&srv))