pthreadpool: Allow multiple jobs to be received

This can avoid syscalls when multiple jobs are finished simultaneously

Signed-off-by: Volker Lendecke <vl@samba.org>
Reviewed-by: Jeremy Allison <jra@samba.org>

source3/lib/asys/asys.c

@@ -295,9 +295,9 @@ int asys_result(struct asys_context *ctx, ssize_t *pret, int *perrno,
 	struct asys_job *job;
 	int ret, jobid;
 
-	ret = pthreadpool_finished_job(ctx->pool, &jobid);
+	ret = pthreadpool_finished_jobs(ctx->pool, &jobid, 1);
-	if (ret != 0) {
+	if (ret < 0) {
-		return ret;
+		return -ret;
 	}
 	if ((jobid < 0) || (jobid >= ctx->num_jobs)) {
 		return EIO;

source3/lib/fncall.c

@@ -287,7 +287,7 @@ static void fncall_handler(struct tevent_context *ev, struct tevent_fd *fde,
 	int i, num_pending;
 	int job_id;
 
-	if (pthreadpool_finished_job(ctx->pool, &job_id) != 0) {
+	if (pthreadpool_finished_jobs(ctx->pool, &job_id, 1) < 0) {
 		return;
 	}
 

source3/lib/pthreadpool/pthreadpool.c

@@ -288,25 +288,26 @@ static void pthreadpool_join_children(struct pthreadpool *pool)
  * Fetch a finished job number from the signal pipe
  */
 
-int pthreadpool_finished_job(struct pthreadpool *pool, int *jobid)
+int pthreadpool_finished_jobs(struct pthreadpool *pool, int *jobids,
+			      unsigned num_jobids)
 {
-	int ret_jobid;
+	ssize_t to_read, nread;
-	ssize_t nread;
 
 	nread = -1;
 	errno = EINTR;
 
+	to_read = sizeof(int) * num_jobids;
+
 	while ((nread == -1) && (errno == EINTR)) {
-		nread = read(pool->sig_pipe[0], &ret_jobid, sizeof(int));
+		nread = read(pool->sig_pipe[0], jobids, to_read);
 	}
 	if (nread == -1) {
-		return errno;
+		return -errno;
 	}
-	if (nread != sizeof(int)) {
+	if ((nread % sizeof(int)) != 0) {
-		return EINVAL;
+		return -EINVAL;
 	}
-	*jobid = ret_jobid;
+	return nread / sizeof(int);
-	return 0;
 }
 
 /*

source3/lib/pthreadpool/pthreadpool.h

@@ -61,7 +61,7 @@ int pthreadpool_destroy(struct pthreadpool *pool);
  *
  * This adds a job to a pthreadpool. The job can be identified by
  * job_id. This integer will be returned from
- * pthreadpool_finished_job() then the job is completed.
+ * pthreadpool_finished_jobs() then the job is completed.
  *
  * @param[in]	pool		The pool to run the job on
  * @param[in]	job_id		A custom identifier
@@ -84,15 +84,18 @@ int pthreadpool_add_job(struct pthreadpool *pool, int job_id,
 int pthreadpool_signal_fd(struct pthreadpool *pool);
 
 /**
- * @brief Get the job_id of a finished job
+ * @brief Get the job_ids of finished jobs
  *
  * This blocks until a job has finished unless the fd returned by
  * pthreadpool_signal_fd() is readable.
  *
  * @param[in]	pool		The pool to query for finished jobs
- * @param[out]  pjobid		The job_id of the finished job
+ * @param[out]  jobids		The job_ids of the finished job
- * @return			success: 0, failure: errno
+ * @param[int]  num_jobids      The job_ids array size
+ * @return			success: >=0, number of finished jobs
+ *                              failure: -errno
  */
-int pthreadpool_finished_job(struct pthreadpool *pool, int *jobid);
+int pthreadpool_finished_jobs(struct pthreadpool *pool, int *jobids,
+			      unsigned num_jobids);
 
 #endif

source3/lib/pthreadpool/pthreadpool_sync.c

@@ -133,27 +133,35 @@ int pthreadpool_add_job(struct pthreadpool *pool, int job_id,
 
 }
 
-int pthreadpool_finished_job(struct pthreadpool *pool, int *jobid)
+int pthreadpool_finished_jobs(struct pthreadpool *pool, int *jobids,
+			      unsigned num_jobids)
 {
-	int ret_jobid;
+	ssize_t to_read, nread;
-	ssize_t nread;
+	int ret;
 
 	nread = -1;
 	errno = EINTR;
 
+	to_read = sizeof(int) * num_jobids;
+
 	while ((nread == -1) && (errno == EINTR)) {
-		nread = read(pool->sig_pipe[0], &ret_jobid, sizeof(int));
+		nread = read(pool->sig_pipe[0], jobids, to_read);
 	}
 	if (nread == -1) {
-		return errno;
+		return -errno;
 	}
-	if (nread != sizeof(int)) {
+	if ((nread % sizeof(int)) != 0) {
-		return EINVAL;
+		return -EINVAL;
 	}
-	*jobid = ret_jobid;
 
 	pool->pipe_busy = 0;
-	return pthreadpool_write_to_pipe(pool);
+
+	ret = pthreadpool_write_to_pipe(pool);
+	if (ret != 0) {
+		return -ret;
+	}
+
+	return nread / sizeof(int);
 }
 
 int pthreadpool_destroy(struct pthreadpool *pool)

source3/lib/pthreadpool/tests.c

@@ -71,8 +71,8 @@ static int test_jobs(int num_threads, int num_jobs)
 
 	for (i=0; i<num_jobs; i++) {
 		int jobid = -1;
-		ret = pthreadpool_finished_job(p, &jobid);
+		ret = pthreadpool_finished_jobs(p, &jobid, 1);
-		if ((ret != 0) || (jobid >= num_jobs)) {
+		if ((ret != 1) || (jobid >= num_jobs)) {
 			fprintf(stderr, "invalid job number %d\n", jobid);
 			return -1;
 		}
@@ -284,8 +284,8 @@ static int test_threaded_addjob(int num_pools, int num_threads, int poolsize,
 				continue;
 			}
 
-			ret = pthreadpool_finished_job(pools[j], &jobid);
+			ret = pthreadpool_finished_jobs(pools[j], &jobid, 1);
-			if ((ret != 0) || (jobid >= num_jobs * num_threads)) {
+			if ((ret != 1) || (jobid >= num_jobs * num_threads)) {
 				fprintf(stderr, "invalid job number %d\n",
 					jobid);
 				return -1;

source3/modules/vfs_aio_pthread.c

@@ -166,8 +166,8 @@ static void aio_open_handle_completion(struct tevent_context *event_ctx,
 		return;
 	}
 
-	ret = pthreadpool_finished_job(open_pool, &jobid);
+	ret = pthreadpool_finished_jobs(open_pool, &jobid, 1);
-	if (ret) {
+	if (ret != 1) {
 		smb_panic("aio_open_handle_completion");
 		/* notreached. */
 		return;

source3/torture/bench_pthreadpool.c

@@ -50,15 +50,15 @@ bool run_bench_pthreadpool(int dummy)
 				  strerror(ret));
 			break;
 		}
-		ret = pthreadpool_finished_job(pool, &jobid);
+		ret = pthreadpool_finished_jobs(pool, &jobid, 1);
-		if (ret != 0) {
+		if (ret < 0) {
 			d_fprintf(stderr, "pthreadpool_finished_job failed: %s\n",
-				  strerror(ret));
+				  strerror(-ret));
 			break;
 		}
 	}
 
 	pthreadpool_destroy(pool);
 
-	return (ret == 0);
+	return (ret == 1);
 }