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"
28 #include "qemu-common.h"
30 #include "block_int.h"
33 #include "block/raw-posix-aio.h"
37 BlockDriverAIOCB common
;
40 struct iovec
*aio_iov
;
45 #define aio_ioctl_cmd aio_nbytes /* for QEMU_AIO_IOCTL */
49 QTAILQ_ENTRY(qemu_paiocb
) node
;
53 struct qemu_paiocb
*next
;
58 typedef struct PosixAioState
{
60 struct qemu_paiocb
*first_aio
;
64 static pthread_mutex_t lock
= PTHREAD_MUTEX_INITIALIZER
;
65 static pthread_cond_t cond
= PTHREAD_COND_INITIALIZER
;
66 static pthread_t thread_id
;
67 static pthread_attr_t attr
;
68 static int max_threads
= 64;
69 static int cur_threads
= 0;
70 static int idle_threads
= 0;
71 static QTAILQ_HEAD(, qemu_paiocb
) request_list
;
74 static int preadv_present
= 1;
76 static int preadv_present
= 0;
79 static void die2(int err
, const char *what
)
81 fprintf(stderr
, "%s failed: %s\n", what
, strerror(err
));
85 static void die(const char *what
)
90 static void mutex_lock(pthread_mutex_t
*mutex
)
92 int ret
= pthread_mutex_lock(mutex
);
93 if (ret
) die2(ret
, "pthread_mutex_lock");
96 static void mutex_unlock(pthread_mutex_t
*mutex
)
98 int ret
= pthread_mutex_unlock(mutex
);
99 if (ret
) die2(ret
, "pthread_mutex_unlock");
102 static int cond_timedwait(pthread_cond_t
*cond
, pthread_mutex_t
*mutex
,
105 int ret
= pthread_cond_timedwait(cond
, mutex
, ts
);
106 if (ret
&& ret
!= ETIMEDOUT
) die2(ret
, "pthread_cond_timedwait");
110 static void cond_signal(pthread_cond_t
*cond
)
112 int ret
= pthread_cond_signal(cond
);
113 if (ret
) die2(ret
, "pthread_cond_signal");
116 static void thread_create(pthread_t
*thread
, pthread_attr_t
*attr
,
117 void *(*start_routine
)(void*), void *arg
)
119 int ret
= pthread_create(thread
, attr
, start_routine
, arg
);
120 if (ret
) die2(ret
, "pthread_create");
123 static ssize_t
handle_aiocb_ioctl(struct qemu_paiocb
*aiocb
)
127 ret
= ioctl(aiocb
->aio_fildes
, aiocb
->aio_ioctl_cmd
, aiocb
->aio_ioctl_buf
);
132 * This looks weird, but the aio code only consideres a request
133 * successful if it has written the number full number of bytes.
135 * Now we overload aio_nbytes as aio_ioctl_cmd for the ioctl command,
136 * so in fact we return the ioctl command here to make posix_aio_read()
139 return aiocb
->aio_nbytes
;
142 static ssize_t
handle_aiocb_flush(struct qemu_paiocb
*aiocb
)
146 ret
= qemu_fdatasync(aiocb
->aio_fildes
);
155 qemu_preadv(int fd
, const struct iovec
*iov
, int nr_iov
, off_t offset
)
157 return preadv(fd
, iov
, nr_iov
, offset
);
161 qemu_pwritev(int fd
, const struct iovec
*iov
, int nr_iov
, off_t offset
)
163 return pwritev(fd
, iov
, nr_iov
, offset
);
169 qemu_preadv(int fd
, const struct iovec
*iov
, int nr_iov
, off_t offset
)
175 qemu_pwritev(int fd
, const struct iovec
*iov
, int nr_iov
, off_t offset
)
182 static ssize_t
handle_aiocb_rw_vector(struct qemu_paiocb
*aiocb
)
188 if (aiocb
->aio_type
& QEMU_AIO_WRITE
)
189 len
= qemu_pwritev(aiocb
->aio_fildes
,
192 aiocb
->aio_offset
+ offset
);
194 len
= qemu_preadv(aiocb
->aio_fildes
,
197 aiocb
->aio_offset
+ offset
);
198 } while (len
== -1 && errno
== EINTR
);
205 static ssize_t
handle_aiocb_rw_linear(struct qemu_paiocb
*aiocb
, char *buf
)
210 while (offset
< aiocb
->aio_nbytes
) {
211 if (aiocb
->aio_type
& QEMU_AIO_WRITE
)
212 len
= pwrite(aiocb
->aio_fildes
,
213 (const char *)buf
+ offset
,
214 aiocb
->aio_nbytes
- offset
,
215 aiocb
->aio_offset
+ offset
);
217 len
= pread(aiocb
->aio_fildes
,
219 aiocb
->aio_nbytes
- offset
,
220 aiocb
->aio_offset
+ offset
);
222 if (len
== -1 && errno
== EINTR
)
224 else if (len
== -1) {
236 static ssize_t
handle_aiocb_rw(struct qemu_paiocb
*aiocb
)
241 if (!(aiocb
->aio_type
& QEMU_AIO_MISALIGNED
)) {
243 * If there is just a single buffer, and it is properly aligned
244 * we can just use plain pread/pwrite without any problems.
246 if (aiocb
->aio_niov
== 1)
247 return handle_aiocb_rw_linear(aiocb
, aiocb
->aio_iov
->iov_base
);
250 * We have more than one iovec, and all are properly aligned.
252 * Try preadv/pwritev first and fall back to linearizing the
253 * buffer if it's not supported.
255 if (preadv_present
) {
256 nbytes
= handle_aiocb_rw_vector(aiocb
);
257 if (nbytes
== aiocb
->aio_nbytes
)
259 if (nbytes
< 0 && nbytes
!= -ENOSYS
)
265 * XXX(hch): short read/write. no easy way to handle the reminder
266 * using these interfaces. For now retry using plain
272 * Ok, we have to do it the hard way, copy all segments into
273 * a single aligned buffer.
275 buf
= qemu_blockalign(aiocb
->common
.bs
, aiocb
->aio_nbytes
);
276 if (aiocb
->aio_type
& QEMU_AIO_WRITE
) {
280 for (i
= 0; i
< aiocb
->aio_niov
; ++i
) {
281 memcpy(p
, aiocb
->aio_iov
[i
].iov_base
, aiocb
->aio_iov
[i
].iov_len
);
282 p
+= aiocb
->aio_iov
[i
].iov_len
;
286 nbytes
= handle_aiocb_rw_linear(aiocb
, buf
);
287 if (!(aiocb
->aio_type
& QEMU_AIO_WRITE
)) {
289 size_t count
= aiocb
->aio_nbytes
, copy
;
292 for (i
= 0; i
< aiocb
->aio_niov
&& count
; ++i
) {
294 if (copy
> aiocb
->aio_iov
[i
].iov_len
)
295 copy
= aiocb
->aio_iov
[i
].iov_len
;
296 memcpy(aiocb
->aio_iov
[i
].iov_base
, p
, copy
);
306 static void *aio_thread(void *unused
)
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)
372 sigset_t set
, oldset
;
377 /* block all signals */
378 if (sigfillset(&set
)) die("sigfillset");
379 if (sigprocmask(SIG_SETMASK
, &set
, &oldset
)) die("sigprocmask");
381 thread_create(&thread_id
, &attr
, aio_thread
, NULL
);
383 if (sigprocmask(SIG_SETMASK
, &oldset
, NULL
)) die("sigprocmask restore");
386 static void qemu_paio_submit(struct qemu_paiocb
*aiocb
)
388 aiocb
->ret
= -EINPROGRESS
;
391 if (idle_threads
== 0 && cur_threads
< max_threads
)
393 QTAILQ_INSERT_TAIL(&request_list
, aiocb
, node
);
398 static ssize_t
qemu_paio_return(struct qemu_paiocb
*aiocb
)
409 static int qemu_paio_error(struct qemu_paiocb
*aiocb
)
411 ssize_t ret
= qemu_paio_return(aiocb
);
421 static int posix_aio_process_queue(void *opaque
)
423 PosixAioState
*s
= opaque
;
424 struct qemu_paiocb
*acb
, **pacb
;
427 int async_context_id
= get_async_context_id();
430 pacb
= &s
->first_aio
;
436 /* we're only interested in requests in the right context */
437 if (acb
->async_context_id
!= async_context_id
) {
442 ret
= qemu_paio_error(acb
);
443 if (ret
== ECANCELED
) {
444 /* remove the request */
446 qemu_aio_release(acb
);
448 } else if (ret
!= EINPROGRESS
) {
451 ret
= qemu_paio_return(acb
);
452 if (ret
== acb
->aio_nbytes
)
459 /* remove the request */
461 /* call the callback */
462 acb
->common
.cb(acb
->common
.opaque
, ret
);
463 qemu_aio_release(acb
);
475 static void posix_aio_read(void *opaque
)
477 PosixAioState
*s
= opaque
;
479 struct qemu_signalfd_siginfo siginfo
;
484 /* try to read from signalfd, don't freak out if we can't read anything */
486 while (offset
< 128) {
489 len
= read(s
->fd
, sig
.buf
+ offset
, 128 - offset
);
490 if (len
== -1 && errno
== EINTR
)
492 if (len
== -1 && errno
== EAGAIN
) {
493 /* there is no natural reason for this to happen,
494 * so we'll spin hard until we get everything just
495 * to be on the safe side. */
503 posix_aio_process_queue(s
);
506 static int posix_aio_flush(void *opaque
)
508 PosixAioState
*s
= opaque
;
509 return !!s
->first_aio
;
512 static PosixAioState
*posix_aio_state
;
514 static void paio_remove(struct qemu_paiocb
*acb
)
516 struct qemu_paiocb
**pacb
;
518 /* remove the callback from the queue */
519 pacb
= &posix_aio_state
->first_aio
;
522 fprintf(stderr
, "paio_remove: aio request not found!\n");
524 } else if (*pacb
== acb
) {
526 qemu_aio_release(acb
);
529 pacb
= &(*pacb
)->next
;
533 static void paio_cancel(BlockDriverAIOCB
*blockacb
)
535 struct qemu_paiocb
*acb
= (struct qemu_paiocb
*)blockacb
;
540 QTAILQ_REMOVE(&request_list
, acb
, node
);
541 acb
->ret
= -ECANCELED
;
542 } else if (acb
->ret
== -EINPROGRESS
) {
548 /* fail safe: if the aio could not be canceled, we wait for
550 while (qemu_paio_error(acb
) == EINPROGRESS
)
557 static AIOPool raw_aio_pool
= {
558 .aiocb_size
= sizeof(struct qemu_paiocb
),
559 .cancel
= paio_cancel
,
562 BlockDriverAIOCB
*paio_submit(BlockDriverState
*bs
, int fd
,
563 int64_t sector_num
, QEMUIOVector
*qiov
, int nb_sectors
,
564 BlockDriverCompletionFunc
*cb
, void *opaque
, int type
)
566 struct qemu_paiocb
*acb
;
568 acb
= qemu_aio_get(&raw_aio_pool
, bs
, cb
, opaque
);
571 acb
->aio_type
= type
;
572 acb
->aio_fildes
= fd
;
573 acb
->ev_signo
= SIGUSR2
;
574 acb
->async_context_id
= get_async_context_id();
577 acb
->aio_iov
= qiov
->iov
;
578 acb
->aio_niov
= qiov
->niov
;
580 acb
->aio_nbytes
= nb_sectors
* 512;
581 acb
->aio_offset
= sector_num
* 512;
583 acb
->next
= posix_aio_state
->first_aio
;
584 posix_aio_state
->first_aio
= acb
;
586 trace_paio_submit(acb
, opaque
, sector_num
, nb_sectors
, type
);
587 qemu_paio_submit(acb
);
591 BlockDriverAIOCB
*paio_ioctl(BlockDriverState
*bs
, int fd
,
592 unsigned long int req
, void *buf
,
593 BlockDriverCompletionFunc
*cb
, void *opaque
)
595 struct qemu_paiocb
*acb
;
597 acb
= qemu_aio_get(&raw_aio_pool
, bs
, cb
, opaque
);
600 acb
->aio_type
= QEMU_AIO_IOCTL
;
601 acb
->aio_fildes
= fd
;
602 acb
->ev_signo
= SIGUSR2
;
603 acb
->async_context_id
= get_async_context_id();
605 acb
->aio_ioctl_buf
= buf
;
606 acb
->aio_ioctl_cmd
= req
;
608 acb
->next
= posix_aio_state
->first_aio
;
609 posix_aio_state
->first_aio
= acb
;
611 qemu_paio_submit(acb
);
624 s
= qemu_malloc(sizeof(PosixAioState
));
626 /* Make sure to block AIO signal */
628 sigaddset(&mask
, SIGUSR2
);
629 sigprocmask(SIG_BLOCK
, &mask
, NULL
);
632 s
->fd
= qemu_signalfd(&mask
);
634 fprintf(stderr
, "failed to create signalfd\n");
638 fcntl(s
->fd
, F_SETFL
, O_NONBLOCK
);
640 qemu_aio_set_fd_handler(s
->fd
, posix_aio_read
, NULL
, posix_aio_flush
,
641 posix_aio_process_queue
, s
);
643 ret
= pthread_attr_init(&attr
);
645 die2(ret
, "pthread_attr_init");
647 ret
= pthread_attr_setdetachstate(&attr
, PTHREAD_CREATE_DETACHED
);
649 die2(ret
, "pthread_attr_setdetachstate");
651 QTAILQ_INIT(&request_list
);