posix-aio-compat.c

   1 /*
   2  * QEMU posix-aio emulation
   3  *
   4  * Copyright IBM, Corp. 2008
   5  *
   6  * Authors:
   7  *  Anthony Liguori   <aliguori@us.ibm.com>
   8  *
   9  * This work is licensed under the terms of the GNU GPL, version 2.  See
  10  * the COPYING file in the top-level directory.
  11  *
  12  */
  13
  14 #include <sys/ioctl.h>
  15 #include <sys/types.h>
  16 #include <pthread.h>
  17 #include <unistd.h>
  18 #include <errno.h>
  19 #include <time.h>
  20 #include <string.h>
  21 #include <stdlib.h>
  22 #include <stdio.h>
  23
  24 #include "qemu-queue.h"
  25 #include "osdep.h"
  26 #include "sysemu.h"
  27 #include "qemu-common.h"
  28 #include "trace.h"
  29 #include "block_int.h"
  30 #include "compatfd.h"
  31
  32 #include "block/raw-posix-aio.h"
  33
  34 static void do_spawn_thread(void);
  35
  36 struct qemu_paiocb {
  37     BlockDriverAIOCB common;
  38     int aio_fildes;
  39     union {
  40         struct iovec *aio_iov;
  41         void *aio_ioctl_buf;
  42     };
  43     int aio_niov;
  44     size_t aio_nbytes;
  45 #define aio_ioctl_cmd   aio_nbytes /* for QEMU_AIO_IOCTL */
  46     int ev_signo;
  47     off_t aio_offset;
  48
  49     QTAILQ_ENTRY(qemu_paiocb) node;
  50     int aio_type;
  51     ssize_t ret;
  52     int active;
  53     struct qemu_paiocb *next;
  54 };
  55
  56 typedef struct PosixAioState {
  57     int fd;
  58     struct qemu_paiocb *first_aio;
  59 } PosixAioState;
  60
  61
  62 static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
  63 static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
  64 static pthread_t thread_id;
  65 static pthread_attr_t attr;
  66 static int max_threads = 64;
  67 static int cur_threads = 0;
  68 static int idle_threads = 0;
  69 static int new_threads = 0;     /* backlog of threads we need to create */
  70 static int pending_threads = 0; /* threads created but not running yet */
  71 static QEMUBH *new_thread_bh;
  72 static QTAILQ_HEAD(, qemu_paiocb) request_list;
  73
  74 #ifdef CONFIG_PREADV
  75 static int preadv_present = 1;
  76 #else
  77 static int preadv_present = 0;
  78 #endif
  79
  80 static void die2(int err, const char *what)
  81 {
  82     fprintf(stderr, "%s failed: %s\n", what, strerror(err));
  83     abort();
  84 }
  85
  86 static void die(const char *what)
  87 {
  88     die2(errno, what);
  89 }
  90
  91 static void mutex_lock(pthread_mutex_t *mutex)
  92 {
  93     int ret = pthread_mutex_lock(mutex);
  94     if (ret) die2(ret, "pthread_mutex_lock");
  95 }
  96
  97 static void mutex_unlock(pthread_mutex_t *mutex)
  98 {
  99     int ret = pthread_mutex_unlock(mutex);
 100     if (ret) die2(ret, "pthread_mutex_unlock");
 101 }
 102
 103 static int cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex,
 104                            struct timespec *ts)
 105 {
 106     int ret = pthread_cond_timedwait(cond, mutex, ts);
 107     if (ret && ret != ETIMEDOUT) die2(ret, "pthread_cond_timedwait");
 108     return ret;
 109 }
 110
 111 static void cond_signal(pthread_cond_t *cond)
 112 {
 113     int ret = pthread_cond_signal(cond);
 114     if (ret) die2(ret, "pthread_cond_signal");
 115 }
 116
 117 static void thread_create(pthread_t *thread, pthread_attr_t *attr,
 118                           void *(*start_routine)(void*), void *arg)
 119 {
 120     int ret = pthread_create(thread, attr, start_routine, arg);
 121     if (ret) die2(ret, "pthread_create");
 122 }
 123
 124 static ssize_t handle_aiocb_ioctl(struct qemu_paiocb *aiocb)
 125 {
 126     int ret;
 127
 128     ret = ioctl(aiocb->aio_fildes, aiocb->aio_ioctl_cmd, aiocb->aio_ioctl_buf);
 129     if (ret == -1)
 130         return -errno;
 131
 132     /*
 133      * This looks weird, but the aio code only consideres a request
 134      * successful if it has written the number full number of bytes.
 135      *
 136      * Now we overload aio_nbytes as aio_ioctl_cmd for the ioctl command,
 137      * so in fact we return the ioctl command here to make posix_aio_read()
 138      * happy..
 139      */
 140     return aiocb->aio_nbytes;
 141 }
 142
 143 static ssize_t handle_aiocb_flush(struct qemu_paiocb *aiocb)
 144 {
 145     int ret;
 146
 147     ret = qemu_fdatasync(aiocb->aio_fildes);
 148     if (ret == -1)
 149         return -errno;
 150     return 0;
 151 }
 152
 153 #ifdef CONFIG_PREADV
 154
 155 static ssize_t
 156 qemu_preadv(int fd, const struct iovec *iov, int nr_iov, off_t offset)
 157 {
 158     return preadv(fd, iov, nr_iov, offset);
 159 }
 160
 161 static ssize_t
 162 qemu_pwritev(int fd, const struct iovec *iov, int nr_iov, off_t offset)
 163 {
 164     return pwritev(fd, iov, nr_iov, offset);
 165 }
 166
 167 #else
 168
 169 static ssize_t
 170 qemu_preadv(int fd, const struct iovec *iov, int nr_iov, off_t offset)
 171 {
 172     return -ENOSYS;
 173 }
 174
 175 static ssize_t
 176 qemu_pwritev(int fd, const struct iovec *iov, int nr_iov, off_t offset)
 177 {
 178     return -ENOSYS;
 179 }
 180
 181 #endif
 182
 183 static ssize_t handle_aiocb_rw_vector(struct qemu_paiocb *aiocb)
 184 {
 185     size_t offset = 0;
 186     ssize_t len;
 187
 188     do {
 189         if (aiocb->aio_type & QEMU_AIO_WRITE)
 190             len = qemu_pwritev(aiocb->aio_fildes,
 191                                aiocb->aio_iov,
 192                                aiocb->aio_niov,
 193                                aiocb->aio_offset + offset);
 194          else
 195             len = qemu_preadv(aiocb->aio_fildes,
 196                               aiocb->aio_iov,
 197                               aiocb->aio_niov,
 198                               aiocb->aio_offset + offset);
 199     } while (len == -1 && errno == EINTR);
 200
 201     if (len == -1)
 202         return -errno;
 203     return len;
 204 }
 205
 206 /*
 207  * Read/writes the data to/from a given linear buffer.
 208  *
 209  * Returns the number of bytes handles or -errno in case of an error. Short
 210  * reads are only returned if the end of the file is reached.
 211  */
 212 static ssize_t handle_aiocb_rw_linear(struct qemu_paiocb *aiocb, char *buf)
 213 {
 214     ssize_t offset = 0;
 215     ssize_t len;
 216
 217     while (offset < aiocb->aio_nbytes) {
 218          if (aiocb->aio_type & QEMU_AIO_WRITE)
 219              len = pwrite(aiocb->aio_fildes,
 220                           (const char *)buf + offset,
 221                           aiocb->aio_nbytes - offset,
 222                           aiocb->aio_offset + offset);
 223          else
 224              len = pread(aiocb->aio_fildes,
 225                          buf + offset,
 226                          aiocb->aio_nbytes - offset,
 227                          aiocb->aio_offset + offset);
 228
 229          if (len == -1 && errno == EINTR)
 230              continue;
 231          else if (len == -1) {
 232              offset = -errno;
 233              break;
 234          } else if (len == 0)
 235              break;
 236
 237          offset += len;
 238     }
 239
 240     return offset;
 241 }
 242
 243 static ssize_t handle_aiocb_rw(struct qemu_paiocb *aiocb)
 244 {
 245     ssize_t nbytes;
 246     char *buf;
 247
 248     if (!(aiocb->aio_type & QEMU_AIO_MISALIGNED)) {
 249         /*
 250          * If there is just a single buffer, and it is properly aligned
 251          * we can just use plain pread/pwrite without any problems.
 252          */
 253         if (aiocb->aio_niov == 1)
 254              return handle_aiocb_rw_linear(aiocb, aiocb->aio_iov->iov_base);
 255
 256         /*
 257          * We have more than one iovec, and all are properly aligned.
 258          *
 259          * Try preadv/pwritev first and fall back to linearizing the
 260          * buffer if it's not supported.
 261          */
 262         if (preadv_present) {
 263             nbytes = handle_aiocb_rw_vector(aiocb);
 264             if (nbytes == aiocb->aio_nbytes)
 265                 return nbytes;
 266             if (nbytes < 0 && nbytes != -ENOSYS)
 267                 return nbytes;
 268             preadv_present = 0;
 269         }
 270
 271         /*
 272          * XXX(hch): short read/write.  no easy way to handle the reminder
 273          * using these interfaces.  For now retry using plain
 274          * pread/pwrite?
 275          */
 276     }
 277
 278     /*
 279      * Ok, we have to do it the hard way, copy all segments into
 280      * a single aligned buffer.
 281      */
 282     buf = qemu_blockalign(aiocb->common.bs, aiocb->aio_nbytes);
 283     if (aiocb->aio_type & QEMU_AIO_WRITE) {
 284         char *p = buf;
 285         int i;
 286
 287         for (i = 0; i < aiocb->aio_niov; ++i) {
 288             memcpy(p, aiocb->aio_iov[i].iov_base, aiocb->aio_iov[i].iov_len);
 289             p += aiocb->aio_iov[i].iov_len;
 290         }
 291     }
 292
 293     nbytes = handle_aiocb_rw_linear(aiocb, buf);
 294     if (!(aiocb->aio_type & QEMU_AIO_WRITE)) {
 295         char *p = buf;
 296         size_t count = aiocb->aio_nbytes, copy;
 297         int i;
 298
 299         for (i = 0; i < aiocb->aio_niov && count; ++i) {
 300             copy = count;
 301             if (copy > aiocb->aio_iov[i].iov_len)
 302                 copy = aiocb->aio_iov[i].iov_len;
 303             memcpy(aiocb->aio_iov[i].iov_base, p, copy);
 304             p     += copy;
 305             count -= copy;
 306         }
 307     }
 308     qemu_vfree(buf);
 309
 310     return nbytes;
 311 }
 312
 313 static void *aio_thread(void *unused)
 314 {
 315     pid_t pid;
 316
 317     pid = getpid();
 318
 319     mutex_lock(&lock);
 320     pending_threads--;
 321     mutex_unlock(&lock);
 322     do_spawn_thread();
 323
 324     while (1) {
 325         struct qemu_paiocb *aiocb;
 326         ssize_t ret = 0;
 327         qemu_timeval tv;
 328         struct timespec ts;
 329
 330         qemu_gettimeofday(&tv);
 331         ts.tv_sec = tv.tv_sec + 10;
 332         ts.tv_nsec = 0;
 333
 334         mutex_lock(&lock);
 335
 336         while (QTAILQ_EMPTY(&request_list) &&
 337                !(ret == ETIMEDOUT)) {
 338             idle_threads++;
 339             ret = cond_timedwait(&cond, &lock, &ts);
 340             idle_threads--;
 341         }
 342
 343         if (QTAILQ_EMPTY(&request_list))
 344             break;
 345
 346         aiocb = QTAILQ_FIRST(&request_list);
 347         QTAILQ_REMOVE(&request_list, aiocb, node);
 348         aiocb->active = 1;
 349         mutex_unlock(&lock);
 350
 351         switch (aiocb->aio_type & QEMU_AIO_TYPE_MASK) {
 352         case QEMU_AIO_READ:
 353             ret = handle_aiocb_rw(aiocb);
 354             if (ret >= 0 && ret < aiocb->aio_nbytes && aiocb->common.bs->growable) {
 355                 /* A short read means that we have reached EOF. Pad the buffer
 356                  * with zeros for bytes after EOF. */
 357                 QEMUIOVector qiov;
 358
 359                 qemu_iovec_init_external(&qiov, aiocb->aio_iov,
 360                                          aiocb->aio_niov);
 361                 qemu_iovec_memset_skip(&qiov, 0, aiocb->aio_nbytes - ret, ret);
 362
 363                 ret = aiocb->aio_nbytes;
 364             }
 365             break;
 366         case QEMU_AIO_WRITE:
 367             ret = handle_aiocb_rw(aiocb);
 368             break;
 369         case QEMU_AIO_FLUSH:
 370             ret = handle_aiocb_flush(aiocb);
 371             break;
 372         case QEMU_AIO_IOCTL:
 373             ret = handle_aiocb_ioctl(aiocb);
 374             break;
 375         default:
 376             fprintf(stderr, "invalid aio request (0x%x)\n", aiocb->aio_type);
 377             ret = -EINVAL;
 378             break;
 379         }
 380
 381         mutex_lock(&lock);
 382         aiocb->ret = ret;
 383         mutex_unlock(&lock);
 384
 385         if (kill(pid, aiocb->ev_signo)) die("kill failed");
 386     }
 387
 388     cur_threads--;
 389     mutex_unlock(&lock);
 390
 391     return NULL;
 392 }
 393
 394 static void do_spawn_thread(void)
 395 {
 396     sigset_t set, oldset;
 397
 398     mutex_lock(&lock);
 399     if (!new_threads) {
 400         mutex_unlock(&lock);
 401         return;
 402     }
 403
 404     new_threads--;
 405     pending_threads++;
 406
 407     mutex_unlock(&lock);
 408
 409     /* block all signals */
 410     if (sigfillset(&set)) die("sigfillset");
 411     if (sigprocmask(SIG_SETMASK, &set, &oldset)) die("sigprocmask");
 412
 413     thread_create(&thread_id, &attr, aio_thread, NULL);
 414
 415     if (sigprocmask(SIG_SETMASK, &oldset, NULL)) die("sigprocmask restore");
 416 }
 417
 418 static void spawn_thread_bh_fn(void *opaque)
 419 {
 420     do_spawn_thread();
 421 }
 422
 423 static void spawn_thread(void)
 424 {
 425     cur_threads++;
 426     new_threads++;
 427     /* If there are threads being created, they will spawn new workers, so
 428      * we don't spend time creating many threads in a loop holding a mutex or
 429      * starving the current vcpu.
 430      *
 431      * If there are no idle threads, ask the main thread to create one, so we
 432      * inherit the correct affinity instead of the vcpu affinity.
 433      */
 434     if (!pending_threads) {
 435         qemu_bh_schedule(new_thread_bh);
 436     }
 437 }
 438
 439 static void qemu_paio_submit(struct qemu_paiocb *aiocb)
 440 {
 441     aiocb->ret = -EINPROGRESS;
 442     aiocb->active = 0;
 443     mutex_lock(&lock);
 444     if (idle_threads == 0 && cur_threads < max_threads)
 445         spawn_thread();
 446     QTAILQ_INSERT_TAIL(&request_list, aiocb, node);
 447     mutex_unlock(&lock);
 448     cond_signal(&cond);
 449 }
 450
 451 static ssize_t qemu_paio_return(struct qemu_paiocb *aiocb)
 452 {
 453     ssize_t ret;
 454
 455     mutex_lock(&lock);
 456     ret = aiocb->ret;
 457     mutex_unlock(&lock);
 458
 459     return ret;
 460 }
 461
 462 static int qemu_paio_error(struct qemu_paiocb *aiocb)
 463 {
 464     ssize_t ret = qemu_paio_return(aiocb);
 465
 466     if (ret < 0)
 467         ret = -ret;
 468     else
 469         ret = 0;
 470
 471     return ret;
 472 }
 473
 474 static int posix_aio_process_queue(void *opaque)
 475 {
 476     PosixAioState *s = opaque;
 477     struct qemu_paiocb *acb, **pacb;
 478     int ret;
 479     int result = 0;
 480
 481     for(;;) {
 482         pacb = &s->first_aio;
 483         for(;;) {
 484             acb = *pacb;
 485             if (!acb)
 486                 return result;
 487
 488             ret = qemu_paio_error(acb);
 489             if (ret == ECANCELED) {
 490                 /* remove the request */
 491                 *pacb = acb->next;
 492                 qemu_aio_release(acb);
 493                 result = 1;
 494             } else if (ret != EINPROGRESS) {
 495                 /* end of aio */
 496                 if (ret == 0) {
 497                     ret = qemu_paio_return(acb);
 498                     if (ret == acb->aio_nbytes)
 499                         ret = 0;
 500                     else
 501                         ret = -EINVAL;
 502                 } else {
 503                     ret = -ret;
 504                 }
 505
 506                 trace_paio_complete(acb, acb->common.opaque, ret);
 507
 508                 /* remove the request */
 509                 *pacb = acb->next;
 510                 /* call the callback */
 511                 acb->common.cb(acb->common.opaque, ret);
 512                 qemu_aio_release(acb);
 513                 result = 1;
 514                 break;
 515             } else {
 516                 pacb = &acb->next;
 517             }
 518         }
 519     }
 520
 521     return result;
 522 }
 523
 524 static void posix_aio_read(void *opaque)
 525 {
 526     PosixAioState *s = opaque;
 527     union {
 528         struct qemu_signalfd_siginfo siginfo;
 529         char buf[128];
 530     } sig;
 531     size_t offset;
 532
 533     /* try to read from signalfd, don't freak out if we can't read anything */
 534     offset = 0;
 535     while (offset < 128) {
 536         ssize_t len;
 537
 538         len = read(s->fd, sig.buf + offset, 128 - offset);
 539         if (len == -1 && errno == EINTR)
 540             continue;
 541         if (len == -1 && errno == EAGAIN) {
 542             /* there is no natural reason for this to happen,
 543              * so we'll spin hard until we get everything just
 544              * to be on the safe side. */
 545             if (offset > 0)
 546                 continue;
 547         }
 548
 549         offset += len;
 550     }
 551
 552     posix_aio_process_queue(s);
 553 }
 554
 555 static int posix_aio_flush(void *opaque)
 556 {
 557     PosixAioState *s = opaque;
 558     return !!s->first_aio;
 559 }
 560
 561 static PosixAioState *posix_aio_state;
 562
 563 static void paio_remove(struct qemu_paiocb *acb)
 564 {
 565     struct qemu_paiocb **pacb;
 566
 567     /* remove the callback from the queue */
 568     pacb = &posix_aio_state->first_aio;
 569     for(;;) {
 570         if (*pacb == NULL) {
 571             fprintf(stderr, "paio_remove: aio request not found!\n");
 572             break;
 573         } else if (*pacb == acb) {
 574             *pacb = acb->next;
 575             qemu_aio_release(acb);
 576             break;
 577         }
 578         pacb = &(*pacb)->next;
 579     }
 580 }
 581
 582 static void paio_cancel(BlockDriverAIOCB *blockacb)
 583 {
 584     struct qemu_paiocb *acb = (struct qemu_paiocb *)blockacb;
 585     int active = 0;
 586
 587     trace_paio_cancel(acb, acb->common.opaque);
 588
 589     mutex_lock(&lock);
 590     if (!acb->active) {
 591         QTAILQ_REMOVE(&request_list, acb, node);
 592         acb->ret = -ECANCELED;
 593     } else if (acb->ret == -EINPROGRESS) {
 594         active = 1;
 595     }
 596     mutex_unlock(&lock);
 597
 598     if (active) {
 599         /* fail safe: if the aio could not be canceled, we wait for
 600            it */
 601         while (qemu_paio_error(acb) == EINPROGRESS)
 602             ;
 603     }
 604
 605     paio_remove(acb);
 606 }
 607
 608 static AIOPool raw_aio_pool = {
 609     .aiocb_size         = sizeof(struct qemu_paiocb),
 610     .cancel             = paio_cancel,
 611 };
 612
 613 BlockDriverAIOCB *paio_submit(BlockDriverState *bs, int fd,
 614         int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
 615         BlockDriverCompletionFunc *cb, void *opaque, int type)
 616 {
 617     struct qemu_paiocb *acb;
 618
 619     acb = qemu_aio_get(&raw_aio_pool, bs, cb, opaque);
 620     if (!acb)
 621         return NULL;
 622     acb->aio_type = type;
 623     acb->aio_fildes = fd;
 624     acb->ev_signo = SIGUSR2;
 625
 626     if (qiov) {
 627         acb->aio_iov = qiov->iov;
 628         acb->aio_niov = qiov->niov;
 629     }
 630     acb->aio_nbytes = nb_sectors * 512;
 631     acb->aio_offset = sector_num * 512;
 632
 633     acb->next = posix_aio_state->first_aio;
 634     posix_aio_state->first_aio = acb;
 635
 636     trace_paio_submit(acb, opaque, sector_num, nb_sectors, type);
 637     qemu_paio_submit(acb);
 638     return &acb->common;
 639 }
 640
 641 BlockDriverAIOCB *paio_ioctl(BlockDriverState *bs, int fd,
 642         unsigned long int req, void *buf,
 643         BlockDriverCompletionFunc *cb, void *opaque)
 644 {
 645     struct qemu_paiocb *acb;
 646
 647     acb = qemu_aio_get(&raw_aio_pool, bs, cb, opaque);
 648     if (!acb)
 649         return NULL;
 650     acb->aio_type = QEMU_AIO_IOCTL;
 651     acb->aio_fildes = fd;
 652     acb->ev_signo = SIGUSR2;
 653     acb->aio_offset = 0;
 654     acb->aio_ioctl_buf = buf;
 655     acb->aio_ioctl_cmd = req;
 656
 657     acb->next = posix_aio_state->first_aio;
 658     posix_aio_state->first_aio = acb;
 659
 660     qemu_paio_submit(acb);
 661     return &acb->common;
 662 }
 663
 664 int paio_init(void)
 665 {
 666     sigset_t mask;
 667     PosixAioState *s;
 668     int ret;
 669
 670     if (posix_aio_state)
 671         return 0;
 672
 673     s = g_malloc(sizeof(PosixAioState));
 674
 675     /* Make sure to block AIO signal */
 676     sigemptyset(&mask);
 677     sigaddset(&mask, SIGUSR2);
 678     sigprocmask(SIG_BLOCK, &mask, NULL);
 679
 680     s->first_aio = NULL;
 681     s->fd = qemu_signalfd(&mask);
 682     if (s->fd == -1) {
 683         fprintf(stderr, "failed to create signalfd\n");
 684         return -1;
 685     }
 686
 687     fcntl(s->fd, F_SETFL, O_NONBLOCK);
 688
 689     qemu_aio_set_fd_handler(s->fd, posix_aio_read, NULL, posix_aio_flush,
 690         posix_aio_process_queue, s);
 691
 692     ret = pthread_attr_init(&attr);
 693     if (ret)
 694         die2(ret, "pthread_attr_init");
 695
 696     ret = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
 697     if (ret)
 698         die2(ret, "pthread_attr_setdetachstate");
 699
 700     QTAILQ_INIT(&request_list);
 701     new_thread_bh = qemu_bh_new(spawn_thread_bh_fn, NULL);
 702
 703     posix_aio_state = s;
 704     return 0;
 705 }