source3/modules/vfs_aio_pthread.c

   1 /*
   2  * Simulate Posix AIO using pthreads.
   3  *
   4  * Based on the aio_fork work from Volker and Volker's pthreadpool library.
   5  *
   6  * Copyright (C) Volker Lendecke 2008
   7  * Copyright (C) Jeremy Allison 2012
   8  *
   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.
  13  *
  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.
  18  *
  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.
  22  */
  23
  24 #include "includes.h"
  25 #include "system/filesys.h"
  26 #include "system/shmem.h"
  27 #include "smbd/smbd.h"
  28 #include "pthreadpool.h"
  29
  30 struct aio_extra;
  31 static struct pthreadpool *pool;
  32 static int aio_pthread_jobid;
  33
  34 struct aio_private_data {
  35         struct aio_private_data *prev, *next;
  36         int jobid;
  37         SMB_STRUCT_AIOCB *aiocb;
  38         ssize_t ret_size;
  39         int ret_errno;
  40         bool cancelled;
  41         bool write_command;
  42 };
  43
  44 /* List of outstanding requests we have. */
  45 struct aio_private_data *pd_list;
  46
  47 static void aio_pthread_handle_completion(struct event_context *event_ctx,
  48                                 struct fd_event *event,
  49                                 uint16 flags,
  50                                 void *p);
  51
  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 ***********************************************************************/
  61
  62 static int aio_get_num_threads(struct vfs_handle_struct *handle)
  63 {
  64         return lp_parm_bool(SNUM(handle->conn),
  65                                 "aio_pthread",
  66                                 "aio num threads",
  67                                 100);
  68 }
  69
  70 /************************************************************************
  71  Ensure thread pool is initialized.
  72 ***********************************************************************/
  73
  74 static bool init_aio_threadpool(struct vfs_handle_struct *handle)
  75 {
  76         struct fd_event *sock_event = NULL;
  77         int ret = 0;
  78         int num_threads;
  79
  80         if (pool) {
  81                 return true;
  82         }
  83
  84         num_threads = aio_get_num_threads(handle);
  85         ret = pthreadpool_init(num_threads, &pool);
  86         if (ret) {
  87                 errno = ret;
  88                 return false;
  89         }
  90         sock_event = tevent_add_fd(server_event_context(),
  91                                 NULL,
  92                                 pthreadpool_signal_fd(pool),
  93                                 TEVENT_FD_READ,
  94                                 aio_pthread_handle_completion,
  95                                 NULL);
  96         if (sock_event == NULL) {
  97                 pthreadpool_destroy(pool);
  98                 pool = NULL;
  99                 return false;
 100         }
 101
 102         DEBUG(10,("init_aio_threadpool: initialized with up to %d threads\n",
 103                         num_threads));
 104
 105         return true;
 106 }
 107
 108
 109 /************************************************************************
 110  Worker function - core of the pthread aio engine.
 111  This is the function that actually does the IO.
 112 ***********************************************************************/
 113
 114 static void aio_worker(void *private_data)
 115 {
 116         struct aio_private_data *pd =
 117                         (struct aio_private_data *)private_data;
 118
 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);
 124         } else {
 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);
 129         }
 130         if (pd->ret_size == -1) {
 131                 pd->ret_errno = errno;
 132         } else {
 133                 pd->ret_errno = 0;
 134         }
 135 }
 136
 137 /************************************************************************
 138  Private data destructor.
 139 ***********************************************************************/
 140
 141 static int pd_destructor(struct aio_private_data *pd)
 142 {
 143         DLIST_REMOVE(pd_list, pd);
 144         return 0;
 145 }
 146
 147 /************************************************************************
 148  Create and initialize a private data struct.
 149 ***********************************************************************/
 150
 151 static struct aio_private_data *create_private_data(TALLOC_CTX *ctx,
 152                                         SMB_STRUCT_AIOCB *aiocb)
 153 {
 154         struct aio_private_data *pd = talloc_zero(ctx, struct aio_private_data);
 155         if (!pd) {
 156                 return NULL;
 157         }
 158         pd->jobid = aio_pthread_jobid++;
 159         pd->aiocb = aiocb;
 160         pd->ret_size = -1;
 161         pd->ret_errno = EINPROGRESS;
 162         talloc_set_destructor(pd, pd_destructor);
 163         DLIST_ADD_END(pd_list, pd, struct aio_private_data *);
 164         return pd;
 165 }
 166
 167 /************************************************************************
 168  Spin off a threadpool (if needed) and initiate a pread call.
 169 ***********************************************************************/
 170
 171 static int aio_pthread_read(struct vfs_handle_struct *handle,
 172                                 struct files_struct *fsp,
 173                                 SMB_STRUCT_AIOCB *aiocb)
 174 {
 175         struct aio_extra *aio_ex = (struct aio_extra *)aiocb->aio_sigevent.sigev_value.sival_ptr;
 176         struct aio_private_data *pd = NULL;
 177         int ret;
 178
 179         if (!init_aio_threadpool(handle)) {
 180                 return -1;
 181         }
 182
 183         pd = create_private_data(aio_ex, aiocb);
 184         if (pd == NULL) {
 185                 DEBUG(10, ("aio_pthread_read: Could not create private data.\n"));
 186                 return -1;
 187         }
 188
 189         ret = pthreadpool_add_job(pool, pd->jobid, aio_worker, (void *)pd);
 190         if (ret) {
 191                 errno = ret;
 192                 return -1;
 193         }
 194
 195         DEBUG(10, ("aio_pthread_read: jobid=%d pread requested "
 196                 "of %llu bytes at offset %llu\n",
 197                 pd->jobid,
 198                 (unsigned long long)pd->aiocb->aio_nbytes,
 199                 (unsigned long long)pd->aiocb->aio_offset));
 200
 201         return 0;
 202 }
 203
 204 /************************************************************************
 205  Spin off a threadpool (if needed) and initiate a pwrite call.
 206 ***********************************************************************/
 207
 208 static int aio_pthread_write(struct vfs_handle_struct *handle,
 209                                 struct files_struct *fsp,
 210                                 SMB_STRUCT_AIOCB *aiocb)
 211 {
 212         struct aio_extra *aio_ex = (struct aio_extra *)aiocb->aio_sigevent.sigev_value.sival_ptr;
 213         struct aio_private_data *pd = NULL;
 214         int ret;
 215
 216         if (!init_aio_threadpool(handle)) {
 217                 return -1;
 218         }
 219
 220         pd = create_private_data(aio_ex, aiocb);
 221         if (pd == NULL) {
 222                 DEBUG(10, ("aio_pthread_write: Could not create private data.\n"));
 223                 return -1;
 224         }
 225
 226         pd->write_command = true;
 227
 228         ret = pthreadpool_add_job(pool, pd->jobid, aio_worker, (void *)pd);
 229         if (ret) {
 230                 errno = ret;
 231                 return -1;
 232         }
 233
 234         DEBUG(10, ("aio_pthread_write: jobid=%d pwrite requested "
 235                 "of %llu bytes at offset %llu\n",
 236                 pd->jobid,
 237                 (unsigned long long)pd->aiocb->aio_nbytes,
 238                 (unsigned long long)pd->aiocb->aio_offset));
 239
 240         return 0;
 241 }
 242
 243 /************************************************************************
 244  Find the private data by jobid.
 245 ***********************************************************************/
 246
 247 static struct aio_private_data *find_private_data_by_jobid(int jobid)
 248 {
 249         struct aio_private_data *pd;
 250
 251         for (pd = pd_list; pd != NULL; pd = pd->next) {
 252                 if (pd->jobid == jobid) {
 253                         return pd;
 254                 }
 255         }
 256
 257         return NULL;
 258 }
 259
 260 /************************************************************************
 261  Callback when an IO completes.
 262 ***********************************************************************/
 263
 264 static void aio_pthread_handle_completion(struct event_context *event_ctx,
 265                                 struct fd_event *event,
 266                                 uint16 flags,
 267                                 void *p)
 268 {
 269         struct aio_extra *aio_ex = NULL;
 270         struct aio_private_data *pd = NULL;
 271         int jobid = 0;
 272         int ret;
 273
 274         DEBUG(10, ("aio_pthread_handle_completion called with flags=%d\n",
 275                         (int)flags));
 276
 277         if ((flags & EVENT_FD_READ) == 0) {
 278                 return;
 279         }
 280
 281         ret = pthreadpool_finished_job(pool, &jobid);
 282         if (ret) {
 283                 smb_panic("aio_pthread_handle_completion");
 284                 return;
 285         }
 286
 287         pd = find_private_data_by_jobid(jobid);
 288         if (pd == NULL) {
 289                 DEBUG(1, ("aio_pthread_handle_completion cannot find jobid %d\n",
 290                           jobid));
 291                 return;
 292         }
 293
 294         aio_ex = (struct aio_extra *)pd->aiocb->aio_sigevent.sigev_value.sival_ptr;
 295         smbd_aio_complete_aio_ex(aio_ex);
 296
 297         DEBUG(10,("aio_pthread_handle_completion: jobid %d completed\n",
 298                 jobid ));
 299
 300 }
 301
 302 /************************************************************************
 303  Find the private data by aiocb.
 304 ***********************************************************************/
 305
 306 static struct aio_private_data *find_private_data_by_aiocb(SMB_STRUCT_AIOCB *aiocb)
 307 {
 308         struct aio_private_data *pd;
 309
 310         for (pd = pd_list; pd != NULL; pd = pd->next) {
 311                 if (pd->aiocb == aiocb) {
 312                         return pd;
 313                 }
 314         }
 315
 316         return NULL;
 317 }
 318
 319 /************************************************************************
 320  Called to return the result of a completed AIO.
 321  Should only be called if aio_error returns something other than EINPROGRESS.
 322  Returns:
 323         Any other value - return from IO operation.
 324 ***********************************************************************/
 325
 326 static ssize_t aio_pthread_return_fn(struct vfs_handle_struct *handle,
 327                                 struct files_struct *fsp,
 328                                 SMB_STRUCT_AIOCB *aiocb)
 329 {
 330         struct aio_private_data *pd = find_private_data_by_aiocb(aiocb);
 331
 332         if (pd == NULL) {
 333                 errno = EINVAL;
 334                 DEBUG(0, ("aio_pthread_return_fn: returning EINVAL\n"));
 335                 return -1;
 336         }
 337
 338         pd->aiocb = NULL;
 339
 340         if (pd->ret_size == -1) {
 341                 errno = pd->ret_errno;
 342         }
 343
 344         return pd->ret_size;
 345 }
 346
 347 /************************************************************************
 348  Called to check the result of an AIO.
 349  Returns:
 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 ***********************************************************************/
 356
 357 static int aio_pthread_error_fn(struct vfs_handle_struct *handle,
 358                              struct files_struct *fsp,
 359                              SMB_STRUCT_AIOCB *aiocb)
 360 {
 361         struct aio_private_data *pd = find_private_data_by_aiocb(aiocb);
 362
 363         if (pd == NULL) {
 364                 return EINVAL;
 365         }
 366         if (pd->cancelled) {
 367                 return ECANCELED;
 368         }
 369         return pd->ret_errno;
 370 }
 371
 372 /************************************************************************
 373  Called to request the cancel of an AIO, or all of them on a specific
 374  fsp if aiocb == NULL.
 375 ***********************************************************************/
 376
 377 static int aio_pthread_cancel(struct vfs_handle_struct *handle,
 378                         struct files_struct *fsp,
 379                         SMB_STRUCT_AIOCB *aiocb)
 380 {
 381         struct aio_private_data *pd = NULL;
 382
 383         for (pd = pd_list; pd != NULL; pd = pd->next) {
 384                 if (pd->aiocb == NULL) {
 385                         continue;
 386                 }
 387                 if (pd->aiocb->aio_fildes != fsp->fh->fd) {
 388                         continue;
 389                 }
 390                 if ((aiocb != NULL) && (pd->aiocb != aiocb)) {
 391                         continue;
 392                 }
 393
 394                 /*
 395                  * We let the child do its job, but we discard the result when
 396                  * it's finished.
 397                  */
 398
 399                 pd->cancelled = true;
 400         }
 401
 402         return AIO_CANCELED;
 403 }
 404
 405 /************************************************************************
 406  Callback for a previously detected job completion.
 407 ***********************************************************************/
 408
 409 static void aio_pthread_handle_immediate(struct tevent_context *ctx,
 410                                 struct tevent_immediate *im,
 411                                 void *private_data)
 412 {
 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);
 416
 417         if (pd == NULL) {
 418                 DEBUG(1, ("aio_pthread_handle_immediate cannot find jobid %d\n",
 419                           *pjobid));
 420                 TALLOC_FREE(pjobid);
 421                 return;
 422         }
 423
 424         TALLOC_FREE(pjobid);
 425         aio_ex = (struct aio_extra *)pd->aiocb->aio_sigevent.sigev_value.sival_ptr;
 426         smbd_aio_complete_aio_ex(aio_ex);
 427 }
 428
 429 /************************************************************************
 430  Private data struct used in suspend completion code.
 431 ***********************************************************************/
 432
 433 struct suspend_private {
 434         int num_entries;
 435         int num_finished;
 436         const SMB_STRUCT_AIOCB * const *aiocb_array;
 437 };
 438
 439 /************************************************************************
 440  Callback when an IO completes from a suspend call.
 441 ***********************************************************************/
 442
 443 static void aio_pthread_handle_suspend_completion(struct event_context *event_ctx,
 444                                 struct fd_event *event,
 445                                 uint16 flags,
 446                                 void *p)
 447 {
 448         struct suspend_private *sp = (struct suspend_private *)p;
 449         struct aio_private_data *pd = NULL;
 450         struct tevent_immediate *im = NULL;
 451         int *pjobid = NULL;
 452         int i;
 453
 454         DEBUG(10, ("aio_pthread_handle_suspend_completion called with flags=%d\n",
 455                         (int)flags));
 456
 457         if ((flags & EVENT_FD_READ) == 0) {
 458                 return;
 459         }
 460
 461         pjobid = talloc_array(NULL, int, 1);
 462         if (pjobid) {
 463                 smb_panic("aio_pthread_handle_suspend_completion: no memory.");
 464         }
 465
 466         if (pthreadpool_finished_job(pool, pjobid)) {
 467                 smb_panic("aio_pthread_handle_suspend_completion: can't find job.");
 468                 return;
 469         }
 470
 471         pd = find_private_data_by_jobid(*pjobid);
 472         if (pd == NULL) {
 473                 DEBUG(1, ("aio_pthread_handle_completion cannot find jobid %d\n",
 474                           *pjobid));
 475                 TALLOC_FREE(pjobid);
 476                 return;
 477         }
 478
 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) {
 482                         sp->num_finished++;
 483                         TALLOC_FREE(pjobid);
 484                         return;
 485                 }
 486         }
 487
 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);
 492         if (!im) {
 493                 exit_server_cleanly("aio_pthread_handle_suspend_completion: no memory");
 494         }
 495
 496         DEBUG(10,("aio_pthread_handle_suspend_completion: "
 497                         "re-scheduling job id %d\n",
 498                         *pjobid));
 499
 500         tevent_schedule_immediate(im,
 501                         server_event_context(),
 502                         aio_pthread_handle_immediate,
 503                         (void *)pjobid);
 504 }
 505
 506
 507 static void aio_pthread_suspend_timed_out(struct tevent_context *event_ctx,
 508                                         struct tevent_timer *te,
 509                                         struct timeval now,
 510                                         void *private_data)
 511 {
 512         bool *timed_out = (bool *)private_data;
 513         /* Remove this timed event handler. */
 514         TALLOC_FREE(te);
 515         *timed_out = true;
 516 }
 517
 518 /************************************************************************
 519  Called to request everything to stop until all IO is completed.
 520 ***********************************************************************/
 521
 522 static int aio_pthread_suspend(struct vfs_handle_struct *handle,
 523                         struct files_struct *fsp,
 524                         const SMB_STRUCT_AIOCB * const aiocb_array[],
 525                         int n,
 526                         const struct timespec *timeout)
 527 {
 528         struct event_context *ev = NULL;
 529         struct fd_event *sock_event = NULL;
 530         int ret = -1;
 531         struct suspend_private sp;
 532         bool timed_out = false;
 533         TALLOC_CTX *frame = talloc_stackframe();
 534
 535         /* This is a blocking call, and has to use a sub-event loop. */
 536         ev = event_context_init(frame);
 537         if (ev == NULL) {
 538                 errno = ENOMEM;
 539                 goto out;
 540         }
 541
 542         if (timeout) {
 543                 struct timeval tv = convert_timespec_to_timeval(*timeout);
 544                 struct tevent_timer *te = tevent_add_timer(ev,
 545                                                 frame,
 546                                                 timeval_current_ofs(tv.tv_sec,
 547                                                                     tv.tv_usec),
 548                                                 aio_pthread_suspend_timed_out,
 549                                                 &timed_out);
 550                 if (!te) {
 551                         errno = ENOMEM;
 552                         goto out;
 553                 }
 554         }
 555
 556         ZERO_STRUCT(sp);
 557         sp.num_entries = n;
 558         sp.aiocb_array = aiocb_array;
 559         sp.num_finished = 0;
 560
 561         sock_event = tevent_add_fd(ev,
 562                                 frame,
 563                                 pthreadpool_signal_fd(pool),
 564                                 TEVENT_FD_READ,
 565                                 aio_pthread_handle_suspend_completion,
 566                                 (void *)&sp);
 567         if (sock_event == NULL) {
 568                 pthreadpool_destroy(pool);
 569                 pool = NULL;
 570                 goto out;
 571         }
 572         /*
 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.
 578          */
 579         while (sp.num_entries != sp.num_finished) {
 580                 if (tevent_loop_once(ev) == -1) {
 581                         goto out;
 582                 }
 583
 584                 if (timed_out) {
 585                         errno = EAGAIN;
 586                         goto out;
 587                 }
 588         }
 589
 590         ret = 0;
 591
 592   out:
 593
 594         TALLOC_FREE(frame);
 595         return ret;
 596 }
 597
 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,
 605 };
 606
 607 NTSTATUS vfs_aio_pthread_init(void);
 608 NTSTATUS vfs_aio_pthread_init(void)
 609 {
 610         return smb_register_vfs(SMB_VFS_INTERFACE_VERSION,
 611                                 "aio_pthread", &vfs_aio_pthread_fns);
 612 }