664 lines
16 KiB
C
664 lines
16 KiB
C
/*
|
|
* QEMU posix-aio emulation
|
|
*
|
|
* Copyright IBM, Corp. 2008
|
|
*
|
|
* Authors:
|
|
* Anthony Liguori <aliguori@us.ibm.com>
|
|
*
|
|
* This work is licensed under the terms of the GNU GPL, version 2. See
|
|
* the COPYING file in the top-level directory.
|
|
*
|
|
*/
|
|
|
|
#include <sys/ioctl.h>
|
|
#include <sys/types.h>
|
|
#include <pthread.h>
|
|
#include <unistd.h>
|
|
#include <errno.h>
|
|
#include <time.h>
|
|
#include <string.h>
|
|
#include <stdlib.h>
|
|
#include <stdio.h>
|
|
|
|
#include "qemu-queue.h"
|
|
#include "osdep.h"
|
|
#include "sysemu.h"
|
|
#include "qemu-common.h"
|
|
#include "trace.h"
|
|
#include "block_int.h"
|
|
|
|
#include "block/raw-posix-aio.h"
|
|
|
|
|
|
struct qemu_paiocb {
|
|
BlockDriverAIOCB common;
|
|
int aio_fildes;
|
|
union {
|
|
struct iovec *aio_iov;
|
|
void *aio_ioctl_buf;
|
|
};
|
|
int aio_niov;
|
|
size_t aio_nbytes;
|
|
#define aio_ioctl_cmd aio_nbytes /* for QEMU_AIO_IOCTL */
|
|
int ev_signo;
|
|
off_t aio_offset;
|
|
|
|
QTAILQ_ENTRY(qemu_paiocb) node;
|
|
int aio_type;
|
|
ssize_t ret;
|
|
int active;
|
|
struct qemu_paiocb *next;
|
|
|
|
int async_context_id;
|
|
};
|
|
|
|
typedef struct PosixAioState {
|
|
int rfd, wfd;
|
|
struct qemu_paiocb *first_aio;
|
|
} PosixAioState;
|
|
|
|
|
|
static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
|
|
static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
|
|
static pthread_t thread_id;
|
|
static pthread_attr_t attr;
|
|
static int max_threads = 64;
|
|
static int cur_threads = 0;
|
|
static int idle_threads = 0;
|
|
static QTAILQ_HEAD(, qemu_paiocb) request_list;
|
|
|
|
#ifdef CONFIG_PREADV
|
|
static int preadv_present = 1;
|
|
#else
|
|
static int preadv_present = 0;
|
|
#endif
|
|
|
|
static void die2(int err, const char *what)
|
|
{
|
|
fprintf(stderr, "%s failed: %s\n", what, strerror(err));
|
|
abort();
|
|
}
|
|
|
|
static void die(const char *what)
|
|
{
|
|
die2(errno, what);
|
|
}
|
|
|
|
static void mutex_lock(pthread_mutex_t *mutex)
|
|
{
|
|
int ret = pthread_mutex_lock(mutex);
|
|
if (ret) die2(ret, "pthread_mutex_lock");
|
|
}
|
|
|
|
static void mutex_unlock(pthread_mutex_t *mutex)
|
|
{
|
|
int ret = pthread_mutex_unlock(mutex);
|
|
if (ret) die2(ret, "pthread_mutex_unlock");
|
|
}
|
|
|
|
static int cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex,
|
|
struct timespec *ts)
|
|
{
|
|
int ret = pthread_cond_timedwait(cond, mutex, ts);
|
|
if (ret && ret != ETIMEDOUT) die2(ret, "pthread_cond_timedwait");
|
|
return ret;
|
|
}
|
|
|
|
static void cond_signal(pthread_cond_t *cond)
|
|
{
|
|
int ret = pthread_cond_signal(cond);
|
|
if (ret) die2(ret, "pthread_cond_signal");
|
|
}
|
|
|
|
static void thread_create(pthread_t *thread, pthread_attr_t *attr,
|
|
void *(*start_routine)(void*), void *arg)
|
|
{
|
|
int ret = pthread_create(thread, attr, start_routine, arg);
|
|
if (ret) die2(ret, "pthread_create");
|
|
}
|
|
|
|
static ssize_t handle_aiocb_ioctl(struct qemu_paiocb *aiocb)
|
|
{
|
|
int ret;
|
|
|
|
ret = ioctl(aiocb->aio_fildes, aiocb->aio_ioctl_cmd, aiocb->aio_ioctl_buf);
|
|
if (ret == -1)
|
|
return -errno;
|
|
|
|
/*
|
|
* This looks weird, but the aio code only consideres a request
|
|
* successful if it has written the number full number of bytes.
|
|
*
|
|
* Now we overload aio_nbytes as aio_ioctl_cmd for the ioctl command,
|
|
* so in fact we return the ioctl command here to make posix_aio_read()
|
|
* happy..
|
|
*/
|
|
return aiocb->aio_nbytes;
|
|
}
|
|
|
|
static ssize_t handle_aiocb_flush(struct qemu_paiocb *aiocb)
|
|
{
|
|
int ret;
|
|
|
|
ret = qemu_fdatasync(aiocb->aio_fildes);
|
|
if (ret == -1)
|
|
return -errno;
|
|
return 0;
|
|
}
|
|
|
|
#ifdef CONFIG_PREADV
|
|
|
|
static ssize_t
|
|
qemu_preadv(int fd, const struct iovec *iov, int nr_iov, off_t offset)
|
|
{
|
|
return preadv(fd, iov, nr_iov, offset);
|
|
}
|
|
|
|
static ssize_t
|
|
qemu_pwritev(int fd, const struct iovec *iov, int nr_iov, off_t offset)
|
|
{
|
|
return pwritev(fd, iov, nr_iov, offset);
|
|
}
|
|
|
|
#else
|
|
|
|
static ssize_t
|
|
qemu_preadv(int fd, const struct iovec *iov, int nr_iov, off_t offset)
|
|
{
|
|
return -ENOSYS;
|
|
}
|
|
|
|
static ssize_t
|
|
qemu_pwritev(int fd, const struct iovec *iov, int nr_iov, off_t offset)
|
|
{
|
|
return -ENOSYS;
|
|
}
|
|
|
|
#endif
|
|
|
|
static ssize_t handle_aiocb_rw_vector(struct qemu_paiocb *aiocb)
|
|
{
|
|
size_t offset = 0;
|
|
ssize_t len;
|
|
|
|
do {
|
|
if (aiocb->aio_type & QEMU_AIO_WRITE)
|
|
len = qemu_pwritev(aiocb->aio_fildes,
|
|
aiocb->aio_iov,
|
|
aiocb->aio_niov,
|
|
aiocb->aio_offset + offset);
|
|
else
|
|
len = qemu_preadv(aiocb->aio_fildes,
|
|
aiocb->aio_iov,
|
|
aiocb->aio_niov,
|
|
aiocb->aio_offset + offset);
|
|
} while (len == -1 && errno == EINTR);
|
|
|
|
if (len == -1)
|
|
return -errno;
|
|
return len;
|
|
}
|
|
|
|
static ssize_t handle_aiocb_rw_linear(struct qemu_paiocb *aiocb, char *buf)
|
|
{
|
|
ssize_t offset = 0;
|
|
ssize_t len;
|
|
|
|
while (offset < aiocb->aio_nbytes) {
|
|
if (aiocb->aio_type & QEMU_AIO_WRITE)
|
|
len = pwrite(aiocb->aio_fildes,
|
|
(const char *)buf + offset,
|
|
aiocb->aio_nbytes - offset,
|
|
aiocb->aio_offset + offset);
|
|
else
|
|
len = pread(aiocb->aio_fildes,
|
|
buf + offset,
|
|
aiocb->aio_nbytes - offset,
|
|
aiocb->aio_offset + offset);
|
|
|
|
if (len == -1 && errno == EINTR)
|
|
continue;
|
|
else if (len == -1) {
|
|
offset = -errno;
|
|
break;
|
|
} else if (len == 0)
|
|
break;
|
|
|
|
offset += len;
|
|
}
|
|
|
|
return offset;
|
|
}
|
|
|
|
static ssize_t handle_aiocb_rw(struct qemu_paiocb *aiocb)
|
|
{
|
|
ssize_t nbytes;
|
|
char *buf;
|
|
|
|
if (!(aiocb->aio_type & QEMU_AIO_MISALIGNED)) {
|
|
/*
|
|
* If there is just a single buffer, and it is properly aligned
|
|
* we can just use plain pread/pwrite without any problems.
|
|
*/
|
|
if (aiocb->aio_niov == 1)
|
|
return handle_aiocb_rw_linear(aiocb, aiocb->aio_iov->iov_base);
|
|
|
|
/*
|
|
* We have more than one iovec, and all are properly aligned.
|
|
*
|
|
* Try preadv/pwritev first and fall back to linearizing the
|
|
* buffer if it's not supported.
|
|
*/
|
|
if (preadv_present) {
|
|
nbytes = handle_aiocb_rw_vector(aiocb);
|
|
if (nbytes == aiocb->aio_nbytes)
|
|
return nbytes;
|
|
if (nbytes < 0 && nbytes != -ENOSYS)
|
|
return nbytes;
|
|
preadv_present = 0;
|
|
}
|
|
|
|
/*
|
|
* XXX(hch): short read/write. no easy way to handle the reminder
|
|
* using these interfaces. For now retry using plain
|
|
* pread/pwrite?
|
|
*/
|
|
}
|
|
|
|
/*
|
|
* Ok, we have to do it the hard way, copy all segments into
|
|
* a single aligned buffer.
|
|
*/
|
|
buf = qemu_blockalign(aiocb->common.bs, aiocb->aio_nbytes);
|
|
if (aiocb->aio_type & QEMU_AIO_WRITE) {
|
|
char *p = buf;
|
|
int i;
|
|
|
|
for (i = 0; i < aiocb->aio_niov; ++i) {
|
|
memcpy(p, aiocb->aio_iov[i].iov_base, aiocb->aio_iov[i].iov_len);
|
|
p += aiocb->aio_iov[i].iov_len;
|
|
}
|
|
}
|
|
|
|
nbytes = handle_aiocb_rw_linear(aiocb, buf);
|
|
if (!(aiocb->aio_type & QEMU_AIO_WRITE)) {
|
|
char *p = buf;
|
|
size_t count = aiocb->aio_nbytes, copy;
|
|
int i;
|
|
|
|
for (i = 0; i < aiocb->aio_niov && count; ++i) {
|
|
copy = count;
|
|
if (copy > aiocb->aio_iov[i].iov_len)
|
|
copy = aiocb->aio_iov[i].iov_len;
|
|
memcpy(aiocb->aio_iov[i].iov_base, p, copy);
|
|
p += copy;
|
|
count -= copy;
|
|
}
|
|
}
|
|
qemu_vfree(buf);
|
|
|
|
return nbytes;
|
|
}
|
|
|
|
static void *aio_thread(void *unused)
|
|
{
|
|
pid_t pid;
|
|
|
|
pid = getpid();
|
|
|
|
while (1) {
|
|
struct qemu_paiocb *aiocb;
|
|
ssize_t ret = 0;
|
|
qemu_timeval tv;
|
|
struct timespec ts;
|
|
|
|
qemu_gettimeofday(&tv);
|
|
ts.tv_sec = tv.tv_sec + 10;
|
|
ts.tv_nsec = 0;
|
|
|
|
mutex_lock(&lock);
|
|
|
|
while (QTAILQ_EMPTY(&request_list) &&
|
|
!(ret == ETIMEDOUT)) {
|
|
idle_threads++;
|
|
ret = cond_timedwait(&cond, &lock, &ts);
|
|
idle_threads--;
|
|
}
|
|
|
|
if (QTAILQ_EMPTY(&request_list))
|
|
break;
|
|
|
|
aiocb = QTAILQ_FIRST(&request_list);
|
|
QTAILQ_REMOVE(&request_list, aiocb, node);
|
|
aiocb->active = 1;
|
|
mutex_unlock(&lock);
|
|
|
|
switch (aiocb->aio_type & QEMU_AIO_TYPE_MASK) {
|
|
case QEMU_AIO_READ:
|
|
case QEMU_AIO_WRITE:
|
|
ret = handle_aiocb_rw(aiocb);
|
|
break;
|
|
case QEMU_AIO_FLUSH:
|
|
ret = handle_aiocb_flush(aiocb);
|
|
break;
|
|
case QEMU_AIO_IOCTL:
|
|
ret = handle_aiocb_ioctl(aiocb);
|
|
break;
|
|
default:
|
|
fprintf(stderr, "invalid aio request (0x%x)\n", aiocb->aio_type);
|
|
ret = -EINVAL;
|
|
break;
|
|
}
|
|
|
|
mutex_lock(&lock);
|
|
aiocb->ret = ret;
|
|
mutex_unlock(&lock);
|
|
|
|
if (kill(pid, aiocb->ev_signo)) die("kill failed");
|
|
}
|
|
|
|
cur_threads--;
|
|
mutex_unlock(&lock);
|
|
|
|
return NULL;
|
|
}
|
|
|
|
static void spawn_thread(void)
|
|
{
|
|
sigset_t set, oldset;
|
|
|
|
cur_threads++;
|
|
|
|
/* block all signals */
|
|
if (sigfillset(&set)) die("sigfillset");
|
|
if (sigprocmask(SIG_SETMASK, &set, &oldset)) die("sigprocmask");
|
|
|
|
thread_create(&thread_id, &attr, aio_thread, NULL);
|
|
|
|
if (sigprocmask(SIG_SETMASK, &oldset, NULL)) die("sigprocmask restore");
|
|
}
|
|
|
|
static void qemu_paio_submit(struct qemu_paiocb *aiocb)
|
|
{
|
|
aiocb->ret = -EINPROGRESS;
|
|
aiocb->active = 0;
|
|
mutex_lock(&lock);
|
|
if (idle_threads == 0 && cur_threads < max_threads)
|
|
spawn_thread();
|
|
QTAILQ_INSERT_TAIL(&request_list, aiocb, node);
|
|
mutex_unlock(&lock);
|
|
cond_signal(&cond);
|
|
}
|
|
|
|
static ssize_t qemu_paio_return(struct qemu_paiocb *aiocb)
|
|
{
|
|
ssize_t ret;
|
|
|
|
mutex_lock(&lock);
|
|
ret = aiocb->ret;
|
|
mutex_unlock(&lock);
|
|
|
|
return ret;
|
|
}
|
|
|
|
static int qemu_paio_error(struct qemu_paiocb *aiocb)
|
|
{
|
|
ssize_t ret = qemu_paio_return(aiocb);
|
|
|
|
if (ret < 0)
|
|
ret = -ret;
|
|
else
|
|
ret = 0;
|
|
|
|
return ret;
|
|
}
|
|
|
|
static int posix_aio_process_queue(void *opaque)
|
|
{
|
|
PosixAioState *s = opaque;
|
|
struct qemu_paiocb *acb, **pacb;
|
|
int ret;
|
|
int result = 0;
|
|
int async_context_id = get_async_context_id();
|
|
|
|
for(;;) {
|
|
pacb = &s->first_aio;
|
|
for(;;) {
|
|
acb = *pacb;
|
|
if (!acb)
|
|
return result;
|
|
|
|
/* we're only interested in requests in the right context */
|
|
if (acb->async_context_id != async_context_id) {
|
|
pacb = &acb->next;
|
|
continue;
|
|
}
|
|
|
|
ret = qemu_paio_error(acb);
|
|
if (ret == ECANCELED) {
|
|
/* remove the request */
|
|
*pacb = acb->next;
|
|
qemu_aio_release(acb);
|
|
result = 1;
|
|
} else if (ret != EINPROGRESS) {
|
|
/* end of aio */
|
|
if (ret == 0) {
|
|
ret = qemu_paio_return(acb);
|
|
if (ret == acb->aio_nbytes)
|
|
ret = 0;
|
|
else
|
|
ret = -EINVAL;
|
|
} else {
|
|
ret = -ret;
|
|
}
|
|
|
|
trace_paio_complete(acb, acb->common.opaque, ret);
|
|
|
|
/* remove the request */
|
|
*pacb = acb->next;
|
|
/* call the callback */
|
|
acb->common.cb(acb->common.opaque, ret);
|
|
qemu_aio_release(acb);
|
|
result = 1;
|
|
break;
|
|
} else {
|
|
pacb = &acb->next;
|
|
}
|
|
}
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
static void posix_aio_read(void *opaque)
|
|
{
|
|
PosixAioState *s = opaque;
|
|
ssize_t len;
|
|
|
|
/* read all bytes from signal pipe */
|
|
for (;;) {
|
|
char bytes[16];
|
|
|
|
len = read(s->rfd, bytes, sizeof(bytes));
|
|
if (len == -1 && errno == EINTR)
|
|
continue; /* try again */
|
|
if (len == sizeof(bytes))
|
|
continue; /* more to read */
|
|
break;
|
|
}
|
|
|
|
posix_aio_process_queue(s);
|
|
}
|
|
|
|
static int posix_aio_flush(void *opaque)
|
|
{
|
|
PosixAioState *s = opaque;
|
|
return !!s->first_aio;
|
|
}
|
|
|
|
static PosixAioState *posix_aio_state;
|
|
|
|
static void aio_signal_handler(int signum)
|
|
{
|
|
if (posix_aio_state) {
|
|
char byte = 0;
|
|
ssize_t ret;
|
|
|
|
ret = write(posix_aio_state->wfd, &byte, sizeof(byte));
|
|
if (ret < 0 && errno != EAGAIN)
|
|
die("write()");
|
|
}
|
|
|
|
qemu_service_io();
|
|
}
|
|
|
|
static void paio_remove(struct qemu_paiocb *acb)
|
|
{
|
|
struct qemu_paiocb **pacb;
|
|
|
|
/* remove the callback from the queue */
|
|
pacb = &posix_aio_state->first_aio;
|
|
for(;;) {
|
|
if (*pacb == NULL) {
|
|
fprintf(stderr, "paio_remove: aio request not found!\n");
|
|
break;
|
|
} else if (*pacb == acb) {
|
|
*pacb = acb->next;
|
|
qemu_aio_release(acb);
|
|
break;
|
|
}
|
|
pacb = &(*pacb)->next;
|
|
}
|
|
}
|
|
|
|
static void paio_cancel(BlockDriverAIOCB *blockacb)
|
|
{
|
|
struct qemu_paiocb *acb = (struct qemu_paiocb *)blockacb;
|
|
int active = 0;
|
|
|
|
trace_paio_cancel(acb, acb->common.opaque);
|
|
|
|
mutex_lock(&lock);
|
|
if (!acb->active) {
|
|
QTAILQ_REMOVE(&request_list, acb, node);
|
|
acb->ret = -ECANCELED;
|
|
} else if (acb->ret == -EINPROGRESS) {
|
|
active = 1;
|
|
}
|
|
mutex_unlock(&lock);
|
|
|
|
if (active) {
|
|
/* fail safe: if the aio could not be canceled, we wait for
|
|
it */
|
|
while (qemu_paio_error(acb) == EINPROGRESS)
|
|
;
|
|
}
|
|
|
|
paio_remove(acb);
|
|
}
|
|
|
|
static AIOPool raw_aio_pool = {
|
|
.aiocb_size = sizeof(struct qemu_paiocb),
|
|
.cancel = paio_cancel,
|
|
};
|
|
|
|
BlockDriverAIOCB *paio_submit(BlockDriverState *bs, int fd,
|
|
int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
|
|
BlockDriverCompletionFunc *cb, void *opaque, int type)
|
|
{
|
|
struct qemu_paiocb *acb;
|
|
|
|
acb = qemu_aio_get(&raw_aio_pool, bs, cb, opaque);
|
|
if (!acb)
|
|
return NULL;
|
|
acb->aio_type = type;
|
|
acb->aio_fildes = fd;
|
|
acb->ev_signo = SIGUSR2;
|
|
acb->async_context_id = get_async_context_id();
|
|
|
|
if (qiov) {
|
|
acb->aio_iov = qiov->iov;
|
|
acb->aio_niov = qiov->niov;
|
|
}
|
|
acb->aio_nbytes = nb_sectors * 512;
|
|
acb->aio_offset = sector_num * 512;
|
|
|
|
acb->next = posix_aio_state->first_aio;
|
|
posix_aio_state->first_aio = acb;
|
|
|
|
trace_paio_submit(acb, opaque, sector_num, nb_sectors, type);
|
|
qemu_paio_submit(acb);
|
|
return &acb->common;
|
|
}
|
|
|
|
BlockDriverAIOCB *paio_ioctl(BlockDriverState *bs, int fd,
|
|
unsigned long int req, void *buf,
|
|
BlockDriverCompletionFunc *cb, void *opaque)
|
|
{
|
|
struct qemu_paiocb *acb;
|
|
|
|
acb = qemu_aio_get(&raw_aio_pool, bs, cb, opaque);
|
|
if (!acb)
|
|
return NULL;
|
|
acb->aio_type = QEMU_AIO_IOCTL;
|
|
acb->aio_fildes = fd;
|
|
acb->ev_signo = SIGUSR2;
|
|
acb->async_context_id = get_async_context_id();
|
|
acb->aio_offset = 0;
|
|
acb->aio_ioctl_buf = buf;
|
|
acb->aio_ioctl_cmd = req;
|
|
|
|
acb->next = posix_aio_state->first_aio;
|
|
posix_aio_state->first_aio = acb;
|
|
|
|
qemu_paio_submit(acb);
|
|
return &acb->common;
|
|
}
|
|
|
|
int paio_init(void)
|
|
{
|
|
struct sigaction act;
|
|
PosixAioState *s;
|
|
int fds[2];
|
|
int ret;
|
|
|
|
if (posix_aio_state)
|
|
return 0;
|
|
|
|
s = qemu_malloc(sizeof(PosixAioState));
|
|
|
|
sigfillset(&act.sa_mask);
|
|
act.sa_flags = 0; /* do not restart syscalls to interrupt select() */
|
|
act.sa_handler = aio_signal_handler;
|
|
sigaction(SIGUSR2, &act, NULL);
|
|
|
|
s->first_aio = NULL;
|
|
if (qemu_pipe(fds) == -1) {
|
|
fprintf(stderr, "failed to create pipe\n");
|
|
return -1;
|
|
}
|
|
|
|
s->rfd = fds[0];
|
|
s->wfd = fds[1];
|
|
|
|
fcntl(s->rfd, F_SETFL, O_NONBLOCK);
|
|
fcntl(s->wfd, F_SETFL, O_NONBLOCK);
|
|
|
|
qemu_aio_set_fd_handler(s->rfd, posix_aio_read, NULL, posix_aio_flush,
|
|
posix_aio_process_queue, s);
|
|
|
|
ret = pthread_attr_init(&attr);
|
|
if (ret)
|
|
die2(ret, "pthread_attr_init");
|
|
|
|
ret = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
|
|
if (ret)
|
|
die2(ret, "pthread_attr_setdetachstate");
|
|
|
|
QTAILQ_INIT(&request_list);
|
|
|
|
posix_aio_state = s;
|
|
return 0;
|
|
}
|