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