tools/diskomizer/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
 */

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

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

#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <sys/mman.h>
#include <signal.h>
#include <pthread.h>
#include <diskomizer/log.h>

#include "async.h"
#include "signal_catch.h"

/*
 * General routines for providing aync operations.
 *
 * Async operations get passed in as async_work structures. A job
 * object is found or created that contains a thread to run the job and that
 * thread then runs the job and puts the result on the done_queue from
 * whence it can be reaped by async_queue_wait(); meanwhile the job
 * puts itself on the queue of free jobs so that it can run the next
 * job.
 */
/*
 * Still to do:
 * 	cancellation is not implemented.
 * 	unemployed jobs should exit after a while
 */

/*
 * madvise flags. Not defined on 5.6 and 5.7 which so have to be added
 * here.  On older releases the use of the flags should just give an error
 */
#ifndef MADV_FREE
#define	MADV_FREE 0x5
#endif

#ifndef MADV_ACCESS_LWP
#define	MADV_ACCESS_LWP 0x7
#endif

#define	roundup(x, y) ((((x)+((y)-1))/(y))*(y))
#define	PTHREAD_STACK_MIN sysconf(_SC_THREAD_STACK_MIN)

enum state {
	READY_TO_RUN,
	COMPLETED
};
struct jobs {
	pthread_mutex_t mlock;
	pthread_cond_t cv;
	pthread_t thread;
	enum state state;
	struct jobs *next;
	struct async_work *work;
};

struct queue {
	pthread_mutex_t mlock;
	pthread_cond_t cv;
	struct jobs *head;
	struct jobs *tail;
};
struct work_done {
	pthread_mutex_t mlock;
	pthread_cond_t cv;
	struct async_work *head;
	struct async_work *tail;
};

struct thr_info {
	pthread_mutex_t mlock;
	int thr_free;
	int thr_running;
};
static struct queue unemployed;
static struct work_done work_done;

/*
 * Job queues.
 *
 * There are three routines for managing the job queues, 2 for adding and
 * one for taking off the queue.
 *
 * Jobs are always removed from the head of the queue.
 *
 * For a FIFO queue jobs are added to the tail of the queue by add_to_job_fifo
 * and for LIFO they are added to the head of the queue by add_to_job_lifo.
 */
#ifdef FIFO_NEEDED
static void
add_to_job_fifo(struct queue *q, struct jobs *job)
{
	pthread_mutex_lock(&q->mlock);
	if (q->head == NULL) {
		q->head = q->tail = job;
	} else {
		q->tail->next = job;
		q->tail = job;
	}
	job->next = NULL;
	pthread_mutex_unlock(&q->mlock);
}
#endif
static void
add_to_job_lifo(struct queue *q, struct jobs *job)
{
	pthread_mutex_lock(&q->mlock);
	if (q->head == NULL) {
		q->head = q->tail = job;
		job->next = NULL;
	} else {
		job->next = q->head;
		q->head = job;
	}
	pthread_mutex_unlock(&q->mlock);
}

static struct jobs *
get_job(struct queue *q)
{
	struct jobs *job;
	pthread_mutex_lock(&q->mlock);
	job = q->head;
	if (job != NULL) {
		q->head = job->next;
		if (job == q->tail)
			q->tail = NULL;
		job->next = NULL;
	}
	pthread_mutex_unlock(&q->mlock);
	return (job);
}
static void
add_to_done(struct async_work *work)
{
	pthread_mutex_lock(&work_done.mlock);
	if (work_done.head == NULL) {
		work_done.head = work_done.tail = work;
		pthread_cond_signal(&work_done.cv);
	} else {
		work_done.tail->u.next = work;
		work_done.tail = work;
	}
	work->flags |= ASYNCWORK_COMPLETE;
	work->u.next = NULL;
	pthread_mutex_unlock(&work_done.mlock);
}
static void
do_work(struct async_work *work)
{
	work->result = work->u.funcs.func(work->arg);
	add_to_done(work);
}
static void *
worker(void *arg)
{
	struct jobs *job = arg;
#ifdef __lint
	int err = 0;
#endif

	for (;;) {
		while (pthread_mutex_lock(&job->mlock) != 0)
#ifdef __lint
			err++;
#else
			;
#endif
		while (job->state != READY_TO_RUN) {
			pthread_cond_wait(&job->cv, &job->mlock);
		}
		do_work(job->work);
		job->work = NULL;
		job->state = COMPLETED;
		pthread_mutex_unlock(&job->mlock);
		add_to_job_lifo(&unemployed, job);
	}
	/*NOTREACHED*/
}

void *
create_stack(size_t len)
{
	int fd;
	int redzone_len = getpagesize();
	char *addr;

	if ((fd = open("/dev/zero", O_RDWR)) == -1)
		return (NULL);

	addr = mmap(NULL, len + redzone_len, PROT_READ|PROT_WRITE,
	    MAP_PRIVATE, fd, 0);
	close(fd);
	if (addr == MAP_FAILED)
		return (NULL);

	madvise(addr, redzone_len, MADV_DONTNEED);
	madvise(addr, redzone_len, MADV_FREE);
	mprotect(addr, redzone_len, PROT_NONE);

	madvise(addr + redzone_len, len, MADV_ACCESS_LWP);

	expect_signal(SIGSEGV, "Thread Buffer Overflow\n", addr, redzone_len);

	plog(LOG_DEBUG, "Stack %p\n", addr + redzone_len);
	return ((void *)(addr + redzone_len));
}

/*
 * create_job.  Creates a new job structure and its thread.
 */
static struct jobs *
create_job(void)
{
	void *stack;
	struct jobs *job;
	pthread_attr_t attr;
	int err;
	size_t stack_size = roundup(getpagesize() + PTHREAD_STACK_MIN,
	    getpagesize());

	if ((job = malloc(sizeof (struct jobs))) == NULL)
		return (NULL);

	if ((stack = create_stack(stack_size)) == NULL) {
		free(job);
		return (NULL);
	}
	pthread_mutex_init(&job->mlock, NULL);
	pthread_cond_init(&job->cv, NULL);
	job->next = NULL;
	job->state = COMPLETED;
	pthread_attr_init(&attr);
	pthread_attr_setscope(&attr, PTHREAD_SCOPE_SYSTEM);
	pthread_attr_setstacksize(&attr, stack_size);
	pthread_attr_setstackaddr(&attr, stack);
	err = pthread_create(&job->thread, &attr, worker, job);
	if (err != 0) {
		pthread_cond_destroy(&job->cv);
		pthread_mutex_destroy(&job->mlock);
		free(job);
		errno = err;
		return (NULL);
	}
	return (job);
}

static struct jobs *
get_free_job(void)
{
	struct jobs *job;

	if ((job = get_job(&unemployed)) == NULL) {
		job = create_job();
	}
	return (job);
}

void *
async_queue_request(struct async_work *work)
{
	struct jobs *job;

	if ((job = get_free_job()) == NULL)
		return (NULL);
	pthread_mutex_lock(&job->mlock);
	work->flags &= (~ASYNCWORK_COMPLETE);
	job->work = work;
	job->state = READY_TO_RUN;
	pthread_cond_signal(&job->cv);
	pthread_mutex_unlock(&job->mlock);
	return (job);
}
/*ARGSUSED*/
struct async_work *
async_queue_wait(void *handle, const struct timespec *abstime, int flags)
{
	struct async_work *work;
	pthread_mutex_lock(&work_done.mlock);
	while (work_done.head == NULL) {
		if (abstime == NULL) {
			pthread_cond_wait(&work_done.cv, &work_done.mlock);
		} else {
			int x;
			x = pthread_cond_timedwait(&work_done.cv,
			    &work_done.mlock, abstime);
			if (x == ETIMEDOUT) {
				pthread_mutex_unlock(&work_done.mlock);
				return (NULL);
			}
		}
	}
	work = work_done.head;
	work_done.head = work->u.next;
	if (work_done.tail == work) {
		work_done.tail = NULL;
	}
	work->u.next = NULL;
	pthread_mutex_unlock(&work_done.mlock);
	return (work);
}

void
find_work(struct async_work *work)
{
	struct async_work *w;

	for (w = work_done.head; w != NULL; w = w->u.next) {
		if (w == work)
			abort();
	}
}
int
async_cancel_request(struct async_work *work, int flags)
{
	int ret;
	pthread_mutex_lock(&work_done.mlock);
	if (work->flags & ASYNCWORK_COMPLETE) {
		ret = -1;
	} else {
		if (work->u.funcs.cancel) {
			ret = work->u.funcs.cancel(work->arg, flags);
		} else {
			find_work(work);
			abort();
		}
	}
	pthread_mutex_unlock(&work_done.mlock);
	return (ret);
}