pthreadpool: add pthreadpool_cancel_job()

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

3 files changed, 78 insertions, 0 deletions

diff --git a/lib/pthreadpool/pthreadpool.c b/lib/pthreadpool/pthreadpool.c
index 6c51bc5272b..55ea36ed023 100644
--- a/lib/pthreadpool/pthreadpool.c
+++ b/lib/pthreadpool/pthreadpool.c
@@ -750,3 +750,49 @@ int pthreadpool_add_job(struct pthreadpool *pool, int job_id,
 
 	return res;
 }
+
+size_t pthreadpool_cancel_job(struct pthreadpool *pool, int job_id,
+			      void (*fn)(void *private_data), void *private_data)
+{
+	int res;
+	size_t i, j;
+	size_t num = 0;
+
+	res = pthread_mutex_lock(&pool->mutex);
+	if (res != 0) {
+		return res;
+	}
+
+	for (i = 0, j = 0; i < pool->num_jobs; i++) {
+		size_t idx = (pool->head + i) % pool->jobs_array_len;
+		size_t new_idx = (pool->head + j) % pool->jobs_array_len;
+		struct pthreadpool_job *job = &pool->jobs[idx];
+
+		if ((job->private_data == private_data) &&
+		    (job->id == job_id) &&
+		    (job->fn == fn))
+		{
+			/*
+			 * Just skip the entry.
+			 */
+			num++;
+			continue;
+		}
+
+		/*
+		 * If we already removed one or more jobs (so j will be smaller
+		 * then i), we need to fill possible gaps in the logical list.
+		 */
+		if (j < i) {
+			pool->jobs[new_idx] = *job;
+		}
+		j++;
+	}
+
+	pool->num_jobs -= num;
+
+	res = pthread_mutex_unlock(&pool->mutex);
+	assert(res == 0);
+
+	return num;
+}
diff --git a/lib/pthreadpool/pthreadpool.h b/lib/pthreadpool/pthreadpool.h
index cb8baffebb1..dd1f9718b23 100644
--- a/lib/pthreadpool/pthreadpool.h
+++ b/lib/pthreadpool/pthreadpool.h
@@ -103,4 +103,30 @@ int pthreadpool_destroy(struct pthreadpool *pool);
 int pthreadpool_add_job(struct pthreadpool *pool, int job_id,
 			void (*fn)(void *private_data), void *private_data);
 
+/**
+ * @brief Try to cancel a job in a pthreadpool
+ *
+ * This tries to cancel a job in a pthreadpool. The same
+ * arguments, which were given to pthreadpool_add_job()
+ * needs to be passed.
+ *
+ * The combination of id, fn, private_data might not be unique.
+ * So the function tries to cancel as much matching jobs as possible.
+ * Note once a job is scheduled in a thread it's to late to
+ * cancel it.
+ *
+ * Canceled jobs that weren't started yet won't be reported via a
+ * pool's signal_fn.
+ *
+ * @param[in]	pool		The pool to run the job on
+ * @param[in]	job_id		A custom identifier
+ * @param[in]	fn		The function to run asynchronously
+ * @param[in]	private_data	Pointer passed to fn
+ * @return			The number of canceled jobs
+ *
+ * @see pthreadpool_add_job()
+ */
+size_t pthreadpool_cancel_job(struct pthreadpool *pool, int job_id,
+			      void (*fn)(void *private_data), void *private_data);
+
 #endif
diff --git a/lib/pthreadpool/pthreadpool_sync.c b/lib/pthreadpool/pthreadpool_sync.c
index a7dce580951..837abac54d7 100644
--- a/lib/pthreadpool/pthreadpool_sync.c
+++ b/lib/pthreadpool/pthreadpool_sync.c
@@ -71,6 +71,12 @@ int pthreadpool_add_job(struct pthreadpool *pool, int job_id,
 			       pool->signal_fn_private_data);
 }
 
+size_t pthreadpool_cancel_job(struct pthreadpool *pool, int job_id,
+			      void (*fn)(void *private_data), void *private_data)
+{
+	return 0;
+}
+
 int pthreadpool_destroy(struct pthreadpool *pool)
 {
 	free(pool);