fb_localfs.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.
 *
 * Portions Copyright 2008 Denis Cheng
 */

#include "filebench.h"
#include "flowop.h"
#include "threadflow.h" /* For aiolist definition */

#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <libgen.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/param.h>
#include <sys/resource.h>
#include <strings.h>

#include "filebench.h"
#include "fsplug.h"

#ifdef HAVE_AIO
#include <aio.h>
#endif /* HAVE_AIO */

/*
 * These routines implement local file access. They are placed into a
 * vector of functions that are called by all I/O operations in fileset.c
 * and flowop_library.c. This represents the default file system plug-in,
 * and may be replaced by vectors for other file system plug-ins.
 */

static void fb_lfs_init_master(void);
static int fb_lfs_freemem(fb_fdesc_t *fd, off64_t size);
static int fb_lfs_open(fb_fdesc_t *, char *, int, int);
static int fb_lfs_pread(fb_fdesc_t *, caddr_t, fbint_t, off64_t);
static int fb_lfs_read(fb_fdesc_t *, caddr_t, fbint_t);
static int fb_lfs_pwrite(fb_fdesc_t *, caddr_t, fbint_t, off64_t);
static int fb_lfs_write(fb_fdesc_t *, caddr_t, fbint_t);
static int fb_lfs_lseek(fb_fdesc_t *, off64_t, int);
static int fb_lfs_truncate(fb_fdesc_t *, off64_t);
static int fb_lfs_rename(const char *, const char *);
static int fb_lfs_close(fb_fdesc_t *);
static int fb_lfs_link(const char *, const char *);
static int fb_lfs_symlink(const char *, const char *);
static int fb_lfs_unlink(char *);
static ssize_t fb_lfs_readlink(const char *, char *, size_t);
static int fb_lfs_mkdir(char *, int);
static int fb_lfs_rmdir(char *);
static struct fsplug_dir *fb_lfs_opendir(char *);
static int fb_lfs_readdir(struct fsplug_dir *, struct fsplug_dirent *);
static int fb_lfs_closedir(struct fsplug_dir *);
static int fb_lfs_fsync(fb_fdesc_t *);
static int fb_lfs_stat(char *, struct stat64 *);
static int fb_lfs_fstat(fb_fdesc_t *, struct stat64 *);
static int fb_lfs_access(const char *, int);
static void fb_lfs_recur_rm(char *);

/*
 * This is named as would be a generic fsplug, even though we are going to
 * statically link this plugin into the executable.  This allows us to test
 * the filesystem and loader as different components.  The name
 * "fsplug_funcs" is never used directly, despite being exported here.
 */
fsplug_func_t fsplug_funcs =
{
	"locfs",

	fb_lfs_init_master,
	NULL,				/* Needs no per-worker initialization */

	fb_lfs_freemem,		/* flush page cache */
	fb_lfs_open,		/* open */
	fb_lfs_pread,		/* pread */
	fb_lfs_read,		/* read */
	fb_lfs_pwrite,		/* pwrite */
	fb_lfs_write,		/* write */
	fb_lfs_lseek,		/* lseek */
	fb_lfs_truncate,	/* ftruncate */
	fb_lfs_rename,		/* rename */
	fb_lfs_close,		/* close */
	fb_lfs_link,		/* link */
	fb_lfs_symlink,		/* symlink */
	fb_lfs_unlink,		/* unlink */
	fb_lfs_readlink,	/* readlink */
	fb_lfs_mkdir,		/* mkdir */
	fb_lfs_rmdir,		/* rmdir */
	fb_lfs_opendir,		/* opendir */
	fb_lfs_readdir,		/* readdir */
	fb_lfs_closedir,	/* closedir */
	fb_lfs_fsync,		/* fsync */
	fb_lfs_stat,		/* stat */
	fb_lfs_fstat,		/* fstat */
	fb_lfs_access,		/* access */
	fb_lfs_recur_rm		/* recursive rm */
};

/* Set the default fs_functions_vec to us */
fsplug_func_t *fs_functions_vec = &fsplug_funcs;

#ifdef HAVE_AIO
/*
 * Local file system asynchronous IO flowops are in this module, as
 * they have a number of local file system specific features.
 */
static int fb_lfsflow_aiowrite(threadflow_t *threadflow, flowop_t *flowop);
static int fb_lfsflow_aiowait(threadflow_t *threadflow, flowop_t *flowop);

static flowop_proto_t fb_lfsflow_funcs[] = {
	{FLOW_TYPE_AIO, FLOW_ATTR_WRITE, "aiowrite", flowop_init_generic,
	fb_lfsflow_aiowrite, flowop_destruct_generic},
	{FLOW_TYPE_AIO, 0, "aiowait", flowop_init_generic,
	fb_lfsflow_aiowait, flowop_destruct_generic}
};

#endif /* HAVE_AIO */

/*
 * Initialize those flowops which implementation is file system specific. It is
 * called only once in the master process.
 */
static void
fb_lfs_init_master(void)
{
#ifdef HAVE_AIO
	int nops;
	nops = sizeof (fb_lfsflow_funcs) / sizeof (flowop_proto_t);
	flowop_add_from_proto(fb_lfsflow_funcs, nops);
#endif /* HAVE_AIO */
}

/*
 * Frees up memory mapped file region of supplied size. The
 * file descriptor "fd" indicates which memory mapped file.
 * If successful, returns 0. Otherwise returns -1 if "size"
 * is zero, or -1 times the number of times msync() failed.
 */
#ifdef HAVE_MMAP64
static int
fb_lfs_freemem(fb_fdesc_t *fd, off64_t size)
{
	off64_t left;
	int ret = 0;

	for (left = size; left > 0; left -= MMAP_SIZE) {
		off64_t thismapsize;
		caddr_t addr;

		thismapsize = MIN(MMAP_SIZE, left);
		addr = mmap64(0, thismapsize, PROT_READ|PROT_WRITE,
		    MAP_SHARED, fd->fd_num, size - left);
		ret += msync(addr, thismapsize, MS_INVALIDATE);
		(void) munmap(addr, thismapsize);
	}
	return (ret);
}
#else
static int
fb_lfs_freemem(fb_fdesc_t *fd, off_t size)
{
	off64_t left;
	int ret = 0;

	for (left = size; left > 0; left -= MMAP_SIZE) {
		off64_t thismapsize;
		caddr_t addr;

		thismapsize = MIN(MMAP_SIZE, left);
		addr = mmap(0, thismapsize, PROT_READ|PROT_WRITE,
		    MAP_SHARED, fd->fd_num, size - left);
		ret += msync(addr, thismapsize, MS_INVALIDATE);
		(void) munmap(addr, thismapsize);
	}
	return (ret);
}
#endif

/*
 * Does a posix pread. Returns what the pread() returns.
 */
#ifdef HAVE_PREAD64
static int
fb_lfs_pread(fb_fdesc_t *fd, caddr_t iobuf, fbint_t iosize, off64_t fileoffset)
{
	return (pread64(fd->fd_num, iobuf, iosize, fileoffset));
}
#else
static int
fb_lfs_pread(fb_fdesc_t *fd, caddr_t iobuf, fbint_t iosize, off_t fileoffset)
{
	return (pread(fd->fd_num, iobuf, iosize, fileoffset));
}
#endif

/*
 * Does a posix read. Returns what the read() returns.
 */
static int
fb_lfs_read(fb_fdesc_t *fd, caddr_t iobuf, fbint_t iosize)
{
	return (read(fd->fd_num, iobuf, iosize));
}

#ifdef HAVE_AIO

/*
 * Asynchronous write section. An Asynchronous IO element
 * (aiolist_t) is used to associate the asynchronous write request with
 * its subsequent completion. This element includes a aiocb64 struct
 * that is used by posix aio_xxx calls to track the asynchronous writes.
 * The flowops aiowrite and aiowait result in calls to these posix
 * aio_xxx system routines to do the actual asynchronous write IO
 * operations.
 */


/*
 * Allocates an asynchronous I/O list (aio, of type
 * aiolist_t) element. Adds it to the flowop thread's
 * threadflow aio list. Returns a pointer to the element.
 */
static aiolist_t *
aio_allocate(flowop_t *flowop)
{
	aiolist_t *aiolist;

	if ((aiolist = malloc(sizeof (aiolist_t))) == NULL) {
		filebench_log(LOG_ERROR, "malloc aiolist failed");
		filebench_shutdown(1);
	}

	bzero(aiolist, sizeof(*aiolist));

	/* Add to list */
	if (flowop->fo_thread->tf_aiolist == NULL) {
		flowop->fo_thread->tf_aiolist = aiolist;
		aiolist->al_next = NULL;
	} else {
		aiolist->al_next = flowop->fo_thread->tf_aiolist;
		flowop->fo_thread->tf_aiolist = aiolist;
	}
	return (aiolist);
}

/*
 * Searches for the aiolist element that has a matching
 * completion block, aiocb. If none found returns FILEBENCH_ERROR. If
 * found, removes the aiolist element from flowop thread's
 * list and returns FILEBENCH_OK.
 */
static int
aio_deallocate(flowop_t *flowop, struct aiocb64 *aiocb)
{
	aiolist_t *aiolist = flowop->fo_thread->tf_aiolist;
	aiolist_t *previous = NULL;
	aiolist_t *match = NULL;

	if (aiocb == NULL) {
		filebench_log(LOG_ERROR, "null aiocb deallocate");
		return (FILEBENCH_OK);
	}

	while (aiolist) {
		if (aiocb == &(aiolist->al_aiocb)) {
			match = aiolist;
			break;
		}
		previous = aiolist;
		aiolist = aiolist->al_next;
	}

	if (match == NULL)
		return (FILEBENCH_ERROR);

	/* Remove from the list */
	if (previous)
		previous->al_next = match->al_next;
	else
		flowop->fo_thread->tf_aiolist = match->al_next;

	return (FILEBENCH_OK);
}

/*
 * Emulate posix aiowrite(). Determines which file to use,
 * either one file of a fileset, or the file associated
 * with a fileobj, allocates and fills an aiolist_t element
 * for the write, and issues the asynchronous write. This
 * operation is only valid for random IO, and returns an
 * error if the flowop is set for sequential IO. Returns
 * FILEBENCH_OK on success, FILEBENCH_NORSC if iosetup can't
 * obtain a file to open, and FILEBENCH_ERROR on any
 * encountered error.
 */
static int
fb_lfsflow_aiowrite(threadflow_t *threadflow, flowop_t *flowop)
{
	caddr_t iobuf;
	fbint_t wss;
	fbint_t iosize;
	fb_fdesc_t *fdesc;
	int ret;

	iosize = avd_get_int(flowop->fo_iosize);

	if ((ret = flowoplib_iosetup(threadflow, flowop, &wss, &iobuf,
	    &fdesc, iosize)) != FILEBENCH_OK)
		return (ret);

	if (avd_get_bool(flowop->fo_random)) {
		uint64_t fileoffset;
		struct aiocb64 *aiocb;
		aiolist_t *aiolist;

		if (wss < iosize) {
			filebench_log(LOG_ERROR,
			    "file size smaller than IO size for thread %s",
			    flowop->fo_name);
			return (FILEBENCH_ERROR);
		}

		fb_random64(&fileoffset, wss, iosize, NULL);

		aiolist = aio_allocate(flowop);
		aiolist->al_type = AL_WRITE;
		aiocb = &aiolist->al_aiocb;

		aiocb->aio_fildes = fdesc->fd_num;
		aiocb->aio_buf = iobuf;
		aiocb->aio_nbytes = (size_t)iosize;
		aiocb->aio_offset = (off64_t)fileoffset;
		aiocb->aio_reqprio = 0;

		filebench_log(LOG_DEBUG_IMPL,
		    "aio fd=%d, bytes=%llu, offset=%llu",
		    fdesc->fd_num, (u_longlong_t)iosize,
		    (u_longlong_t)fileoffset);

		flowop_beginop(threadflow, flowop);
		if (aio_write64(aiocb) < 0) {
			filebench_log(LOG_ERROR, "aiowrite failed: %s",
			    strerror(errno));
			filebench_shutdown(1);
		}
		flowop_endop(threadflow, flowop, iosize);
	} else {
		return (FILEBENCH_ERROR);
	}

	return (FILEBENCH_OK);
}



#define	MAXREAP 4096

/*
 * Emulate posix aiowait(). Waits for the completion of half the
 * outstanding asynchronous IOs, or a single IO, which ever is
 * larger. The routine will return after a sufficient number of
 * completed calls issued by any thread in the procflow have
 * completed, or a 1 second timout elapses. All completed
 * IO operations are deleted from the thread's aiolist.
 */
static int
fb_lfsflow_aiowait(threadflow_t *threadflow, flowop_t *flowop)
{
	struct aiocb64 **worklist;
	aiolist_t *aio = flowop->fo_thread->tf_aiolist;
	int uncompleted = 0;
#ifdef HAVE_AIOWAITN
	int i;
#endif

	worklist = calloc(MAXREAP, sizeof (struct aiocb64 *));

	/* Count the list of pending aios */
	while (aio) {
		uncompleted++;
		aio = aio->al_next;
	}

	do {
		uint_t ncompleted = 0;
		uint_t todo;
		int inprogress;
#ifdef HAVE_AIOWAITN
		struct timespec timeout;

		/* Wait for half of the outstanding requests */
		timeout.tv_sec = 1;
		timeout.tv_nsec = 0;
#endif

		if (uncompleted > MAXREAP)
			todo = MAXREAP;
		else
			todo = uncompleted / 2;

		if (todo == 0)
			todo = 1;

		flowop_beginop(threadflow, flowop);

#ifdef HAVE_AIOWAITN
		if (((aio_waitn64((struct aiocb64 **)worklist,
		    MAXREAP, &todo, &timeout)) == -1) &&
		    errno && (errno != ETIME)) {
			filebench_log(LOG_ERROR,
			    "aiowait failed: %s, outstanding = %d, "
			    "ncompleted = %d ",
			    strerror(errno), uncompleted, todo);
		}

		ncompleted = todo;
		/* Take the  completed I/Os from the list */
		inprogress = 0;
		for (i = 0; i < ncompleted; i++) {
			if ((aio_return64(worklist[i]) == -1) &&
			    (errno == EINPROGRESS)) {
				inprogress++;
				continue;
			}
			if (aio_deallocate(flowop, worklist[i])
			    == FILEBENCH_ERROR) {
				filebench_log(LOG_ERROR, "Could not remove "
				    "aio from list ");
				flowop_endop(threadflow, flowop, 0);
				return (FILEBENCH_ERROR);
			}
		}

		uncompleted -= ncompleted;
		uncompleted += inprogress;

#else

		for (ncompleted = 0, inprogress = 0,
		    aio = flowop->fo_thread->tf_aiolist;
		    ncompleted < todo && aio != NULL; aio = aio->al_next) {
			int result = aio_error64(&aio->al_aiocb);

			if (result == EINPROGRESS) {
				inprogress++;
				continue;
			}

			if ((aio_return64(&aio->al_aiocb) == -1) || result) {
				filebench_log(LOG_ERROR, "aio failed: %s",
				    strerror(result));
				continue;
			}

			ncompleted++;

			if (aio_deallocate(flowop, &aio->al_aiocb) < 0) {
				filebench_log(LOG_ERROR, "Could not remove "
				    "aio from list ");
				flowop_endop(threadflow, flowop, 0);
				return (FILEBENCH_ERROR);
			}
		}

		uncompleted -= ncompleted;

#endif
		filebench_log(LOG_DEBUG_SCRIPT,
		    "aio2 completed %d ios, uncompleted = %d, inprogress = %d",
		    ncompleted, uncompleted, inprogress);

	} while (uncompleted > MAXREAP);

	flowop_endop(threadflow, flowop, 0);

	free(worklist);

	return (FILEBENCH_OK);
}

#endif /* HAVE_AIO */

/*
 * Does an open64 of a file. Inserts the file descriptor number returned
 * by open() into the supplied filebench fd. Returns FILEBENCH_OK on
 * successs, and FILEBENCH_ERROR on failure.
 */

#ifdef HAVE_OPEN64
static int
fb_lfs_open(fb_fdesc_t *fd, char *path, int flags, int perms)
{
	if ((fd->fd_num = open64(path, flags, perms)) < 0)
		return (FILEBENCH_ERROR);
	else
		return (FILEBENCH_OK);
}
#else
static int
fb_lfs_open(fb_fdesc_t *fd, char *path, int flags, int perms)
{
	if ((fd->fd_num = open(path, flags, perms)) < 0)
		return (FILEBENCH_ERROR);
	else
		return (FILEBENCH_OK);
}
#endif

/*
 * Does an unlink (delete) of a file.
 */
static int
fb_lfs_unlink(char *path)
{
	return (unlink(path));
}

/*
 * Does a readlink of a symbolic link.
 */
static ssize_t
fb_lfs_readlink(const char *path, char *buf, size_t buf_size)
{
	return (readlink(path, buf, buf_size));
}

/*
 * Does fsync of a file. Returns with fsync return info.
 */
static int
fb_lfs_fsync(fb_fdesc_t *fd)
{
	return (fsync(fd->fd_num));
}

/*
 * Do a posix lseek of a file. Return what lseek() returns.
 */
#ifdef HAVE_LSEEK64
static int
fb_lfs_lseek(fb_fdesc_t *fd, off64_t offset, int whence)
{
	return (lseek64(fd->fd_num, offset, whence));
}
#else
static int
fb_lfs_lseek(fb_fdesc_t *fd, off_t offset, int whence)
{
	return (lseek(fd->fd_num, offset, whence));
}
#endif

/*
 * Do a posix rename of a file. Return what rename() returns.
 */
static int
fb_lfs_rename(const char *old, const char *new)
{
	return (rename(old, new));
}


/*
 * Do a posix close of a file. Return what close() returns.
 */
static int
fb_lfs_close(fb_fdesc_t *fd)
{
	return (close(fd->fd_num));
}

/*
 * Use mkdir to create a directory.
 */
static int
fb_lfs_mkdir(char *path, int perm)
{
	return (mkdir(path, perm));
}

/*
 * Use rmdir to delete a directory. Returns what rmdir() returns.
 */
static int
fb_lfs_rmdir(char *path)
{
	return (rmdir(path));
}

/*
 * does a recursive rm to remove an entire directory tree (i.e. a fileset).
 * Supplied with the path to the root of the tree.
 */
static void
fb_lfs_recur_rm(char *path)
{
	char cmd[MAXPATHLEN];

	(void) snprintf(cmd, sizeof (cmd), "rm -rf %s", path);

	/* We ignore system()'s return value */
	(void) system(cmd);
	return;
}

/*
 * Does a posix opendir(), Returns a directory handle on success,
 * NULL on failure.
 */
static struct fsplug_dir *
fb_lfs_opendir(char *path)
{
	return (struct fsplug_dir *)(opendir(path));
}

/*
 * Emulates a readdir() call.  Populates dirent on success (and returns 0)
 * or returns -1.
 */
static int
fb_lfs_readdir(struct fsplug_dir *dirp, struct fsplug_dirent *dirent)
{
	struct dirent *d = readdir((DIR*)dirp);

	if (d != NULL) {
		dirent->bytecost = sizeof(*d) + strlen(d->d_name) - 1;

		/* We don't actually care about the name at the moment, except as it
		 * influences the bytecost above.
		 */
#if 0
		strncpy(dirent->d_name, d->d_name, sizeof(dirent->d_name)-1);
		dirent->d_name[sizeof(dirent->d_name)-1] = '\0';
#endif

		return 0;
	} else {
		return -1;
	}
}

/*
 * Does a closedir() call.
 */
static int
fb_lfs_closedir(struct fsplug_dir *dirp)
{
	return (closedir((DIR*)dirp));
}

/*
 * Does an fstat of a file.
 */
#ifdef HAVE_FSTAT64
static int
fb_lfs_fstat(fb_fdesc_t *fd, struct stat64 *statbufp)
{
	return (fstat64(fd->fd_num, statbufp));
}
#else
static int
fb_lfs_fstat(fb_fdesc_t *fd, struct stat *statbufp)
{
	return (fstat(fd->fd_num, statbufp));
}
#endif


/*
 * Does a stat of a file.
 */
#ifdef HAVE_STAT64
static int
fb_lfs_stat(char *path, struct stat64 *statbufp)
{
	return (stat64(path, statbufp));
}
#else
static int
fb_lfs_stat(char *path, struct stat *statbufp)
{
	return (stat(path, statbufp));
}
#endif

/*
 * Do a pwrite64 to a file.
 */
#ifdef HAVE_PWRITE64
static int
fb_lfs_pwrite(fb_fdesc_t *fd, caddr_t iobuf, fbint_t iosize, off64_t offset)
{
	return (pwrite64(fd->fd_num, iobuf, iosize, offset));
}
#else
static int
fb_lfs_pwrite(fb_fdesc_t *fd, caddr_t iobuf, fbint_t iosize, off_t offset)
{
	return (pwrite(fd->fd_num, iobuf, iosize, offset));
}
#endif

/*
 * Do a write to a file.
 */
static int
fb_lfs_write(fb_fdesc_t *fd, caddr_t iobuf, fbint_t iosize)
{
	return (write(fd->fd_num, iobuf, iosize));
}

/*
 * Does a truncate operation and returns the result
 */
static int
fb_lfs_truncate(fb_fdesc_t *fd, off64_t fse_size)
{
#ifdef HAVE_FTRUNCATE64
	return (ftruncate64(fd->fd_num, fse_size));
#else
	filebench_log(LOG_ERROR, "Converting off64_t to off_t in ftruncate,"
			" might be a possible problem");
	return (ftruncate(fd->fd_num, (off_t)fse_size));
#endif
}

/*
 * Does a link operation and returns the result
 */
static int
fb_lfs_link(const char *existing, const char *new)
{
	return (link(existing, new));
}

/*
 * Does a symlink operation and returns the result
 */
static int
fb_lfs_symlink(const char *existing, const char *new)
{
	return (symlink(existing, new));
}

/*
 * Does an access() check on a file.
 */
static int
fb_lfs_access(const char *path, int amode)
{
	return (access(path, amode));
}