The branch, master has been updated
       via  3a18a42 Add S3 vfs_aio_pthread module to replace broken glibc aio 
code.
      from  d2d4dfd waf: Use git repository.

http://gitweb.samba.org/?p=samba.git;a=shortlog;h=master


- Log -----------------------------------------------------------------
commit 3a18a42d13b87a5a6c41d8a70e961313fd6d0a86
Author: Jeremy Allison <[email protected]>
Date:   Wed Jan 4 12:54:16 2012 -0800

    Add S3 vfs_aio_pthread module to replace broken glibc aio code.
    
    Compiles but not yet tested.
    
    Autobuild-User: Jeremy Allison <[email protected]>
    Autobuild-Date: Thu Jan  5 01:43:51 CET 2012 on sn-devel-104

-----------------------------------------------------------------------

Summary of changes:
 source3/Makefile.in               |    5 +
 source3/configure.in              |    4 +
 source3/modules/vfs_aio_pthread.c |  637 +++++++++++++++++++++++++++++++++++++
 source3/modules/wscript_build     |   10 +
 source3/wscript                   |    3 +
 5 files changed, 659 insertions(+), 0 deletions(-)
 create mode 100644 source3/modules/vfs_aio_pthread.c


Changeset truncated at 500 lines:

diff --git a/source3/Makefile.in b/source3/Makefile.in
index cd73263..4bd1b40 100644
--- a/source3/Makefile.in
+++ b/source3/Makefile.in
@@ -868,6 +868,7 @@ VFS_READAHEAD_OBJ = modules/vfs_readahead.o
 VFS_TSMSM_OBJ = modules/vfs_tsmsm.o
 VFS_FILEID_OBJ = modules/vfs_fileid.o
 VFS_AIO_FORK_OBJ = modules/vfs_aio_fork.o
+VFS_AIO_PTHREAD_OBJ = modules/vfs_aio_pthread.o
 VFS_PREOPEN_OBJ = modules/vfs_preopen.o
 VFS_SYNCOPS_OBJ = modules/vfs_syncops.o
 VFS_ACL_XATTR_OBJ = modules/vfs_acl_xattr.o
@@ -3054,6 +3055,10 @@ bin/aio_fork.@SHLIBEXT@: $(BINARY_PREREQS) 
$(VFS_AIO_FORK_OBJ)
        @echo "Building plugin $@"
        @$(SHLD_MODULE) $(VFS_AIO_FORK_OBJ)
 
+bin/aio_pthread.@SHLIBEXT@: $(BINARY_PREREQS) $(VFS_AIO_PTHREAD_OBJ)
+       @echo "Building plugin $@"
+       @$(SHLD_MODULE) $(VFS_AIO_PTHREAD_OBJ)
+
 bin/preopen.@SHLIBEXT@: $(BINARY_PREREQS) $(VFS_PREOPEN_OBJ)
        @echo "Building plugin $@"
        @$(SHLD_MODULE) $(VFS_PREOPEN_OBJ)
diff --git a/source3/configure.in b/source3/configure.in
index 6f4300c..7ed093c 100644
--- a/source3/configure.in
+++ b/source3/configure.in
@@ -6724,6 +6724,9 @@ if test x"$enable_pthreadpool" = x"yes" -a 
x"$samba_cv_HAVE_PTHREAD" = x"yes"; t
     AC_SUBST(PTHREADPOOL_OBJ, "lib/pthreadpool/pthreadpool.o")
     PTHREADPOOLTEST="bin/pthreadpooltest\$(EXEEXT)"
     AC_SUBST(PTHREADPOOLTEST)
+    if test x"$samba_cv_HAVE_AIO" = x"yes"; then
+        default_shared_modules="$default_shared_modules vfs_aio_pthread"
+    fi
 fi
 
 #################################################
@@ -6947,6 +6950,7 @@ SMB_MODULE(vfs_readahead, \$(VFS_READAHEAD_OBJ), 
"bin/readahead.$SHLIBEXT", VFS)
 SMB_MODULE(vfs_tsmsm, \$(VFS_TSMSM_OBJ), "bin/tsmsm.$SHLIBEXT", VFS)
 SMB_MODULE(vfs_fileid, \$(VFS_FILEID_OBJ), "bin/fileid.$SHLIBEXT", VFS)
 SMB_MODULE(vfs_aio_fork, \$(VFS_AIO_FORK_OBJ), "bin/aio_fork.$SHLIBEXT", VFS)
+SMB_MODULE(vfs_aio_pthread, \$(VFS_AIO_PTHREAD_OBJ), 
"bin/aio_pthread.$SHLIBEXT", VFS)
 SMB_MODULE(vfs_preopen, \$(VFS_PREOPEN_OBJ), "bin/preopen.$SHLIBEXT", VFS)
 SMB_MODULE(vfs_syncops, \$(VFS_SYNCOPS_OBJ), "bin/syncops.$SHLIBEXT", VFS)
 SMB_MODULE(vfs_zfsacl, \$(VFS_ZFSACL_OBJ), "bin/zfsacl.$SHLIBEXT", VFS)
diff --git a/source3/modules/vfs_aio_pthread.c 
b/source3/modules/vfs_aio_pthread.c
new file mode 100644
index 0000000..56a2abd
--- /dev/null
+++ b/source3/modules/vfs_aio_pthread.c
@@ -0,0 +1,637 @@
+/*
+ * Simulate Posix AIO using pthreads.
+ *
+ * Based on the aio_fork work from Volker and Volker's pthreadpool library.
+ *
+ * Copyright (C) Volker Lendecke 2008
+ * Copyright (C) Jeremy Allison 2012
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+#include "includes.h"
+#include "system/filesys.h"
+#include "system/shmem.h"
+#include "smbd/smbd.h"
+#include "pthreadpool.h"
+
+struct aio_extra;
+static struct pthreadpool *pool;
+static int aio_pthread_jobid;
+
+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;
+};
+
+/* List of outstanding requests we have. */
+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);
+
+/************************************************************************
+ How many threads to initialize ?
+***********************************************************************/
+
+static int aio_get_num_threads(void)
+{
+       return 10;
+}
+
+#if 0
+/************************************************************************
+ Called every 30 seconds to destroy pool if it's idle.
+***********************************************************************/
+
+static void idle_pool_destroy_timer(struct tevent_context *ev,
+                       struct tevent_timer *te,
+                       struct timeval current_time,
+                       void *private_data)
+{
+       struct timeval ne;
+
+       TALLOC_FREE(te);
+
+       if (pool && pd_list == NULL) {
+               if (pthreadpool_destroy(pool) == 0) {
+                       pool = NULL;
+               }
+               DEBUG(10,("idle_pool_destroy_timer: destroyed AIO pool.\n"));
+               return;
+       }
+
+       /* Here, the IO is still active. */
+
+       /* Set an event up for 30 seconds time - if we have
+          no outstanding IO at this time shut the threadpool
+          down. */
+       ne = tevent_timeval_current_ofs(30, 0);
+       tevent_add_timer(server_event_context(),
+                       NULL,
+                       ne,
+                       idle_pool_destroy_timer,
+                       NULL);
+}
+#endif
+
+/************************************************************************
+ Ensure thread pool is initialized.
+***********************************************************************/
+
+static bool init_aio_threadpool(void)
+{
+       struct fd_event *sock_event = NULL;
+       int ret = 0;
+       int num_threads = aio_get_num_threads();
+#if 0
+       struct timeval ne;
+#endif
+
+       if (pool) {
+               return true;
+       }
+
+       ret = pthreadpool_init(num_threads, &pool);
+       if (ret) {
+               errno = ret;
+               return false;
+       }
+       sock_event = tevent_add_fd(server_event_context(),
+                               NULL,
+                               pthreadpool_signal_fd(pool),
+                               TEVENT_FD_READ,
+                               aio_pthread_handle_completion,
+                               NULL);
+       if (sock_event == NULL) {
+               pthreadpool_destroy(pool);
+               pool = NULL;
+               return false;
+       }
+
+#if 0
+       /* Set an event up for 30 seconds time - if we have
+          no outstanding IO at this time shut the threadpool
+          down. */
+       ne = tevent_timeval_current_ofs(30, 0);
+       tevent_add_timer(server_event_context(),
+                       NULL,
+                       ne,
+                       idle_pool_destroy_timer,
+                       NULL);
+#endif
+
+       DEBUG(10,("init_aio_threadpool: initialized with %d threads\n",
+                       num_threads));
+
+       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 = pwrite(pd->aiocb->aio_fildes,
+                               (const void *)pd->aiocb->aio_buf,
+                               pd->aiocb->aio_nbytes,
+                               pd->aiocb->aio_offset);
+       } else {
+               pd->ret_size = pread(pd->aiocb->aio_fildes,
+                               (void *)pd->aiocb->aio_buf,
+                               pd->aiocb->aio_nbytes,
+                               pd->aiocb->aio_offset);
+       }
+       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)
+{
+       DLIST_REMOVE(pd_list, pd);
+       return 0;
+}
+
+/************************************************************************
+ Create and initialize a private data struct.
+***********************************************************************/
+
+static struct aio_private_data *create_private_data(TALLOC_CTX *ctx,
+                                       SMB_STRUCT_AIOCB *aiocb)
+{
+       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;
+       int ret;
+
+       if (!init_aio_threadpool()) {
+               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;
+       }
+
+       ret = pthreadpool_add_job(pool, pd->jobid, aio_worker, (void *)pd);
+       if (ret) {
+               errno = ret;
+               return -1;
+       }
+
+       return 0;
+}
+
+/************************************************************************
+ 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)
+{
+       struct aio_extra *aio_ex = (struct aio_extra 
*)aiocb->aio_sigevent.sigev_value.sival_ptr;
+       struct aio_private_data *pd = NULL;
+       int ret;
+
+       if (!init_aio_threadpool()) {
+               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;
+
+       ret = pthreadpool_add_job(pool, pd->jobid, aio_worker, (void *)pd);
+       if (ret) {
+               errno = ret;
+               return -1;
+       }
+
+       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;
+               }
+       }
+
+       return NULL;
+}
+
+/************************************************************************
+ Callback when an IO completes.
+***********************************************************************/
+
+static void aio_pthread_handle_completion(struct event_context *event_ctx,
+                               struct fd_event *event,
+                               uint16 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) {
+               return;
+       }
+
+       ret = pthreadpool_finished_job(pool, &jobid);
+       if (ret) {
+               smb_panic("aio_pthread_handle_completion");
+               return;
+       }
+
+       pd = find_private_data_by_jobid(jobid);
+       if (pd == NULL) {
+               DEBUG(1, ("aio_pthread_handle_completion cannot find jobid 
%d\n",
+                         jobid));
+               return;
+       }
+
+       aio_ex = (struct aio_extra 
*)pd->aiocb->aio_sigevent.sigev_value.sival_ptr;
+       smbd_aio_complete_aio_ex(aio_ex);
+}
+
+/************************************************************************
+ 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;
+}
+
+/************************************************************************
+ 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)
+{
+       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;
+       }
+
+       pd->aiocb = NULL;
+
+       if (pd->ret_size == -1) {
+               errno = pd->ret_errno;
+       }
+
+       return pd->ret_size;
+}
+
+/************************************************************************
+ 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;
+       int *pjobid = (int *)private_data;
+       struct aio_private_data *pd = find_private_data_by_jobid(*pjobid);
+
+       if (pd == NULL) {
+               DEBUG(1, ("aio_pthread_handle_immediate cannot find jobid %d\n",
+                         *pjobid));
+               TALLOC_FREE(pjobid);
+               return;
+       }
+
+       TALLOC_FREE(pjobid);


-- 
Samba Shared Repository

Reply via email to