2 * QEMU posix-aio emulation
4 * Copyright IBM, Corp. 2008
7 * Anthony Liguori <aliguori@us.ibm.com>
9 * This work is licensed under the terms of the GNU GPL, version 2. See
10 * the COPYING file in the top-level directory.
14 #include <sys/ioctl.h>
15 #include <sys/types.h>
25 #include "qemu-queue.h"
27 #include "qemu-common.h"
28 #include "block_int.h"
30 #include "block/raw-posix-aio.h"
34 BlockDriverAIOCB common;
37 struct iovec *aio_iov;
42 #define aio_ioctl_cmd aio_nbytes /* for QEMU_AIO_IOCTL */
46 QTAILQ_ENTRY(qemu_paiocb) node;
50 struct qemu_paiocb *next;
53 typedef struct PosixAioState {
55 struct qemu_paiocb *first_aio;
59 static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
60 static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
61 static pthread_t thread_id;
62 static pthread_attr_t attr;
63 static int max_threads = 64;
64 static int cur_threads = 0;
65 static int idle_threads = 0;
66 static QTAILQ_HEAD(, qemu_paiocb) request_list;
69 static int preadv_present = 1;
71 static int preadv_present = 0;
74 static void die2(int err, const char *what)
76 fprintf(stderr, "%s failed: %s\n", what, strerror(err));
80 static void die(const char *what)
85 static void mutex_lock(pthread_mutex_t *mutex)
87 int ret = pthread_mutex_lock(mutex);
88 if (ret) die2(ret, "pthread_mutex_lock");
91 static void mutex_unlock(pthread_mutex_t *mutex)
93 int ret = pthread_mutex_unlock(mutex);
94 if (ret) die2(ret, "pthread_mutex_unlock");
97 static int cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex,
100 int ret = pthread_cond_timedwait(cond, mutex, ts);
101 if (ret && ret != ETIMEDOUT) die2(ret, "pthread_cond_timedwait");
105 static void cond_signal(pthread_cond_t *cond)
107 int ret = pthread_cond_signal(cond);
108 if (ret) die2(ret, "pthread_cond_signal");
111 static void thread_create(pthread_t *thread, pthread_attr_t *attr,
112 void *(*start_routine)(void*), void *arg)
114 int ret = pthread_create(thread, attr, start_routine, arg);
115 if (ret) die2(ret, "pthread_create");
118 static size_t handle_aiocb_ioctl(struct qemu_paiocb *aiocb)
122 ret = ioctl(aiocb->aio_fildes, aiocb->aio_ioctl_cmd, aiocb->aio_ioctl_buf);
127 * This looks weird, but the aio code only consideres a request
128 * successfull if it has written the number full number of bytes.
130 * Now we overload aio_nbytes as aio_ioctl_cmd for the ioctl command,
131 * so in fact we return the ioctl command here to make posix_aio_read()
134 return aiocb->aio_nbytes;
137 static size_t handle_aiocb_flush(struct qemu_paiocb *aiocb)
141 ret = qemu_fdatasync(aiocb->aio_fildes);
150 qemu_preadv(int fd, const struct iovec *iov, int nr_iov, off_t offset)
152 return preadv(fd, iov, nr_iov, offset);
156 qemu_pwritev(int fd, const struct iovec *iov, int nr_iov, off_t offset)
158 return pwritev(fd, iov, nr_iov, offset);
164 qemu_preadv(int fd, const struct iovec *iov, int nr_iov, off_t offset)
170 qemu_pwritev(int fd, const struct iovec *iov, int nr_iov, off_t offset)
177 static size_t handle_aiocb_rw_vector(struct qemu_paiocb *aiocb)
183 if (aiocb->aio_type & QEMU_AIO_WRITE)
184 len = qemu_pwritev(aiocb->aio_fildes,
187 aiocb->aio_offset + offset);
189 len = qemu_preadv(aiocb->aio_fildes,
192 aiocb->aio_offset + offset);
193 } while (len == -1 && errno == EINTR);
200 static size_t handle_aiocb_rw_linear(struct qemu_paiocb *aiocb, char *buf)
205 while (offset < aiocb->aio_nbytes) {
206 if (aiocb->aio_type & QEMU_AIO_WRITE)
207 len = pwrite(aiocb->aio_fildes,
208 (const char *)buf + offset,
209 aiocb->aio_nbytes - offset,
210 aiocb->aio_offset + offset);
212 len = pread(aiocb->aio_fildes,
214 aiocb->aio_nbytes - offset,
215 aiocb->aio_offset + offset);
217 if (len == -1 && errno == EINTR)
219 else if (len == -1) {
231 static size_t handle_aiocb_rw(struct qemu_paiocb *aiocb)
236 if (!(aiocb->aio_type & QEMU_AIO_MISALIGNED)) {
238 * If there is just a single buffer, and it is properly aligned
239 * we can just use plain pread/pwrite without any problems.
241 if (aiocb->aio_niov == 1)
242 return handle_aiocb_rw_linear(aiocb, aiocb->aio_iov->iov_base);
245 * We have more than one iovec, and all are properly aligned.
247 * Try preadv/pwritev first and fall back to linearizing the
248 * buffer if it's not supported.
250 if (preadv_present) {
251 nbytes = handle_aiocb_rw_vector(aiocb);
252 if (nbytes == aiocb->aio_nbytes)
254 if (nbytes < 0 && nbytes != -ENOSYS)
260 * XXX(hch): short read/write. no easy way to handle the reminder
261 * using these interfaces. For now retry using plain
267 * Ok, we have to do it the hard way, copy all segments into
268 * a single aligned buffer.
270 buf = qemu_memalign(512, aiocb->aio_nbytes);
271 if (aiocb->aio_type & QEMU_AIO_WRITE) {
275 for (i = 0; i < aiocb->aio_niov; ++i) {
276 memcpy(p, aiocb->aio_iov[i].iov_base, aiocb->aio_iov[i].iov_len);
277 p += aiocb->aio_iov[i].iov_len;
281 nbytes = handle_aiocb_rw_linear(aiocb, buf);
282 if (!(aiocb->aio_type & QEMU_AIO_WRITE)) {
284 size_t count = aiocb->aio_nbytes, copy;
287 for (i = 0; i < aiocb->aio_niov && count; ++i) {
289 if (copy > aiocb->aio_iov[i].iov_len)
290 copy = aiocb->aio_iov[i].iov_len;
291 memcpy(aiocb->aio_iov[i].iov_base, p, copy);
301 static void *aio_thread(void *unused)
308 /* block all signals */
309 if (sigfillset(&set)) die("sigfillset");
310 if (sigprocmask(SIG_BLOCK, &set, NULL)) die("sigprocmask");
313 struct qemu_paiocb *aiocb;
318 qemu_gettimeofday(&tv);
319 ts.tv_sec = tv.tv_sec + 10;
324 while (QTAILQ_EMPTY(&request_list) &&
325 !(ret == ETIMEDOUT)) {
326 ret = cond_timedwait(&cond, &lock, &ts);
329 if (QTAILQ_EMPTY(&request_list))
332 aiocb = QTAILQ_FIRST(&request_list);
333 QTAILQ_REMOVE(&request_list, aiocb, node);
338 switch (aiocb->aio_type & QEMU_AIO_TYPE_MASK) {
341 ret = handle_aiocb_rw(aiocb);
344 ret = handle_aiocb_flush(aiocb);
347 ret = handle_aiocb_ioctl(aiocb);
350 fprintf(stderr, "invalid aio request (0x%x)\n", aiocb->aio_type);
360 if (kill(pid, aiocb->ev_signo)) die("kill failed");
370 static void spawn_thread(void)
374 thread_create(&thread_id, &attr, aio_thread, NULL);
377 static void qemu_paio_submit(struct qemu_paiocb *aiocb)
379 aiocb->ret = -EINPROGRESS;
382 if (idle_threads == 0 && cur_threads < max_threads)
384 QTAILQ_INSERT_TAIL(&request_list, aiocb, node);
389 static ssize_t qemu_paio_return(struct qemu_paiocb *aiocb)
400 static int qemu_paio_error(struct qemu_paiocb *aiocb)
402 ssize_t ret = qemu_paio_return(aiocb);
412 static void posix_aio_read(void *opaque)
414 PosixAioState *s = opaque;
415 struct qemu_paiocb *acb, **pacb;
419 /* read all bytes from signal pipe */
423 len = read(s->rfd, bytes, sizeof(bytes));
424 if (len == -1 && errno == EINTR)
425 continue; /* try again */
426 if (len == sizeof(bytes))
427 continue; /* more to read */
432 pacb = &s->first_aio;
437 ret = qemu_paio_error(acb);
438 if (ret == ECANCELED) {
439 /* remove the request */
441 qemu_aio_release(acb);
442 } else if (ret != EINPROGRESS) {
445 ret = qemu_paio_return(acb);
446 if (ret == acb->aio_nbytes)
453 /* remove the request */
455 /* call the callback */
456 acb->common.cb(acb->common.opaque, ret);
457 qemu_aio_release(acb);
467 static int posix_aio_flush(void *opaque)
469 PosixAioState *s = opaque;
470 return !!s->first_aio;
473 static PosixAioState *posix_aio_state;
475 static void aio_signal_handler(int signum)
477 if (posix_aio_state) {
480 write(posix_aio_state->wfd, &byte, sizeof(byte));
486 static void paio_remove(struct qemu_paiocb *acb)
488 struct qemu_paiocb **pacb;
490 /* remove the callback from the queue */
491 pacb = &posix_aio_state->first_aio;
494 fprintf(stderr, "paio_remove: aio request not found!\n");
496 } else if (*pacb == acb) {
498 qemu_aio_release(acb);
501 pacb = &(*pacb)->next;
505 static void paio_cancel(BlockDriverAIOCB *blockacb)
507 struct qemu_paiocb *acb = (struct qemu_paiocb *)blockacb;
512 QTAILQ_REMOVE(&request_list, acb, node);
513 acb->ret = -ECANCELED;
514 } else if (acb->ret == -EINPROGRESS) {
520 /* fail safe: if the aio could not be canceled, we wait for
522 while (qemu_paio_error(acb) == EINPROGRESS)
529 static AIOPool raw_aio_pool = {
530 .aiocb_size = sizeof(struct qemu_paiocb),
531 .cancel = paio_cancel,
534 BlockDriverAIOCB *paio_submit(BlockDriverState *bs, void *aio_ctx, int fd,
535 int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
536 BlockDriverCompletionFunc *cb, void *opaque, int type)
538 struct qemu_paiocb *acb;
540 acb = qemu_aio_get(&raw_aio_pool, bs, cb, opaque);
543 acb->aio_type = type;
544 acb->aio_fildes = fd;
545 acb->ev_signo = SIGUSR2;
547 acb->aio_iov = qiov->iov;
548 acb->aio_niov = qiov->niov;
550 acb->aio_nbytes = nb_sectors * 512;
551 acb->aio_offset = sector_num * 512;
553 acb->next = posix_aio_state->first_aio;
554 posix_aio_state->first_aio = acb;
556 qemu_paio_submit(acb);
560 BlockDriverAIOCB *paio_ioctl(BlockDriverState *bs, int fd,
561 unsigned long int req, void *buf,
562 BlockDriverCompletionFunc *cb, void *opaque)
564 struct qemu_paiocb *acb;
566 acb = qemu_aio_get(&raw_aio_pool, bs, cb, opaque);
569 acb->aio_type = QEMU_AIO_IOCTL;
570 acb->aio_fildes = fd;
571 acb->ev_signo = SIGUSR2;
573 acb->aio_ioctl_buf = buf;
574 acb->aio_ioctl_cmd = req;
576 acb->next = posix_aio_state->first_aio;
577 posix_aio_state->first_aio = acb;
579 qemu_paio_submit(acb);
583 void *paio_init(void)
585 struct sigaction act;
591 return posix_aio_state;
593 s = qemu_malloc(sizeof(PosixAioState));
595 sigfillset(&act.sa_mask);
596 act.sa_flags = 0; /* do not restart syscalls to interrupt select() */
597 act.sa_handler = aio_signal_handler;
598 sigaction(SIGUSR2, &act, NULL);
601 if (pipe(fds) == -1) {
602 fprintf(stderr, "failed to create pipe\n");
609 fcntl(s->rfd, F_SETFL, O_NONBLOCK);
610 fcntl(s->wfd, F_SETFL, O_NONBLOCK);
612 qemu_aio_set_fd_handler(s->rfd, posix_aio_read, NULL, posix_aio_flush, s);
614 ret = pthread_attr_init(&attr);
616 die2(ret, "pthread_attr_init");
618 ret = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
620 die2(ret, "pthread_attr_setdetachstate");
622 QTAILQ_INIT(&request_list);
626 return posix_aio_state;