[fio.git] / engines / pmemblk.c

/*
 * pmemblk: IO engine that uses NVML libpmemblk to read and write data
 *
 * Copyright (C) 2016 Hewlett Packard Enterprise Development LP
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License,
 * version 2 as published by the Free Software Foundation..
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public
 * License along with this program; if not, write to the Free
 * Software Foundation, Inc., 59 Temple Place, Suite 330,
 * Boston, MA 02111-1307 USA
 */

/*
 * pmemblk engine
 *
 * IO engine that uses libpmemblk to read and write data
 *
 * To use:
 *   ioengine=pmemblk
 *
 * Other relevant settings:
 *   iodepth=1
 *   direct=1
 *   thread=1   REQUIRED
 *   unlink=1
 *   filename=/pmem0/fiotestfile,BSIZE,FSIZEMB
 *
 *   thread must be set to 1 for pmemblk as multiple processes cannot
 *     open the same block pool file.
 *
 *   iodepth should be set to 1 as pmemblk is always synchronous.
 *   Use numjobs to scale up.
 *
 *   direct=1 is implied as pmemblk is always direct.
 *
 *   Can set unlink to 1 to remove the block pool file after testing.
 *
 *   When specifying the filename, if the block pool file does not already
 *   exist, then the pmemblk engine can create the pool file if you specify
 *   the block and file sizes.  BSIZE is the block size in bytes.
 *   FSIZEMB is the pool file size in MB.
 *
 *   See examples/pmemblk.fio for more.
 *
 */

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/uio.h>
#include <errno.h>
#include <assert.h>
#include <string.h>
#include <libpmem.h>
#include <libpmemblk.h>

#include "../fio.h"

/*
 * libpmemblk
 */
typedef struct fio_pmemblk_file *fio_pmemblk_file_t;

struct fio_pmemblk_file {
	fio_pmemblk_file_t pmb_next;
	char *pmb_filename;
	uint64_t pmb_refcnt;
	PMEMblkpool *pmb_pool;
	size_t pmb_bsize;
	size_t pmb_nblocks;
};
#define FIOFILEPMBSET(_f, _v)  do {                 \
	(_f)->engine_data = (uint64_t)(uintptr_t)(_v);  \
} while(0)
#define FIOFILEPMBGET(_f)  ((fio_pmemblk_file_t)((_f)->engine_data))

static fio_pmemblk_file_t Cache;

static pthread_mutex_t CacheLock = PTHREAD_MUTEX_INITIALIZER;

#define PMB_CREATE   (0x0001)	/* should create file */

fio_pmemblk_file_t fio_pmemblk_cache_lookup(const char *filename)
{
	fio_pmemblk_file_t i;

	for (i = Cache; i != NULL; i = i->pmb_next)
		if (!strcmp(filename, i->pmb_filename))
			return i;

	return NULL;
}

static void fio_pmemblk_cache_insert(fio_pmemblk_file_t pmb)
{
	pmb->pmb_next = Cache;
	Cache = pmb;
}

static void fio_pmemblk_cache_remove(fio_pmemblk_file_t pmb)
{
	fio_pmemblk_file_t i;

	if (pmb == Cache) {
		Cache = Cache->pmb_next;
		pmb->pmb_next = NULL;
		return;
	}

	for (i = Cache; i != NULL; i = i->pmb_next)
		if (pmb == i->pmb_next) {
			i->pmb_next = i->pmb_next->pmb_next;
			pmb->pmb_next = NULL;
			return;
		}
}

/*
 * to control block size and gross file size at the libpmemblk
 * level, we allow the block size and file size to be appended
 * to the file name:
 *
 *   path[,bsize,fsizemb]
 *
 * note that we do not use the fio option "filesize" to dictate
 * the file size because we can only give libpmemblk the gross
 * file size, which is different from the net or usable file
 * size (which is probably what fio wants).
 *
 * the final path without the parameters is returned in ppath.
 * the block size and file size are returned in pbsize and fsize.
 *
 * note that the user should specify the file size in MiB, but
 * we return bytes from here.
 */
static void pmb_parse_path(const char *pathspec, char **ppath, uint64_t *pbsize,
			   uint64_t *pfsize)
{
	char *path;
	char *s;
	uint64_t bsize;
	uint64_t fsizemb;

	path = strdup(pathspec);
	if (!path) {
		*ppath = NULL;
		return;
	}

	/* extract sizes, if given */
	s = strrchr(path, ',');
	if (s && (fsizemb = strtoull(s + 1, NULL, 10))) {
		*s = 0;
		s = strrchr(path, ',');
		if (s && (bsize = strtoull(s + 1, NULL, 10))) {
			*s = 0;
			*ppath = path;
			*pbsize = bsize;
			*pfsize = fsizemb << 20;
			return;
		}
	}

	/* size specs not found */
	strcpy(path, pathspec);
	*ppath = path;
	*pbsize = 0;
	*pfsize = 0;
}

static fio_pmemblk_file_t pmb_open(const char *pathspec, int flags)
{
	fio_pmemblk_file_t pmb;
	char *path = NULL;
	uint64_t bsize = 0;
	uint64_t fsize = 0;

	pmb_parse_path(pathspec, &path, &bsize, &fsize);
	if (!path)
		return NULL;

	pthread_mutex_lock(&CacheLock);

	pmb = fio_pmemblk_cache_lookup(path);
	if (!pmb) {
		pmb = malloc(sizeof(*pmb));
		if (!pmb)
			goto error;

		/* try opening existing first, create it if needed */
		pmb->pmb_pool = pmemblk_open(path, bsize);
		if (!pmb->pmb_pool && (errno == ENOENT) &&
		    (flags & PMB_CREATE) && (0 < fsize) && (0 < bsize)) {
			pmb->pmb_pool =
			    pmemblk_create(path, bsize, fsize, 0644);
		}
		if (!pmb->pmb_pool) {
			log_err
			    ("fio: enable to open pmemblk pool file (errno %d)\n",
			     errno);
			goto error;
		}

		pmb->pmb_filename = path;
		pmb->pmb_next = NULL;
		pmb->pmb_refcnt = 0;
		pmb->pmb_bsize = pmemblk_bsize(pmb->pmb_pool);
		pmb->pmb_nblocks = pmemblk_nblock(pmb->pmb_pool);

		fio_pmemblk_cache_insert(pmb);
	}

	pmb->pmb_refcnt += 1;

	pthread_mutex_unlock(&CacheLock);

	return pmb;

error:
	if (pmb) {
		if (pmb->pmb_pool)
			pmemblk_close(pmb->pmb_pool);
		pmb->pmb_pool = NULL;
		pmb->pmb_filename = NULL;
		free(pmb);
	}
	if (path)
		free(path);

	pthread_mutex_unlock(&CacheLock);
	return NULL;
}

static void pmb_close(fio_pmemblk_file_t pmb, const bool keep)
{
	pthread_mutex_lock(&CacheLock);

	pmb->pmb_refcnt--;

	if (!keep && !pmb->pmb_refcnt) {
		pmemblk_close(pmb->pmb_pool);
		pmb->pmb_pool = NULL;
		free(pmb->pmb_filename);
		pmb->pmb_filename = NULL;
		fio_pmemblk_cache_remove(pmb);
		free(pmb);
	}

	pthread_mutex_unlock(&CacheLock);
}

static int pmb_get_flags(struct thread_data *td, uint64_t *pflags)
{
	static int thread_warned = 0;
	static int odirect_warned = 0;

	uint64_t flags = 0;

	if (!td->o.use_thread) {
		if (!thread_warned) {
			thread_warned = 1;
			log_err("fio: must set thread=1 for pmemblk engine\n");
		}
		return 1;
	}

	if (!td->o.odirect && !odirect_warned) {
		odirect_warned = 1;
		log_info("fio: direct == 0, but pmemblk is always direct\n");
	}

	if (td->o.allow_create)
		flags |= PMB_CREATE;

	(*pflags) = flags;
	return 0;
}

static int fio_pmemblk_open_file(struct thread_data *td, struct fio_file *f)
{
	uint64_t flags = 0;
	fio_pmemblk_file_t pmb;

	if (pmb_get_flags(td, &flags))
		return 1;

	pmb = pmb_open(f->file_name, flags);
	if (!pmb)
		return 1;

	FIOFILEPMBSET(f, pmb);
	return 0;
}

static int fio_pmemblk_close_file(struct thread_data fio_unused *td,
				  struct fio_file *f)
{
	fio_pmemblk_file_t pmb = FIOFILEPMBGET(f);

	if (pmb)
		pmb_close(pmb, false);

	FIOFILEPMBSET(f, NULL);
	return 0;
}

static int fio_pmemblk_get_file_size(struct thread_data *td, struct fio_file *f)
{
	uint64_t flags = 0;
	fio_pmemblk_file_t pmb = FIOFILEPMBGET(f);

	if (fio_file_size_known(f))
		return 0;

	if (!pmb) {
		if (pmb_get_flags(td, &flags))
			return 1;
		pmb = pmb_open(f->file_name, flags);
		if (!pmb)
			return 1;
	}

	f->real_file_size = pmb->pmb_bsize * pmb->pmb_nblocks;

	fio_file_set_size_known(f);

	if (!FIOFILEPMBGET(f))
		pmb_close(pmb, true);

	return 0;
}

static int fio_pmemblk_queue(struct thread_data *td, struct io_u *io_u)
{
	struct fio_file *f = io_u->file;
	fio_pmemblk_file_t pmb = FIOFILEPMBGET(f);

	unsigned long long off;
	unsigned long len;
	void *buf;

	fio_ro_check(td, io_u);

	switch (io_u->ddir) {
	case DDIR_READ:
	case DDIR_WRITE:
		off = io_u->offset;
		len = io_u->xfer_buflen;

		io_u->error = EINVAL;
		if (off % pmb->pmb_bsize)
			break;
		if (len % pmb->pmb_bsize)
			break;
		if ((off + len) / pmb->pmb_bsize > pmb->pmb_nblocks)
			break;

		io_u->error = 0;
		buf = io_u->xfer_buf;
		off /= pmb->pmb_bsize;
		len /= pmb->pmb_bsize;
		while (0 < len) {
			if (io_u->ddir == DDIR_READ &&
			   0 != pmemblk_read(pmb->pmb_pool, buf, off)) {
				io_u->error = errno;
				break;
			} else if (0 != pmemblk_write(pmb->pmb_pool, buf, off)) {
				io_u->error = errno;
				break;
			}
			buf += pmb->pmb_bsize;
			off++;
			len--;
		}
		off *= pmb->pmb_bsize;
		len *= pmb->pmb_bsize;
		io_u->resid = io_u->xfer_buflen - (off - io_u->offset);
		break;
	case DDIR_SYNC:
	case DDIR_DATASYNC:
	case DDIR_SYNC_FILE_RANGE:
		/* we're always sync'd */
		io_u->error = 0;
		break;
	default:
		io_u->error = EINVAL;
		break;
	}

	return FIO_Q_COMPLETED;
}

static int fio_pmemblk_unlink_file(struct thread_data *td, struct fio_file *f)
{
	char *path = NULL;
	uint64_t bsize = 0;
	uint64_t fsize = 0;

	/*
	 * we need our own unlink in case the user has specified
	 * the block and file sizes in the path name.  we parse
	 * the file_name to determine the file name we actually used.
	 */

	pmb_parse_path(f->file_name, &path, &bsize, &fsize);
	if (!path)
		return ENOENT;

	unlink(path);
	free(path);
	return 0;
}

static struct ioengine_ops ioengine = {
	.name = "pmemblk",
	.version = FIO_IOOPS_VERSION,
	.queue = fio_pmemblk_queue,
	.open_file = fio_pmemblk_open_file,
	.close_file = fio_pmemblk_close_file,
	.get_file_size = fio_pmemblk_get_file_size,
	.unlink_file = fio_pmemblk_unlink_file,
	.flags = FIO_SYNCIO | FIO_DISKLESSIO | FIO_NOEXTEND | FIO_NODISKUTIL,
};

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

static void fio_exit fio_pmemblk_unregister(void)
{
	unregister_ioengine(&ioengine);
}
Commit	Line	Data
	1	/*
	2	* pmemblk: IO engine that uses NVML libpmemblk to read and write data
	3	*
	4	* Copyright (C) 2016 Hewlett Packard Enterprise Development LP
	5	*
	6	* This program is free software; you can redistribute it and/or
	7	* modify it under the terms of the GNU General Public License,
	8	* version 2 as published by the Free Software Foundation..
	9	*
	10	* This program is distributed in the hope that it will be useful,
	11	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	12	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	13	* GNU General Public License for more details.
	14	*
	15	* You should have received a copy of the GNU General Public
	16	* License along with this program; if not, write to the Free
	17	* Software Foundation, Inc., 59 Temple Place, Suite 330,
	18	* Boston, MA 02111-1307 USA
	19	*/
	20
	21	/*
	22	* pmemblk engine
	23	*
	24	* IO engine that uses libpmemblk to read and write data
	25	*
	26	* To use:
	27	* ioengine=pmemblk
	28	*
	29	* Other relevant settings:
	30	* iodepth=1
	31	* direct=1
	32	* thread=1 REQUIRED
	33	* unlink=1
	34	* filename=/pmem0/fiotestfile,BSIZE,FSIZEMB
	35	*
	36	* thread must be set to 1 for pmemblk as multiple processes cannot
	37	* open the same block pool file.
	38	*
	39	* iodepth should be set to 1 as pmemblk is always synchronous.
	40	* Use numjobs to scale up.
	41	*
	42	* direct=1 is implied as pmemblk is always direct.
	43	*
	44	* Can set unlink to 1 to remove the block pool file after testing.
	45	*
	46	* When specifying the filename, if the block pool file does not already
	47	* exist, then the pmemblk engine can create the pool file if you specify
	48	* the block and file sizes. BSIZE is the block size in bytes.
	49	* FSIZEMB is the pool file size in MB.
	50	*
	51	* See examples/pmemblk.fio for more.
	52	*
	53	*/
	54
	55	#include <stdio.h>
	56	#include <stdlib.h>
	57	#include <unistd.h>
	58	#include <sys/uio.h>
	59	#include <errno.h>
	60	#include <assert.h>
	61	#include <string.h>
	62	#include <libpmem.h>
	63	#include <libpmemblk.h>
	64
	65	#include "../fio.h"
	66
	67	/*
	68	* libpmemblk
	69	*/
	70	typedef struct fio_pmemblk_file *fio_pmemblk_file_t;
	71
	72	struct fio_pmemblk_file {
	73	fio_pmemblk_file_t pmb_next;
	74	char *pmb_filename;
	75	uint64_t pmb_refcnt;
	76	PMEMblkpool *pmb_pool;
	77	size_t pmb_bsize;
	78	size_t pmb_nblocks;
	79	};
	80	#define FIOFILEPMBSET(_f, _v) do { \
	81	(_f)->engine_data = (uint64_t)(uintptr_t)(_v); \
	82	} while(0)
	83	#define FIOFILEPMBGET(_f) ((fio_pmemblk_file_t)((_f)->engine_data))
	84
	85	static fio_pmemblk_file_t Cache;
	86
	87	static pthread_mutex_t CacheLock = PTHREAD_MUTEX_INITIALIZER;
	88
	89	#define PMB_CREATE (0x0001) /* should create file */
	90
	91	fio_pmemblk_file_t fio_pmemblk_cache_lookup(const char *filename)
	92	{
	93	fio_pmemblk_file_t i;
	94
	95	for (i = Cache; i != NULL; i = i->pmb_next)
	96	if (!strcmp(filename, i->pmb_filename))
	97	return i;
	98
	99	return NULL;
	100	}
	101
	102	static void fio_pmemblk_cache_insert(fio_pmemblk_file_t pmb)
	103	{
	104	pmb->pmb_next = Cache;
	105	Cache = pmb;
	106	}
	107
	108	static void fio_pmemblk_cache_remove(fio_pmemblk_file_t pmb)
	109	{
	110	fio_pmemblk_file_t i;
	111
	112	if (pmb == Cache) {
	113	Cache = Cache->pmb_next;
	114	pmb->pmb_next = NULL;
	115	return;
	116	}
	117
	118	for (i = Cache; i != NULL; i = i->pmb_next)
	119	if (pmb == i->pmb_next) {
	120	i->pmb_next = i->pmb_next->pmb_next;
	121	pmb->pmb_next = NULL;
	122	return;
	123	}
	124	}
	125
	126	/*
	127	* to control block size and gross file size at the libpmemblk
	128	* level, we allow the block size and file size to be appended
	129	* to the file name:
	130	*
	131	* path[,bsize,fsizemb]
	132	*
	133	* note that we do not use the fio option "filesize" to dictate
	134	* the file size because we can only give libpmemblk the gross
	135	* file size, which is different from the net or usable file
	136	* size (which is probably what fio wants).
	137	*
	138	* the final path without the parameters is returned in ppath.
	139	* the block size and file size are returned in pbsize and fsize.
	140	*
	141	* note that the user should specify the file size in MiB, but
	142	* we return bytes from here.
	143	*/
	144	static void pmb_parse_path(const char pathspec, char ppath, uint64_t pbsize,
	145	uint64_t *pfsize)
	146	{
	147	char *path;
	148	char *s;
	149	uint64_t bsize;
	150	uint64_t fsizemb;
	151
	152	path = strdup(pathspec);
	153	if (!path) {
	154	*ppath = NULL;
	155	return;
	156	}
	157
	158	/* extract sizes, if given */
	159	s = strrchr(path, ',');
	160	if (s && (fsizemb = strtoull(s + 1, NULL, 10))) {
	161	*s = 0;
	162	s = strrchr(path, ',');
	163	if (s && (bsize = strtoull(s + 1, NULL, 10))) {
	164	*s = 0;
	165	*ppath = path;
	166	*pbsize = bsize;
	167	*pfsize = fsizemb << 20;
	168	return;
	169	}
	170	}
	171
	172	/* size specs not found */
	173	strcpy(path, pathspec);
	174	*ppath = path;
	175	*pbsize = 0;
	176	*pfsize = 0;
	177	}
	178
	179	static fio_pmemblk_file_t pmb_open(const char *pathspec, int flags)
	180	{
	181	fio_pmemblk_file_t pmb;
	182	char *path = NULL;
	183	uint64_t bsize = 0;
	184	uint64_t fsize = 0;
	185
	186	pmb_parse_path(pathspec, &path, &bsize, &fsize);
	187	if (!path)
	188	return NULL;
	189
	190	pthread_mutex_lock(&CacheLock);
	191
	192	pmb = fio_pmemblk_cache_lookup(path);
	193	if (!pmb) {
	194	pmb = malloc(sizeof(*pmb));
	195	if (!pmb)
	196	goto error;
	197
	198	/* try opening existing first, create it if needed */
	199	pmb->pmb_pool = pmemblk_open(path, bsize);
	200	if (!pmb->pmb_pool && (errno == ENOENT) &&
	201	(flags & PMB_CREATE) && (0 < fsize) && (0 < bsize)) {
	202	pmb->pmb_pool =
	203	pmemblk_create(path, bsize, fsize, 0644);
	204	}
	205	if (!pmb->pmb_pool) {
	206	log_err
	207	("fio: enable to open pmemblk pool file (errno %d)\n",
	208	errno);
	209	goto error;
	210	}
	211
	212	pmb->pmb_filename = path;
	213	pmb->pmb_next = NULL;
	214	pmb->pmb_refcnt = 0;
	215	pmb->pmb_bsize = pmemblk_bsize(pmb->pmb_pool);
	216	pmb->pmb_nblocks = pmemblk_nblock(pmb->pmb_pool);
	217
	218	fio_pmemblk_cache_insert(pmb);
	219	}
	220
	221	pmb->pmb_refcnt += 1;
	222
	223	pthread_mutex_unlock(&CacheLock);
	224
	225	return pmb;
	226
	227	error:
	228	if (pmb) {
	229	if (pmb->pmb_pool)
	230	pmemblk_close(pmb->pmb_pool);
	231	pmb->pmb_pool = NULL;
	232	pmb->pmb_filename = NULL;
	233	free(pmb);
	234	}
	235	if (path)
	236	free(path);
	237
	238	pthread_mutex_unlock(&CacheLock);
	239	return NULL;
	240	}
	241
	242	static void pmb_close(fio_pmemblk_file_t pmb, const bool keep)
	243	{
	244	pthread_mutex_lock(&CacheLock);
	245
	246	pmb->pmb_refcnt--;
	247
	248	if (!keep && !pmb->pmb_refcnt) {
	249	pmemblk_close(pmb->pmb_pool);
	250	pmb->pmb_pool = NULL;
	251	free(pmb->pmb_filename);
	252	pmb->pmb_filename = NULL;
	253	fio_pmemblk_cache_remove(pmb);
	254	free(pmb);
	255	}
	256
	257	pthread_mutex_unlock(&CacheLock);
	258	}
	259
	260	static int pmb_get_flags(struct thread_data td, uint64_t pflags)
	261	{
	262	static int thread_warned = 0;
	263	static int odirect_warned = 0;
	264
	265	uint64_t flags = 0;
	266
	267	if (!td->o.use_thread) {
	268	if (!thread_warned) {
	269	thread_warned = 1;
	270	log_err("fio: must set thread=1 for pmemblk engine\n");
	271	}
	272	return 1;
	273	}
	274
	275	if (!td->o.odirect && !odirect_warned) {
	276	odirect_warned = 1;
	277	log_info("fio: direct == 0, but pmemblk is always direct\n");
	278	}
	279
	280	if (td->o.allow_create)
	281	flags \|= PMB_CREATE;
	282
	283	(*pflags) = flags;
	284	return 0;
	285	}
	286
	287	static int fio_pmemblk_open_file(struct thread_data td, struct fio_file f)
	288	{
	289	uint64_t flags = 0;
	290	fio_pmemblk_file_t pmb;
	291
	292	if (pmb_get_flags(td, &flags))
	293	return 1;
	294
	295	pmb = pmb_open(f->file_name, flags);
	296	if (!pmb)
	297	return 1;
	298
	299	FIOFILEPMBSET(f, pmb);
	300	return 0;
	301	}
	302
	303	static int fio_pmemblk_close_file(struct thread_data fio_unused *td,
	304	struct fio_file *f)
	305	{
	306	fio_pmemblk_file_t pmb = FIOFILEPMBGET(f);
	307
	308	if (pmb)
	309	pmb_close(pmb, false);
	310
	311	FIOFILEPMBSET(f, NULL);
	312	return 0;
	313	}
	314
	315	static int fio_pmemblk_get_file_size(struct thread_data td, struct fio_file f)
	316	{
	317	uint64_t flags = 0;
	318	fio_pmemblk_file_t pmb = FIOFILEPMBGET(f);
	319
	320	if (fio_file_size_known(f))
	321	return 0;
	322
	323	if (!pmb) {
	324	if (pmb_get_flags(td, &flags))
	325	return 1;
	326	pmb = pmb_open(f->file_name, flags);
	327	if (!pmb)
	328	return 1;
	329	}
	330
	331	f->real_file_size = pmb->pmb_bsize * pmb->pmb_nblocks;
	332
	333	fio_file_set_size_known(f);
	334
	335	if (!FIOFILEPMBGET(f))
	336	pmb_close(pmb, true);
	337
	338	return 0;
	339	}
	340
	341	static int fio_pmemblk_queue(struct thread_data td, struct io_u io_u)
	342	{
	343	struct fio_file *f = io_u->file;
	344	fio_pmemblk_file_t pmb = FIOFILEPMBGET(f);
	345
	346	unsigned long long off;
	347	unsigned long len;
	348	void *buf;
	349
	350	fio_ro_check(td, io_u);
	351
	352	switch (io_u->ddir) {
	353	case DDIR_READ:
	354	case DDIR_WRITE:
	355	off = io_u->offset;
	356	len = io_u->xfer_buflen;
	357
	358	io_u->error = EINVAL;
	359	if (off % pmb->pmb_bsize)
	360	break;
	361	if (len % pmb->pmb_bsize)
	362	break;
	363	if ((off + len) / pmb->pmb_bsize > pmb->pmb_nblocks)
	364	break;
	365
	366	io_u->error = 0;
	367	buf = io_u->xfer_buf;
	368	off /= pmb->pmb_bsize;
	369	len /= pmb->pmb_bsize;
	370	while (0 < len) {
	371	if (io_u->ddir == DDIR_READ &&
	372	0 != pmemblk_read(pmb->pmb_pool, buf, off)) {
	373	io_u->error = errno;
	374	break;
	375	} else if (0 != pmemblk_write(pmb->pmb_pool, buf, off)) {
	376	io_u->error = errno;
	377	break;
	378	}
	379	buf += pmb->pmb_bsize;
	380	off++;
	381	len--;
	382	}
	383	off *= pmb->pmb_bsize;
	384	len *= pmb->pmb_bsize;
	385	io_u->resid = io_u->xfer_buflen - (off - io_u->offset);
	386	break;
	387	case DDIR_SYNC:
	388	case DDIR_DATASYNC:
	389	case DDIR_SYNC_FILE_RANGE:
	390	/* we're always sync'd */
	391	io_u->error = 0;
	392	break;
	393	default:
	394	io_u->error = EINVAL;
	395	break;
	396	}
	397
	398	return FIO_Q_COMPLETED;
	399	}
	400
	401	static int fio_pmemblk_unlink_file(struct thread_data td, struct fio_file f)
	402	{
	403	char *path = NULL;
	404	uint64_t bsize = 0;
	405	uint64_t fsize = 0;
	406
	407	/*
	408	* we need our own unlink in case the user has specified
	409	* the block and file sizes in the path name. we parse
	410	* the file_name to determine the file name we actually used.
	411	*/
	412
	413	pmb_parse_path(f->file_name, &path, &bsize, &fsize);
	414	if (!path)
	415	return ENOENT;
	416
	417	unlink(path);
	418	free(path);
	419	return 0;
	420	}
	421
	422	static struct ioengine_ops ioengine = {
	423	.name = "pmemblk",
	424	.version = FIO_IOOPS_VERSION,
	425	.queue = fio_pmemblk_queue,
	426	.open_file = fio_pmemblk_open_file,
	427	.close_file = fio_pmemblk_close_file,
	428	.get_file_size = fio_pmemblk_get_file_size,
	429	.unlink_file = fio_pmemblk_unlink_file,
	430	.flags = FIO_SYNCIO \| FIO_DISKLESSIO \| FIO_NOEXTEND \| FIO_NODISKUTIL,
	431	};
	432
	433	static void fio_init fio_pmemblk_register(void)
	434	{
	435	register_ioengine(&ioengine);
	436	}
	437
	438	static void fio_exit fio_pmemblk_unregister(void)
	439	{
	440	unregister_ioengine(&ioengine);
	441	}