tools/diskomizer/daio_async.c

/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License (the "License").
 * You may not use this file except in compliance with the License.
 *
 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
 * or http://www.opensolaris.org/os/licensing.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
 * If applicable, add the following below this CDDL HEADER, with the
 * fields enclosed by brackets "[]" replaced with your own identifying
 * information: Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 */

#pragma ident	"@(#)daio_async.c	1.8	09/05/26 SMI"

/*
 * Copyright 2009 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 */

#include <diskomizer/daio.h>
#include "async.h"
#include "list.h"
/*LINTED*/
#include "daio_async.h"
#include <pthread.h>
#include <stdlib.h>
#include <errno.h>
#include <sys/time.h>
#include <unistd.h>

struct daio_async {
	struct async_work work;
	/* lock protects the timestamps and status */
	pthread_mutex_t mlock;
	daio_status_t status;
	int fildes;
	int bufs;
	int is_read:1;
	off_t offset;
	uchar_t *bufp;
	struct daio_id *id;
	ssize_t (*rw)(int fildes, void  *buf,  size_t  nbyte,  off_t offset);
	daio_result_t *resultp;
	daio_result_t res;
	struct daio_async *next;
	struct daio_async *prev;
};
struct daio_async_list {
	struct daio_async *head;
	struct daio_async *tail;
};

static struct daio_async_list free_list;
static struct daio_async_list work_list;
static int saved_count;
static struct daio_async *base;

static int (*data_check)(uchar_t *buf, size_t bufs, struct daio_id *id);

static int
bounds_check(struct daio_async *x)
{
	int off = base - x;
	if (off < 0 || off >= saved_count || &base[off] != x) {
		return (0);
	} else {
		return (1);
	}
}

static struct timespec
timespec_plus_timeval(const struct timespec *ts, const struct timeval *tv)
{
	struct timespec res;

	res.tv_nsec = ts->tv_nsec + (tv->tv_usec * 1000);
	if (res.tv_nsec >= 1000000000) {
		res.tv_sec = 1 + ts->tv_sec + tv->tv_sec;
		res.tv_nsec %= 1000000000;
	} else {
		res.tv_sec = ts->tv_sec + tv->tv_sec;
	}
	return (res);
}
void
free_daio_async(struct daio_async *x)
{
	LIST_REMOVE(&work_list, x);
	LIST_ADD(&free_list, x);
}
void
daio_async_fini(void *y)
{
	/* To be general purpose this should wait for the work_list to drain */
	int i;
	struct daio_async *x = (struct daio_async *)y;

	for (i = 0; i < saved_count; i++) {
		pthread_mutex_destroy(&x[i].mlock);
	}
	free(x);
}

void *
daio_async_init(int count)
{
	int i;

	if ((base = calloc(count, sizeof (struct daio_async))) == NULL) {
		return (NULL);
	}
	saved_count = count;

	for (i = 0; i < count; i++) {
		pthread_mutex_init(&base[i].mlock, NULL);
		LIST_ADD(&free_list, &base[i]);
	}
	return (base);
}

/*ARGSUSED*/
void
daio_async_init_checker(const char *checker, off64_t max_block_size)
{
	data_check = choose_data_checker(checker);
}

data_checker_t
daio_async_get_checker(void)
{
	return (data_check);
}

static struct daio_async *
get_daio_async(void)
{
	struct daio_async *x;

	x = LIST_HEAD(&free_list);
	if (x == NULL) {
		errno = ENOMEM;
		return (NULL);
	}
	LIST_REMOVE(&free_list, x);
	LIST_ADD(&work_list, x);
	return (x);
}
hrtime_t
daio_async_end_time(daio_result_t *y)
{
	struct daio_async *x = (struct daio_async *)y->private_data;
	hrtime_t res;

	if (!bounds_check(x)) {
		return (0);
	}

	pthread_mutex_lock(&x->mlock);
	res = DAIO_GET_END_TIME(x->res);
	pthread_mutex_unlock(&x->mlock);
	return (res);
}
daio_status_t
daio_async_status(daio_result_t *y)
{
	struct daio_async *x = (struct daio_async *)y->private_data;
	daio_status_t res;

	if (!bounds_check(x)) {
		return (DAIO_ERROR);
	}

	pthread_mutex_lock(&x->mlock);
	res = x->status;
	pthread_mutex_unlock(&x->mlock);
	return (res);
}
hrtime_t
daio_async_start_time(daio_result_t *y)
{
	struct daio_async *x = (struct daio_async *)y->private_data;
	hrtime_t res;

	if (!bounds_check(x)) {
		return (0);
	}

	pthread_mutex_lock(&x->mlock);
	res = DAIO_GET_START_TIME(x->res);
	pthread_mutex_unlock(&x->mlock);
	return (res);
}
static void *
do_daio_async_rw(void *y)
{
	struct daio_async *x = (struct daio_async *)y;
	hrtime_t et;

	pthread_mutex_lock(&x->mlock);
	x->status = DAIO_INPROGRESS;
	DAIO_SET_START_TIME(x->res, gethrtime());
	pthread_mutex_unlock(&x->mlock);
	DAIO_SET_RETURN(x->res, x->rw(x->fildes, x->bufp, x->bufs, x->offset));
	DAIO_SET_ERROR(x->res, errno);
	et =  gethrtime();
	if (DAIO_RETURN(x->res) < 0) {
		pthread_mutex_lock(&x->mlock);
		x->status = DAIO_COMPLETE_ERROR;
	} else {
		if (x->is_read &&
		    data_check(x->bufp, DAIO_RETURN(x->res), x->id) != 0) {
			DAIO_SET_RETURN(x->res, DAIO_CORRUPT);
		}
		pthread_mutex_lock(&x->mlock);
		x->status = DAIO_COMPLETE_OK;
	}
	DAIO_SET_END_TIME(x->res, et);
	pthread_mutex_unlock(&x->mlock);
	return (x);
}

int
daio_async_cancel(daio_result_t *y)
{
	struct daio_async *x = (struct daio_async *)y->private_data;

	if (bounds_check(x)) {
		return (async_cancel_request(&x->work, 0));
	} else {
		errno = EINVAL;
		return (-1);
	}
}

static int
/*ARGSUSED*/
do_daio_async_cancel(void *x, int flags)
{
	errno = ENOTSUP;
	return (-1);
}
ssize_t
daio_async_pread(int fildes, uchar_t *bufp, size_t bufs, off_t off,
		struct daio_id *id)
{
	ssize_t ret;

	ret = pread(fildes, bufp, bufs, off);

	if (id != NULL && ret >= 0) {
		if (data_check(bufp, ret, id) != 0)
			ret = DAIO_CORRUPT;
	}
	return (ret);
}
/*ARGSUSED4*/
ssize_t
daio_async_pwrite(int fildes, const uchar_t *bufp, size_t bufs, off_t off,
		struct daio_id *id)
{
	return (pwrite(fildes, bufp, bufs, off));
}
int
daio_async_rw(int fildes, uchar_t *bufp, int  bufs, off_t
	offset, daio_result_t *resultp,  struct daio_id *id,
	ssize_t (*rw)(int fildes, void  *buf,  size_t  nbyte,  off_t offset),
	read_or_write_t is_read)
{
	struct daio_async *x;

	if ((x = get_daio_async()) == NULL) {
		errno = ENOMEM;
		return (-1);
	}

	x->work.u.funcs.func = (void * (*)(void *))do_daio_async_rw;
	x->work.u.funcs.cancel = (int (*)(void *, int))do_daio_async_cancel;
	x->work.arg = x;
	x->work.flags = 0;
	x->fildes = fildes;
	x->bufs = bufs;
	x->offset = offset;
	x->bufp = bufp;
	x->id = id;
	x->rw = rw;
	x->is_read = is_read == IS_READ ? 1 : 0;
	x->resultp = resultp;
	DAIO_SET_START_TIME(x->res, DAIO_NOT_STARTED);
	resultp->private_data = (void *)x;

	return (async_queue_request(&x->work) == NULL ? -1 : 0);

}
daio_result_t *
daio_async_wait(const struct timeval *timeout)
{
	struct async_work *work;
	struct daio_async *x;
	daio_result_t *res;

	if (LIST_HEAD(&work_list) == NULL) {
		errno = EINVAL;
		return ((daio_result_t *)-1);
	}

	if (timeout == NULL) {
		work = async_queue_wait(NULL, NULL, 0);
	} else {
		struct timespec abstime;
		clock_gettime(CLOCK_REALTIME, &abstime);

		abstime = timespec_plus_timeval(&abstime, timeout);
		work = async_queue_wait(NULL, &abstime, 0);
	}
	if (work == NULL)
		return ((daio_result_t *)NULL);

	x = work->arg;

	*x->resultp = x->res;
	res = x->resultp;

	free_daio_async(x);

	return (res);
}