1 /*
2 * QEMU posix-aio emulation
3 *
4 * Copyright IBM, Corp. 2008
5 *
6 * Authors:
7 * Anthony Liguori <aliguori@us.ibm.com>
8 *
9 * This work is licensed under the terms of the GNU GPL, version 2. See
10 * the COPYING file in the top-level directory.
11 *
12 */
13
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 <signal.h>
21 #include <string.h>
22 #include <stdlib.h>
23 #include <stdio.h>
24
25 #include "qemu-queue.h"
26 #include "osdep.h"
27 #include "sysemu.h"
28 #include "qemu-common.h"
29 #include "block_int.h"
30
31 #include "block/raw-posix-aio.h"
32
33
34 struct qemu_paiocb {
35 BlockDriverAIOCB common;
36 int aio_fildes;
37 union {
38 struct iovec *aio_iov;
39 void *aio_ioctl_buf;
40 };
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;
46
47 QTAILQ_ENTRY(qemu_paiocb) node;
48 int aio_type;
49 ssize_t ret;
50 int active;
51 struct qemu_paiocb *next;
52
53 int async_context_id;
54 };
55
56 typedef struct PosixAioState {
57 int rfd, wfd;
58 struct qemu_paiocb *first_aio;
59 } PosixAioState;
60
61
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;
70
71 #ifdef CONFIG_PREADV
72 static int preadv_present = 1;
73 #else
74 static int preadv_present = 0;
75 #endif
76
die2(int err,const char * what)77 static void die2(int err, const char *what)
78 {
79 fprintf(stderr, "%s failed: %s\n", what, strerror(err));
80 abort();
81 }
82
die(const char * what)83 static void die(const char *what)
84 {
85 die2(errno, what);
86 }
87
mutex_lock(pthread_mutex_t * mutex)88 static void mutex_lock(pthread_mutex_t *mutex)
89 {
90 int ret = pthread_mutex_lock(mutex);
91 if (ret) die2(ret, "pthread_mutex_lock");
92 }
93
mutex_unlock(pthread_mutex_t * mutex)94 static void mutex_unlock(pthread_mutex_t *mutex)
95 {
96 int ret = pthread_mutex_unlock(mutex);
97 if (ret) die2(ret, "pthread_mutex_unlock");
98 }
99
cond_timedwait(pthread_cond_t * cond,pthread_mutex_t * mutex,struct timespec * ts)100 static int cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex,
101 struct timespec *ts)
102 {
103 int ret = pthread_cond_timedwait(cond, mutex, ts);
104 if (ret && ret != ETIMEDOUT) die2(ret, "pthread_cond_timedwait");
105 return ret;
106 }
107
cond_signal(pthread_cond_t * cond)108 static void cond_signal(pthread_cond_t *cond)
109 {
110 int ret = pthread_cond_signal(cond);
111 if (ret) die2(ret, "pthread_cond_signal");
112 }
113
thread_create(pthread_t * thread,pthread_attr_t * attr,void * (* start_routine)(void *),void * arg)114 static void thread_create(pthread_t *thread, pthread_attr_t *attr,
115 void *(*start_routine)(void*), void *arg)
116 {
117 int ret = pthread_create(thread, attr, start_routine, arg);
118 if (ret) die2(ret, "pthread_create");
119 }
120
handle_aiocb_ioctl(struct qemu_paiocb * aiocb)121 static ssize_t handle_aiocb_ioctl(struct qemu_paiocb *aiocb)
122 {
123 int ret;
124
125 ret = ioctl(aiocb->aio_fildes, aiocb->aio_ioctl_cmd, aiocb->aio_ioctl_buf);
126 if (ret == -1)
127 return -errno;
128
129 /*
130 * This looks weird, but the aio code only consideres a request
131 * successful if it has written the number full number of bytes.
132 *
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..
136 */
137 return aiocb->aio_nbytes;
138 }
139
handle_aiocb_flush(struct qemu_paiocb * aiocb)140 static ssize_t handle_aiocb_flush(struct qemu_paiocb *aiocb)
141 {
142 int ret;
143
144 ret = qemu_fdatasync(aiocb->aio_fildes);
145 if (ret == -1)
146 return -errno;
147 return 0;
148 }
149
150 #ifdef CONFIG_PREADV
151
152 static ssize_t
qemu_preadv(int fd,const struct iovec * iov,int nr_iov,off_t offset)153 qemu_preadv(int fd, const struct iovec *iov, int nr_iov, off_t offset)
154 {
155 return preadv(fd, iov, nr_iov, offset);
156 }
157
158 static ssize_t
qemu_pwritev(int fd,const struct iovec * iov,int nr_iov,off_t offset)159 qemu_pwritev(int fd, const struct iovec *iov, int nr_iov, off_t offset)
160 {
161 return pwritev(fd, iov, nr_iov, offset);
162 }
163
164 #else
165
166 static ssize_t
qemu_preadv(int fd,const struct iovec * iov,int nr_iov,off_t offset)167 qemu_preadv(int fd, const struct iovec *iov, int nr_iov, off_t offset)
168 {
169 return -ENOSYS;
170 }
171
172 static ssize_t
qemu_pwritev(int fd,const struct iovec * iov,int nr_iov,off_t offset)173 qemu_pwritev(int fd, const struct iovec *iov, int nr_iov, off_t offset)
174 {
175 return -ENOSYS;
176 }
177
178 #endif
179
handle_aiocb_rw_vector(struct qemu_paiocb * aiocb)180 static ssize_t handle_aiocb_rw_vector(struct qemu_paiocb *aiocb)
181 {
182 size_t offset = 0;
183 ssize_t len;
184
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);
197
198 if (len == -1)
199 return -errno;
200 return len;
201 }
202
handle_aiocb_rw_linear(struct qemu_paiocb * aiocb,char * buf)203 static ssize_t handle_aiocb_rw_linear(struct qemu_paiocb *aiocb, char *buf)
204 {
205 ssize_t offset = 0;
206 ssize_t len;
207
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);
219
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;
227
228 offset += len;
229 }
230
231 return offset;
232 }
233
handle_aiocb_rw(struct qemu_paiocb * aiocb)234 static ssize_t handle_aiocb_rw(struct qemu_paiocb *aiocb)
235 {
236 ssize_t nbytes;
237 char *buf;
238
239 if (!(aiocb->aio_type & QEMU_AIO_MISALIGNED)) {
240 /*
241 * If there is just a single buffer, and it is properly aligned
242 * we can just use plain pread/pwrite without any problems.
243 */
244 if (aiocb->aio_niov == 1)
245 return handle_aiocb_rw_linear(aiocb, aiocb->aio_iov->iov_base);
246
247 /*
248 * We have more than one iovec, and all are properly aligned.
249 *
250 * Try preadv/pwritev first and fall back to linearizing the
251 * buffer if it's not supported.
252 */
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;
260 }
261
262 /*
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?
266 */
267 }
268
269 /*
270 * Ok, we have to do it the hard way, copy all segments into
271 * a single aligned buffer.
272 */
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;
277
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;
281 }
282 }
283
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;
289
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;
297 }
298 }
299 qemu_vfree(buf);
300
301 return nbytes;
302 }
303
aio_thread(void * unused)304 static void *aio_thread(void *unused)
305 {
306 pid_t pid;
307
308 pid = getpid();
309
310 while (1) {
311 struct qemu_paiocb *aiocb;
312 ssize_t ret = 0;
313 qemu_timeval tv;
314 struct timespec ts;
315
316 qemu_gettimeofday(&tv);
317 ts.tv_sec = tv.tv_sec + 10;
318 ts.tv_nsec = 0;
319
320 mutex_lock(&lock);
321
322 while (QTAILQ_EMPTY(&request_list) &&
323 !(ret == ETIMEDOUT)) {
324 ret = cond_timedwait(&cond, &lock, &ts);
325 }
326
327 if (QTAILQ_EMPTY(&request_list))
328 break;
329
330 aiocb = QTAILQ_FIRST(&request_list);
331 QTAILQ_REMOVE(&request_list, aiocb, node);
332 aiocb->active = 1;
333 idle_threads--;
334 mutex_unlock(&lock);
335
336 switch (aiocb->aio_type & QEMU_AIO_TYPE_MASK) {
337 case QEMU_AIO_READ:
338 case QEMU_AIO_WRITE:
339 ret = handle_aiocb_rw(aiocb);
340 break;
341 case QEMU_AIO_FLUSH:
342 ret = handle_aiocb_flush(aiocb);
343 break;
344 case QEMU_AIO_IOCTL:
345 ret = handle_aiocb_ioctl(aiocb);
346 break;
347 default:
348 fprintf(stderr, "invalid aio request (0x%x)\n", aiocb->aio_type);
349 ret = -EINVAL;
350 break;
351 }
352
353 mutex_lock(&lock);
354 aiocb->ret = ret;
355 idle_threads++;
356 mutex_unlock(&lock);
357
358 if (kill(pid, aiocb->ev_signo)) die("kill failed");
359 }
360
361 idle_threads--;
362 cur_threads--;
363 mutex_unlock(&lock);
364
365 return NULL;
366 }
367
spawn_thread(void)368 static void spawn_thread(void)
369 {
370 sigset_t set, oldset;
371
372 cur_threads++;
373 idle_threads++;
374
375 /* block all signals */
376 if (sigfillset(&set)) die("sigfillset");
377 if (sigprocmask(SIG_SETMASK, &set, &oldset)) die("sigprocmask");
378
379 thread_create(&thread_id, &attr, aio_thread, NULL);
380
381 if (sigprocmask(SIG_SETMASK, &oldset, NULL)) die("sigprocmask restore");
382 }
383
qemu_paio_submit(struct qemu_paiocb * aiocb)384 static void qemu_paio_submit(struct qemu_paiocb *aiocb)
385 {
386 aiocb->ret = -EINPROGRESS;
387 aiocb->active = 0;
388 mutex_lock(&lock);
389 if (idle_threads == 0 && cur_threads < max_threads)
390 spawn_thread();
391 QTAILQ_INSERT_TAIL(&request_list, aiocb, node);
392 mutex_unlock(&lock);
393 cond_signal(&cond);
394 }
395
qemu_paio_return(struct qemu_paiocb * aiocb)396 static ssize_t qemu_paio_return(struct qemu_paiocb *aiocb)
397 {
398 ssize_t ret;
399
400 mutex_lock(&lock);
401 ret = aiocb->ret;
402 mutex_unlock(&lock);
403
404 return ret;
405 }
406
qemu_paio_error(struct qemu_paiocb * aiocb)407 static int qemu_paio_error(struct qemu_paiocb *aiocb)
408 {
409 ssize_t ret = qemu_paio_return(aiocb);
410
411 if (ret < 0)
412 ret = -ret;
413 else
414 ret = 0;
415
416 return ret;
417 }
418
posix_aio_process_queue(void * opaque)419 static int posix_aio_process_queue(void *opaque)
420 {
421 PosixAioState *s = opaque;
422 struct qemu_paiocb *acb, **pacb;
423 int ret;
424 int result = 0;
425 int async_context_id = get_async_context_id();
426
427 for(;;) {
428 pacb = &s->first_aio;
429 for(;;) {
430 acb = *pacb;
431 if (!acb)
432 return result;
433
434 /* we're only interested in requests in the right context */
435 if (acb->async_context_id != async_context_id) {
436 pacb = &acb->next;
437 continue;
438 }
439
440 ret = qemu_paio_error(acb);
441 if (ret == ECANCELED) {
442 /* remove the request */
443 *pacb = acb->next;
444 qemu_aio_release(acb);
445 result = 1;
446 } else if (ret != EINPROGRESS) {
447 /* end of aio */
448 if (ret == 0) {
449 ret = qemu_paio_return(acb);
450 if (ret == acb->aio_nbytes)
451 ret = 0;
452 else
453 ret = -EINVAL;
454 } else {
455 ret = -ret;
456 }
457
458 //trace_paio_complete(acb, acb->common.opaque, ret);
459
460 /* remove the request */
461 *pacb = acb->next;
462 /* call the callback */
463 acb->common.cb(acb->common.opaque, ret);
464 qemu_aio_release(acb);
465 result = 1;
466 break;
467 } else {
468 pacb = &acb->next;
469 }
470 }
471 }
472
473 return result;
474 }
475
posix_aio_read(void * opaque)476 static void posix_aio_read(void *opaque)
477 {
478 PosixAioState *s = opaque;
479 ssize_t len;
480
481 /* read all bytes from signal pipe */
482 for (;;) {
483 char bytes[16];
484
485 len = read(s->rfd, bytes, sizeof(bytes));
486 if (len == -1 && errno == EINTR)
487 continue; /* try again */
488 if (len == sizeof(bytes))
489 continue; /* more to read */
490 break;
491 }
492
493 posix_aio_process_queue(s);
494 }
495
posix_aio_flush(void * opaque)496 static int posix_aio_flush(void *opaque)
497 {
498 PosixAioState *s = opaque;
499 return !!s->first_aio;
500 }
501
502 static PosixAioState *posix_aio_state;
503
aio_signal_handler(int signum)504 static void aio_signal_handler(int signum)
505 {
506 if (posix_aio_state) {
507 char byte = 0;
508 ssize_t ret;
509
510 ret = write(posix_aio_state->wfd, &byte, sizeof(byte));
511 if (ret < 0 && errno != EAGAIN)
512 die("write()");
513 }
514
515 qemu_service_io();
516 }
517
paio_remove(struct qemu_paiocb * acb)518 static void paio_remove(struct qemu_paiocb *acb)
519 {
520 struct qemu_paiocb **pacb;
521
522 /* remove the callback from the queue */
523 pacb = &posix_aio_state->first_aio;
524 for(;;) {
525 if (*pacb == NULL) {
526 fprintf(stderr, "paio_remove: aio request not found!\n");
527 break;
528 } else if (*pacb == acb) {
529 *pacb = acb->next;
530 qemu_aio_release(acb);
531 break;
532 }
533 pacb = &(*pacb)->next;
534 }
535 }
536
paio_cancel(BlockDriverAIOCB * blockacb)537 static void paio_cancel(BlockDriverAIOCB *blockacb)
538 {
539 struct qemu_paiocb *acb = (struct qemu_paiocb *)blockacb;
540 int active = 0;
541
542 //trace_paio_cancel(acb, acb->common.opaque);
543
544 mutex_lock(&lock);
545 if (!acb->active) {
546 QTAILQ_REMOVE(&request_list, acb, node);
547 acb->ret = -ECANCELED;
548 } else if (acb->ret == -EINPROGRESS) {
549 active = 1;
550 }
551 mutex_unlock(&lock);
552
553 if (active) {
554 /* fail safe: if the aio could not be canceled, we wait for
555 it */
556 while (qemu_paio_error(acb) == EINPROGRESS)
557 ;
558 }
559
560 paio_remove(acb);
561 }
562
563 static AIOPool raw_aio_pool = {
564 .aiocb_size = sizeof(struct qemu_paiocb),
565 .cancel = paio_cancel,
566 };
567
paio_submit(BlockDriverState * bs,int fd,int64_t sector_num,QEMUIOVector * qiov,int nb_sectors,BlockDriverCompletionFunc * cb,void * opaque,int type)568 BlockDriverAIOCB *paio_submit(BlockDriverState *bs, int fd,
569 int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
570 BlockDriverCompletionFunc *cb, void *opaque, int type)
571 {
572 struct qemu_paiocb *acb;
573
574 acb = qemu_aio_get(&raw_aio_pool, bs, cb, opaque);
575 if (!acb)
576 return NULL;
577 acb->aio_type = type;
578 acb->aio_fildes = fd;
579 acb->ev_signo = SIGUSR2;
580 acb->async_context_id = get_async_context_id();
581
582 if (qiov) {
583 acb->aio_iov = qiov->iov;
584 acb->aio_niov = qiov->niov;
585 }
586 acb->aio_nbytes = nb_sectors * 512;
587 acb->aio_offset = sector_num * 512;
588
589 acb->next = posix_aio_state->first_aio;
590 posix_aio_state->first_aio = acb;
591
592 //trace_paio_submit(acb, opaque, sector_num, nb_sectors, type);
593 qemu_paio_submit(acb);
594 return &acb->common;
595 }
596
paio_ioctl(BlockDriverState * bs,int fd,unsigned long int req,void * buf,BlockDriverCompletionFunc * cb,void * opaque)597 BlockDriverAIOCB *paio_ioctl(BlockDriverState *bs, int fd,
598 unsigned long int req, void *buf,
599 BlockDriverCompletionFunc *cb, void *opaque)
600 {
601 struct qemu_paiocb *acb;
602
603 acb = qemu_aio_get(&raw_aio_pool, bs, cb, opaque);
604 if (!acb)
605 return NULL;
606 acb->aio_type = QEMU_AIO_IOCTL;
607 acb->aio_fildes = fd;
608 acb->ev_signo = SIGUSR2;
609 acb->async_context_id = get_async_context_id();
610 acb->aio_offset = 0;
611 acb->aio_ioctl_buf = buf;
612 acb->aio_ioctl_cmd = req;
613
614 acb->next = posix_aio_state->first_aio;
615 posix_aio_state->first_aio = acb;
616
617 qemu_paio_submit(acb);
618 return &acb->common;
619 }
620
paio_init(void)621 int paio_init(void)
622 {
623 struct sigaction act;
624 PosixAioState *s;
625 int fds[2];
626 int ret;
627
628 if (posix_aio_state)
629 return 0;
630
631 s = qemu_malloc(sizeof(PosixAioState));
632
633 sigfillset(&act.sa_mask);
634 act.sa_flags = 0; /* do not restart syscalls to interrupt select() */
635 act.sa_handler = aio_signal_handler;
636 sigaction(SIGUSR2, &act, NULL);
637
638 s->first_aio = NULL;
639 if (qemu_pipe(fds) == -1) {
640 fprintf(stderr, "failed to create pipe\n");
641 return -1;
642 }
643
644 s->rfd = fds[0];
645 s->wfd = fds[1];
646
647 fcntl(s->rfd, F_SETFL, O_NONBLOCK);
648 fcntl(s->wfd, F_SETFL, O_NONBLOCK);
649
650 qemu_aio_set_fd_handler(s->rfd, posix_aio_read, NULL, posix_aio_flush,
651 posix_aio_process_queue, s);
652
653 ret = pthread_attr_init(&attr);
654 if (ret)
655 die2(ret, "pthread_attr_init");
656
657 ret = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
658 if (ret)
659 die2(ret, "pthread_attr_setdetachstate");
660
661 QTAILQ_INIT(&request_list);
662
663 posix_aio_state = s;
664 return 0;
665 }
666