[fio.git] / engines / windowsaio.c

/*
 * Native Windows async IO engine
 * Copyright (C) 2011 Bruce Cran <bruce@cran.org.uk>
 */

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <signal.h>
#include <errno.h>
#include <windows.h>

#include "../fio.h"

typedef BOOL (WINAPI *CANCELIOEX)(HANDLE hFile, LPOVERLAPPED lpOverlapped);

struct fio_overlapped {
	OVERLAPPED o;
	struct io_u *io_u;
	BOOL io_complete;
    BOOL io_free;
};

struct windowsaio_data {
	struct fio_overlapped *ovls;
	struct io_u **aio_events;
	HANDLE iothread;
	HANDLE iocomplete_event;
	CANCELIOEX pCancelIoEx;
	BOOL iothread_running;
	BOOL use_iocp;
};

struct thread_ctx {
	HANDLE iocp;
	struct windowsaio_data *wd;
};

static void PrintError(LPCSTR lpszFunction);
static int fio_windowsaio_cancel(struct thread_data *td,
			       struct io_u *io_u);
static BOOL timeout_expired(DWORD start_count, DWORD end_count);
static int fio_windowsaio_getevents(struct thread_data *td, unsigned int min,
				    unsigned int max, struct timespec *t);
static struct io_u *fio_windowsaio_event(struct thread_data *td, int event);
static int fio_windowsaio_queue(struct thread_data *td,
			      struct io_u *io_u);
static void fio_windowsaio_cleanup(struct thread_data *td);
static DWORD WINAPI IoCompletionRoutine(LPVOID lpParameter);
static int fio_windowsaio_init(struct thread_data *td);
static int fio_windowsaio_open_file(struct thread_data *td, struct fio_file *f);
static int fio_windowsaio_close_file(struct thread_data fio_unused *td, struct fio_file *f);

int sync_file_range(int fd, off64_t offset, off64_t nbytes,
			   unsigned int flags)
{
	errno = ENOSYS;
	return -1;
}

static void PrintError(LPCSTR lpszFunction)
{
	// Retrieve the system error message for the last-error code

	LPSTR lpMsgBuf;
	DWORD dw = GetLastError();

	FormatMessage(
		FORMAT_MESSAGE_ALLOCATE_BUFFER |
		FORMAT_MESSAGE_FROM_SYSTEM |
		FORMAT_MESSAGE_IGNORE_INSERTS,
		NULL,
		dw,
		MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
		(LPTSTR)&lpMsgBuf,
		0, NULL );

	log_err("%s - %s", lpszFunction, lpMsgBuf);
	LocalFree(lpMsgBuf);
}

static int fio_windowsaio_init(struct thread_data *td)
{
	struct windowsaio_data *wd;
	HANDLE hKernel32Dll;
	int rc = 0;
	int i;

	wd = malloc(sizeof(struct windowsaio_data));
	if (wd != NULL)
		ZeroMemory(wd, sizeof(struct windowsaio_data));
	else
		rc = 1;

	if (!rc) {
		wd->aio_events = malloc(td->o.iodepth * sizeof(struct io_u*));
		if (wd->aio_events == NULL)
			rc = 1;
	}

	if (!rc) {
		wd->ovls = malloc(td->o.iodepth * sizeof(struct fio_overlapped));
		if (wd->ovls == NULL)
			rc = 1;
	}

	if (!rc) {
	    for (i = 0; i < td->o.iodepth; i++) {
	        wd->ovls[i].io_free = TRUE;
	        wd->ovls[i].io_complete = FALSE;

			wd->ovls[i].o.hEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
			if (wd->ovls[i].o.hEvent == NULL) {
				rc = 1;
				break;
			}
	    }
	}

	if (!rc) {
		/* Create an auto-reset event */
		wd->iocomplete_event = CreateEvent(NULL, FALSE, FALSE, NULL);
		if (wd->iocomplete_event == NULL)
			rc = 1;
	}

	if (rc) {
		PrintError(__func__);
		if (wd != NULL) {
			if (wd->ovls != NULL)
				free(wd->ovls);
			if (wd->aio_events != NULL)
				free(wd->aio_events);

			free(wd);
		}
	}

	hKernel32Dll = GetModuleHandle("kernel32.dll");
	wd->pCancelIoEx = GetProcAddress(hKernel32Dll, "CancelIoEx");

	td->io_ops->data = wd;
	return 0;
}

static void fio_windowsaio_cleanup(struct thread_data *td)
{
	int i;
	struct windowsaio_data *wd;

	wd = td->io_ops->data;

	if (wd != NULL) {
	    wd->iothread_running = FALSE;
    	WaitForSingleObject(wd->iothread, INFINITE);

		CloseHandle(wd->iothread);
		CloseHandle(wd->iocomplete_event);

		for (i = 0; i < td->o.iodepth; i++) {
			CloseHandle(wd->ovls[i].o.hEvent);
		}

		free(wd->aio_events);
		free(wd->ovls);
		free(wd);

		td->io_ops->data = NULL;
	}
}


static int fio_windowsaio_open_file(struct thread_data *td, struct fio_file *f)
{
	int rc = 0;
	HANDLE hFile;
	DWORD flags = FILE_FLAG_POSIX_SEMANTICS;
	DWORD sharemode = FILE_SHARE_READ | FILE_SHARE_WRITE;
	DWORD openmode = OPEN_ALWAYS;
	DWORD access;

	dprint(FD_FILE, "fd open %s\n", f->file_name);

	if (f->filetype == FIO_TYPE_PIPE) {
		log_err("fio: windowsaio doesn't support pipes\n");
		return 1;
	}

	if (!strcmp(f->file_name, "-")) {
		log_err("fio: can't read/write to stdin/out\n");
		return 1;
	}

    if (!td->o.odirect && !td->o.sync_io && td->io_ops->data != NULL)
	    flags |= FILE_FLAG_OVERLAPPED;

	if (td->o.odirect)
		flags |= FILE_FLAG_NO_BUFFERING;
	if (td->o.sync_io)
		flags |= FILE_FLAG_WRITE_THROUGH;


	if (td->o.td_ddir == TD_DDIR_READ  ||
		td->o.td_ddir == TD_DDIR_WRITE)
		flags |= FILE_FLAG_SEQUENTIAL_SCAN;
	else
		flags |= FILE_FLAG_RANDOM_ACCESS;

	if (!td_write(td) || read_only)
		access = GENERIC_READ;
	else
		access = (GENERIC_READ | GENERIC_WRITE);

	if (td->o.create_on_open > 0)
		openmode = OPEN_ALWAYS;
	else
		openmode = OPEN_EXISTING;

	f->hFile = CreateFile(f->file_name, access, sharemode,
		NULL, openmode, flags, NULL);

	if (f->hFile == INVALID_HANDLE_VALUE) {
		PrintError(__func__);
		rc = 1;
	}

	/* Only set up the competion port and thread if we're not just
	 * querying the device size */
    if (!rc && td->io_ops->data != NULL && !td->o.odirect && !td->o.sync_io) {
		struct thread_ctx *ctx;
        struct windowsaio_data *wd;
		hFile = CreateIoCompletionPort(f->hFile, NULL, 0, 0);

        wd = td->io_ops->data;

        if (!td->o.odirect && !td->o.sync_io)
            wd->use_iocp = 1;
        else
            wd->use_iocp = 0;

		wd->iothread_running = TRUE;

		if (!rc) {
			ctx = malloc(sizeof(struct thread_ctx));
			ctx->iocp = hFile;
			ctx->wd = wd;

			wd->iothread = CreateThread(NULL, 0, IoCompletionRoutine, ctx, 0, NULL);
		}

		if (rc || wd->iothread == NULL) {
			PrintError(__func__);
			rc = 1;
		}
	}

	return rc;
}

static int fio_windowsaio_close_file(struct thread_data fio_unused *td, struct fio_file *f)
{
	dprint(FD_FILE, "fd close %s\n", f->file_name);

	if (f->hFile != INVALID_HANDLE_VALUE) {
		if (!CloseHandle(f->hFile))
			PrintError(__func__);
	}

	f->hFile = INVALID_HANDLE_VALUE;
	return 0;
}

static BOOL timeout_expired(DWORD start_count, DWORD end_count)
{
	BOOL expired = FALSE;
	DWORD current_time;

	current_time = GetTickCount();

	if ((end_count > start_count) && current_time >= end_count)
		expired = TRUE;
	else if (current_time < start_count && current_time > end_count)
		expired = TRUE;

	return expired;
}

static struct io_u* fio_windowsaio_event(struct thread_data *td, int event)
{
	struct windowsaio_data *wd = td->io_ops->data;
	return wd->aio_events[event];
}

static int fio_windowsaio_getevents(struct thread_data *td, unsigned int min,
				    unsigned int max, struct timespec *t)
{
	struct windowsaio_data *wd = td->io_ops->data;
	struct flist_head *entry;
	unsigned int dequeued = 0;
	struct io_u *io_u;
	struct fio_overlapped *fov;
	DWORD start_count = 0;
	DWORD end_count = 0;
	DWORD status;
	DWORD mswait = 250;

	if (t != NULL) {
		mswait = (t->tv_sec * 1000) + (t->tv_nsec / 1000000);
		start_count = GetTickCount();
		end_count = start_count + (t->tv_sec * 1000) + (t->tv_nsec / 1000000);
	}

	do {
		flist_for_each(entry, &td->io_u_busylist) {
			io_u = flist_entry(entry, struct io_u, list);
			fov = (struct fio_overlapped*)io_u->engine_data;

			if (fov->io_complete) {
                fov->io_complete = FALSE;
				fov->io_free  = TRUE;
				wd->aio_events[dequeued] = io_u;
				dequeued++;
			}

			if (dequeued >= min)
				break;
		}

        if (dequeued < min) {
			status = WaitForSingleObject(wd->iocomplete_event, mswait);
			if (status != WAIT_OBJECT_0 && dequeued > 0)
			    break;
		}

		if (dequeued >= min || (t != NULL && timeout_expired(start_count, end_count)))
			break;
	} while (1);

	return dequeued;
}

static int fio_windowsaio_queue(struct thread_data *td,
			      struct io_u *io_u)
{
    LPOVERLAPPED lpOvl = NULL;
	struct windowsaio_data *wd;
	DWORD iobytes;
	BOOL success;
	int index;
	int rc = FIO_Q_COMPLETED;

	fio_ro_check(td, io_u);

	wd = td->io_ops->data;

	if (wd->use_iocp) {
	    for (index = 0; index < td->o.iodepth; index++) {
	        if (wd->ovls[index].io_free) {
                wd->ovls[index].io_free = FALSE;
	            ResetEvent(wd->ovls[index].o.hEvent);
	            break;
	        }
        }

        assert(index < td->o.iodepth);

        lpOvl = &wd->ovls[index].o;
	    wd->ovls[index].io_u = io_u;
    	lpOvl->Internal = STATUS_PENDING;
    	lpOvl->InternalHigh = 0;
    	lpOvl->Offset = io_u->offset & 0xFFFFFFFF;
    	lpOvl->OffsetHigh = io_u->offset >> 32;
    	lpOvl->Pointer = NULL;
        io_u->engine_data = &wd->ovls[index];
	}

	switch (io_u->ddir) {
    case DDIR_WRITE:
		success = WriteFile(io_u->file->hFile, io_u->xfer_buf, io_u->xfer_buflen, &iobytes, lpOvl);
		break;
	case DDIR_READ:
		success = ReadFile(io_u->file->hFile, io_u->xfer_buf, io_u->xfer_buflen, &iobytes, lpOvl);
		break;
	case DDIR_SYNC:
	case DDIR_DATASYNC:
	case DDIR_SYNC_FILE_RANGE:
		success = FlushFileBuffers(io_u->file->hFile);
		if (!success)
		    io_u->error = GetLastError();

		return FIO_Q_COMPLETED;
		break;
	case DDIR_TRIM:
		log_err("manual TRIM isn't supported on Windows");
		io_u->error = 1;
		io_u->resid = io_u->xfer_buflen;
		return FIO_Q_COMPLETED;
		break;
	default:
		assert(0);
	}

    if (wd->use_iocp && (success || GetLastError() == ERROR_IO_PENDING)) {
		rc = FIO_Q_QUEUED;
	} else if (success && !wd->use_iocp) {
		io_u->resid = io_u->xfer_buflen - iobytes;
		io_u->error = 0;
	} else {
		PrintError(__func__);
		io_u->error = GetLastError();
		io_u->resid = io_u->xfer_buflen;
	}

	return rc;
}

/* Runs as a thread and waits for queued IO to complete */
static DWORD WINAPI IoCompletionRoutine(LPVOID lpParameter)
{
	OVERLAPPED *ovl;
	struct fio_overlapped *fov;
	struct io_u *io_u;
	struct windowsaio_data *wd;
	struct thread_ctx *ctx;
	ULONG_PTR ulKey = 0;
	DWORD bytes;

	ctx = (struct thread_ctx*)lpParameter;
	wd = ctx->wd;

	do {
		if (!GetQueuedCompletionStatus(ctx->iocp, &bytes, &ulKey, &ovl, 250))
			continue;

		fov = CONTAINING_RECORD(ovl, struct fio_overlapped, o);
		io_u = fov->io_u;

		if (ovl->Internal == ERROR_SUCCESS) {
			io_u->resid = io_u->xfer_buflen - ovl->InternalHigh;
			io_u->error = 0;
		} else {
			io_u->resid = io_u->xfer_buflen;
			io_u->error = ovl->Internal;
		}

        fov->io_complete = TRUE;
		SetEvent(wd->iocomplete_event);
	} while (ctx->wd->iothread_running);

	CloseHandle(ctx->iocp);
	free(ctx);
	return 0;
}

static int fio_windowsaio_cancel(struct thread_data *td,
			       struct io_u *io_u)
{
	int rc = 0;

	struct windowsaio_data *wd = td->io_ops->data;

	/* If we're running on Vista or newer, we can cancel individual IO requests */
	if (wd->pCancelIoEx != NULL) {
		struct fio_overlapped *ovl = io_u->engine_data;
		if (!wd->pCancelIoEx(io_u->file->hFile, &ovl->o))
			rc = 1;
	} else
		rc = 1;

	return rc;
}

static struct ioengine_ops ioengine = {
	.name		= "windowsaio",
	.version	= FIO_IOOPS_VERSION,
	.init		= fio_windowsaio_init,
	.queue		= fio_windowsaio_queue,
	.cancel		= fio_windowsaio_cancel,
	.getevents	= fio_windowsaio_getevents,
	.event		= fio_windowsaio_event,
	.cleanup	= fio_windowsaio_cleanup,
	.open_file	= fio_windowsaio_open_file,
	.close_file	= fio_windowsaio_close_file,
	.get_file_size	= generic_get_file_size
};

static void fio_init fio_posixaio_register(void)
{
	register_ioengine(&ioengine);
}

static void fio_exit fio_posixaio_unregister(void)
{
	unregister_ioengine(&ioengine);
}
Commit	Line	Data
	1	/*
	2	* Native Windows async IO engine
	3	* Copyright (C) 2011 Bruce Cran <bruce@cran.org.uk>
	4	*/
	5
	6	#include <stdio.h>
	7	#include <stdlib.h>
	8	#include <unistd.h>
	9	#include <signal.h>
	10	#include <errno.h>
	11	#include <windows.h>
	12
	13	#include "../fio.h"
	14
	15	typedef BOOL (WINAPI *CANCELIOEX)(HANDLE hFile, LPOVERLAPPED lpOverlapped);
	16
	17	struct fio_overlapped {
	18	OVERLAPPED o;
	19	struct io_u *io_u;
	20	BOOL io_complete;
	21	BOOL io_free;
	22	};
	23
	24	struct windowsaio_data {
	25	struct fio_overlapped *ovls;
	26	struct io_u **aio_events;
	27	HANDLE iothread;
	28	HANDLE iocomplete_event;
	29	CANCELIOEX pCancelIoEx;
	30	BOOL iothread_running;
	31	BOOL use_iocp;
	32	};
	33
	34	struct thread_ctx {
	35	HANDLE iocp;
	36	struct windowsaio_data *wd;
	37	};
	38
	39	static void PrintError(LPCSTR lpszFunction);
	40	static int fio_windowsaio_cancel(struct thread_data *td,
	41	struct io_u *io_u);
	42	static BOOL timeout_expired(DWORD start_count, DWORD end_count);
	43	static int fio_windowsaio_getevents(struct thread_data *td, unsigned int min,
	44	unsigned int max, struct timespec *t);
	45	static struct io_u fio_windowsaio_event(struct thread_data td, int event);
	46	static int fio_windowsaio_queue(struct thread_data *td,
	47	struct io_u *io_u);
	48	static void fio_windowsaio_cleanup(struct thread_data *td);
	49	static DWORD WINAPI IoCompletionRoutine(LPVOID lpParameter);
	50	static int fio_windowsaio_init(struct thread_data *td);
	51	static int fio_windowsaio_open_file(struct thread_data td, struct fio_file f);
	52	static int fio_windowsaio_close_file(struct thread_data fio_unused td, struct fio_file f);
	53
	54	int sync_file_range(int fd, off64_t offset, off64_t nbytes,
	55	unsigned int flags)
	56	{
	57	errno = ENOSYS;
	58	return -1;
	59	}
	60
	61	static void PrintError(LPCSTR lpszFunction)
	62	{
	63	// Retrieve the system error message for the last-error code
	64
	65	LPSTR lpMsgBuf;
	66	DWORD dw = GetLastError();
	67
	68	FormatMessage(
	69	FORMAT_MESSAGE_ALLOCATE_BUFFER \|
	70	FORMAT_MESSAGE_FROM_SYSTEM \|
	71	FORMAT_MESSAGE_IGNORE_INSERTS,
	72	NULL,
	73	dw,
	74	MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
	75	(LPTSTR)&lpMsgBuf,
	76	0, NULL );
	77
	78	log_err("%s - %s", lpszFunction, lpMsgBuf);
	79	LocalFree(lpMsgBuf);
	80	}
	81
	82	static int fio_windowsaio_init(struct thread_data *td)
	83	{
	84	struct windowsaio_data *wd;
	85	HANDLE hKernel32Dll;
	86	int rc = 0;
	87	int i;
	88
	89	wd = malloc(sizeof(struct windowsaio_data));
	90	if (wd != NULL)
	91	ZeroMemory(wd, sizeof(struct windowsaio_data));
	92	else
	93	rc = 1;
	94
	95	if (!rc) {
	96	wd->aio_events = malloc(td->o.iodepth * sizeof(struct io_u*));
	97	if (wd->aio_events == NULL)
	98	rc = 1;
	99	}
	100
	101	if (!rc) {
	102	wd->ovls = malloc(td->o.iodepth * sizeof(struct fio_overlapped));
	103	if (wd->ovls == NULL)
	104	rc = 1;
	105	}
	106
	107	if (!rc) {
	108	for (i = 0; i < td->o.iodepth; i++) {
	109	wd->ovls[i].io_free = TRUE;
	110	wd->ovls[i].io_complete = FALSE;
	111
	112	wd->ovls[i].o.hEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
	113	if (wd->ovls[i].o.hEvent == NULL) {
	114	rc = 1;
	115	break;
	116	}
	117	}
	118	}
	119
	120	if (!rc) {
	121	/* Create an auto-reset event */
	122	wd->iocomplete_event = CreateEvent(NULL, FALSE, FALSE, NULL);
	123	if (wd->iocomplete_event == NULL)
	124	rc = 1;
	125	}
	126
	127	if (rc) {
	128	PrintError(__func__);
	129	if (wd != NULL) {
	130	if (wd->ovls != NULL)
	131	free(wd->ovls);
	132	if (wd->aio_events != NULL)
	133	free(wd->aio_events);
	134
	135	free(wd);
	136	}
	137	}
	138
	139	hKernel32Dll = GetModuleHandle("kernel32.dll");
	140	wd->pCancelIoEx = GetProcAddress(hKernel32Dll, "CancelIoEx");
	141
	142	td->io_ops->data = wd;
	143	return 0;
	144	}
	145
	146	static void fio_windowsaio_cleanup(struct thread_data *td)
	147	{
	148	int i;
	149	struct windowsaio_data *wd;
	150
	151	wd = td->io_ops->data;
	152
	153	if (wd != NULL) {
	154	wd->iothread_running = FALSE;
	155	WaitForSingleObject(wd->iothread, INFINITE);
	156
	157	CloseHandle(wd->iothread);
	158	CloseHandle(wd->iocomplete_event);
	159
	160	for (i = 0; i < td->o.iodepth; i++) {
	161	CloseHandle(wd->ovls[i].o.hEvent);
	162	}
	163
	164	free(wd->aio_events);
	165	free(wd->ovls);
	166	free(wd);
	167
	168	td->io_ops->data = NULL;
	169	}
	170	}
	171
	172
	173	static int fio_windowsaio_open_file(struct thread_data td, struct fio_file f)
	174	{
	175	int rc = 0;
	176	HANDLE hFile;
	177	DWORD flags = FILE_FLAG_POSIX_SEMANTICS;
	178	DWORD sharemode = FILE_SHARE_READ \| FILE_SHARE_WRITE;
	179	DWORD openmode = OPEN_ALWAYS;
	180	DWORD access;
	181
	182	dprint(FD_FILE, "fd open %s\n", f->file_name);
	183
	184	if (f->filetype == FIO_TYPE_PIPE) {
	185	log_err("fio: windowsaio doesn't support pipes\n");
	186	return 1;
	187	}
	188
	189	if (!strcmp(f->file_name, "-")) {
	190	log_err("fio: can't read/write to stdin/out\n");
	191	return 1;
	192	}
	193
	194	if (!td->o.odirect && !td->o.sync_io && td->io_ops->data != NULL)
	195	flags \|= FILE_FLAG_OVERLAPPED;
	196
	197	if (td->o.odirect)
	198	flags \|= FILE_FLAG_NO_BUFFERING;
	199	if (td->o.sync_io)
	200	flags \|= FILE_FLAG_WRITE_THROUGH;
	201
	202
	203	if (td->o.td_ddir == TD_DDIR_READ \|\|
	204	td->o.td_ddir == TD_DDIR_WRITE)
	205	flags \|= FILE_FLAG_SEQUENTIAL_SCAN;
	206	else
	207	flags \|= FILE_FLAG_RANDOM_ACCESS;
	208
	209	if (!td_write(td) \|\| read_only)
	210	access = GENERIC_READ;
	211	else
	212	access = (GENERIC_READ \| GENERIC_WRITE);
	213
	214	if (td->o.create_on_open > 0)
	215	openmode = OPEN_ALWAYS;
	216	else
	217	openmode = OPEN_EXISTING;
	218
	219	f->hFile = CreateFile(f->file_name, access, sharemode,
	220	NULL, openmode, flags, NULL);
	221
	222	if (f->hFile == INVALID_HANDLE_VALUE) {
	223	PrintError(__func__);
	224	rc = 1;
	225	}
	226
	227	/* Only set up the competion port and thread if we're not just
	228	* querying the device size */
	229	if (!rc && td->io_ops->data != NULL && !td->o.odirect && !td->o.sync_io) {
	230	struct thread_ctx *ctx;
	231	struct windowsaio_data *wd;
	232	hFile = CreateIoCompletionPort(f->hFile, NULL, 0, 0);
	233
	234	wd = td->io_ops->data;
	235
	236	if (!td->o.odirect && !td->o.sync_io)
	237	wd->use_iocp = 1;
	238	else
	239	wd->use_iocp = 0;
	240
	241	wd->iothread_running = TRUE;
	242
	243	if (!rc) {
	244	ctx = malloc(sizeof(struct thread_ctx));
	245	ctx->iocp = hFile;
	246	ctx->wd = wd;
	247
	248	wd->iothread = CreateThread(NULL, 0, IoCompletionRoutine, ctx, 0, NULL);
	249	}
	250
	251	if (rc \|\| wd->iothread == NULL) {
	252	PrintError(__func__);
	253	rc = 1;
	254	}
	255	}
	256
	257	return rc;
	258	}
	259
	260	static int fio_windowsaio_close_file(struct thread_data fio_unused td, struct fio_file f)
	261	{
	262	dprint(FD_FILE, "fd close %s\n", f->file_name);
	263
	264	if (f->hFile != INVALID_HANDLE_VALUE) {
	265	if (!CloseHandle(f->hFile))
	266	PrintError(__func__);
	267	}
	268
	269	f->hFile = INVALID_HANDLE_VALUE;
	270	return 0;
	271	}
	272
	273	static BOOL timeout_expired(DWORD start_count, DWORD end_count)
	274	{
	275	BOOL expired = FALSE;
	276	DWORD current_time;
	277
	278	current_time = GetTickCount();
	279
	280	if ((end_count > start_count) && current_time >= end_count)
	281	expired = TRUE;
	282	else if (current_time < start_count && current_time > end_count)
	283	expired = TRUE;
	284
	285	return expired;
	286	}
	287
	288	static struct io_u* fio_windowsaio_event(struct thread_data *td, int event)
	289	{
	290	struct windowsaio_data *wd = td->io_ops->data;
	291	return wd->aio_events[event];
	292	}
	293
	294	static int fio_windowsaio_getevents(struct thread_data *td, unsigned int min,
	295	unsigned int max, struct timespec *t)
	296	{
	297	struct windowsaio_data *wd = td->io_ops->data;
	298	struct flist_head *entry;
	299	unsigned int dequeued = 0;
	300	struct io_u *io_u;
	301	struct fio_overlapped *fov;
	302	DWORD start_count = 0;
	303	DWORD end_count = 0;
	304	DWORD status;
	305	DWORD mswait = 250;
	306
	307	if (t != NULL) {
	308	mswait = (t->tv_sec * 1000) + (t->tv_nsec / 1000000);
	309	start_count = GetTickCount();
	310	end_count = start_count + (t->tv_sec * 1000) + (t->tv_nsec / 1000000);
	311	}
	312
	313	do {
	314	flist_for_each(entry, &td->io_u_busylist) {
	315	io_u = flist_entry(entry, struct io_u, list);
	316	fov = (struct fio_overlapped*)io_u->engine_data;
	317
	318	if (fov->io_complete) {
	319	fov->io_complete = FALSE;
	320	fov->io_free = TRUE;
	321	wd->aio_events[dequeued] = io_u;
	322	dequeued++;
	323	}
	324
	325	if (dequeued >= min)
	326	break;
	327	}
	328
	329	if (dequeued < min) {
	330	status = WaitForSingleObject(wd->iocomplete_event, mswait);
	331	if (status != WAIT_OBJECT_0 && dequeued > 0)
	332	break;
	333	}
	334
	335	if (dequeued >= min \|\| (t != NULL && timeout_expired(start_count, end_count)))
	336	break;
	337	} while (1);
	338
	339	return dequeued;
	340	}
	341
	342	static int fio_windowsaio_queue(struct thread_data *td,
	343	struct io_u *io_u)
	344	{
	345	LPOVERLAPPED lpOvl = NULL;
	346	struct windowsaio_data *wd;
	347	DWORD iobytes;
	348	BOOL success;
	349	int index;
	350	int rc = FIO_Q_COMPLETED;
	351
	352	fio_ro_check(td, io_u);
	353
	354	wd = td->io_ops->data;
	355
	356	if (wd->use_iocp) {
	357	for (index = 0; index < td->o.iodepth; index++) {
	358	if (wd->ovls[index].io_free) {
	359	wd->ovls[index].io_free = FALSE;
	360	ResetEvent(wd->ovls[index].o.hEvent);
	361	break;
	362	}
	363	}
	364
	365	assert(index < td->o.iodepth);
	366
	367	lpOvl = &wd->ovls[index].o;
	368	wd->ovls[index].io_u = io_u;
	369	lpOvl->Internal = STATUS_PENDING;
	370	lpOvl->InternalHigh = 0;
	371	lpOvl->Offset = io_u->offset & 0xFFFFFFFF;
	372	lpOvl->OffsetHigh = io_u->offset >> 32;
	373	lpOvl->Pointer = NULL;
	374	io_u->engine_data = &wd->ovls[index];
	375	}
	376
	377	switch (io_u->ddir) {
	378	case DDIR_WRITE:
	379	success = WriteFile(io_u->file->hFile, io_u->xfer_buf, io_u->xfer_buflen, &iobytes, lpOvl);
	380	break;
	381	case DDIR_READ:
	382	success = ReadFile(io_u->file->hFile, io_u->xfer_buf, io_u->xfer_buflen, &iobytes, lpOvl);
	383	break;
	384	case DDIR_SYNC:
	385	case DDIR_DATASYNC:
	386	case DDIR_SYNC_FILE_RANGE:
	387	success = FlushFileBuffers(io_u->file->hFile);
	388	if (!success)
	389	io_u->error = GetLastError();
	390
	391	return FIO_Q_COMPLETED;
	392	break;
	393	case DDIR_TRIM:
	394	log_err("manual TRIM isn't supported on Windows");
	395	io_u->error = 1;
	396	io_u->resid = io_u->xfer_buflen;
	397	return FIO_Q_COMPLETED;
	398	break;
	399	default:
	400	assert(0);
	401	}
	402
	403	if (wd->use_iocp && (success \|\| GetLastError() == ERROR_IO_PENDING)) {
	404	rc = FIO_Q_QUEUED;
	405	} else if (success && !wd->use_iocp) {
	406	io_u->resid = io_u->xfer_buflen - iobytes;
	407	io_u->error = 0;
	408	} else {
	409	PrintError(__func__);
	410	io_u->error = GetLastError();
	411	io_u->resid = io_u->xfer_buflen;
	412	}
	413
	414	return rc;
	415	}
	416
	417	/* Runs as a thread and waits for queued IO to complete */
	418	static DWORD WINAPI IoCompletionRoutine(LPVOID lpParameter)
	419	{
	420	OVERLAPPED *ovl;
	421	struct fio_overlapped *fov;
	422	struct io_u *io_u;
	423	struct windowsaio_data *wd;
	424	struct thread_ctx *ctx;
	425	ULONG_PTR ulKey = 0;
	426	DWORD bytes;
	427
	428	ctx = (struct thread_ctx*)lpParameter;
	429	wd = ctx->wd;
	430
	431	do {
	432	if (!GetQueuedCompletionStatus(ctx->iocp, &bytes, &ulKey, &ovl, 250))
	433	continue;
	434
	435	fov = CONTAINING_RECORD(ovl, struct fio_overlapped, o);
	436	io_u = fov->io_u;
	437
	438	if (ovl->Internal == ERROR_SUCCESS) {
	439	io_u->resid = io_u->xfer_buflen - ovl->InternalHigh;
	440	io_u->error = 0;
	441	} else {
	442	io_u->resid = io_u->xfer_buflen;
	443	io_u->error = ovl->Internal;
	444	}
	445
	446	fov->io_complete = TRUE;
	447	SetEvent(wd->iocomplete_event);
	448	} while (ctx->wd->iothread_running);
	449
	450	CloseHandle(ctx->iocp);
	451	free(ctx);
	452	return 0;
	453	}
	454
	455	static int fio_windowsaio_cancel(struct thread_data *td,
	456	struct io_u *io_u)
	457	{
	458	int rc = 0;
	459
	460	struct windowsaio_data *wd = td->io_ops->data;
	461
	462	/* If we're running on Vista or newer, we can cancel individual IO requests */
	463	if (wd->pCancelIoEx != NULL) {
	464	struct fio_overlapped *ovl = io_u->engine_data;
	465	if (!wd->pCancelIoEx(io_u->file->hFile, &ovl->o))
	466	rc = 1;
	467	} else
	468	rc = 1;
	469
	470	return rc;
	471	}
	472
	473	static struct ioengine_ops ioengine = {
	474	.name = "windowsaio",
	475	.version = FIO_IOOPS_VERSION,
	476	.init = fio_windowsaio_init,
	477	.queue = fio_windowsaio_queue,
	478	.cancel = fio_windowsaio_cancel,
	479	.getevents = fio_windowsaio_getevents,
	480	.event = fio_windowsaio_event,
	481	.cleanup = fio_windowsaio_cleanup,
	482	.open_file = fio_windowsaio_open_file,
	483	.close_file = fio_windowsaio_close_file,
	484	.get_file_size = generic_get_file_size
	485	};
	486
	487	static void fio_init fio_posixaio_register(void)
	488	{
	489	register_ioengine(&ioengine);
	490	}
	491
	492	static void fio_exit fio_posixaio_unregister(void)
	493	{
	494	unregister_ioengine(&ioengine);
	495	}