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 <signal.h>
  21 #include <string.h>
  22 #include <stdlib.h>
  23 #include <stdio.h>
  24
  25 #include "qemu-queue.h"
  26 #include "osdep.h"
  27 #include "qemu-common.h"
  28 #include "block_int.h"
  29 #include "compatfd.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 fd;
  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 size_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          * successfull 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 size_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 size_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 static size_t handle_aiocb_rw_linear(struct qemu_paiocb *aiocb, char *buf)
 202 {
 203     size_t offset = 0;
 204     size_t len;
 205
 206     while (offset < aiocb->aio_nbytes) {
 207          if (aiocb->aio_type & QEMU_AIO_WRITE)
 208              len = pwrite(aiocb->aio_fildes,
 209                           (const char *)buf + offset,
 210                           aiocb->aio_nbytes - offset,
 211                           aiocb->aio_offset + offset);
 212          else
 213              len = pread(aiocb->aio_fildes,
 214                          buf + offset,
 215                          aiocb->aio_nbytes - offset,
 216                          aiocb->aio_offset + offset);
 217
 218          if (len == -1 && errno == EINTR)
 219              continue;
 220          else if (len == -1) {
 221              offset = -errno;
 222              break;
 223          } else if (len == 0)
 224              break;
 225
 226          offset += len;
 227     }
 228
 229     return offset;
 230 }
 231
 232 static size_t handle_aiocb_rw(struct qemu_paiocb *aiocb)
 233 {
 234     size_t nbytes;
 235     char *buf;
 236
 237     if (!(aiocb->aio_type & QEMU_AIO_MISALIGNED)) {
 238         /*
 239          * If there is just a single buffer, and it is properly aligned
 240          * we can just use plain pread/pwrite without any problems.
 241          */
 242         if (aiocb->aio_niov == 1)
 243              return handle_aiocb_rw_linear(aiocb, aiocb->aio_iov->iov_base);
 244
 245         /*
 246          * We have more than one iovec, and all are properly aligned.
 247          *
 248          * Try preadv/pwritev first and fall back to linearizing the
 249          * buffer if it's not supported.
 250          */
 251         if (preadv_present) {
 252             nbytes = handle_aiocb_rw_vector(aiocb);
 253             if (nbytes == aiocb->aio_nbytes)
 254                 return nbytes;
 255             if (nbytes < 0 && nbytes != -ENOSYS)
 256                 return nbytes;
 257             preadv_present = 0;
 258         }
 259
 260         /*
 261          * XXX(hch): short read/write.  no easy way to handle the reminder
 262          * using these interfaces.  For now retry using plain
 263          * pread/pwrite?
 264          */
 265     }
 266
 267     /*
 268      * Ok, we have to do it the hard way, copy all segments into
 269      * a single aligned buffer.
 270      */
 271     buf = qemu_memalign(512, aiocb->aio_nbytes);
 272     if (aiocb->aio_type & QEMU_AIO_WRITE) {
 273         char *p = buf;
 274         int i;
 275
 276         for (i = 0; i < aiocb->aio_niov; ++i) {
 277             memcpy(p, aiocb->aio_iov[i].iov_base, aiocb->aio_iov[i].iov_len);
 278             p += aiocb->aio_iov[i].iov_len;
 279         }
 280     }
 281
 282     nbytes = handle_aiocb_rw_linear(aiocb, buf);
 283     if (!(aiocb->aio_type & QEMU_AIO_WRITE)) {
 284         char *p = buf;
 285         size_t count = aiocb->aio_nbytes, copy;
 286         int i;
 287
 288         for (i = 0; i < aiocb->aio_niov && count; ++i) {
 289             copy = count;
 290             if (copy > aiocb->aio_iov[i].iov_len)
 291                 copy = aiocb->aio_iov[i].iov_len;
 292             memcpy(aiocb->aio_iov[i].iov_base, p, copy);
 293             p     += copy;
 294             count -= copy;
 295         }
 296     }
 297     qemu_vfree(buf);
 298
 299     return nbytes;
 300 }
 301
 302 static void *aio_thread(void *unused)
 303 {
 304     pid_t pid;
 305
 306     pid = getpid();
 307
 308     while (1) {
 309         struct qemu_paiocb *aiocb;
 310         size_t ret = 0;
 311         qemu_timeval tv;
 312         struct timespec ts;
 313
 314         qemu_gettimeofday(&tv);
 315         ts.tv_sec = tv.tv_sec + 10;
 316         ts.tv_nsec = 0;
 317
 318         mutex_lock(&lock);
 319
 320         while (QTAILQ_EMPTY(&request_list) &&
 321                !(ret == ETIMEDOUT)) {
 322             ret = cond_timedwait(&cond, &lock, &ts);
 323         }
 324
 325         if (QTAILQ_EMPTY(&request_list))
 326             break;
 327
 328         aiocb = QTAILQ_FIRST(&request_list);
 329         QTAILQ_REMOVE(&request_list, aiocb, node);
 330         aiocb->active = 1;
 331         idle_threads--;
 332         mutex_unlock(&lock);
 333
 334         switch (aiocb->aio_type & QEMU_AIO_TYPE_MASK) {
 335         case QEMU_AIO_READ:
 336         case QEMU_AIO_WRITE:
 337                 ret = handle_aiocb_rw(aiocb);
 338                 break;
 339         case QEMU_AIO_FLUSH:
 340                 ret = handle_aiocb_flush(aiocb);
 341                 break;
 342         case QEMU_AIO_IOCTL:
 343                 ret = handle_aiocb_ioctl(aiocb);
 344                 break;
 345         default:
 346                 fprintf(stderr, "invalid aio request (0x%x)\n", aiocb->aio_type);
 347                 ret = -EINVAL;
 348                 break;
 349         }
 350
 351         mutex_lock(&lock);
 352         aiocb->ret = ret;
 353         idle_threads++;
 354         mutex_unlock(&lock);
 355
 356         if (kill(pid, aiocb->ev_signo)) die("kill failed");
 357     }
 358
 359     idle_threads--;
 360     cur_threads--;
 361     mutex_unlock(&lock);
 362
 363     return NULL;
 364 }
 365
 366 static void spawn_thread(void)
 367 {
 368     sigset_t set, oldset;
 369
 370     cur_threads++;
 371     idle_threads++;
 372
 373     /* block all signals */
 374     if (sigfillset(&set)) die("sigfillset");
 375     if (sigprocmask(SIG_SETMASK, &set, &oldset)) die("sigprocmask");
 376
 377     thread_create(&thread_id, &attr, aio_thread, NULL);
 378
 379     if (sigprocmask(SIG_SETMASK, &oldset, NULL)) die("sigprocmask restore");
 380 }
 381
 382 static void qemu_paio_submit(struct qemu_paiocb *aiocb)
 383 {
 384     aiocb->ret = -EINPROGRESS;
 385     aiocb->active = 0;
 386     mutex_lock(&lock);
 387     if (idle_threads == 0 && cur_threads < max_threads)
 388         spawn_thread();
 389     QTAILQ_INSERT_TAIL(&request_list, aiocb, node);
 390     mutex_unlock(&lock);
 391     cond_signal(&cond);
 392 }
 393
 394 static ssize_t qemu_paio_return(struct qemu_paiocb *aiocb)
 395 {
 396     ssize_t ret;
 397
 398     mutex_lock(&lock);
 399     ret = aiocb->ret;
 400     mutex_unlock(&lock);
 401
 402     return ret;
 403 }
 404
 405 static int qemu_paio_error(struct qemu_paiocb *aiocb)
 406 {
 407     ssize_t ret = qemu_paio_return(aiocb);
 408
 409     if (ret < 0)
 410         ret = -ret;
 411     else
 412         ret = 0;
 413
 414     return ret;
 415 }
 416
 417 static void posix_aio_read(void *opaque)
 418 {
 419     PosixAioState *s = opaque;
 420     struct qemu_paiocb *acb, **pacb;
 421     int ret;
 422     union {
 423         struct qemu_signalfd_siginfo siginfo;
 424         char buf[128];
 425     } sig;
 426     size_t offset;
 427
 428     /* try to read from signalfd, don't freak out if we can't read anything */
 429     offset = 0;
 430     while (offset < 128) {
 431         ssize_t len;
 432
 433         len = read(s->fd, sig.buf + offset, 128 - offset);
 434         if (len == -1 && errno == EINTR)
 435             continue;
 436         if (len == -1 && errno == EAGAIN) {
 437             /* there is no natural reason for this to happen,
 438              * so we'll spin hard until we get everything just
 439              * to be on the safe side. */
 440             if (offset > 0)
 441                 continue;
 442         }
 443
 444         offset += len;
 445     }
 446
 447     for(;;) {
 448         pacb = &s->first_aio;
 449         for(;;) {
 450             acb = *pacb;
 451             if (!acb)
 452                 goto the_end;
 453             ret = qemu_paio_error(acb);
 454             if (ret == ECANCELED) {
 455                 /* remove the request */
 456                 *pacb = acb->next;
 457                 qemu_aio_release(acb);
 458             } else if (ret != EINPROGRESS) {
 459                 /* end of aio */
 460                 if (ret == 0) {
 461                     ret = qemu_paio_return(acb);
 462                     if (ret == acb->aio_nbytes)
 463                         ret = 0;
 464                     else
 465                         ret = -EINVAL;
 466                 } else {
 467                     ret = -ret;
 468                 }
 469                 /* remove the request */
 470                 *pacb = acb->next;
 471                 /* call the callback */
 472                 acb->common.cb(acb->common.opaque, ret);
 473                 qemu_aio_release(acb);
 474                 break;
 475             } else {
 476                 pacb = &acb->next;
 477             }
 478         }
 479     }
 480  the_end: ;
 481 }
 482
 483 static int posix_aio_flush(void *opaque)
 484 {
 485     PosixAioState *s = opaque;
 486     return !!s->first_aio;
 487 }
 488
 489 static PosixAioState *posix_aio_state;
 490
 491 static void paio_remove(struct qemu_paiocb *acb)
 492 {
 493     struct qemu_paiocb **pacb;
 494
 495     /* remove the callback from the queue */
 496     pacb = &posix_aio_state->first_aio;
 497     for(;;) {
 498         if (*pacb == NULL) {
 499             fprintf(stderr, "paio_remove: aio request not found!\n");
 500             break;
 501         } else if (*pacb == acb) {
 502             *pacb = acb->next;
 503             qemu_aio_release(acb);
 504             break;
 505         }
 506         pacb = &(*pacb)->next;
 507     }
 508 }
 509
 510 static void paio_cancel(BlockDriverAIOCB *blockacb)
 511 {
 512     struct qemu_paiocb *acb = (struct qemu_paiocb *)blockacb;
 513     int active = 0;
 514
 515     mutex_lock(&lock);
 516     if (!acb->active) {
 517         QTAILQ_REMOVE(&request_list, acb, node);
 518         acb->ret = -ECANCELED;
 519     } else if (acb->ret == -EINPROGRESS) {
 520         active = 1;
 521     }
 522     mutex_unlock(&lock);
 523
 524     if (active) {
 525         /* fail safe: if the aio could not be canceled, we wait for
 526            it */
 527         while (qemu_paio_error(acb) == EINPROGRESS)
 528             ;
 529     }
 530
 531     paio_remove(acb);
 532 }
 533
 534 static AIOPool raw_aio_pool = {
 535     .aiocb_size         = sizeof(struct qemu_paiocb),
 536     .cancel             = paio_cancel,
 537 };
 538
 539 BlockDriverAIOCB *paio_submit(BlockDriverState *bs, void *aio_ctx, int fd,
 540         int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
 541         BlockDriverCompletionFunc *cb, void *opaque, int type)
 542 {
 543     struct qemu_paiocb *acb;
 544
 545     acb = qemu_aio_get(&raw_aio_pool, bs, cb, opaque);
 546     if (!acb)
 547         return NULL;
 548     acb->aio_type = type;
 549     acb->aio_fildes = fd;
 550     acb->ev_signo = SIGUSR2;
 551     if (qiov) {
 552         acb->aio_iov = qiov->iov;
 553         acb->aio_niov = qiov->niov;
 554     }
 555     acb->aio_nbytes = nb_sectors * 512;
 556     acb->aio_offset = sector_num * 512;
 557
 558     acb->next = posix_aio_state->first_aio;
 559     posix_aio_state->first_aio = acb;
 560
 561     qemu_paio_submit(acb);
 562     return &acb->common;
 563 }
 564
 565 BlockDriverAIOCB *paio_ioctl(BlockDriverState *bs, int fd,
 566         unsigned long int req, void *buf,
 567         BlockDriverCompletionFunc *cb, void *opaque)
 568 {
 569     struct qemu_paiocb *acb;
 570
 571     acb = qemu_aio_get(&raw_aio_pool, bs, cb, opaque);
 572     if (!acb)
 573         return NULL;
 574     acb->aio_type = QEMU_AIO_IOCTL;
 575     acb->aio_fildes = fd;
 576     acb->ev_signo = SIGUSR2;
 577     acb->aio_offset = 0;
 578     acb->aio_ioctl_buf = buf;
 579     acb->aio_ioctl_cmd = req;
 580
 581     acb->next = posix_aio_state->first_aio;
 582     posix_aio_state->first_aio = acb;
 583
 584     qemu_paio_submit(acb);
 585     return &acb->common;
 586 }
 587
 588 void *paio_init(void)
 589 {
 590     sigset_t mask;
 591     PosixAioState *s;
 592     int ret;
 593
 594     if (posix_aio_state)
 595         return posix_aio_state;
 596
 597     s = qemu_malloc(sizeof(PosixAioState));
 598
 599     /* Make sure to block AIO signal */
 600     sigemptyset(&mask);
 601     sigaddset(&mask, SIGUSR2);
 602     sigprocmask(SIG_BLOCK, &mask, NULL);
 603
 604     s->first_aio = NULL;
 605     s->fd = qemu_signalfd(&mask);
 606     if (s->fd == -1) {
 607         fprintf(stderr, "failed to create signalfd\n");
 608         return NULL;
 609     }
 610
 611     fcntl(s->fd, F_SETFL, O_NONBLOCK);
 612
 613     qemu_aio_set_fd_handler(s->fd, posix_aio_read, NULL, posix_aio_flush, s);
 614
 615     ret = pthread_attr_init(&attr);
 616     if (ret)
 617         die2(ret, "pthread_attr_init");
 618
 619     ret = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
 620     if (ret)
 621         die2(ret, "pthread_attr_setdetachstate");
 622
 623     QTAILQ_INIT(&request_list);
 624
 625     posix_aio_state = s;
 626
 627     return posix_aio_state;
 628 }