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"
32 #include "block/raw-posix-aio.h"
36 BlockDriverAIOCB common
;
39 struct iovec
*aio_iov
;
44 #define aio_ioctl_cmd aio_nbytes /* for QEMU_AIO_IOCTL */
48 QTAILQ_ENTRY(qemu_paiocb
) node
;
52 struct qemu_paiocb
*next
;
57 typedef struct PosixAioState
{
59 struct qemu_paiocb
*first_aio
;
63 static pthread_mutex_t lock
= PTHREAD_MUTEX_INITIALIZER
;
64 static pthread_cond_t cond
= PTHREAD_COND_INITIALIZER
;
65 static pthread_t thread_id
;
66 static pthread_attr_t attr
;
67 static int max_threads
= 64;
68 static int cur_threads
= 0;
69 static int idle_threads
= 0;
70 static QTAILQ_HEAD(, qemu_paiocb
) request_list
;
73 static int preadv_present
= 1;
75 static int preadv_present
= 0;
78 static void die2(int err
, const char *what
)
80 fprintf(stderr
, "%s failed: %s\n", what
, strerror(err
));
84 static void die(const char *what
)
89 static void mutex_lock(pthread_mutex_t
*mutex
)
91 int ret
= pthread_mutex_lock(mutex
);
92 if (ret
) die2(ret
, "pthread_mutex_lock");
95 static void mutex_unlock(pthread_mutex_t
*mutex
)
97 int ret
= pthread_mutex_unlock(mutex
);
98 if (ret
) die2(ret
, "pthread_mutex_unlock");
101 static int cond_timedwait(pthread_cond_t
*cond
, pthread_mutex_t
*mutex
,
104 int ret
= pthread_cond_timedwait(cond
, mutex
, ts
);
105 if (ret
&& ret
!= ETIMEDOUT
) die2(ret
, "pthread_cond_timedwait");
109 static void cond_signal(pthread_cond_t
*cond
)
111 int ret
= pthread_cond_signal(cond
);
112 if (ret
) die2(ret
, "pthread_cond_signal");
115 static void thread_create(pthread_t
*thread
, pthread_attr_t
*attr
,
116 void *(*start_routine
)(void*), void *arg
)
118 int ret
= pthread_create(thread
, attr
, start_routine
, arg
);
119 if (ret
) die2(ret
, "pthread_create");
122 static ssize_t
handle_aiocb_ioctl(struct qemu_paiocb
*aiocb
)
126 ret
= ioctl(aiocb
->aio_fildes
, aiocb
->aio_ioctl_cmd
, aiocb
->aio_ioctl_buf
);
131 * This looks weird, but the aio code only consideres a request
132 * successful if it has written the number full number of bytes.
134 * Now we overload aio_nbytes as aio_ioctl_cmd for the ioctl command,
135 * so in fact we return the ioctl command here to make posix_aio_read()
138 return aiocb
->aio_nbytes
;
141 static ssize_t
handle_aiocb_flush(struct qemu_paiocb
*aiocb
)
145 ret
= qemu_fdatasync(aiocb
->aio_fildes
);
154 qemu_preadv(int fd
, const struct iovec
*iov
, int nr_iov
, off_t offset
)
156 return preadv(fd
, iov
, nr_iov
, offset
);
160 qemu_pwritev(int fd
, const struct iovec
*iov
, int nr_iov
, off_t offset
)
162 return pwritev(fd
, iov
, nr_iov
, offset
);
168 qemu_preadv(int fd
, const struct iovec
*iov
, int nr_iov
, off_t offset
)
174 qemu_pwritev(int fd
, const struct iovec
*iov
, int nr_iov
, off_t offset
)
181 static ssize_t
handle_aiocb_rw_vector(struct qemu_paiocb
*aiocb
)
187 if (aiocb
->aio_type
& QEMU_AIO_WRITE
)
188 len
= qemu_pwritev(aiocb
->aio_fildes
,
191 aiocb
->aio_offset
+ offset
);
193 len
= qemu_preadv(aiocb
->aio_fildes
,
196 aiocb
->aio_offset
+ offset
);
197 } while (len
== -1 && errno
== EINTR
);
204 static ssize_t
handle_aiocb_rw_linear(struct qemu_paiocb
*aiocb
, char *buf
)
209 while (offset
< aiocb
->aio_nbytes
) {
210 if (aiocb
->aio_type
& QEMU_AIO_WRITE
)
211 len
= pwrite(aiocb
->aio_fildes
,
212 (const char *)buf
+ offset
,
213 aiocb
->aio_nbytes
- offset
,
214 aiocb
->aio_offset
+ offset
);
216 len
= pread(aiocb
->aio_fildes
,
218 aiocb
->aio_nbytes
- offset
,
219 aiocb
->aio_offset
+ offset
);
221 if (len
== -1 && errno
== EINTR
)
223 else if (len
== -1) {
235 static ssize_t
handle_aiocb_rw(struct qemu_paiocb
*aiocb
)
240 if (!(aiocb
->aio_type
& QEMU_AIO_MISALIGNED
)) {
242 * If there is just a single buffer, and it is properly aligned
243 * we can just use plain pread/pwrite without any problems.
245 if (aiocb
->aio_niov
== 1)
246 return handle_aiocb_rw_linear(aiocb
, aiocb
->aio_iov
->iov_base
);
249 * We have more than one iovec, and all are properly aligned.
251 * Try preadv/pwritev first and fall back to linearizing the
252 * buffer if it's not supported.
254 if (preadv_present
) {
255 nbytes
= handle_aiocb_rw_vector(aiocb
);
256 if (nbytes
== aiocb
->aio_nbytes
)
258 if (nbytes
< 0 && nbytes
!= -ENOSYS
)
264 * XXX(hch): short read/write. no easy way to handle the reminder
265 * using these interfaces. For now retry using plain
271 * Ok, we have to do it the hard way, copy all segments into
272 * a single aligned buffer.
274 buf
= qemu_blockalign(aiocb
->common
.bs
, aiocb
->aio_nbytes
);
275 if (aiocb
->aio_type
& QEMU_AIO_WRITE
) {
279 for (i
= 0; i
< aiocb
->aio_niov
; ++i
) {
280 memcpy(p
, aiocb
->aio_iov
[i
].iov_base
, aiocb
->aio_iov
[i
].iov_len
);
281 p
+= aiocb
->aio_iov
[i
].iov_len
;
285 nbytes
= handle_aiocb_rw_linear(aiocb
, buf
);
286 if (!(aiocb
->aio_type
& QEMU_AIO_WRITE
)) {
288 size_t count
= aiocb
->aio_nbytes
, copy
;
291 for (i
= 0; i
< aiocb
->aio_niov
&& count
; ++i
) {
293 if (copy
> aiocb
->aio_iov
[i
].iov_len
)
294 copy
= aiocb
->aio_iov
[i
].iov_len
;
295 memcpy(aiocb
->aio_iov
[i
].iov_base
, p
, copy
);
305 static void *aio_thread(void *unused
)
312 struct qemu_paiocb
*aiocb
;
317 qemu_gettimeofday(&tv
);
318 ts
.tv_sec
= tv
.tv_sec
+ 10;
323 while (QTAILQ_EMPTY(&request_list
) &&
324 !(ret
== ETIMEDOUT
)) {
325 ret
= cond_timedwait(&cond
, &lock
, &ts
);
328 if (QTAILQ_EMPTY(&request_list
))
331 aiocb
= QTAILQ_FIRST(&request_list
);
332 QTAILQ_REMOVE(&request_list
, aiocb
, node
);
337 switch (aiocb
->aio_type
& QEMU_AIO_TYPE_MASK
) {
340 ret
= handle_aiocb_rw(aiocb
);
343 ret
= handle_aiocb_flush(aiocb
);
346 ret
= handle_aiocb_ioctl(aiocb
);
349 fprintf(stderr
, "invalid aio request (0x%x)\n", aiocb
->aio_type
);
359 if (kill(pid
, aiocb
->ev_signo
)) die("kill failed");
369 static void spawn_thread(void)
371 sigset_t set
, oldset
;
376 /* block all signals */
377 if (sigfillset(&set
)) die("sigfillset");
378 if (sigprocmask(SIG_SETMASK
, &set
, &oldset
)) die("sigprocmask");
380 thread_create(&thread_id
, &attr
, aio_thread
, NULL
);
382 if (sigprocmask(SIG_SETMASK
, &oldset
, NULL
)) die("sigprocmask restore");
385 static void qemu_paio_submit(struct qemu_paiocb
*aiocb
)
387 aiocb
->ret
= -EINPROGRESS
;
390 if (idle_threads
== 0 && cur_threads
< max_threads
)
392 QTAILQ_INSERT_TAIL(&request_list
, aiocb
, node
);
397 static ssize_t
qemu_paio_return(struct qemu_paiocb
*aiocb
)
408 static int qemu_paio_error(struct qemu_paiocb
*aiocb
)
410 ssize_t ret
= qemu_paio_return(aiocb
);
420 static int posix_aio_process_queue(void *opaque
)
422 PosixAioState
*s
= opaque
;
423 struct qemu_paiocb
*acb
, **pacb
;
426 int async_context_id
= get_async_context_id();
429 pacb
= &s
->first_aio
;
435 /* we're only interested in requests in the right context */
436 if (acb
->async_context_id
!= async_context_id
) {
441 ret
= qemu_paio_error(acb
);
442 if (ret
== ECANCELED
) {
443 /* remove the request */
445 qemu_aio_release(acb
);
447 } else if (ret
!= EINPROGRESS
) {
450 ret
= qemu_paio_return(acb
);
451 if (ret
== acb
->aio_nbytes
)
458 /* remove the request */
460 /* call the callback */
461 acb
->common
.cb(acb
->common
.opaque
, ret
);
462 qemu_aio_release(acb
);
474 static void posix_aio_read(void *opaque
)
476 PosixAioState
*s
= opaque
;
479 /* read all bytes from signal pipe */
483 len
= read(s
->rfd
, bytes
, sizeof(bytes
));
484 if (len
== -1 && errno
== EINTR
)
485 continue; /* try again */
486 if (len
== sizeof(bytes
))
487 continue; /* more to read */
491 posix_aio_process_queue(s
);
494 static int posix_aio_flush(void *opaque
)
496 PosixAioState
*s
= opaque
;
497 return !!s
->first_aio
;
500 static PosixAioState
*posix_aio_state
;
502 static void aio_signal_handler(int signum
)
504 if (posix_aio_state
) {
508 ret
= write(posix_aio_state
->wfd
, &byte
, sizeof(byte
));
509 if (ret
< 0 && errno
!= EAGAIN
)
516 static void paio_remove(struct qemu_paiocb
*acb
)
518 struct qemu_paiocb
**pacb
;
520 /* remove the callback from the queue */
521 pacb
= &posix_aio_state
->first_aio
;
524 fprintf(stderr
, "paio_remove: aio request not found!\n");
526 } else if (*pacb
== acb
) {
528 qemu_aio_release(acb
);
531 pacb
= &(*pacb
)->next
;
535 static void paio_cancel(BlockDriverAIOCB
*blockacb
)
537 struct qemu_paiocb
*acb
= (struct qemu_paiocb
*)blockacb
;
542 QTAILQ_REMOVE(&request_list
, acb
, node
);
543 acb
->ret
= -ECANCELED
;
544 } else if (acb
->ret
== -EINPROGRESS
) {
550 /* fail safe: if the aio could not be canceled, we wait for
552 while (qemu_paio_error(acb
) == EINPROGRESS
)
559 static AIOPool raw_aio_pool
= {
560 .aiocb_size
= sizeof(struct qemu_paiocb
),
561 .cancel
= paio_cancel
,
564 BlockDriverAIOCB
*paio_submit(BlockDriverState
*bs
, int fd
,
565 int64_t sector_num
, QEMUIOVector
*qiov
, int nb_sectors
,
566 BlockDriverCompletionFunc
*cb
, void *opaque
, int type
)
568 struct qemu_paiocb
*acb
;
570 acb
= qemu_aio_get(&raw_aio_pool
, bs
, cb
, opaque
);
573 acb
->aio_type
= type
;
574 acb
->aio_fildes
= fd
;
575 acb
->ev_signo
= SIGUSR2
;
576 acb
->async_context_id
= get_async_context_id();
579 acb
->aio_iov
= qiov
->iov
;
580 acb
->aio_niov
= qiov
->niov
;
582 acb
->aio_nbytes
= nb_sectors
* 512;
583 acb
->aio_offset
= sector_num
* 512;
585 acb
->next
= posix_aio_state
->first_aio
;
586 posix_aio_state
->first_aio
= acb
;
588 trace_paio_submit(acb
, opaque
, sector_num
, nb_sectors
, type
);
589 qemu_paio_submit(acb
);
593 BlockDriverAIOCB
*paio_ioctl(BlockDriverState
*bs
, int fd
,
594 unsigned long int req
, void *buf
,
595 BlockDriverCompletionFunc
*cb
, void *opaque
)
597 struct qemu_paiocb
*acb
;
599 acb
= qemu_aio_get(&raw_aio_pool
, bs
, cb
, opaque
);
602 acb
->aio_type
= QEMU_AIO_IOCTL
;
603 acb
->aio_fildes
= fd
;
604 acb
->ev_signo
= SIGUSR2
;
605 acb
->async_context_id
= get_async_context_id();
607 acb
->aio_ioctl_buf
= buf
;
608 acb
->aio_ioctl_cmd
= req
;
610 acb
->next
= posix_aio_state
->first_aio
;
611 posix_aio_state
->first_aio
= acb
;
613 qemu_paio_submit(acb
);
619 struct sigaction act
;
627 s
= qemu_malloc(sizeof(PosixAioState
));
629 sigfillset(&act
.sa_mask
);
630 act
.sa_flags
= 0; /* do not restart syscalls to interrupt select() */
631 act
.sa_handler
= aio_signal_handler
;
632 sigaction(SIGUSR2
, &act
, NULL
);
635 if (qemu_pipe(fds
) == -1) {
636 fprintf(stderr
, "failed to create pipe\n");
643 fcntl(s
->rfd
, F_SETFL
, O_NONBLOCK
);
644 fcntl(s
->wfd
, F_SETFL
, O_NONBLOCK
);
646 qemu_aio_set_fd_handler(s
->rfd
, posix_aio_read
, NULL
, posix_aio_flush
,
647 posix_aio_process_queue
, s
);
649 ret
= pthread_attr_init(&attr
);
651 die2(ret
, "pthread_attr_init");
653 ret
= pthread_attr_setdetachstate(&attr
, PTHREAD_CREATE_DETACHED
);
655 die2(ret
, "pthread_attr_setdetachstate");
657 QTAILQ_INIT(&request_list
);