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 "sys-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 TAILQ_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 TAILQ_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;
140 qemu_preadv(int fd, const struct iovec *iov, int nr_iov, off_t offset)
142 return preadv(fd, iov, nr_iov, offset);
146 qemu_pwritev(int fd, const struct iovec *iov, int nr_iov, off_t offset)
148 return pwritev(fd, iov, nr_iov, offset);
154 qemu_preadv(int fd, const struct iovec *iov, int nr_iov, off_t offset)
160 qemu_pwritev(int fd, const struct iovec *iov, int nr_iov, off_t offset)
167 static size_t handle_aiocb_rw_vector(struct qemu_paiocb *aiocb)
173 if (aiocb->aio_type & QEMU_AIO_WRITE)
174 len = qemu_pwritev(aiocb->aio_fildes,
177 aiocb->aio_offset + offset);
179 len = qemu_preadv(aiocb->aio_fildes,
182 aiocb->aio_offset + offset);
183 } while (len == -1 && errno == EINTR);
190 static size_t handle_aiocb_rw_linear(struct qemu_paiocb *aiocb, char *buf)
195 while (offset < aiocb->aio_nbytes) {
196 if (aiocb->aio_type & QEMU_AIO_WRITE)
197 len = pwrite(aiocb->aio_fildes,
198 (const char *)buf + offset,
199 aiocb->aio_nbytes - offset,
200 aiocb->aio_offset + offset);
202 len = pread(aiocb->aio_fildes,
204 aiocb->aio_nbytes - offset,
205 aiocb->aio_offset + offset);
207 if (len == -1 && errno == EINTR)
209 else if (len == -1) {
221 static size_t handle_aiocb_rw(struct qemu_paiocb *aiocb)
226 if (!(aiocb->aio_type & QEMU_AIO_MISALIGNED)) {
228 * If there is just a single buffer, and it is properly aligned
229 * we can just use plain pread/pwrite without any problems.
231 if (aiocb->aio_niov == 1)
232 return handle_aiocb_rw_linear(aiocb, aiocb->aio_iov->iov_base);
235 * We have more than one iovec, and all are properly aligned.
237 * Try preadv/pwritev first and fall back to linearizing the
238 * buffer if it's not supported.
240 if (preadv_present) {
241 nbytes = handle_aiocb_rw_vector(aiocb);
242 if (nbytes == aiocb->aio_nbytes)
244 if (nbytes < 0 && nbytes != -ENOSYS)
250 * XXX(hch): short read/write. no easy way to handle the reminder
251 * using these interfaces. For now retry using plain
257 * Ok, we have to do it the hard way, copy all segments into
258 * a single aligned buffer.
260 buf = qemu_memalign(512, aiocb->aio_nbytes);
261 if (aiocb->aio_type & QEMU_AIO_WRITE) {
265 for (i = 0; i < aiocb->aio_niov; ++i) {
266 memcpy(p, aiocb->aio_iov[i].iov_base, aiocb->aio_iov[i].iov_len);
267 p += aiocb->aio_iov[i].iov_len;
271 nbytes = handle_aiocb_rw_linear(aiocb, buf);
272 if (!(aiocb->aio_type & QEMU_AIO_WRITE)) {
274 size_t count = aiocb->aio_nbytes, copy;
277 for (i = 0; i < aiocb->aio_niov && count; ++i) {
279 if (copy > aiocb->aio_iov[i].iov_len)
280 copy = aiocb->aio_iov[i].iov_len;
281 memcpy(aiocb->aio_iov[i].iov_base, p, copy);
291 static void *aio_thread(void *unused)
298 /* block all signals */
299 if (sigfillset(&set)) die("sigfillset");
300 if (sigprocmask(SIG_BLOCK, &set, NULL)) die("sigprocmask");
303 struct qemu_paiocb *aiocb;
308 qemu_gettimeofday(&tv);
309 ts.tv_sec = tv.tv_sec + 10;
314 while (TAILQ_EMPTY(&request_list) &&
315 !(ret == ETIMEDOUT)) {
316 ret = cond_timedwait(&cond, &lock, &ts);
319 if (TAILQ_EMPTY(&request_list))
322 aiocb = TAILQ_FIRST(&request_list);
323 TAILQ_REMOVE(&request_list, aiocb, node);
328 switch (aiocb->aio_type & QEMU_AIO_TYPE_MASK) {
331 ret = handle_aiocb_rw(aiocb);
334 ret = handle_aiocb_ioctl(aiocb);
337 fprintf(stderr, "invalid aio request (0x%x)\n", aiocb->aio_type);
347 if (kill(pid, aiocb->ev_signo)) die("kill failed");
357 static void spawn_thread(void)
361 thread_create(&thread_id, &attr, aio_thread, NULL);
364 static void qemu_paio_submit(struct qemu_paiocb *aiocb)
366 aiocb->ret = -EINPROGRESS;
369 if (idle_threads == 0 && cur_threads < max_threads)
371 TAILQ_INSERT_TAIL(&request_list, aiocb, node);
376 static ssize_t qemu_paio_return(struct qemu_paiocb *aiocb)
387 static int qemu_paio_error(struct qemu_paiocb *aiocb)
389 ssize_t ret = qemu_paio_return(aiocb);
399 static void posix_aio_read(void *opaque)
401 PosixAioState *s = opaque;
402 struct qemu_paiocb *acb, **pacb;
406 /* read all bytes from signal pipe */
410 len = read(s->rfd, bytes, sizeof(bytes));
411 if (len == -1 && errno == EINTR)
412 continue; /* try again */
413 if (len == sizeof(bytes))
414 continue; /* more to read */
419 pacb = &s->first_aio;
424 ret = qemu_paio_error(acb);
425 if (ret == ECANCELED) {
426 /* remove the request */
428 qemu_aio_release(acb);
429 } else if (ret != EINPROGRESS) {
432 ret = qemu_paio_return(acb);
433 if (ret == acb->aio_nbytes)
440 /* remove the request */
442 /* call the callback */
443 acb->common.cb(acb->common.opaque, ret);
444 qemu_aio_release(acb);
454 static int posix_aio_flush(void *opaque)
456 PosixAioState *s = opaque;
457 return !!s->first_aio;
460 static PosixAioState *posix_aio_state;
462 static void aio_signal_handler(int signum)
464 if (posix_aio_state) {
467 write(posix_aio_state->wfd, &byte, sizeof(byte));
473 static void paio_remove(struct qemu_paiocb *acb)
475 struct qemu_paiocb **pacb;
477 /* remove the callback from the queue */
478 pacb = &posix_aio_state->first_aio;
481 fprintf(stderr, "paio_remove: aio request not found!\n");
483 } else if (*pacb == acb) {
485 qemu_aio_release(acb);
488 pacb = &(*pacb)->next;
492 static void paio_cancel(BlockDriverAIOCB *blockacb)
494 struct qemu_paiocb *acb = (struct qemu_paiocb *)blockacb;
499 TAILQ_REMOVE(&request_list, acb, node);
500 acb->ret = -ECANCELED;
501 } else if (acb->ret == -EINPROGRESS) {
507 /* fail safe: if the aio could not be canceled, we wait for
509 while (qemu_paio_error(acb) == EINPROGRESS)
516 static AIOPool raw_aio_pool = {
517 .aiocb_size = sizeof(struct qemu_paiocb),
518 .cancel = paio_cancel,
521 BlockDriverAIOCB *paio_submit(BlockDriverState *bs, void *aio_ctx, int fd,
522 int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
523 BlockDriverCompletionFunc *cb, void *opaque, int type)
525 struct qemu_paiocb *acb;
527 acb = qemu_aio_get(&raw_aio_pool, bs, cb, opaque);
530 acb->aio_type = type;
531 acb->aio_fildes = fd;
532 acb->ev_signo = SIGUSR2;
533 acb->aio_iov = qiov->iov;
534 acb->aio_niov = qiov->niov;
535 acb->aio_nbytes = nb_sectors * 512;
536 acb->aio_offset = sector_num * 512;
538 acb->next = posix_aio_state->first_aio;
539 posix_aio_state->first_aio = acb;
541 qemu_paio_submit(acb);
545 BlockDriverAIOCB *paio_ioctl(BlockDriverState *bs, int fd,
546 unsigned long int req, void *buf,
547 BlockDriverCompletionFunc *cb, void *opaque)
549 struct qemu_paiocb *acb;
551 acb = qemu_aio_get(&raw_aio_pool, bs, cb, opaque);
554 acb->aio_type = QEMU_AIO_IOCTL;
555 acb->aio_fildes = fd;
556 acb->ev_signo = SIGUSR2;
558 acb->aio_ioctl_buf = buf;
559 acb->aio_ioctl_cmd = req;
561 acb->next = posix_aio_state->first_aio;
562 posix_aio_state->first_aio = acb;
564 qemu_paio_submit(acb);
568 void *paio_init(void)
570 struct sigaction act;
576 return posix_aio_state;
578 s = qemu_malloc(sizeof(PosixAioState));
580 sigfillset(&act.sa_mask);
581 act.sa_flags = 0; /* do not restart syscalls to interrupt select() */
582 act.sa_handler = aio_signal_handler;
583 sigaction(SIGUSR2, &act, NULL);
586 if (pipe(fds) == -1) {
587 fprintf(stderr, "failed to create pipe\n");
594 fcntl(s->rfd, F_SETFL, O_NONBLOCK);
595 fcntl(s->wfd, F_SETFL, O_NONBLOCK);
597 qemu_aio_set_fd_handler(s->rfd, posix_aio_read, NULL, posix_aio_flush, s);
599 ret = pthread_attr_init(&attr);
601 die2(ret, "pthread_attr_init");
603 ret = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
605 die2(ret, "pthread_attr_setdetachstate");
607 TAILQ_INIT(&request_list);
611 return posix_aio_state;