2 * Simulate Posix AIO using pthreads.
4 * Based on the aio_fork work from Volker and Volker's pthreadpool library.
6 * Copyright (C) Volker Lendecke 2008
7 * Copyright (C) Jeremy Allison 2012
9 * This program is free software; you can redistribute it and/or modify
10 * it under the terms of the GNU General Public License as published by
11 * the Free Software Foundation; either version 3 of the License, or
12 * (at your option) any later version.
14 * This program is distributed in the hope that it will be useful,
15 * but WITHOUT ANY WARRANTY; without even the implied warranty of
16 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17 * GNU General Public License for more details.
19 * You should have received a copy of the GNU General Public License
20 * along with this program; if not, write to the Free Software
21 * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
25 #include "system/filesys.h"
26 #include "system/shmem.h"
27 #include "smbd/smbd.h"
28 #include "lib/pthreadpool/pthreadpool.h"
31 static struct pthreadpool *pool;
32 static int aio_pthread_jobid;
34 struct aio_private_data {
35 struct aio_private_data *prev, *next;
37 SMB_STRUCT_AIOCB *aiocb;
44 /* List of outstanding requests we have. */
45 static struct aio_private_data *pd_list;
47 static void aio_pthread_handle_completion(struct event_context *event_ctx,
48 struct fd_event *event,
52 /************************************************************************
53 How many threads to initialize ?
54 100 per process seems insane as a default until you realize that
55 (a) Threads terminate after 1 second when idle.
56 (b) Throttling is done in SMB2 via the crediting algorithm.
57 (c) SMB1 clients are limited to max_mux (50) outstanding requests and
58 Windows clients don't use this anyway.
59 Essentially we want this to be unlimited unless smb.conf says different.
60 ***********************************************************************/
62 static int aio_get_num_threads(struct vfs_handle_struct *handle)
64 return lp_parm_bool(SNUM(handle->conn),
70 /************************************************************************
71 Ensure thread pool is initialized.
72 ***********************************************************************/
74 static bool init_aio_threadpool(struct vfs_handle_struct *handle)
76 struct fd_event *sock_event = NULL;
84 num_threads = aio_get_num_threads(handle);
85 ret = pthreadpool_init(num_threads, &pool);
90 sock_event = tevent_add_fd(server_event_context(),
92 pthreadpool_signal_fd(pool),
94 aio_pthread_handle_completion,
96 if (sock_event == NULL) {
97 pthreadpool_destroy(pool);
102 DEBUG(10,("init_aio_threadpool: initialized with up to %d threads\n",
109 /************************************************************************
110 Worker function - core of the pthread aio engine.
111 This is the function that actually does the IO.
112 ***********************************************************************/
114 static void aio_worker(void *private_data)
116 struct aio_private_data *pd =
117 (struct aio_private_data *)private_data;
119 if (pd->write_command) {
120 pd->ret_size = pwrite(pd->aiocb->aio_fildes,
121 (const void *)pd->aiocb->aio_buf,
122 pd->aiocb->aio_nbytes,
123 pd->aiocb->aio_offset);
125 pd->ret_size = pread(pd->aiocb->aio_fildes,
126 (void *)pd->aiocb->aio_buf,
127 pd->aiocb->aio_nbytes,
128 pd->aiocb->aio_offset);
130 if (pd->ret_size == -1) {
131 pd->ret_errno = errno;
137 /************************************************************************
138 Private data destructor.
139 ***********************************************************************/
141 static int pd_destructor(struct aio_private_data *pd)
143 DLIST_REMOVE(pd_list, pd);
147 /************************************************************************
148 Create and initialize a private data struct.
149 ***********************************************************************/
151 static struct aio_private_data *create_private_data(TALLOC_CTX *ctx,
152 SMB_STRUCT_AIOCB *aiocb)
154 struct aio_private_data *pd = talloc_zero(ctx, struct aio_private_data);
158 pd->jobid = aio_pthread_jobid++;
161 pd->ret_errno = EINPROGRESS;
162 talloc_set_destructor(pd, pd_destructor);
163 DLIST_ADD_END(pd_list, pd, struct aio_private_data *);
167 /************************************************************************
168 Spin off a threadpool (if needed) and initiate a pread call.
169 ***********************************************************************/
171 static int aio_pthread_read(struct vfs_handle_struct *handle,
172 struct files_struct *fsp,
173 SMB_STRUCT_AIOCB *aiocb)
175 struct aio_extra *aio_ex = (struct aio_extra *)aiocb->aio_sigevent.sigev_value.sival_ptr;
176 struct aio_private_data *pd = NULL;
179 if (!init_aio_threadpool(handle)) {
183 pd = create_private_data(aio_ex, aiocb);
185 DEBUG(10, ("aio_pthread_read: Could not create private data.\n"));
189 ret = pthreadpool_add_job(pool, pd->jobid, aio_worker, (void *)pd);
195 DEBUG(10, ("aio_pthread_read: jobid=%d pread requested "
196 "of %llu bytes at offset %llu\n",
198 (unsigned long long)pd->aiocb->aio_nbytes,
199 (unsigned long long)pd->aiocb->aio_offset));
204 /************************************************************************
205 Spin off a threadpool (if needed) and initiate a pwrite call.
206 ***********************************************************************/
208 static int aio_pthread_write(struct vfs_handle_struct *handle,
209 struct files_struct *fsp,
210 SMB_STRUCT_AIOCB *aiocb)
212 struct aio_extra *aio_ex = (struct aio_extra *)aiocb->aio_sigevent.sigev_value.sival_ptr;
213 struct aio_private_data *pd = NULL;
216 if (!init_aio_threadpool(handle)) {
220 pd = create_private_data(aio_ex, aiocb);
222 DEBUG(10, ("aio_pthread_write: Could not create private data.\n"));
226 pd->write_command = true;
228 ret = pthreadpool_add_job(pool, pd->jobid, aio_worker, (void *)pd);
234 DEBUG(10, ("aio_pthread_write: jobid=%d pwrite requested "
235 "of %llu bytes at offset %llu\n",
237 (unsigned long long)pd->aiocb->aio_nbytes,
238 (unsigned long long)pd->aiocb->aio_offset));
243 /************************************************************************
244 Find the private data by jobid.
245 ***********************************************************************/
247 static struct aio_private_data *find_private_data_by_jobid(int jobid)
249 struct aio_private_data *pd;
251 for (pd = pd_list; pd != NULL; pd = pd->next) {
252 if (pd->jobid == jobid) {
260 /************************************************************************
261 Callback when an IO completes.
262 ***********************************************************************/
264 static void aio_pthread_handle_completion(struct event_context *event_ctx,
265 struct fd_event *event,
269 struct aio_extra *aio_ex = NULL;
270 struct aio_private_data *pd = NULL;
274 DEBUG(10, ("aio_pthread_handle_completion called with flags=%d\n",
277 if ((flags & EVENT_FD_READ) == 0) {
281 ret = pthreadpool_finished_job(pool, &jobid);
283 smb_panic("aio_pthread_handle_completion");
287 pd = find_private_data_by_jobid(jobid);
289 DEBUG(1, ("aio_pthread_handle_completion cannot find jobid %d\n",
294 aio_ex = (struct aio_extra *)pd->aiocb->aio_sigevent.sigev_value.sival_ptr;
295 smbd_aio_complete_aio_ex(aio_ex);
297 DEBUG(10,("aio_pthread_handle_completion: jobid %d completed\n",
302 /************************************************************************
303 Find the private data by aiocb.
304 ***********************************************************************/
306 static struct aio_private_data *find_private_data_by_aiocb(SMB_STRUCT_AIOCB *aiocb)
308 struct aio_private_data *pd;
310 for (pd = pd_list; pd != NULL; pd = pd->next) {
311 if (pd->aiocb == aiocb) {
319 /************************************************************************
320 Called to return the result of a completed AIO.
321 Should only be called if aio_error returns something other than EINPROGRESS.
323 Any other value - return from IO operation.
324 ***********************************************************************/
326 static ssize_t aio_pthread_return_fn(struct vfs_handle_struct *handle,
327 struct files_struct *fsp,
328 SMB_STRUCT_AIOCB *aiocb)
330 struct aio_private_data *pd = find_private_data_by_aiocb(aiocb);
334 DEBUG(0, ("aio_pthread_return_fn: returning EINVAL\n"));
340 if (pd->ret_size == -1) {
341 errno = pd->ret_errno;
347 /************************************************************************
348 Called to check the result of an AIO.
350 EINPROGRESS - still in progress.
351 EINVAL - invalid aiocb.
352 ECANCELED - request was cancelled.
353 0 - request completed successfully.
354 Any other value - errno from IO operation.
355 ***********************************************************************/
357 static int aio_pthread_error_fn(struct vfs_handle_struct *handle,
358 struct files_struct *fsp,
359 SMB_STRUCT_AIOCB *aiocb)
361 struct aio_private_data *pd = find_private_data_by_aiocb(aiocb);
369 return pd->ret_errno;
372 /************************************************************************
373 Called to request the cancel of an AIO, or all of them on a specific
374 fsp if aiocb == NULL.
375 ***********************************************************************/
377 static int aio_pthread_cancel(struct vfs_handle_struct *handle,
378 struct files_struct *fsp,
379 SMB_STRUCT_AIOCB *aiocb)
381 struct aio_private_data *pd = NULL;
383 for (pd = pd_list; pd != NULL; pd = pd->next) {
384 if (pd->aiocb == NULL) {
387 if (pd->aiocb->aio_fildes != fsp->fh->fd) {
390 if ((aiocb != NULL) && (pd->aiocb != aiocb)) {
395 * We let the child do its job, but we discard the result when
399 pd->cancelled = true;
405 /************************************************************************
406 Callback for a previously detected job completion.
407 ***********************************************************************/
409 static void aio_pthread_handle_immediate(struct tevent_context *ctx,
410 struct tevent_immediate *im,
413 struct aio_extra *aio_ex = NULL;
414 int *pjobid = (int *)private_data;
415 struct aio_private_data *pd = find_private_data_by_jobid(*pjobid);
418 DEBUG(1, ("aio_pthread_handle_immediate cannot find jobid %d\n",
425 aio_ex = (struct aio_extra *)pd->aiocb->aio_sigevent.sigev_value.sival_ptr;
426 smbd_aio_complete_aio_ex(aio_ex);
430 /************************************************************************
431 Private data struct used in suspend completion code.
432 ***********************************************************************/
434 struct suspend_private {
437 const SMB_STRUCT_AIOCB * const *aiocb_array;
440 /************************************************************************
441 Callback when an IO completes from a suspend call.
442 ***********************************************************************/
444 static void aio_pthread_handle_suspend_completion(struct event_context *event_ctx,
445 struct fd_event *event,
449 struct suspend_private *sp = (struct suspend_private *)p;
450 struct aio_private_data *pd = NULL;
451 struct tevent_immediate *im = NULL;
455 DEBUG(10, ("aio_pthread_handle_suspend_completion called with flags=%d\n",
458 if ((flags & EVENT_FD_READ) == 0) {
462 pjobid = talloc_array(NULL, int, 1);
464 smb_panic("aio_pthread_handle_suspend_completion: no memory.");
467 if (pthreadpool_finished_job(pool, pjobid)) {
468 smb_panic("aio_pthread_handle_suspend_completion: can't find job.");
472 pd = find_private_data_by_jobid(*pjobid);
474 DEBUG(1, ("aio_pthread_handle_completion cannot find jobid %d\n",
480 /* Is this a jobid with an aiocb we're interested in ? */
481 for (i = 0; i < sp->num_entries; i++) {
482 if (sp->aiocb_array[i] == pd->aiocb) {
489 /* Jobid completed we weren't waiting for.
490 We must reshedule this as an immediate event
491 on the main event context. */
492 im = tevent_create_immediate(NULL);
494 exit_server_cleanly("aio_pthread_handle_suspend_completion: no memory");
497 DEBUG(10,("aio_pthread_handle_suspend_completion: "
498 "re-scheduling job id %d\n",
501 tevent_schedule_immediate(im,
502 server_event_context(),
503 aio_pthread_handle_immediate,
508 static void aio_pthread_suspend_timed_out(struct tevent_context *event_ctx,
509 struct tevent_timer *te,
513 bool *timed_out = (bool *)private_data;
514 /* Remove this timed event handler. */
519 /************************************************************************
520 Called to request everything to stop until all IO is completed.
521 ***********************************************************************/
523 static int aio_pthread_suspend(struct vfs_handle_struct *handle,
524 struct files_struct *fsp,
525 const SMB_STRUCT_AIOCB * const aiocb_array[],
527 const struct timespec *timeout)
529 struct event_context *ev = NULL;
530 struct fd_event *sock_event = NULL;
532 struct suspend_private sp;
533 bool timed_out = false;
534 TALLOC_CTX *frame = talloc_stackframe();
536 /* This is a blocking call, and has to use a sub-event loop. */
537 ev = event_context_init(frame);
544 struct timeval tv = convert_timespec_to_timeval(*timeout);
545 struct tevent_timer *te = tevent_add_timer(ev,
547 timeval_current_ofs(tv.tv_sec,
549 aio_pthread_suspend_timed_out,
559 sp.aiocb_array = aiocb_array;
562 sock_event = tevent_add_fd(ev,
564 pthreadpool_signal_fd(pool),
566 aio_pthread_handle_suspend_completion,
568 if (sock_event == NULL) {
569 pthreadpool_destroy(pool);
574 * We're going to cheat here. We know that smbd/aio.c
575 * only calls this when it's waiting for every single
576 * outstanding call to finish on a close, so just wait
577 * individually for each IO to complete. We don't care
578 * what order they finish - only that they all do. JRA.
580 while (sp.num_entries != sp.num_finished) {
581 if (tevent_loop_once(ev) == -1) {
599 static struct vfs_fn_pointers vfs_aio_pthread_fns = {
600 .aio_read_fn = aio_pthread_read,
601 .aio_write_fn = aio_pthread_write,
602 .aio_return_fn = aio_pthread_return_fn,
603 .aio_cancel_fn = aio_pthread_cancel,
604 .aio_error_fn = aio_pthread_error_fn,
605 .aio_suspend_fn = aio_pthread_suspend,
608 NTSTATUS vfs_aio_pthread_init(void);
609 NTSTATUS vfs_aio_pthread_init(void)
611 return smb_register_vfs(SMB_VFS_INTERFACE_VERSION,
612 "aio_pthread", &vfs_aio_pthread_fns);