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 "smbd/globals.h"
29 #include "lib/pthreadpool/pthreadpool.h"
32 static struct pthreadpool
*pool
;
33 static int aio_pthread_jobid
;
35 struct aio_private_data
{
36 struct aio_private_data
*prev
, *next
;
38 SMB_STRUCT_AIOCB
*aiocb
;
45 /* List of outstanding requests we have. */
46 static struct aio_private_data
*pd_list
;
48 static void aio_pthread_handle_completion(struct event_context
*event_ctx
,
49 struct fd_event
*event
,
54 /************************************************************************
55 Ensure thread pool is initialized.
56 ***********************************************************************/
58 static bool init_aio_threadpool(struct vfs_handle_struct
*handle
)
60 struct fd_event
*sock_event
= NULL
;
67 ret
= pthreadpool_init(aio_pending_size
, &pool
);
72 sock_event
= tevent_add_fd(server_event_context(),
74 pthreadpool_signal_fd(pool
),
76 aio_pthread_handle_completion
,
78 if (sock_event
== NULL
) {
79 pthreadpool_destroy(pool
);
84 DEBUG(10,("init_aio_threadpool: initialized with up to %d threads\n",
91 /************************************************************************
92 Worker function - core of the pthread aio engine.
93 This is the function that actually does the IO.
94 ***********************************************************************/
96 static void aio_worker(void *private_data
)
98 struct aio_private_data
*pd
=
99 (struct aio_private_data
*)private_data
;
101 if (pd
->write_command
) {
102 pd
->ret_size
= sys_pwrite(pd
->aiocb
->aio_fildes
,
103 (const void *)pd
->aiocb
->aio_buf
,
104 pd
->aiocb
->aio_nbytes
,
105 pd
->aiocb
->aio_offset
);
106 if (pd
->ret_size
== -1 && errno
== ESPIPE
) {
107 /* Maintain the fiction that pipes can
108 be seeked (sought?) on. */
109 pd
->ret_size
= sys_write(pd
->aiocb
->aio_fildes
,
110 (const void *)pd
->aiocb
->aio_buf
,
111 pd
->aiocb
->aio_nbytes
);
114 pd
->ret_size
= sys_pread(pd
->aiocb
->aio_fildes
,
115 (void *)pd
->aiocb
->aio_buf
,
116 pd
->aiocb
->aio_nbytes
,
117 pd
->aiocb
->aio_offset
);
118 if (pd
->ret_size
== -1 && errno
== ESPIPE
) {
119 /* Maintain the fiction that pipes can
120 be seeked (sought?) on. */
121 pd
->ret_size
= sys_read(pd
->aiocb
->aio_fildes
,
122 (void *)pd
->aiocb
->aio_buf
,
123 pd
->aiocb
->aio_nbytes
);
126 if (pd
->ret_size
== -1) {
127 pd
->ret_errno
= errno
;
133 /************************************************************************
134 Private data destructor.
135 ***********************************************************************/
137 static int pd_destructor(struct aio_private_data
*pd
)
139 DLIST_REMOVE(pd_list
, pd
);
143 /************************************************************************
144 Create and initialize a private data struct.
145 ***********************************************************************/
147 static struct aio_private_data
*create_private_data(TALLOC_CTX
*ctx
,
148 SMB_STRUCT_AIOCB
*aiocb
)
150 struct aio_private_data
*pd
= talloc_zero(ctx
, struct aio_private_data
);
154 pd
->jobid
= aio_pthread_jobid
++;
157 pd
->ret_errno
= EINPROGRESS
;
158 talloc_set_destructor(pd
, pd_destructor
);
159 DLIST_ADD_END(pd_list
, pd
, struct aio_private_data
*);
163 /************************************************************************
164 Spin off a threadpool (if needed) and initiate a pread call.
165 ***********************************************************************/
167 static int aio_pthread_read(struct vfs_handle_struct
*handle
,
168 struct files_struct
*fsp
,
169 SMB_STRUCT_AIOCB
*aiocb
)
171 struct aio_extra
*aio_ex
= (struct aio_extra
*)aiocb
->aio_sigevent
.sigev_value
.sival_ptr
;
172 struct aio_private_data
*pd
= NULL
;
175 if (!init_aio_threadpool(handle
)) {
179 pd
= create_private_data(aio_ex
, aiocb
);
181 DEBUG(10, ("aio_pthread_read: Could not create private data.\n"));
185 ret
= pthreadpool_add_job(pool
, pd
->jobid
, aio_worker
, (void *)pd
);
191 DEBUG(10, ("aio_pthread_read: jobid=%d pread requested "
192 "of %llu bytes at offset %llu\n",
194 (unsigned long long)pd
->aiocb
->aio_nbytes
,
195 (unsigned long long)pd
->aiocb
->aio_offset
));
200 /************************************************************************
201 Spin off a threadpool (if needed) and initiate a pwrite call.
202 ***********************************************************************/
204 static int aio_pthread_write(struct vfs_handle_struct
*handle
,
205 struct files_struct
*fsp
,
206 SMB_STRUCT_AIOCB
*aiocb
)
208 struct aio_extra
*aio_ex
= (struct aio_extra
*)aiocb
->aio_sigevent
.sigev_value
.sival_ptr
;
209 struct aio_private_data
*pd
= NULL
;
212 if (!init_aio_threadpool(handle
)) {
216 pd
= create_private_data(aio_ex
, aiocb
);
218 DEBUG(10, ("aio_pthread_write: Could not create private data.\n"));
222 pd
->write_command
= true;
224 ret
= pthreadpool_add_job(pool
, pd
->jobid
, aio_worker
, (void *)pd
);
230 DEBUG(10, ("aio_pthread_write: jobid=%d pwrite requested "
231 "of %llu bytes at offset %llu\n",
233 (unsigned long long)pd
->aiocb
->aio_nbytes
,
234 (unsigned long long)pd
->aiocb
->aio_offset
));
239 /************************************************************************
240 Find the private data by jobid.
241 ***********************************************************************/
243 static struct aio_private_data
*find_private_data_by_jobid(int jobid
)
245 struct aio_private_data
*pd
;
247 for (pd
= pd_list
; pd
!= NULL
; pd
= pd
->next
) {
248 if (pd
->jobid
== jobid
) {
256 /************************************************************************
257 Callback when an IO completes.
258 ***********************************************************************/
260 static void aio_pthread_handle_completion(struct event_context
*event_ctx
,
261 struct fd_event
*event
,
265 struct aio_extra
*aio_ex
= NULL
;
266 struct aio_private_data
*pd
= NULL
;
270 DEBUG(10, ("aio_pthread_handle_completion called with flags=%d\n",
273 if ((flags
& EVENT_FD_READ
) == 0) {
277 ret
= pthreadpool_finished_job(pool
, &jobid
);
279 smb_panic("aio_pthread_handle_completion");
283 pd
= find_private_data_by_jobid(jobid
);
285 DEBUG(1, ("aio_pthread_handle_completion cannot find jobid %d\n",
290 aio_ex
= (struct aio_extra
*)pd
->aiocb
->aio_sigevent
.sigev_value
.sival_ptr
;
291 smbd_aio_complete_aio_ex(aio_ex
);
293 DEBUG(10,("aio_pthread_handle_completion: jobid %d completed\n",
298 /************************************************************************
299 Find the private data by aiocb.
300 ***********************************************************************/
302 static struct aio_private_data
*find_private_data_by_aiocb(SMB_STRUCT_AIOCB
*aiocb
)
304 struct aio_private_data
*pd
;
306 for (pd
= pd_list
; pd
!= NULL
; pd
= pd
->next
) {
307 if (pd
->aiocb
== aiocb
) {
315 /************************************************************************
316 Called to return the result of a completed AIO.
317 Should only be called if aio_error returns something other than EINPROGRESS.
319 Any other value - return from IO operation.
320 ***********************************************************************/
322 static ssize_t
aio_pthread_return_fn(struct vfs_handle_struct
*handle
,
323 struct files_struct
*fsp
,
324 SMB_STRUCT_AIOCB
*aiocb
)
326 struct aio_private_data
*pd
= find_private_data_by_aiocb(aiocb
);
330 DEBUG(0, ("aio_pthread_return_fn: returning EINVAL\n"));
336 if (pd
->ret_size
== -1) {
337 errno
= pd
->ret_errno
;
343 /************************************************************************
344 Called to check the result of an AIO.
346 EINPROGRESS - still in progress.
347 EINVAL - invalid aiocb.
348 ECANCELED - request was cancelled.
349 0 - request completed successfully.
350 Any other value - errno from IO operation.
351 ***********************************************************************/
353 static int aio_pthread_error_fn(struct vfs_handle_struct
*handle
,
354 struct files_struct
*fsp
,
355 SMB_STRUCT_AIOCB
*aiocb
)
357 struct aio_private_data
*pd
= find_private_data_by_aiocb(aiocb
);
365 return pd
->ret_errno
;
368 /************************************************************************
369 Called to request the cancel of an AIO, or all of them on a specific
370 fsp if aiocb == NULL.
371 ***********************************************************************/
373 static int aio_pthread_cancel(struct vfs_handle_struct
*handle
,
374 struct files_struct
*fsp
,
375 SMB_STRUCT_AIOCB
*aiocb
)
377 struct aio_private_data
*pd
= NULL
;
379 for (pd
= pd_list
; pd
!= NULL
; pd
= pd
->next
) {
380 if (pd
->aiocb
== NULL
) {
383 if (pd
->aiocb
->aio_fildes
!= fsp
->fh
->fd
) {
386 if ((aiocb
!= NULL
) && (pd
->aiocb
!= aiocb
)) {
391 * We let the child do its job, but we discard the result when
395 pd
->cancelled
= true;
401 /************************************************************************
402 Callback for a previously detected job completion.
403 ***********************************************************************/
405 static void aio_pthread_handle_immediate(struct tevent_context
*ctx
,
406 struct tevent_immediate
*im
,
409 struct aio_extra
*aio_ex
= NULL
;
410 struct aio_private_data
*pd
= (struct aio_private_data
*)private_data
;
412 aio_ex
= (struct aio_extra
*)pd
->aiocb
->aio_sigevent
.sigev_value
.sival_ptr
;
413 smbd_aio_complete_aio_ex(aio_ex
);
417 /************************************************************************
418 Private data struct used in suspend completion code.
419 ***********************************************************************/
421 struct suspend_private
{
424 const SMB_STRUCT_AIOCB
* const *aiocb_array
;
427 /************************************************************************
428 Callback when an IO completes from a suspend call.
429 ***********************************************************************/
431 static void aio_pthread_handle_suspend_completion(struct event_context
*event_ctx
,
432 struct fd_event
*event
,
436 struct suspend_private
*sp
= (struct suspend_private
*)p
;
437 struct aio_private_data
*pd
= NULL
;
438 struct tevent_immediate
*im
= NULL
;
442 DEBUG(10, ("aio_pthread_handle_suspend_completion called with flags=%d\n",
445 if ((flags
& EVENT_FD_READ
) == 0) {
449 if (pthreadpool_finished_job(pool
, &jobid
)) {
450 smb_panic("aio_pthread_handle_suspend_completion: can't find job.");
454 pd
= find_private_data_by_jobid(jobid
);
456 DEBUG(1, ("aio_pthread_handle_completion cannot find jobid %d\n",
461 /* Is this a jobid with an aiocb we're interested in ? */
462 for (i
= 0; i
< sp
->num_entries
; i
++) {
463 if (sp
->aiocb_array
[i
] == pd
->aiocb
) {
469 /* Jobid completed we weren't waiting for.
470 We must reshedule this as an immediate event
471 on the main event context. */
472 im
= tevent_create_immediate(NULL
);
474 exit_server_cleanly("aio_pthread_handle_suspend_completion: no memory");
477 DEBUG(10,("aio_pthread_handle_suspend_completion: "
478 "re-scheduling job id %d\n",
481 tevent_schedule_immediate(im
,
482 server_event_context(),
483 aio_pthread_handle_immediate
,
488 static void aio_pthread_suspend_timed_out(struct tevent_context
*event_ctx
,
489 struct tevent_timer
*te
,
493 bool *timed_out
= (bool *)private_data
;
494 /* Remove this timed event handler. */
499 /************************************************************************
500 Called to request everything to stop until all IO is completed.
501 ***********************************************************************/
503 static int aio_pthread_suspend(struct vfs_handle_struct
*handle
,
504 struct files_struct
*fsp
,
505 const SMB_STRUCT_AIOCB
* const aiocb_array
[],
507 const struct timespec
*timeout
)
509 struct event_context
*ev
= NULL
;
510 struct fd_event
*sock_event
= NULL
;
512 struct suspend_private sp
;
513 bool timed_out
= false;
514 TALLOC_CTX
*frame
= talloc_stackframe();
516 /* This is a blocking call, and has to use a sub-event loop. */
517 ev
= event_context_init(frame
);
524 struct timeval tv
= convert_timespec_to_timeval(*timeout
);
525 struct tevent_timer
*te
= tevent_add_timer(ev
,
527 timeval_current_ofs(tv
.tv_sec
,
529 aio_pthread_suspend_timed_out
,
539 sp
.aiocb_array
= aiocb_array
;
542 sock_event
= tevent_add_fd(ev
,
544 pthreadpool_signal_fd(pool
),
546 aio_pthread_handle_suspend_completion
,
548 if (sock_event
== NULL
) {
549 pthreadpool_destroy(pool
);
554 * We're going to cheat here. We know that smbd/aio.c
555 * only calls this when it's waiting for every single
556 * outstanding call to finish on a close, so just wait
557 * individually for each IO to complete. We don't care
558 * what order they finish - only that they all do. JRA.
560 while (sp
.num_entries
!= sp
.num_finished
) {
561 if (tevent_loop_once(ev
) == -1) {
579 static int aio_pthread_connect(vfs_handle_struct
*handle
, const char *service
,
582 /*********************************************************************
583 * How many threads to initialize ?
584 * 100 per process seems insane as a default until you realize that
585 * (a) Threads terminate after 1 second when idle.
586 * (b) Throttling is done in SMB2 via the crediting algorithm.
587 * (c) SMB1 clients are limited to max_mux (50) outstanding
588 * requests and Windows clients don't use this anyway.
589 * Essentially we want this to be unlimited unless smb.conf
591 *********************************************************************/
592 aio_pending_size
= lp_parm_int(
593 SNUM(handle
->conn
), "aio_pthread", "aio num threads", 100);
594 return SMB_VFS_NEXT_CONNECT(handle
, service
, user
);
597 static struct vfs_fn_pointers vfs_aio_pthread_fns
= {
598 .connect_fn
= aio_pthread_connect
,
599 .aio_read_fn
= aio_pthread_read
,
600 .aio_write_fn
= aio_pthread_write
,
601 .aio_return_fn
= aio_pthread_return_fn
,
602 .aio_cancel_fn
= aio_pthread_cancel
,
603 .aio_error_fn
= aio_pthread_error_fn
,
604 .aio_suspend_fn
= aio_pthread_suspend
,
607 NTSTATUS
vfs_aio_pthread_init(void);
608 NTSTATUS
vfs_aio_pthread_init(void)
610 return smb_register_vfs(SMB_VFS_INTERFACE_VERSION
,
611 "aio_pthread", &vfs_aio_pthread_fns
);