pthreadpool: maintain a global list of orphaned pthreadpool_tevent_jobs

Instead of leaking the memory forever, we retry the cleanup,
if other pthreadpool_tevent_*() functions are used.

pthreadpool_tevent_cleanup_orphaned_jobs() could also be called
by external callers.

Signed-off-by: Stefan Metzmacher <metze@samba.org>
Reviewed-by: Ralph Boehme <slow@samba.org>

lib/pthreadpool/pthreadpool_tevent.c

@@ -81,6 +81,24 @@ static int pthreadpool_tevent_destructor(struct pthreadpool_tevent *pool);
 
 static void pthreadpool_tevent_job_orphan(struct pthreadpool_tevent_job *job);
 
+static struct pthreadpool_tevent_job *orphaned_jobs;
+
+void pthreadpool_tevent_cleanup_orphaned_jobs(void)
+{
+	struct pthreadpool_tevent_job *job = NULL;
+	struct pthreadpool_tevent_job *njob = NULL;
+
+	for (job = orphaned_jobs; job != NULL; job = njob) {
+		njob = job->next;
+
+		/*
+		 * The job destructor keeps the job alive
+		 * (and in the list) or removes it from the list.
+		 */
+		TALLOC_FREE(job);
+	}
+}
+
 static int pthreadpool_tevent_job_signal(int jobid,
 					 void (*job_fn)(void *private_data),
 					 void *job_private_data,
@@ -92,6 +110,8 @@ int pthreadpool_tevent_init(TALLOC_CTX *mem_ctx, unsigned max_threads,
 	struct pthreadpool_tevent *pool;
 	int ret;
 
+	pthreadpool_tevent_cleanup_orphaned_jobs();
+
 	pool = talloc_zero(mem_ctx, struct pthreadpool_tevent);
 	if (pool == NULL) {
 		return ENOMEM;
@@ -164,6 +184,8 @@ static int pthreadpool_tevent_destructor(struct pthreadpool_tevent *pool)
 	}
 	pool->pool = NULL;
 
+	pthreadpool_tevent_cleanup_orphaned_jobs();
+
 	return 0;
 }
 
@@ -317,10 +339,18 @@ static int pthreadpool_tevent_job_destructor(struct pthreadpool_tevent_job *job)
 		/*
 		 * state->im still there means, we need to wait for the
 		 * immediate event to be triggered or just leak the memory.
+		 *
+		 * Move it to the orphaned list, if it's not already there.
 		 */
 		return -1;
 	}
 
+	/*
+	 * Finally remove from the orphaned_jobs list
+	 * and let talloc destroy us.
+	 */
+	DLIST_REMOVE(orphaned_jobs, job);
+
 	return 0;
 }
 
@@ -362,6 +392,15 @@ static void pthreadpool_tevent_job_orphan(struct pthreadpool_tevent_job *job)
 	 */
 	DLIST_REMOVE(job->pool->jobs, job);
 
+	/*
+	 * Add it to the list of orphaned jobs,
+	 * which may be cleaned up later.
+	 *
+	 * The destructor removes it from the list
+	 * when possible or it denies the free
+	 * and keep it in the list.
+	 */
+	DLIST_ADD_END(orphaned_jobs, job);
 	TALLOC_FREE(job);
 }
 
@@ -400,6 +439,8 @@ struct tevent_req *pthreadpool_tevent_job_send(
 	struct pthreadpool_tevent_job *job = NULL;
 	int ret;
 
+	pthreadpool_tevent_cleanup_orphaned_jobs();
+
 	req = tevent_req_create(mem_ctx, &state,
 				struct pthreadpool_tevent_job_state);
 	if (req == NULL) {

lib/pthreadpool/pthreadpool_tevent.h

@@ -22,6 +22,8 @@
 
 #include <tevent.h>
 
+void pthreadpool_tevent_cleanup_orphaned_jobs(void);
+
 struct pthreadpool_tevent;
 
 int pthreadpool_tevent_init(TALLOC_CTX *mem_ctx, unsigned max_threads,