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"
31 #include "block/raw-posix-aio.h"
35 BlockDriverAIOCB common
;
38 struct iovec
*aio_iov
;
43 #define aio_ioctl_cmd aio_nbytes /* for QEMU_AIO_IOCTL */
47 QTAILQ_ENTRY(qemu_paiocb
) node
;
51 struct qemu_paiocb
*next
;
54 typedef struct PosixAioState
{
56 struct qemu_paiocb
*first_aio
;
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
;
70 static int preadv_present
= 1;
72 static int preadv_present
= 0;
75 static void die2(int err
, const char *what
)
77 fprintf(stderr
, "%s failed: %s\n", what
, strerror(err
));
81 static void die(const char *what
)
86 static void mutex_lock(pthread_mutex_t
*mutex
)
88 int ret
= pthread_mutex_lock(mutex
);
89 if (ret
) die2(ret
, "pthread_mutex_lock");
92 static void mutex_unlock(pthread_mutex_t
*mutex
)
94 int ret
= pthread_mutex_unlock(mutex
);
95 if (ret
) die2(ret
, "pthread_mutex_unlock");
98 static int cond_timedwait(pthread_cond_t
*cond
, pthread_mutex_t
*mutex
,
101 int ret
= pthread_cond_timedwait(cond
, mutex
, ts
);
102 if (ret
&& ret
!= ETIMEDOUT
) die2(ret
, "pthread_cond_timedwait");
106 static void cond_signal(pthread_cond_t
*cond
)
108 int ret
= pthread_cond_signal(cond
);
109 if (ret
) die2(ret
, "pthread_cond_signal");
112 static void thread_create(pthread_t
*thread
, pthread_attr_t
*attr
,
113 void *(*start_routine
)(void*), void *arg
)
115 int ret
= pthread_create(thread
, attr
, start_routine
, arg
);
116 if (ret
) die2(ret
, "pthread_create");
119 static size_t handle_aiocb_ioctl(struct qemu_paiocb
*aiocb
)
123 ret
= ioctl(aiocb
->aio_fildes
, aiocb
->aio_ioctl_cmd
, aiocb
->aio_ioctl_buf
);
128 * This looks weird, but the aio code only consideres a request
129 * successfull if it has written the number full number of bytes.
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()
135 return aiocb
->aio_nbytes
;
138 static size_t handle_aiocb_flush(struct qemu_paiocb
*aiocb
)
142 ret
= qemu_fdatasync(aiocb
->aio_fildes
);
151 qemu_preadv(int fd
, const struct iovec
*iov
, int nr_iov
, off_t offset
)
153 return preadv(fd
, iov
, nr_iov
, offset
);
157 qemu_pwritev(int fd
, const struct iovec
*iov
, int nr_iov
, off_t offset
)
159 return pwritev(fd
, iov
, nr_iov
, offset
);
165 qemu_preadv(int fd
, const struct iovec
*iov
, int nr_iov
, off_t offset
)
171 qemu_pwritev(int fd
, const struct iovec
*iov
, int nr_iov
, off_t offset
)
178 static size_t handle_aiocb_rw_vector(struct qemu_paiocb
*aiocb
)
184 if (aiocb
->aio_type
& QEMU_AIO_WRITE
)
185 len
= qemu_pwritev(aiocb
->aio_fildes
,
188 aiocb
->aio_offset
+ offset
);
190 len
= qemu_preadv(aiocb
->aio_fildes
,
193 aiocb
->aio_offset
+ offset
);
194 } while (len
== -1 && errno
== EINTR
);
201 static size_t handle_aiocb_rw_linear(struct qemu_paiocb
*aiocb
, char *buf
)
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
);
213 len
= pread(aiocb
->aio_fildes
,
215 aiocb
->aio_nbytes
- offset
,
216 aiocb
->aio_offset
+ offset
);
218 if (len
== -1 && errno
== EINTR
)
220 else if (len
== -1) {
232 static size_t handle_aiocb_rw(struct qemu_paiocb
*aiocb
)
237 if (!(aiocb
->aio_type
& QEMU_AIO_MISALIGNED
)) {
239 * If there is just a single buffer, and it is properly aligned
240 * we can just use plain pread/pwrite without any problems.
242 if (aiocb
->aio_niov
== 1)
243 return handle_aiocb_rw_linear(aiocb
, aiocb
->aio_iov
->iov_base
);
246 * We have more than one iovec, and all are properly aligned.
248 * Try preadv/pwritev first and fall back to linearizing the
249 * buffer if it's not supported.
251 if (preadv_present
) {
252 nbytes
= handle_aiocb_rw_vector(aiocb
);
253 if (nbytes
== aiocb
->aio_nbytes
)
255 if (nbytes
< 0 && nbytes
!= -ENOSYS
)
261 * XXX(hch): short read/write. no easy way to handle the reminder
262 * using these interfaces. For now retry using plain
268 * Ok, we have to do it the hard way, copy all segments into
269 * a single aligned buffer.
271 buf
= qemu_memalign(512, aiocb
->aio_nbytes
);
272 if (aiocb
->aio_type
& QEMU_AIO_WRITE
) {
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
;
282 nbytes
= handle_aiocb_rw_linear(aiocb
, buf
);
283 if (!(aiocb
->aio_type
& QEMU_AIO_WRITE
)) {
285 size_t count
= aiocb
->aio_nbytes
, copy
;
288 for (i
= 0; i
< aiocb
->aio_niov
&& count
; ++i
) {
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
);
302 static void *aio_thread(void *unused
)
309 /* block all signals */
310 if (sigfillset(&set
)) die("sigfillset");
311 if (sigprocmask(SIG_BLOCK
, &set
, NULL
)) die("sigprocmask");
314 struct qemu_paiocb
*aiocb
;
319 qemu_gettimeofday(&tv
);
320 ts
.tv_sec
= tv
.tv_sec
+ 10;
325 while (QTAILQ_EMPTY(&request_list
) &&
326 !(ret
== ETIMEDOUT
)) {
327 ret
= cond_timedwait(&cond
, &lock
, &ts
);
330 if (QTAILQ_EMPTY(&request_list
))
333 aiocb
= QTAILQ_FIRST(&request_list
);
334 QTAILQ_REMOVE(&request_list
, aiocb
, node
);
339 switch (aiocb
->aio_type
& QEMU_AIO_TYPE_MASK
) {
342 ret
= handle_aiocb_rw(aiocb
);
345 ret
= handle_aiocb_flush(aiocb
);
348 ret
= handle_aiocb_ioctl(aiocb
);
351 fprintf(stderr
, "invalid aio request (0x%x)\n", aiocb
->aio_type
);
361 if (kill(pid
, aiocb
->ev_signo
)) die("kill failed");
371 static void spawn_thread(void)
375 thread_create(&thread_id
, &attr
, aio_thread
, NULL
);
378 static void qemu_paio_submit(struct qemu_paiocb
*aiocb
)
380 aiocb
->ret
= -EINPROGRESS
;
383 if (idle_threads
== 0 && cur_threads
< max_threads
)
385 QTAILQ_INSERT_TAIL(&request_list
, aiocb
, node
);
390 static ssize_t
qemu_paio_return(struct qemu_paiocb
*aiocb
)
401 static int qemu_paio_error(struct qemu_paiocb
*aiocb
)
403 ssize_t ret
= qemu_paio_return(aiocb
);
413 static void posix_aio_read(void *opaque
)
415 PosixAioState
*s
= opaque
;
416 struct qemu_paiocb
*acb
, **pacb
;
419 struct qemu_signalfd_siginfo siginfo
;
424 /* try to read from signalfd, don't freak out if we can't read anything */
426 while (offset
< 128) {
429 len
= read(s
->fd
, sig
.buf
+ offset
, 128 - offset
);
430 if (len
== -1 && errno
== EINTR
)
432 if (len
== -1 && errno
== EAGAIN
) {
433 /* there is no natural reason for this to happen,
434 * so we'll spin hard until we get everything just
435 * to be on the safe side. */
444 pacb
= &s
->first_aio
;
449 ret
= qemu_paio_error(acb
);
450 if (ret
== ECANCELED
) {
451 /* remove the request */
453 qemu_aio_release(acb
);
454 } else if (ret
!= EINPROGRESS
) {
457 ret
= qemu_paio_return(acb
);
458 if (ret
== acb
->aio_nbytes
)
465 /* remove the request */
467 /* call the callback */
468 acb
->common
.cb(acb
->common
.opaque
, ret
);
469 qemu_aio_release(acb
);
479 static int posix_aio_flush(void *opaque
)
481 PosixAioState
*s
= opaque
;
482 return !!s
->first_aio
;
485 static PosixAioState
*posix_aio_state
;
487 static void paio_remove(struct qemu_paiocb
*acb
)
489 struct qemu_paiocb
**pacb
;
491 /* remove the callback from the queue */
492 pacb
= &posix_aio_state
->first_aio
;
495 fprintf(stderr
, "paio_remove: aio request not found!\n");
497 } else if (*pacb
== acb
) {
499 qemu_aio_release(acb
);
502 pacb
= &(*pacb
)->next
;
506 static void paio_cancel(BlockDriverAIOCB
*blockacb
)
508 struct qemu_paiocb
*acb
= (struct qemu_paiocb
*)blockacb
;
513 QTAILQ_REMOVE(&request_list
, acb
, node
);
514 acb
->ret
= -ECANCELED
;
515 } else if (acb
->ret
== -EINPROGRESS
) {
521 /* fail safe: if the aio could not be canceled, we wait for
523 while (qemu_paio_error(acb
) == EINPROGRESS
)
530 static AIOPool raw_aio_pool
= {
531 .aiocb_size
= sizeof(struct qemu_paiocb
),
532 .cancel
= paio_cancel
,
535 BlockDriverAIOCB
*paio_submit(BlockDriverState
*bs
, void *aio_ctx
, int fd
,
536 int64_t sector_num
, QEMUIOVector
*qiov
, int nb_sectors
,
537 BlockDriverCompletionFunc
*cb
, void *opaque
, int type
)
539 struct qemu_paiocb
*acb
;
541 acb
= qemu_aio_get(&raw_aio_pool
, bs
, cb
, opaque
);
544 acb
->aio_type
= type
;
545 acb
->aio_fildes
= fd
;
546 acb
->ev_signo
= SIGUSR2
;
548 acb
->aio_iov
= qiov
->iov
;
549 acb
->aio_niov
= qiov
->niov
;
551 acb
->aio_nbytes
= nb_sectors
* 512;
552 acb
->aio_offset
= sector_num
* 512;
554 acb
->next
= posix_aio_state
->first_aio
;
555 posix_aio_state
->first_aio
= acb
;
557 qemu_paio_submit(acb
);
561 BlockDriverAIOCB
*paio_ioctl(BlockDriverState
*bs
, int fd
,
562 unsigned long int req
, void *buf
,
563 BlockDriverCompletionFunc
*cb
, void *opaque
)
565 struct qemu_paiocb
*acb
;
567 acb
= qemu_aio_get(&raw_aio_pool
, bs
, cb
, opaque
);
570 acb
->aio_type
= QEMU_AIO_IOCTL
;
571 acb
->aio_fildes
= fd
;
572 acb
->ev_signo
= SIGUSR2
;
574 acb
->aio_ioctl_buf
= buf
;
575 acb
->aio_ioctl_cmd
= req
;
577 acb
->next
= posix_aio_state
->first_aio
;
578 posix_aio_state
->first_aio
= acb
;
580 qemu_paio_submit(acb
);
584 void *paio_init(void)
591 return posix_aio_state
;
593 s
= qemu_malloc(sizeof(PosixAioState
));
595 /* Make sure to block AIO signal */
597 sigaddset(&mask
, SIGUSR2
);
598 sigprocmask(SIG_BLOCK
, &mask
, NULL
);
601 s
->fd
= qemu_signalfd(&mask
);
603 fprintf(stderr
, "failed to create signalfd\n");
607 fcntl(s
->fd
, F_SETFL
, O_NONBLOCK
);
609 qemu_aio_set_fd_handler(s
->fd
, posix_aio_read
, NULL
, posix_aio_flush
, s
);
611 ret
= pthread_attr_init(&attr
);
613 die2(ret
, "pthread_attr_init");
615 ret
= pthread_attr_setdetachstate(&attr
, PTHREAD_CREATE_DETACHED
);
617 die2(ret
, "pthread_attr_setdetachstate");
619 QTAILQ_INIT(&request_list
);
623 return posix_aio_state
;