#include "system/shmem.h"
#include "smbd/smbd.h"
#include "smbd/globals.h"
-#include "lib/pthreadpool/pthreadpool.h"
+#include "../lib/pthreadpool/pthreadpool_pipe.h"
#ifdef HAVE_LINUX_FALLOC_H
#include <linux/falloc.h>
#endif
-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;
- bool flush_write;
-};
-
-/* 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);
-
+#if defined(HAVE_OPENAT) && defined(USE_LINUX_THREAD_CREDENTIALS)
/************************************************************************
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,
+static bool init_aio_threadpool(struct tevent_context *ev_ctx,
+ struct pthreadpool_pipe **pp_pool,
+ void (*completion_fn)(struct tevent_context *,
+ struct tevent_fd *,
+ uint16_t,
void *))
{
- struct fd_event *sock_event = NULL;
+ struct tevent_fd *sock_event = NULL;
int ret = 0;
if (*pp_pool) {
return true;
}
- ret = pthreadpool_init(aio_pending_size, pp_pool);
+ ret = pthreadpool_pipe_init(lp_aio_max_threads(), pp_pool);
if (ret) {
errno = ret;
return false;
}
sock_event = tevent_add_fd(ev_ctx,
NULL,
- pthreadpool_signal_fd(*pp_pool),
+ pthreadpool_pipe_signal_fd(*pp_pool),
TEVENT_FD_READ,
completion_fn,
NULL);
if (sock_event == NULL) {
- pthreadpool_destroy(*pp_pool);
+ pthreadpool_pipe_destroy(*pp_pool);
*pp_pool = NULL;
return false;
}
DEBUG(10,("init_aio_threadpool: initialized with up to %d threads\n",
- aio_pending_size));
+ (int)lp_aio_max_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 = 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)
-{
- 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(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;
- }
-
- ret = pthreadpool_add_job(pool, pd->jobid, aio_worker, (void *)pd);
- if (ret) {
- errno = ret;
- return -1;
- }
-
- 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;
-}
-
-/************************************************************************
- 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(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;
- }
-
- 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;
- }
- }
-
- 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);
-
- DEBUG(10,("aio_pthread_handle_completion: jobid %d completed\n",
- jobid ));
- TALLOC_FREE(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->cancelled) {
- errno = ECANCELED;
- return -1;
- }
-
- 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;
- 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;
-}
-
-/************************************************************************
- Called to request everything to stop until all IO is completed.
-***********************************************************************/
-
-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)
-{
- 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;
- }
-
- 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;
- }
- }
-
- 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),
- TEVENT_FD_READ,
- aio_pthread_handle_suspend_completion,
- (void *)&sp);
- 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;
- }
- }
-
- ret = 0;
-
- out:
-
- TALLOC_FREE(frame);
- return ret;
-}
-
-#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
* process, as is the current jobid.
*/
-static struct pthreadpool *open_pool;
+static struct pthreadpool_pipe *open_pool;
static int aio_pthread_open_jobid;
struct aio_open_private_data {
Callback when an open completes.
***********************************************************************/
-static void aio_open_handle_completion(struct event_context *event_ctx,
- struct fd_event *event,
- uint16 flags,
+static void aio_open_handle_completion(struct tevent_context *event_ctx,
+ struct tevent_fd *event,
+ uint16_t flags,
void *p)
{
struct aio_open_private_data *opd = NULL;
int jobid = 0;
int ret;
+ struct smbXsrv_connection *xconn;
DEBUG(10, ("aio_open_handle_completion called with flags=%d\n",
(int)flags));
- if ((flags & EVENT_FD_READ) == 0) {
+ if ((flags & TEVENT_FD_READ) == 0) {
return;
}
- ret = pthreadpool_finished_job(open_pool, &jobid);
- if (ret) {
+ ret = pthreadpool_pipe_finished_jobs(open_pool, &jobid, 1);
+ if (ret != 1) {
smb_panic("aio_open_handle_completion");
/* notreached. */
return;
opd->in_progress = false;
- /* Find outstanding event and reschdule. */
- if (!schedule_deferred_open_message_smb(opd->sconn, opd->mid)) {
+ /*
+ * TODO: In future we need a proper algorithm
+ * to find the correct connection for a fsp.
+ * For now we only have one connection, so this is correct...
+ */
+ xconn = opd->sconn->client->connections;
+
+ /* Find outstanding event and reschedule. */
+ if (!schedule_deferred_open_message_smb(xconn, opd->mid)) {
/*
* Outstanding event didn't exist or was
* cancelled. Free up the fd and throw
}
talloc_set_destructor(opd, opd_destructor);
- DLIST_ADD_END(open_pd_list, opd, struct aio_open_private_data *);
+ DLIST_ADD_END(open_pd_list, opd);
return opd;
}
return -1;
}
- ret = pthreadpool_add_job(open_pool,
- opd->jobid,
- aio_open_worker,
- (void *)opd);
+ ret = pthreadpool_pipe_add_job(open_pool,
+ opd->jobid,
+ aio_open_worker,
+ (void *)opd);
if (ret) {
errno = ret;
return -1;
}
#endif
-static int aio_pthread_connect(vfs_handle_struct *handle, const char *service,
- const char *user)
-{
- /*********************************************************************
- * How many threads to initialize ?
- * 100 per process seems insane as a default until you realize that
- * (a) Threads terminate after 1 second when idle.
- * (b) Throttling is done in SMB2 via the crediting algorithm.
- * (c) SMB1 clients are limited to max_mux (50) outstanding
- * requests and Windows clients don't use this anyway.
- * Essentially we want this to be unlimited unless smb.conf
- * says different.
- *********************************************************************/
- aio_pending_size = lp_parm_int(
- SNUM(handle->conn), "aio_pthread", "aio num threads", 100);
- return SMB_VFS_NEXT_CONNECT(handle, service, user);
-}
-
static struct vfs_fn_pointers vfs_aio_pthread_fns = {
- .connect_fn = aio_pthread_connect,
#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,
};
NTSTATUS vfs_aio_pthread_init(void);
git.samba.org / slow / samba.git / blobdiff