libcacard: introduce NEXT_TOKEN macro
[qemu.git] / posix-aio-compat.c
blobc4116e30f2e2391762b5528cd46a179d53843ba0
1 /*
2 * QEMU posix-aio emulation
4 * Copyright IBM, Corp. 2008
6 * Authors:
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>
16 #include <pthread.h>
17 #include <unistd.h>
18 #include <errno.h>
19 #include <time.h>
20 #include <string.h>
21 #include <stdlib.h>
22 #include <stdio.h>
24 #include "qemu-queue.h"
25 #include "osdep.h"
26 #include "sysemu.h"
27 #include "qemu-common.h"
28 #include "trace.h"
29 #include "block_int.h"
31 #include "block/raw-posix-aio.h"
34 struct qemu_paiocb {
35 BlockDriverAIOCB common;
36 int aio_fildes;
37 union {
38 struct iovec *aio_iov;
39 void *aio_ioctl_buf;
41 int aio_niov;
42 size_t aio_nbytes;
43 #define aio_ioctl_cmd aio_nbytes /* for QEMU_AIO_IOCTL */
44 int ev_signo;
45 off_t aio_offset;
47 QTAILQ_ENTRY(qemu_paiocb) node;
48 int aio_type;
49 ssize_t ret;
50 int active;
51 struct qemu_paiocb *next;
53 int async_context_id;
56 typedef struct PosixAioState {
57 int rfd, wfd;
58 struct qemu_paiocb *first_aio;
59 } PosixAioState;
62 static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
63 static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
64 static pthread_t thread_id;
65 static pthread_attr_t attr;
66 static int max_threads = 64;
67 static int cur_threads = 0;
68 static int idle_threads = 0;
69 static QTAILQ_HEAD(, qemu_paiocb) request_list;
71 #ifdef CONFIG_PREADV
72 static int preadv_present = 1;
73 #else
74 static int preadv_present = 0;
75 #endif
77 static void die2(int err, const char *what)
79 fprintf(stderr, "%s failed: %s\n", what, strerror(err));
80 abort();
83 static void die(const char *what)
85 die2(errno, what);
88 static void mutex_lock(pthread_mutex_t *mutex)
90 int ret = pthread_mutex_lock(mutex);
91 if (ret) die2(ret, "pthread_mutex_lock");
94 static void mutex_unlock(pthread_mutex_t *mutex)
96 int ret = pthread_mutex_unlock(mutex);
97 if (ret) die2(ret, "pthread_mutex_unlock");
100 static int cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex,
101 struct timespec *ts)
103 int ret = pthread_cond_timedwait(cond, mutex, ts);
104 if (ret && ret != ETIMEDOUT) die2(ret, "pthread_cond_timedwait");
105 return ret;
108 static void cond_signal(pthread_cond_t *cond)
110 int ret = pthread_cond_signal(cond);
111 if (ret) die2(ret, "pthread_cond_signal");
114 static void thread_create(pthread_t *thread, pthread_attr_t *attr,
115 void *(*start_routine)(void*), void *arg)
117 int ret = pthread_create(thread, attr, start_routine, arg);
118 if (ret) die2(ret, "pthread_create");
121 static ssize_t handle_aiocb_ioctl(struct qemu_paiocb *aiocb)
123 int ret;
125 ret = ioctl(aiocb->aio_fildes, aiocb->aio_ioctl_cmd, aiocb->aio_ioctl_buf);
126 if (ret == -1)
127 return -errno;
130 * This looks weird, but the aio code only consideres a request
131 * successful if it has written the number full number of bytes.
133 * Now we overload aio_nbytes as aio_ioctl_cmd for the ioctl command,
134 * so in fact we return the ioctl command here to make posix_aio_read()
135 * happy..
137 return aiocb->aio_nbytes;
140 static ssize_t handle_aiocb_flush(struct qemu_paiocb *aiocb)
142 int ret;
144 ret = qemu_fdatasync(aiocb->aio_fildes);
145 if (ret == -1)
146 return -errno;
147 return 0;
150 #ifdef CONFIG_PREADV
152 static ssize_t
153 qemu_preadv(int fd, const struct iovec *iov, int nr_iov, off_t offset)
155 return preadv(fd, iov, nr_iov, offset);
158 static ssize_t
159 qemu_pwritev(int fd, const struct iovec *iov, int nr_iov, off_t offset)
161 return pwritev(fd, iov, nr_iov, offset);
164 #else
166 static ssize_t
167 qemu_preadv(int fd, const struct iovec *iov, int nr_iov, off_t offset)
169 return -ENOSYS;
172 static ssize_t
173 qemu_pwritev(int fd, const struct iovec *iov, int nr_iov, off_t offset)
175 return -ENOSYS;
178 #endif
180 static ssize_t handle_aiocb_rw_vector(struct qemu_paiocb *aiocb)
182 size_t offset = 0;
183 ssize_t len;
185 do {
186 if (aiocb->aio_type & QEMU_AIO_WRITE)
187 len = qemu_pwritev(aiocb->aio_fildes,
188 aiocb->aio_iov,
189 aiocb->aio_niov,
190 aiocb->aio_offset + offset);
191 else
192 len = qemu_preadv(aiocb->aio_fildes,
193 aiocb->aio_iov,
194 aiocb->aio_niov,
195 aiocb->aio_offset + offset);
196 } while (len == -1 && errno == EINTR);
198 if (len == -1)
199 return -errno;
200 return len;
203 static ssize_t handle_aiocb_rw_linear(struct qemu_paiocb *aiocb, char *buf)
205 ssize_t offset = 0;
206 ssize_t len;
208 while (offset < aiocb->aio_nbytes) {
209 if (aiocb->aio_type & QEMU_AIO_WRITE)
210 len = pwrite(aiocb->aio_fildes,
211 (const char *)buf + offset,
212 aiocb->aio_nbytes - offset,
213 aiocb->aio_offset + offset);
214 else
215 len = pread(aiocb->aio_fildes,
216 buf + offset,
217 aiocb->aio_nbytes - offset,
218 aiocb->aio_offset + offset);
220 if (len == -1 && errno == EINTR)
221 continue;
222 else if (len == -1) {
223 offset = -errno;
224 break;
225 } else if (len == 0)
226 break;
228 offset += len;
231 return offset;
234 static ssize_t handle_aiocb_rw(struct qemu_paiocb *aiocb)
236 ssize_t nbytes;
237 char *buf;
239 if (!(aiocb->aio_type & QEMU_AIO_MISALIGNED)) {
241 * If there is just a single buffer, and it is properly aligned
242 * we can just use plain pread/pwrite without any problems.
244 if (aiocb->aio_niov == 1)
245 return handle_aiocb_rw_linear(aiocb, aiocb->aio_iov->iov_base);
248 * We have more than one iovec, and all are properly aligned.
250 * Try preadv/pwritev first and fall back to linearizing the
251 * buffer if it's not supported.
253 if (preadv_present) {
254 nbytes = handle_aiocb_rw_vector(aiocb);
255 if (nbytes == aiocb->aio_nbytes)
256 return nbytes;
257 if (nbytes < 0 && nbytes != -ENOSYS)
258 return nbytes;
259 preadv_present = 0;
263 * XXX(hch): short read/write. no easy way to handle the reminder
264 * using these interfaces. For now retry using plain
265 * pread/pwrite?
270 * Ok, we have to do it the hard way, copy all segments into
271 * a single aligned buffer.
273 buf = qemu_blockalign(aiocb->common.bs, aiocb->aio_nbytes);
274 if (aiocb->aio_type & QEMU_AIO_WRITE) {
275 char *p = buf;
276 int i;
278 for (i = 0; i < aiocb->aio_niov; ++i) {
279 memcpy(p, aiocb->aio_iov[i].iov_base, aiocb->aio_iov[i].iov_len);
280 p += aiocb->aio_iov[i].iov_len;
284 nbytes = handle_aiocb_rw_linear(aiocb, buf);
285 if (!(aiocb->aio_type & QEMU_AIO_WRITE)) {
286 char *p = buf;
287 size_t count = aiocb->aio_nbytes, copy;
288 int i;
290 for (i = 0; i < aiocb->aio_niov && count; ++i) {
291 copy = count;
292 if (copy > aiocb->aio_iov[i].iov_len)
293 copy = aiocb->aio_iov[i].iov_len;
294 memcpy(aiocb->aio_iov[i].iov_base, p, copy);
295 p += copy;
296 count -= copy;
299 qemu_vfree(buf);
301 return nbytes;
304 static void *aio_thread(void *unused)
306 pid_t pid;
308 pid = getpid();
310 while (1) {
311 struct qemu_paiocb *aiocb;
312 ssize_t ret = 0;
313 qemu_timeval tv;
314 struct timespec ts;
316 qemu_gettimeofday(&tv);
317 ts.tv_sec = tv.tv_sec + 10;
318 ts.tv_nsec = 0;
320 mutex_lock(&lock);
322 while (QTAILQ_EMPTY(&request_list) &&
323 !(ret == ETIMEDOUT)) {
324 idle_threads++;
325 ret = cond_timedwait(&cond, &lock, &ts);
326 idle_threads--;
329 if (QTAILQ_EMPTY(&request_list))
330 break;
332 aiocb = QTAILQ_FIRST(&request_list);
333 QTAILQ_REMOVE(&request_list, aiocb, node);
334 aiocb->active = 1;
335 mutex_unlock(&lock);
337 switch (aiocb->aio_type & QEMU_AIO_TYPE_MASK) {
338 case QEMU_AIO_READ:
339 case QEMU_AIO_WRITE:
340 ret = handle_aiocb_rw(aiocb);
341 break;
342 case QEMU_AIO_FLUSH:
343 ret = handle_aiocb_flush(aiocb);
344 break;
345 case QEMU_AIO_IOCTL:
346 ret = handle_aiocb_ioctl(aiocb);
347 break;
348 default:
349 fprintf(stderr, "invalid aio request (0x%x)\n", aiocb->aio_type);
350 ret = -EINVAL;
351 break;
354 mutex_lock(&lock);
355 aiocb->ret = ret;
356 mutex_unlock(&lock);
358 if (kill(pid, aiocb->ev_signo)) die("kill failed");
361 cur_threads--;
362 mutex_unlock(&lock);
364 return NULL;
367 static void spawn_thread(void)
369 sigset_t set, oldset;
371 cur_threads++;
373 /* block all signals */
374 if (sigfillset(&set)) die("sigfillset");
375 if (sigprocmask(SIG_SETMASK, &set, &oldset)) die("sigprocmask");
377 thread_create(&thread_id, &attr, aio_thread, NULL);
379 if (sigprocmask(SIG_SETMASK, &oldset, NULL)) die("sigprocmask restore");
382 static void qemu_paio_submit(struct qemu_paiocb *aiocb)
384 aiocb->ret = -EINPROGRESS;
385 aiocb->active = 0;
386 mutex_lock(&lock);
387 if (idle_threads == 0 && cur_threads < max_threads)
388 spawn_thread();
389 QTAILQ_INSERT_TAIL(&request_list, aiocb, node);
390 mutex_unlock(&lock);
391 cond_signal(&cond);
394 static ssize_t qemu_paio_return(struct qemu_paiocb *aiocb)
396 ssize_t ret;
398 mutex_lock(&lock);
399 ret = aiocb->ret;
400 mutex_unlock(&lock);
402 return ret;
405 static int qemu_paio_error(struct qemu_paiocb *aiocb)
407 ssize_t ret = qemu_paio_return(aiocb);
409 if (ret < 0)
410 ret = -ret;
411 else
412 ret = 0;
414 return ret;
417 static int posix_aio_process_queue(void *opaque)
419 PosixAioState *s = opaque;
420 struct qemu_paiocb *acb, **pacb;
421 int ret;
422 int result = 0;
423 int async_context_id = get_async_context_id();
425 for(;;) {
426 pacb = &s->first_aio;
427 for(;;) {
428 acb = *pacb;
429 if (!acb)
430 return result;
432 /* we're only interested in requests in the right context */
433 if (acb->async_context_id != async_context_id) {
434 pacb = &acb->next;
435 continue;
438 ret = qemu_paio_error(acb);
439 if (ret == ECANCELED) {
440 /* remove the request */
441 *pacb = acb->next;
442 qemu_aio_release(acb);
443 result = 1;
444 } else if (ret != EINPROGRESS) {
445 /* end of aio */
446 if (ret == 0) {
447 ret = qemu_paio_return(acb);
448 if (ret == acb->aio_nbytes)
449 ret = 0;
450 else
451 ret = -EINVAL;
452 } else {
453 ret = -ret;
456 trace_paio_complete(acb, acb->common.opaque, ret);
458 /* remove the request */
459 *pacb = acb->next;
460 /* call the callback */
461 acb->common.cb(acb->common.opaque, ret);
462 qemu_aio_release(acb);
463 result = 1;
464 break;
465 } else {
466 pacb = &acb->next;
471 return result;
474 static void posix_aio_read(void *opaque)
476 PosixAioState *s = opaque;
477 ssize_t len;
479 /* read all bytes from signal pipe */
480 for (;;) {
481 char bytes[16];
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 */
488 break;
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) {
505 char byte = 0;
506 ssize_t ret;
508 ret = write(posix_aio_state->wfd, &byte, sizeof(byte));
509 if (ret < 0 && errno != EAGAIN)
510 die("write()");
513 qemu_service_io();
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;
522 for(;;) {
523 if (*pacb == NULL) {
524 fprintf(stderr, "paio_remove: aio request not found!\n");
525 break;
526 } else if (*pacb == acb) {
527 *pacb = acb->next;
528 qemu_aio_release(acb);
529 break;
531 pacb = &(*pacb)->next;
535 static void paio_cancel(BlockDriverAIOCB *blockacb)
537 struct qemu_paiocb *acb = (struct qemu_paiocb *)blockacb;
538 int active = 0;
540 trace_paio_cancel(acb, acb->common.opaque);
542 mutex_lock(&lock);
543 if (!acb->active) {
544 QTAILQ_REMOVE(&request_list, acb, node);
545 acb->ret = -ECANCELED;
546 } else if (acb->ret == -EINPROGRESS) {
547 active = 1;
549 mutex_unlock(&lock);
551 if (active) {
552 /* fail safe: if the aio could not be canceled, we wait for
553 it */
554 while (qemu_paio_error(acb) == EINPROGRESS)
558 paio_remove(acb);
561 static AIOPool raw_aio_pool = {
562 .aiocb_size = sizeof(struct qemu_paiocb),
563 .cancel = paio_cancel,
566 BlockDriverAIOCB *paio_submit(BlockDriverState *bs, int fd,
567 int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
568 BlockDriverCompletionFunc *cb, void *opaque, int type)
570 struct qemu_paiocb *acb;
572 acb = qemu_aio_get(&raw_aio_pool, bs, cb, opaque);
573 if (!acb)
574 return NULL;
575 acb->aio_type = type;
576 acb->aio_fildes = fd;
577 acb->ev_signo = SIGUSR2;
578 acb->async_context_id = get_async_context_id();
580 if (qiov) {
581 acb->aio_iov = qiov->iov;
582 acb->aio_niov = qiov->niov;
584 acb->aio_nbytes = nb_sectors * 512;
585 acb->aio_offset = sector_num * 512;
587 acb->next = posix_aio_state->first_aio;
588 posix_aio_state->first_aio = acb;
590 trace_paio_submit(acb, opaque, sector_num, nb_sectors, type);
591 qemu_paio_submit(acb);
592 return &acb->common;
595 BlockDriverAIOCB *paio_ioctl(BlockDriverState *bs, int fd,
596 unsigned long int req, void *buf,
597 BlockDriverCompletionFunc *cb, void *opaque)
599 struct qemu_paiocb *acb;
601 acb = qemu_aio_get(&raw_aio_pool, bs, cb, opaque);
602 if (!acb)
603 return NULL;
604 acb->aio_type = QEMU_AIO_IOCTL;
605 acb->aio_fildes = fd;
606 acb->ev_signo = SIGUSR2;
607 acb->async_context_id = get_async_context_id();
608 acb->aio_offset = 0;
609 acb->aio_ioctl_buf = buf;
610 acb->aio_ioctl_cmd = req;
612 acb->next = posix_aio_state->first_aio;
613 posix_aio_state->first_aio = acb;
615 qemu_paio_submit(acb);
616 return &acb->common;
619 int paio_init(void)
621 struct sigaction act;
622 PosixAioState *s;
623 int fds[2];
624 int ret;
626 if (posix_aio_state)
627 return 0;
629 s = qemu_malloc(sizeof(PosixAioState));
631 sigfillset(&act.sa_mask);
632 act.sa_flags = 0; /* do not restart syscalls to interrupt select() */
633 act.sa_handler = aio_signal_handler;
634 sigaction(SIGUSR2, &act, NULL);
636 s->first_aio = NULL;
637 if (qemu_pipe(fds) == -1) {
638 fprintf(stderr, "failed to create pipe\n");
639 return -1;
642 s->rfd = fds[0];
643 s->wfd = fds[1];
645 fcntl(s->rfd, F_SETFL, O_NONBLOCK);
646 fcntl(s->wfd, F_SETFL, O_NONBLOCK);
648 qemu_aio_set_fd_handler(s->rfd, posix_aio_read, NULL, posix_aio_flush,
649 posix_aio_process_queue, s);
651 ret = pthread_attr_init(&attr);
652 if (ret)
653 die2(ret, "pthread_attr_init");
655 ret = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
656 if (ret)
657 die2(ret, "pthread_attr_setdetachstate");
659 QTAILQ_INIT(&request_list);
661 posix_aio_state = s;
662 return 0;