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>
24 #include "qemu-queue.h"
27 #include "qemu-common.h"
29 #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 ssize_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 * successful 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 ssize_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 ssize_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 ssize_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 ssize_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_blockalign(aiocb
->common
.bs
, 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 struct qemu_paiocb
*aiocb
;
314 qemu_gettimeofday(&tv
);
315 ts
.tv_sec
= tv
.tv_sec
+ 10;
320 while (QTAILQ_EMPTY(&request_list
) &&
321 !(ret
== ETIMEDOUT
)) {
323 ret
= cond_timedwait(&cond
, &lock
, &ts
);
327 if (QTAILQ_EMPTY(&request_list
))
330 aiocb
= QTAILQ_FIRST(&request_list
);
331 QTAILQ_REMOVE(&request_list
, aiocb
, node
);
335 switch (aiocb
->aio_type
& QEMU_AIO_TYPE_MASK
) {
338 ret
= handle_aiocb_rw(aiocb
);
341 ret
= handle_aiocb_flush(aiocb
);
344 ret
= handle_aiocb_ioctl(aiocb
);
347 fprintf(stderr
, "invalid aio request (0x%x)\n", aiocb
->aio_type
);
356 if (kill(pid
, aiocb
->ev_signo
)) die("kill failed");
365 static void spawn_thread(void)
367 sigset_t set
, oldset
;
371 /* block all signals */
372 if (sigfillset(&set
)) die("sigfillset");
373 if (sigprocmask(SIG_SETMASK
, &set
, &oldset
)) die("sigprocmask");
375 thread_create(&thread_id
, &attr
, aio_thread
, NULL
);
377 if (sigprocmask(SIG_SETMASK
, &oldset
, NULL
)) die("sigprocmask restore");
380 static void qemu_paio_submit(struct qemu_paiocb
*aiocb
)
382 aiocb
->ret
= -EINPROGRESS
;
385 if (idle_threads
== 0 && cur_threads
< max_threads
)
387 QTAILQ_INSERT_TAIL(&request_list
, aiocb
, node
);
392 static ssize_t
qemu_paio_return(struct qemu_paiocb
*aiocb
)
403 static int qemu_paio_error(struct qemu_paiocb
*aiocb
)
405 ssize_t ret
= qemu_paio_return(aiocb
);
415 static int posix_aio_process_queue(void *opaque
)
417 PosixAioState
*s
= opaque
;
418 struct qemu_paiocb
*acb
, **pacb
;
423 pacb
= &s
->first_aio
;
429 ret
= qemu_paio_error(acb
);
430 if (ret
== ECANCELED
) {
431 /* remove the request */
433 qemu_aio_release(acb
);
435 } else if (ret
!= EINPROGRESS
) {
438 ret
= qemu_paio_return(acb
);
439 if (ret
== acb
->aio_nbytes
)
447 trace_paio_complete(acb
, acb
->common
.opaque
, ret
);
449 /* remove the request */
451 /* call the callback */
452 acb
->common
.cb(acb
->common
.opaque
, ret
);
453 qemu_aio_release(acb
);
465 static void posix_aio_read(void *opaque
)
467 PosixAioState
*s
= opaque
;
470 /* read all bytes from signal pipe */
474 len
= read(s
->rfd
, bytes
, sizeof(bytes
));
475 if (len
== -1 && errno
== EINTR
)
476 continue; /* try again */
477 if (len
== sizeof(bytes
))
478 continue; /* more to read */
482 posix_aio_process_queue(s
);
485 static int posix_aio_flush(void *opaque
)
487 PosixAioState
*s
= opaque
;
488 return !!s
->first_aio
;
491 static PosixAioState
*posix_aio_state
;
493 static void aio_signal_handler(int signum
)
495 if (posix_aio_state
) {
499 ret
= write(posix_aio_state
->wfd
, &byte
, sizeof(byte
));
500 if (ret
< 0 && errno
!= EAGAIN
)
507 static void paio_remove(struct qemu_paiocb
*acb
)
509 struct qemu_paiocb
**pacb
;
511 /* remove the callback from the queue */
512 pacb
= &posix_aio_state
->first_aio
;
515 fprintf(stderr
, "paio_remove: aio request not found!\n");
517 } else if (*pacb
== acb
) {
519 qemu_aio_release(acb
);
522 pacb
= &(*pacb
)->next
;
526 static void paio_cancel(BlockDriverAIOCB
*blockacb
)
528 struct qemu_paiocb
*acb
= (struct qemu_paiocb
*)blockacb
;
531 trace_paio_cancel(acb
, acb
->common
.opaque
);
535 QTAILQ_REMOVE(&request_list
, acb
, node
);
536 acb
->ret
= -ECANCELED
;
537 } else if (acb
->ret
== -EINPROGRESS
) {
543 /* fail safe: if the aio could not be canceled, we wait for
545 while (qemu_paio_error(acb
) == EINPROGRESS
)
552 static AIOPool raw_aio_pool
= {
553 .aiocb_size
= sizeof(struct qemu_paiocb
),
554 .cancel
= paio_cancel
,
557 BlockDriverAIOCB
*paio_submit(BlockDriverState
*bs
, int fd
,
558 int64_t sector_num
, QEMUIOVector
*qiov
, int nb_sectors
,
559 BlockDriverCompletionFunc
*cb
, void *opaque
, int type
)
561 struct qemu_paiocb
*acb
;
563 acb
= qemu_aio_get(&raw_aio_pool
, bs
, cb
, opaque
);
566 acb
->aio_type
= type
;
567 acb
->aio_fildes
= fd
;
568 acb
->ev_signo
= SIGUSR2
;
571 acb
->aio_iov
= qiov
->iov
;
572 acb
->aio_niov
= qiov
->niov
;
574 acb
->aio_nbytes
= nb_sectors
* 512;
575 acb
->aio_offset
= sector_num
* 512;
577 acb
->next
= posix_aio_state
->first_aio
;
578 posix_aio_state
->first_aio
= acb
;
580 trace_paio_submit(acb
, opaque
, sector_num
, nb_sectors
, type
);
581 qemu_paio_submit(acb
);
585 BlockDriverAIOCB
*paio_ioctl(BlockDriverState
*bs
, int fd
,
586 unsigned long int req
, void *buf
,
587 BlockDriverCompletionFunc
*cb
, void *opaque
)
589 struct qemu_paiocb
*acb
;
591 acb
= qemu_aio_get(&raw_aio_pool
, bs
, cb
, opaque
);
594 acb
->aio_type
= QEMU_AIO_IOCTL
;
595 acb
->aio_fildes
= fd
;
596 acb
->ev_signo
= SIGUSR2
;
598 acb
->aio_ioctl_buf
= buf
;
599 acb
->aio_ioctl_cmd
= req
;
601 acb
->next
= posix_aio_state
->first_aio
;
602 posix_aio_state
->first_aio
= acb
;
604 qemu_paio_submit(acb
);
610 struct sigaction act
;
618 s
= qemu_malloc(sizeof(PosixAioState
));
620 sigfillset(&act
.sa_mask
);
621 act
.sa_flags
= 0; /* do not restart syscalls to interrupt select() */
622 act
.sa_handler
= aio_signal_handler
;
623 sigaction(SIGUSR2
, &act
, NULL
);
626 if (qemu_pipe(fds
) == -1) {
627 fprintf(stderr
, "failed to create pipe\n");
634 fcntl(s
->rfd
, F_SETFL
, O_NONBLOCK
);
635 fcntl(s
->wfd
, F_SETFL
, O_NONBLOCK
);
637 qemu_aio_set_fd_handler(s
->rfd
, posix_aio_read
, NULL
, posix_aio_flush
,
638 posix_aio_process_queue
, s
);
640 ret
= pthread_attr_init(&attr
);
642 die2(ret
, "pthread_attr_init");
644 ret
= pthread_attr_setdetachstate(&attr
, PTHREAD_CREATE_DETACHED
);
646 die2(ret
, "pthread_attr_setdetachstate");
648 QTAILQ_INIT(&request_list
);