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