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 "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 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
);
429 /************************************************************************
430 Private data struct used in suspend completion code.
431 ***********************************************************************/
433 struct suspend_private
{
436 const SMB_STRUCT_AIOCB
* const *aiocb_array
;
439 /************************************************************************
440 Callback when an IO completes from a suspend call.
441 ***********************************************************************/
443 static void aio_pthread_handle_suspend_completion(struct event_context
*event_ctx
,
444 struct fd_event
*event
,
448 struct suspend_private
*sp
= (struct suspend_private
*)p
;
449 struct aio_private_data
*pd
= NULL
;
450 struct tevent_immediate
*im
= NULL
;
454 DEBUG(10, ("aio_pthread_handle_suspend_completion called with flags=%d\n",
457 if ((flags
& EVENT_FD_READ
) == 0) {
461 pjobid
= talloc_array(NULL
, int, 1);
463 smb_panic("aio_pthread_handle_suspend_completion: no memory.");
466 if (pthreadpool_finished_job(pool
, pjobid
)) {
467 smb_panic("aio_pthread_handle_suspend_completion: can't find job.");
471 pd
= find_private_data_by_jobid(*pjobid
);
473 DEBUG(1, ("aio_pthread_handle_completion cannot find jobid %d\n",
479 /* Is this a jobid with an aiocb we're interested in ? */
480 for (i
= 0; i
< sp
->num_entries
; i
++) {
481 if (sp
->aiocb_array
[i
] == pd
->aiocb
) {
488 /* Jobid completed we weren't waiting for.
489 We must reshedule this as an immediate event
490 on the main event context. */
491 im
= tevent_create_immediate(NULL
);
493 exit_server_cleanly("aio_pthread_handle_suspend_completion: no memory");
496 DEBUG(10,("aio_pthread_handle_suspend_completion: "
497 "re-scheduling job id %d\n",
500 tevent_schedule_immediate(im
,
501 server_event_context(),
502 aio_pthread_handle_immediate
,
507 static void aio_pthread_suspend_timed_out(struct tevent_context
*event_ctx
,
508 struct tevent_timer
*te
,
512 bool *timed_out
= (bool *)private_data
;
513 /* Remove this timed event handler. */
518 /************************************************************************
519 Called to request everything to stop until all IO is completed.
520 ***********************************************************************/
522 static int aio_pthread_suspend(struct vfs_handle_struct
*handle
,
523 struct files_struct
*fsp
,
524 const SMB_STRUCT_AIOCB
* const aiocb_array
[],
526 const struct timespec
*timeout
)
528 struct event_context
*ev
= NULL
;
529 struct fd_event
*sock_event
= NULL
;
531 struct suspend_private sp
;
532 bool timed_out
= false;
533 TALLOC_CTX
*frame
= talloc_stackframe();
535 /* This is a blocking call, and has to use a sub-event loop. */
536 ev
= event_context_init(frame
);
543 struct timeval tv
= convert_timespec_to_timeval(*timeout
);
544 struct tevent_timer
*te
= tevent_add_timer(ev
,
546 timeval_current_ofs(tv
.tv_sec
,
548 aio_pthread_suspend_timed_out
,
558 sp
.aiocb_array
= aiocb_array
;
561 sock_event
= tevent_add_fd(ev
,
563 pthreadpool_signal_fd(pool
),
565 aio_pthread_handle_suspend_completion
,
567 if (sock_event
== NULL
) {
568 pthreadpool_destroy(pool
);
573 * We're going to cheat here. We know that smbd/aio.c
574 * only calls this when it's waiting for every single
575 * outstanding call to finish on a close, so just wait
576 * individually for each IO to complete. We don't care
577 * what order they finish - only that they all do. JRA.
579 while (sp
.num_entries
!= sp
.num_finished
) {
580 if (tevent_loop_once(ev
) == -1) {
598 static struct vfs_fn_pointers vfs_aio_pthread_fns
= {
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
);