s3:vfs_aio_pthread: Convert to libasys

Signed-off-by: Jeremy Allison <jra@samba.org>

1 files changed, 122 insertions, 524 deletions

diff --git a/source3/modules/vfs_aio_pthread.c b/source3/modules/vfs_aio_pthread.c
index ae5963b768..06ac8b8667 100644
--- a/source3/modules/vfs_aio_pthread.c
+++ b/source3/modules/vfs_aio_pthread.c
@@ -27,590 +27,168 @@
 #include "smbd/smbd.h"
 #include "smbd/globals.h"
 #include "lib/pthreadpool/pthreadpool.h"
+#include "lib/asys/asys.h"
+#include "lib/util/tevent_unix.h"
 #ifdef HAVE_LINUX_FALLOC_H
 #include <linux/falloc.h>
 #endif
 
-struct aio_extra;
-static struct pthreadpool *pool;
-static int aio_pthread_jobid;
+static struct asys_context *asys_ctx;
+struct tevent_fd *asys_fde;
 
-struct aio_private_data {
-	struct aio_private_data *prev, *next;
-	int jobid;
-	SMB_STRUCT_AIOCB *aiocb;
-	ssize_t ret_size;
-	int ret_errno;
-	bool cancelled;
-	bool write_command;
-	bool flush_write;
+struct aio_pthread_state {
+	struct tevent_req *req;
+	ssize_t ret;
+	int err;
 };
 
-/* List of outstanding requests we have. */
-static struct aio_private_data *pd_list;
-
-static void aio_pthread_handle_completion(struct event_context *event_ctx,
-				struct fd_event *event,
-				uint16 flags,
-				void *p);
-
-
-/************************************************************************
- Ensure thread pool is initialized.
-***********************************************************************/
-
-static bool init_aio_threadpool(struct event_context *ev_ctx,
-				struct pthreadpool **pp_pool,
-				void (*completion_fn)(struct event_context *,
-						struct fd_event *,
-						uint16,
-						void *))
-{
-	struct fd_event *sock_event = NULL;
-	int ret = 0;
-
-	if (*pp_pool) {
-		return true;
-	}
-
-	ret = pthreadpool_init(aio_pending_size, pp_pool);
-	if (ret) {
-		errno = ret;
-		return false;
-	}
-	sock_event = tevent_add_fd(ev_ctx,
-				NULL,
-				pthreadpool_signal_fd(*pp_pool),
-				TEVENT_FD_READ,
-				completion_fn,
-				NULL);
-	if (sock_event == NULL) {
-		pthreadpool_destroy(*pp_pool);
-		*pp_pool = NULL;
-		return false;
-	}
-
-	DEBUG(10,("init_aio_threadpool: initialized with up to %d threads\n",
-		  aio_pending_size));
-
-	return true;
-}
-
-
-/************************************************************************
- Worker function - core of the pthread aio engine.
- This is the function that actually does the IO.
-***********************************************************************/
-
-static void aio_worker(void *private_data)
-{
-	struct aio_private_data *pd =
-			(struct aio_private_data *)private_data;
-
-	if (pd->write_command) {
-		pd->ret_size = sys_pwrite(pd->aiocb->aio_fildes,
-				(const void *)pd->aiocb->aio_buf,
-				pd->aiocb->aio_nbytes,
-				pd->aiocb->aio_offset);
-		if (pd->ret_size == -1 && errno == ESPIPE) {
-			/* Maintain the fiction that pipes can
-			   be seeked (sought?) on. */
-			pd->ret_size = sys_write(pd->aiocb->aio_fildes,
-					(const void *)pd->aiocb->aio_buf,
-					pd->aiocb->aio_nbytes);
-		}
-#if defined(HAVE_FSYNC)
-		if (pd->ret_size != -1 && pd->flush_write) {
-			/*
-			 * Optimization - flush if requested.
-			 * Ignore error as upper layer will
-			 * also do this.
-			 */
-			(void)fsync(pd->aiocb->aio_fildes);
-		}
-#endif
-	} else {
-		pd->ret_size = sys_pread(pd->aiocb->aio_fildes,
-				(void *)pd->aiocb->aio_buf,
-				pd->aiocb->aio_nbytes,
-				pd->aiocb->aio_offset);
-		if (pd->ret_size == -1 && errno == ESPIPE) {
-			/* Maintain the fiction that pipes can
-			   be seeked (sought?) on. */
-			pd->ret_size = sys_read(pd->aiocb->aio_fildes,
-					(void *)pd->aiocb->aio_buf,
-					pd->aiocb->aio_nbytes);
-		}
-	}
-	if (pd->ret_size == -1) {
-		pd->ret_errno = errno;
-	} else {
-		pd->ret_errno = 0;
-	}
-}
-
-/************************************************************************
- Private data destructor.
-***********************************************************************/
-
-static int pd_destructor(struct aio_private_data *pd)
+static int aio_pthread_state_destructor(struct aio_pthread_state *s)
 {
-	DLIST_REMOVE(pd_list, pd);
+	asys_cancel(asys_ctx, s->req);
 	return 0;
 }
 
-/************************************************************************
- Create and initialize a private data struct.
-***********************************************************************/
-
-static struct aio_private_data *create_private_data(TALLOC_CTX *ctx,
-					SMB_STRUCT_AIOCB *aiocb)
+static struct tevent_req *aio_pthread_pread_send(
+	struct vfs_handle_struct *handle,
+	TALLOC_CTX *mem_ctx, struct tevent_context *ev,
+	struct files_struct *fsp, void *data, size_t n, off_t offset)
 {
-	struct aio_private_data *pd = talloc_zero(ctx, struct aio_private_data);
-	if (!pd) {
-		return NULL;
-	}
-	pd->jobid = aio_pthread_jobid++;
-	pd->aiocb = aiocb;
-	pd->ret_size = -1;
-	pd->ret_errno = EINPROGRESS;
-	talloc_set_destructor(pd, pd_destructor);
-	DLIST_ADD_END(pd_list, pd, struct aio_private_data *);
-	return pd;
-}
-
-/************************************************************************
- Spin off a threadpool (if needed) and initiate a pread call.
-***********************************************************************/
-
-static int aio_pthread_read(struct vfs_handle_struct *handle,
-				struct files_struct *fsp,
-				SMB_STRUCT_AIOCB *aiocb)
-{
-	struct aio_extra *aio_ex = (struct aio_extra *)aiocb->aio_sigevent.sigev_value.sival_ptr;
-	struct aio_private_data *pd = NULL;
+	struct tevent_req *req;
+	struct aio_pthread_state *state;
 	int ret;
 
-	if (!init_aio_threadpool(handle->conn->sconn->ev_ctx,
-				&pool,
-				aio_pthread_handle_completion)) {
-		return -1;
-	}
-
-	pd = create_private_data(aio_ex, aiocb);
-	if (pd == NULL) {
-		DEBUG(10, ("aio_pthread_read: Could not create private data.\n"));
-		return -1;
+	req = tevent_req_create(mem_ctx, &state, struct aio_pthread_state);
+	if (req == NULL) {
+		return NULL;
 	}
+	state->req = req;
 
-	ret = pthreadpool_add_job(pool, pd->jobid, aio_worker, (void *)pd);
-	if (ret) {
-		errno = ret;
-		return -1;
+	ret = asys_pread(asys_ctx, fsp->fh->fd, data, n, offset, req);
+	if (ret != 0) {
+		tevent_req_error(req, ret);
+		return tevent_req_post(req, ev);
 	}
+	talloc_set_destructor(state, aio_pthread_state_destructor);
 
-	DEBUG(10, ("aio_pthread_read: jobid=%d pread requested "
-		"of %llu bytes at offset %llu\n",
-		pd->jobid,
-		(unsigned long long)pd->aiocb->aio_nbytes,
-		(unsigned long long)pd->aiocb->aio_offset));
-
-	return 0;
+	return req;
 }
 
-/************************************************************************
- Spin off a threadpool (if needed) and initiate a pwrite call.
-***********************************************************************/
-
-static int aio_pthread_write(struct vfs_handle_struct *handle,
-				struct files_struct *fsp,
-				SMB_STRUCT_AIOCB *aiocb)
+static struct tevent_req *aio_pthread_pwrite_send(
+	struct vfs_handle_struct *handle,
+	TALLOC_CTX *mem_ctx, struct tevent_context *ev,
+	struct files_struct *fsp, const void *data, size_t n, off_t offset)
 {
-	struct aio_extra *aio_ex = (struct aio_extra *)aiocb->aio_sigevent.sigev_value.sival_ptr;
-	struct aio_private_data *pd = NULL;
+	struct tevent_req *req;
+	struct aio_pthread_state *state;
 	int ret;
 
-	if (!init_aio_threadpool(handle->conn->sconn->ev_ctx,
-				&pool,
-				aio_pthread_handle_completion)) {
-		return -1;
-	}
-
-	pd = create_private_data(aio_ex, aiocb);
-	if (pd == NULL) {
-		DEBUG(10, ("aio_pthread_write: Could not create private data.\n"));
-		return -1;
-	}
-
-	pd->write_command = true;
-	if (lp_strict_sync(SNUM(fsp->conn)) &&
-			(lp_syncalways(SNUM(fsp->conn)) ||
-				aio_write_through_requested(aio_ex))) {
-		pd->flush_write = true;
-	}
-
-
-	ret = pthreadpool_add_job(pool, pd->jobid, aio_worker, (void *)pd);
-	if (ret) {
-		errno = ret;
-		return -1;
+	req = tevent_req_create(mem_ctx, &state, struct aio_pthread_state);
+	if (req == NULL) {
+		return NULL;
 	}
+	state->req = req;
 
-	DEBUG(10, ("aio_pthread_write: jobid=%d pwrite requested "
-		"of %llu bytes at offset %llu\n",
-		pd->jobid,
-		(unsigned long long)pd->aiocb->aio_nbytes,
-		(unsigned long long)pd->aiocb->aio_offset));
-
-	return 0;
-}
-
-/************************************************************************
- Find the private data by jobid.
-***********************************************************************/
-
-static struct aio_private_data *find_private_data_by_jobid(int jobid)
-{
-	struct aio_private_data *pd;
-
-	for (pd = pd_list; pd != NULL; pd = pd->next) {
-		if (pd->jobid == jobid) {
-			return pd;
-		}
+	ret = asys_pwrite(asys_ctx, fsp->fh->fd, data, n, offset, req);
+	if (ret != 0) {
+		tevent_req_error(req, ret);
+		return tevent_req_post(req, ev);
 	}
+	talloc_set_destructor(state, aio_pthread_state_destructor);
 
-	return NULL;
+	return req;
 }
 
-/************************************************************************
- Callback when an IO completes.
-***********************************************************************/
-
-static void aio_pthread_handle_completion(struct event_context *event_ctx,
-				struct fd_event *event,
-				uint16 flags,
-				void *p)
+static void aio_pthread_finished(struct tevent_context *ev,
+				 struct tevent_fd *fde,
+				 uint16_t flags, void *p)
 {
-	struct aio_extra *aio_ex = NULL;
-	struct aio_private_data *pd = NULL;
-	int jobid = 0;
-	int ret;
-
-	DEBUG(10, ("aio_pthread_handle_completion called with flags=%d\n",
-			(int)flags));
-
-	if ((flags & EVENT_FD_READ) == 0) {
+	struct tevent_req *req;
+	struct aio_pthread_state *state;
+	int res;
+	ssize_t ret;
+	int err;
+	void *private_data;
+
+	if ((flags & TEVENT_FD_READ) == 0) {
 		return;
 	}
 
-	ret = pthreadpool_finished_job(pool, &jobid);
-	if (ret) {
-		smb_panic("aio_pthread_handle_completion");
+	res = asys_result(asys_ctx, &ret, &err, &private_data);
+	if (res == ECANCELED) {
 		return;
 	}
 
-	pd = find_private_data_by_jobid(jobid);
-	if (pd == NULL) {
-		DEBUG(1, ("aio_pthread_handle_completion cannot find jobid %d\n",
-			  jobid));
+	if (res != 0) {
+		DEBUG(1, ("asys_result returned %s\n", strerror(res)));
 		return;
 	}
 
-	aio_ex = (struct aio_extra *)pd->aiocb->aio_sigevent.sigev_value.sival_ptr;
-	smbd_aio_complete_aio_ex(aio_ex);
+	req = talloc_get_type_abort(private_data, struct tevent_req);
+	state = tevent_req_data(req, struct aio_pthread_state);
 
-	DEBUG(10,("aio_pthread_handle_completion: jobid %d completed\n",
-		jobid ));
-	TALLOC_FREE(aio_ex);
-}
+	talloc_set_destructor(state, NULL);
 
-/************************************************************************
- Find the private data by aiocb.
-***********************************************************************/
-
-static struct aio_private_data *find_private_data_by_aiocb(SMB_STRUCT_AIOCB *aiocb)
-{
-	struct aio_private_data *pd;
-
-	for (pd = pd_list; pd != NULL; pd = pd->next) {
-		if (pd->aiocb == aiocb) {
-			return pd;
-		}
-	}
-
-	return NULL;
+	state->ret = ret;
+	state->err = err;
+	tevent_req_done(req);
 }
 
-/************************************************************************
- Called to return the result of a completed AIO.
- Should only be called if aio_error returns something other than EINPROGRESS.
- Returns:
-	Any other value - return from IO operation.
-***********************************************************************/
-
-static ssize_t aio_pthread_return_fn(struct vfs_handle_struct *handle,
-				struct files_struct *fsp,
-				SMB_STRUCT_AIOCB *aiocb)
+static ssize_t aio_pthread_recv(struct tevent_req *req, int *err)
 {
-	struct aio_private_data *pd = find_private_data_by_aiocb(aiocb);
-
-	if (pd == NULL) {
-		errno = EINVAL;
-		DEBUG(0, ("aio_pthread_return_fn: returning EINVAL\n"));
-		return -1;
-	}
+	struct aio_pthread_state *state = tevent_req_data(
+		req, struct aio_pthread_state);
 
-	pd->aiocb = NULL;
-
-	if (pd->cancelled) {
-		errno = ECANCELED;
+	if (tevent_req_is_unix_error(req, err)) {
 		return -1;
 	}
-
-	if (pd->ret_size == -1) {
-		errno = pd->ret_errno;
-	}
-
-	return pd->ret_size;
+	*err = state->err;
+	return state->ret;
 }
 
-/************************************************************************
- Called to check the result of an AIO.
- Returns:
-	EINPROGRESS - still in progress.
-	EINVAL - invalid aiocb.
-	ECANCELED - request was cancelled.
-	0 - request completed successfully.
-	Any other value - errno from IO operation.
-***********************************************************************/
-
-static int aio_pthread_error_fn(struct vfs_handle_struct *handle,
-			     struct files_struct *fsp,
-			     SMB_STRUCT_AIOCB *aiocb)
-{
-	struct aio_private_data *pd = find_private_data_by_aiocb(aiocb);
-
-	if (pd == NULL) {
-		return EINVAL;
-	}
-	if (pd->cancelled) {
-		return ECANCELED;
-	}
-	return pd->ret_errno;
-}
 
-/************************************************************************
- Called to request the cancel of an AIO, or all of them on a specific
- fsp if aiocb == NULL.
-***********************************************************************/
-
-static int aio_pthread_cancel(struct vfs_handle_struct *handle,
-			struct files_struct *fsp,
-			SMB_STRUCT_AIOCB *aiocb)
-{
-	struct aio_private_data *pd = NULL;
-
-	for (pd = pd_list; pd != NULL; pd = pd->next) {
-		if (pd->aiocb == NULL) {
-			continue;
-		}
-		if (pd->aiocb->aio_fildes != fsp->fh->fd) {
-			continue;
-		}
-		if ((aiocb != NULL) && (pd->aiocb != aiocb)) {
-			continue;
-		}
-
-		/*
-		 * We let the child do its job, but we discard the result when
-		 * it's finished.
-		 */
-
-		pd->cancelled = true;
-	}
-
-	return AIO_CANCELED;
-}
-
-/************************************************************************
- Callback for a previously detected job completion.
-***********************************************************************/
-
-static void aio_pthread_handle_immediate(struct tevent_context *ctx,
-				struct tevent_immediate *im,
-				void *private_data)
-{
-	struct aio_extra *aio_ex = NULL;
-	struct aio_private_data *pd = (struct aio_private_data *)private_data;
-
-	aio_ex = (struct aio_extra *)pd->aiocb->aio_sigevent.sigev_value.sival_ptr;
-	smbd_aio_complete_aio_ex(aio_ex);
-	TALLOC_FREE(aio_ex);
-}
-
-/************************************************************************
- Private data struct used in suspend completion code.
-***********************************************************************/
-
-struct suspend_private {
-	int num_entries;
-	int num_finished;
-	const SMB_STRUCT_AIOCB * const *aiocb_array;
-};
-
-/************************************************************************
- Callback when an IO completes from a suspend call.
-***********************************************************************/
-
-static void aio_pthread_handle_suspend_completion(struct event_context *event_ctx,
-				struct fd_event *event,
-				uint16 flags,
-				void *p)
-{
-	struct suspend_private *sp = (struct suspend_private *)p;
-	struct aio_private_data *pd = NULL;
-	struct tevent_immediate *im = NULL;
-	int jobid;
-	int i;
-
-	DEBUG(10, ("aio_pthread_handle_suspend_completion called with flags=%d\n",
-			(int)flags));
-
-	if ((flags & EVENT_FD_READ) == 0) {
-		return;
-	}
-
-	if (pthreadpool_finished_job(pool, &jobid)) {
-		smb_panic("aio_pthread_handle_suspend_completion: can't find job.");
-		return;
-	}
-
-	pd = find_private_data_by_jobid(jobid);
-	if (pd == NULL) {
-		DEBUG(1, ("aio_pthread_handle_completion cannot find jobid %d\n",
-			  jobid));
-		return;
-	}
-
-	/* Is this a jobid with an aiocb we're interested in ? */
-	for (i = 0; i < sp->num_entries; i++) {
-		if (sp->aiocb_array[i] == pd->aiocb) {
-			sp->num_finished++;
-			return;
-		}
-	}
-
-	/* Jobid completed we weren't waiting for.
-	   We must reschedule this as an immediate event
-	   on the main event context. */
-	im = tevent_create_immediate(NULL);
-	if (!im) {
-		exit_server_cleanly("aio_pthread_handle_suspend_completion: no memory");
-	}
-
-	DEBUG(10,("aio_pthread_handle_suspend_completion: "
-			"re-scheduling job id %d\n",
-			jobid));
-
-	tevent_schedule_immediate(im,
-			server_event_context(),
-			aio_pthread_handle_immediate,
-			(void *)pd);
-}
-
-
-static void aio_pthread_suspend_timed_out(struct tevent_context *event_ctx,
-					struct tevent_timer *te,
-					struct timeval now,
-					void *private_data)
-{
-	bool *timed_out = (bool *)private_data;
-	/* Remove this timed event handler. */
-	TALLOC_FREE(te);
-	*timed_out = true;
-}
+#if defined(HAVE_OPENAT) && defined(USE_LINUX_THREAD_CREDENTIALS)
 
 /************************************************************************
- Called to request everything to stop until all IO is completed.
+ Ensure thread pool is initialized.
 ***********************************************************************/
 
-static int aio_pthread_suspend(struct vfs_handle_struct *handle,
-			struct files_struct *fsp,
-			const SMB_STRUCT_AIOCB * const aiocb_array[],
-			int n,
-			const struct timespec *timeout)
+static bool init_aio_threadpool(struct event_context *ev_ctx,
+				struct pthreadpool **pp_pool,
+				void (*completion_fn)(struct event_context *,
+						struct fd_event *,
+						uint16,
+						void *))
 {
-	struct event_context *ev = NULL;
 	struct fd_event *sock_event = NULL;
-	int ret = -1;
-	struct suspend_private sp;
-	bool timed_out = false;
-	TALLOC_CTX *frame = talloc_stackframe();
-
-	/* This is a blocking call, and has to use a sub-event loop. */
-	ev = event_context_init(frame);
-	if (ev == NULL) {
-		errno = ENOMEM;
-		goto out;
-	}
+	int ret = 0;
 
-	if (timeout) {
-		struct timeval tv = convert_timespec_to_timeval(*timeout);
-		struct tevent_timer *te = tevent_add_timer(ev,
-						frame,
-						timeval_current_ofs(tv.tv_sec,
-								    tv.tv_usec),
-						aio_pthread_suspend_timed_out,
-						&timed_out);
-		if (!te) {
-			errno = ENOMEM;
-			goto out;
-		}
+	if (*pp_pool) {
+		return true;
 	}
 
-	ZERO_STRUCT(sp);
-	sp.num_entries = n;
-	sp.aiocb_array = aiocb_array;
-	sp.num_finished = 0;
-
-	sock_event = tevent_add_fd(ev,
-				frame,
-				pthreadpool_signal_fd(pool),
+	ret = pthreadpool_init(aio_pending_size, pp_pool);
+	if (ret) {
+		errno = ret;
+		return false;
+	}
+	sock_event = tevent_add_fd(ev_ctx,
+				NULL,
+				pthreadpool_signal_fd(*pp_pool),
 				TEVENT_FD_READ,
-				aio_pthread_handle_suspend_completion,
-				(void *)&sp);
+				completion_fn,
+				NULL);
 	if (sock_event == NULL) {
-		pthreadpool_destroy(pool);
-		pool = NULL;
-		goto out;
-	}
-	/*
-	 * We're going to cheat here. We know that smbd/aio.c
-	 * only calls this when it's waiting for every single
-	 * outstanding call to finish on a close, so just wait
-	 * individually for each IO to complete. We don't care
-	 * what order they finish - only that they all do. JRA.
-	 */
-	while (sp.num_entries != sp.num_finished) {
-		if (tevent_loop_once(ev) == -1) {
-			goto out;
-		}
-
-		if (timed_out) {
-			errno = EAGAIN;
-			goto out;
-		}
+		pthreadpool_destroy(*pp_pool);
+		*pp_pool = NULL;
+		return false;
 	}
 
-	ret = 0;
-
-  out:
+	DEBUG(10,("init_aio_threadpool: initialized with up to %d threads\n",
+		  aio_pending_size));
 
-	TALLOC_FREE(frame);
-	return ret;
+	return true;
 }
 
-#if defined(HAVE_OPENAT) && defined(USE_LINUX_THREAD_CREDENTIALS)
 /*
  * We must have openat() to do any thread-based
  * asynchronous opens. We also must be using
@@ -1034,6 +612,28 @@ static int aio_pthread_connect(vfs_handle_struct *handle, const char *service,
 	 *********************************************************************/
 	aio_pending_size = lp_parm_int(
 		SNUM(handle->conn), "aio_pthread", "aio num threads", 100);
+
+	if (asys_ctx == NULL) {
+		int ret;
+
+		ret = asys_context_init(&asys_ctx, aio_pending_size);
+		if (ret != 0) {
+			DEBUG(1, ("asys_context_init failed: %s\n",
+				  strerror(ret)));
+			return -1;
+		}
+
+		asys_fde = tevent_add_fd(handle->conn->sconn->ev_ctx, NULL,
+					 asys_signalfd(asys_ctx),
+					 TEVENT_FD_READ, aio_pthread_finished,
+					 NULL);
+		if (asys_fde == NULL) {
+			DEBUG(1, ("tevent_add_fd failed\n"));
+			asys_context_destroy(asys_ctx);
+			asys_ctx = NULL;
+			return -1;
+		}
+	}
 	return SMB_VFS_NEXT_CONNECT(handle, service, user);
 }
 
@@ -1042,12 +642,10 @@ static struct vfs_fn_pointers vfs_aio_pthread_fns = {
 #if defined(HAVE_OPENAT) && defined(USE_LINUX_THREAD_CREDENTIALS)
 	.open_fn = aio_pthread_open_fn,
 #endif
-	.aio_read_fn = aio_pthread_read,
-	.aio_write_fn = aio_pthread_write,
-	.aio_return_fn = aio_pthread_return_fn,
-	.aio_cancel_fn = aio_pthread_cancel,
-	.aio_error_fn = aio_pthread_error_fn,
-	.aio_suspend_fn = aio_pthread_suspend,
+	.pread_send_fn = aio_pthread_pread_send,
+	.pread_recv_fn = aio_pthread_recv,
+	.pwrite_send_fn = aio_pthread_pwrite_send,
+	.pwrite_recv_fn = aio_pthread_recv,
 };
 
 NTSTATUS vfs_aio_pthread_init(void);