[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 build as an external engine:
 *   Copy pmemblk.c to the engines/ directory of the fio source code, then:
 *   cd .../engines/
 *   gcc -Wall -shared -rdynamic -fPIC -o pmemblk.o -D_GNU_SOURCE pmemblk.c
 *
 * To use as an external engine:
 *   ioengine=/full/path/to/pmemblk.o
 *
 * 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.
 *
 * libpmemblk.so
 *   By default, the pmemblk engine will let the system find the libpmemblk.so
 *   that it uses.  You can use an alternative libpmemblk by setting the
 *   FIO_PMEMBLK_LIB environment variable to the full path to the desired
 *   libpmemblk.so.
 *
 */

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

#include "../fio.h"


/*
 * libpmemblk
 */
struct PMEMblkpool_s;
typedef struct PMEMblkpool_s PMEMblkpool;

PMEMblkpool* (*pmemblk_create_ptr)(const char*, size_t, size_t, mode_t) = NULL;
PMEMblkpool* (*pmemblk_open_ptr)(const char*, size_t) = NULL;
void (*pmemblk_close_ptr)(PMEMblkpool*) = NULL;
size_t (*pmemblk_nblock_ptr)(PMEMblkpool*) = NULL;
size_t (*pmemblk_bsize_ptr)(PMEMblkpool*) = NULL;
int (*pmemblk_read_ptr)(PMEMblkpool*, void*, off_t) = NULL;
int (*pmemblk_write_ptr)(PMEMblkpool*, const void*, off_t) = NULL;

int
load_libpmemblk(
	const char* path
)
{
	void*   dl;

	if (NULL == path)
		path = "libpmemblk.so";

	dl = dlopen(path, RTLD_NOW | RTLD_NODELETE);
	if (NULL == dl)
		goto errorout;

	if (NULL == (pmemblk_create_ptr = dlsym(dl, "pmemblk_create")))
		goto errorout;
	if (NULL == (pmemblk_open_ptr = dlsym(dl, "pmemblk_open")))
		goto errorout;
	if (NULL == (pmemblk_close_ptr = dlsym(dl, "pmemblk_close")))
		goto errorout;
	if (NULL == (pmemblk_nblock_ptr = dlsym(dl, "pmemblk_nblock")))
		goto errorout;
	if (NULL == (pmemblk_bsize_ptr = dlsym(dl, "pmemblk_bsize")))
		goto errorout;
	if (NULL == (pmemblk_read_ptr = dlsym(dl, "pmemblk_read")))
		goto errorout;
	if (NULL == (pmemblk_write_ptr = dlsym(dl, "pmemblk_write")))
		goto errorout;
	
	return 0;

errorout:
	log_err("fio: unable to load libpmemblk: %s\n", dlerror());
	if (NULL != dl)
		dlclose(dl);

	return (-1);
	
}  /* load_libpmemblk() */

#define pmemblk_create   pmemblk_create_ptr
#define pmemblk_open     pmemblk_open_ptr
#define pmemblk_close    pmemblk_close_ptr
#define pmemblk_nblock   pmemblk_nblock_ptr
#define pmemblk_bsize    pmemblk_bsize_ptr
#define pmemblk_read     pmemblk_read_ptr
#define pmemblk_write    pmemblk_write_ptr


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 = NULL;

static pthread_mutex_t      CacheLock = PTHREAD_MUTEX_INITIALIZER;
#define CACHE_LOCK()  \
	(void)pthread_mutex_lock(&CacheLock)
#define CACHE_UNLOCK()  \
	(void)pthread_mutex_unlock(&CacheLock)

#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 (0 == strcmp(filename, i->pmb_filename))
			return i;

	return NULL;

}  /* fio_pmemblk_cache_lookup() */


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

	return;

}  /* fio_pmemblk_cache_insert() */


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;
		}

	return;

}  /* fio_pmemblk_cache_remove() */


/*
 * 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 (NULL == 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;
	return;

}  /* pmb_parse_path() */


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 (NULL == path)
		return NULL;

	CACHE_LOCK();

	pmb = fio_pmemblk_cache_lookup(path);

	if (NULL == pmb) {
		/* load libpmemblk if needed */
		if (NULL == pmemblk_open)
			if (0 != load_libpmemblk(getenv("FIO_PMEMBLK_LIB")))
				goto error;

		pmb = malloc(sizeof(*pmb));
		if (NULL == pmb)
			goto error;

		/* try opening existing first, create it if needed */
		pmb->pmb_pool = pmemblk_open(path, bsize);
		if ((NULL == pmb->pmb_pool) &&
		    (ENOENT == errno) &&
		    (flags & PMB_CREATE) &&
		    (0 < fsize) &&
		    (0 < bsize)) {
			pmb->pmb_pool = pmemblk_create(path, bsize, fsize, 0644);
		}
		if (NULL == 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;

	CACHE_UNLOCK();
	
	return pmb;

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

}  /* pmb_open() */


static void
pmb_close(
	fio_pmemblk_file_t pmb,
	const int          keep
)
{
	CACHE_LOCK();

	pmb->pmb_refcnt--;

	if (!keep && (0 == 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);
	}

	CACHE_UNLOCK();

}  /* pmb_close() */


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;

}  /* pmb_get_flags() */


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

	if (0 != pmb_get_flags(td, &flags))
		return 1;

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

	FIOFILEPMBSET(f, pmb);

	return 0;

}  /* fio_pmemblk_open_file() */


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, 0);

	FIOFILEPMBSET(f, NULL);

	return 0;

}  /* fio_pmemblk_close_file() */


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 (NULL == pmb) {
		if (0 != pmb_get_flags(td, &flags))
			return 1;
		pmb = pmb_open(f->file_name, flags);
		if (NULL == pmb)
			return 1;
	}

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

	fio_file_set_size_known(f);

	if (NULL == FIOFILEPMBGET(f))
		pmb_close(pmb, 1);

	return 0;

}  /* fio_pmemblk_get_file_size() */


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;
	int (*blkop)(PMEMblkpool*, void*, off_t) = (void*)pmemblk_write;

	fio_ro_check(td, io_u);

	switch (io_u->ddir) {
	case DDIR_READ:
		blkop = pmemblk_read;
		/* fall through */
	case DDIR_WRITE:
		off = io_u->offset;
		len = io_u->xfer_buflen;

		io_u->error = EINVAL;
		if (0 != (off % pmb->pmb_bsize))
			break;
		if (0 != (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 (0 != blkop(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;

}  /* fio_pmemblk_queue() */


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 (NULL == path)
		return 1;

	unlink(path);
	free(path);

	return 0;

}  /* fio_pmemblk_unlink_file() */


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	/*
	23	* pmemblk engine
	24	*
	25	* IO engine that uses libpmemblk to read and write data
	26	*
	27	* To build as an external engine:
	28	* Copy pmemblk.c to the engines/ directory of the fio source code, then:
	29	* cd .../engines/
	30	* gcc -Wall -shared -rdynamic -fPIC -o pmemblk.o -D_GNU_SOURCE pmemblk.c
	31	*
	32	* To use as an external engine:
	33	* ioengine=/full/path/to/pmemblk.o
	34	*
	35	* Other relevant settings:
	36	* iodepth=1
	37	* direct=1
	38	* thread=1 REQUIRED
	39	* unlink=1
	40	* filename=/pmem0/fiotestfile,BSIZE,FSIZEMB
	41	*
	42	* thread must be set to 1 for pmemblk as multiple processes cannot
	43	* open the same block pool file.
	44	*
	45	* iodepth should be set to 1 as pmemblk is always synchronous.
	46	* Use numjobs to scale up.
	47	*
	48	* direct=1 is implied as pmemblk is always direct.
	49	*
	50	* Can set unlink to 1 to remove the block pool file after testing.
	51	*
	52	* When specifying the filename, if the block pool file does not already
	53	* exist, then the pmemblk engine can create the pool file if you specify
	54	* the block and file sizes. BSIZE is the block size in bytes.
	55	* FSIZEMB is the pool file size in MB.
	56	*
	57	* libpmemblk.so
	58	* By default, the pmemblk engine will let the system find the libpmemblk.so
	59	* that it uses. You can use an alternative libpmemblk by setting the
	60	* FIO_PMEMBLK_LIB environment variable to the full path to the desired
	61	* libpmemblk.so.
	62	*
	63	*/
	64
	65	#include <stdio.h>
	66	#include <stdlib.h>
	67	#include <unistd.h>
	68	#include <sys/uio.h>
	69	#include <errno.h>
	70	#include <assert.h>
	71	#include <dlfcn.h>
	72	#include <string.h>
	73
	74	#include "../fio.h"
	75
	76
	77
	78	/*
	79	* libpmemblk
	80	*/
	81	struct PMEMblkpool_s;
	82	typedef struct PMEMblkpool_s PMEMblkpool;
	83
	84	PMEMblkpool* (pmemblk_create_ptr)(const char, size_t, size_t, mode_t) = NULL;
	85	PMEMblkpool* (pmemblk_open_ptr)(const char, size_t) = NULL;
	86	void (pmemblk_close_ptr)(PMEMblkpool) = NULL;
	87	size_t (pmemblk_nblock_ptr)(PMEMblkpool) = NULL;
	88	size_t (pmemblk_bsize_ptr)(PMEMblkpool) = NULL;
	89	int (pmemblk_read_ptr)(PMEMblkpool, void*, off_t) = NULL;
	90	int (pmemblk_write_ptr)(PMEMblkpool, const void*, off_t) = NULL;
	91
	92	int
	93	load_libpmemblk(
	94	const char* path
	95	)
	96	{
	97	void* dl;
	98
	99	if (NULL == path)
	100	path = "libpmemblk.so";
	101
	102	dl = dlopen(path, RTLD_NOW \| RTLD_NODELETE);
	103	if (NULL == dl)
	104	goto errorout;
	105
	106	if (NULL == (pmemblk_create_ptr = dlsym(dl, "pmemblk_create")))
	107	goto errorout;
	108	if (NULL == (pmemblk_open_ptr = dlsym(dl, "pmemblk_open")))
	109	goto errorout;
	110	if (NULL == (pmemblk_close_ptr = dlsym(dl, "pmemblk_close")))
	111	goto errorout;
	112	if (NULL == (pmemblk_nblock_ptr = dlsym(dl, "pmemblk_nblock")))
	113	goto errorout;
	114	if (NULL == (pmemblk_bsize_ptr = dlsym(dl, "pmemblk_bsize")))
	115	goto errorout;
	116	if (NULL == (pmemblk_read_ptr = dlsym(dl, "pmemblk_read")))
	117	goto errorout;
	118	if (NULL == (pmemblk_write_ptr = dlsym(dl, "pmemblk_write")))
	119	goto errorout;
	120
	121	return 0;
	122
	123	errorout:
	124	log_err("fio: unable to load libpmemblk: %s\n", dlerror());
	125	if (NULL != dl)
	126	dlclose(dl);
	127
	128	return (-1);
	129
	130	} /* load_libpmemblk() */
	131
	132	#define pmemblk_create pmemblk_create_ptr
	133	#define pmemblk_open pmemblk_open_ptr
	134	#define pmemblk_close pmemblk_close_ptr
	135	#define pmemblk_nblock pmemblk_nblock_ptr
	136	#define pmemblk_bsize pmemblk_bsize_ptr
	137	#define pmemblk_read pmemblk_read_ptr
	138	#define pmemblk_write pmemblk_write_ptr
	139
	140
	141	typedef struct fio_pmemblk_file* fio_pmemblk_file_t;
	142	struct fio_pmemblk_file {
	143	fio_pmemblk_file_t pmb_next;
	144	char* pmb_filename;
	145	uint64_t pmb_refcnt;
	146	PMEMblkpool* pmb_pool;
	147	size_t pmb_bsize;
	148	size_t pmb_nblocks;
	149	};
	150	#define FIOFILEPMBSET(_f, _v) do { \
	151	(_f)->engine_data = (uint64_t)(uintptr_t)(_v); \
	152	} while(0)
	153	#define FIOFILEPMBGET(_f) ((fio_pmemblk_file_t)((_f)->engine_data))
	154
	155	static fio_pmemblk_file_t Cache = NULL;
	156
	157	static pthread_mutex_t CacheLock = PTHREAD_MUTEX_INITIALIZER;
	158	#define CACHE_LOCK() \
	159	(void)pthread_mutex_lock(&CacheLock)
	160	#define CACHE_UNLOCK() \
	161	(void)pthread_mutex_unlock(&CacheLock)
	162
	163	#define PMB_CREATE (0x0001) /* should create file */
	164
	165
	166	fio_pmemblk_file_t
	167	fio_pmemblk_cache_lookup(
	168	const char* filename
	169	)
	170	{
	171	fio_pmemblk_file_t i;
	172
	173	for (i = Cache; i != NULL; i = i->pmb_next)
	174	if (0 == strcmp(filename, i->pmb_filename))
	175	return i;
	176
	177	return NULL;
	178
	179	} /* fio_pmemblk_cache_lookup() */
	180
	181
	182	static void
	183	fio_pmemblk_cache_insert(
	184	fio_pmemblk_file_t pmb
	185	)
	186	{
	187	pmb->pmb_next = Cache;
	188	Cache = pmb;
	189
	190	return;
	191
	192	} /* fio_pmemblk_cache_insert() */
	193
	194
	195	static void
	196	fio_pmemblk_cache_remove(
	197	fio_pmemblk_file_t pmb
	198	)
	199	{
	200	fio_pmemblk_file_t i;
	201
	202	if (pmb == Cache) {
	203	Cache = Cache->pmb_next;
	204	pmb->pmb_next = NULL;
	205	return;
	206	}
	207
	208	for (i = Cache; i != NULL; i = i->pmb_next)
	209	if (pmb == i->pmb_next) {
	210	i->pmb_next = i->pmb_next->pmb_next;
	211	pmb->pmb_next = NULL;
	212	return;
	213	}
	214
	215	return;
	216
	217	} /* fio_pmemblk_cache_remove() */
	218
	219
	220	/*
	221	* to control block size and gross file size at the libpmemblk
	222	* level, we allow the block size and file size to be appended
	223	* to the file name:
	224	*
	225	* path[,bsize,fsizemb]
	226	*
	227	* note that we do not use the fio option "filesize" to dictate
	228	* the file size because we can only give libpmemblk the gross
	229	* file size, which is different from the net or usable file
	230	* size (which is probably what fio wants).
	231	*
	232	* the final path without the parameters is returned in ppath.
	233	* the block size and file size are returned in pbsize and fsize.
	234	*
	235	* note that the user should specify the file size in MiB, but
	236	* we return bytes from here.
	237	*/
	238	static void
	239	pmb_parse_path(
	240	const char* pathspec,
	241	char** ppath,
	242	uint64_t* pbsize,
	243	uint64_t* pfsize
	244	)
	245	{
	246	char* path;
	247	char* s;
	248	uint64_t bsize;
	249	uint64_t fsizemb;
	250
	251	path = strdup(pathspec);
	252	if (NULL == path) {
	253	*ppath = NULL;
	254	return;
	255	}
	256
	257	/* extract sizes, if given */
	258	s = strrchr(path, ',');
	259	if (s && (fsizemb = strtoull(s+1, NULL, 10))) {
	260	*s = 0;
	261	s = strrchr(path, ',');
	262	if (s && (bsize = strtoull(s+1, NULL, 10))) {
	263	*s = 0;
	264	*ppath = path;
	265	*pbsize = bsize;
	266	*pfsize = fsizemb << 20;
	267	return;
	268	}
	269	}
	270
	271	/* size specs not found */
	272	strcpy(path, pathspec);
	273	*ppath = path;
	274	*pbsize = 0;
	275	*pfsize = 0;
	276	return;
	277
	278	} /* pmb_parse_path() */
	279
	280
	281	static
	282	fio_pmemblk_file_t
	283	pmb_open(
	284	const char* pathspec,
	285	int flags
	286	)
	287	{
	288	fio_pmemblk_file_t pmb;
	289	char* path = NULL;
	290	uint64_t bsize = 0;
	291	uint64_t fsize = 0;
	292
	293	pmb_parse_path(pathspec, &path, &bsize, &fsize);
	294	if (NULL == path)
	295	return NULL;
	296
	297	CACHE_LOCK();
	298
	299	pmb = fio_pmemblk_cache_lookup(path);
	300
	301	if (NULL == pmb) {
	302	/* load libpmemblk if needed */
	303	if (NULL == pmemblk_open)
	304	if (0 != load_libpmemblk(getenv("FIO_PMEMBLK_LIB")))
	305	goto error;
	306
	307	pmb = malloc(sizeof(*pmb));
	308	if (NULL == pmb)
	309	goto error;
	310
	311	/* try opening existing first, create it if needed */
	312	pmb->pmb_pool = pmemblk_open(path, bsize);
	313	if ((NULL == pmb->pmb_pool) &&
	314	(ENOENT == errno) &&
	315	(flags & PMB_CREATE) &&
	316	(0 < fsize) &&
	317	(0 < bsize)) {
	318	pmb->pmb_pool = pmemblk_create(path, bsize, fsize, 0644);
	319	}
	320	if (NULL == pmb->pmb_pool) {
	321	log_err("fio: enable to open pmemblk pool file (errno %d)\n",
	322	errno);
	323	goto error;
	324	}
	325
	326	pmb->pmb_filename = path;
	327	pmb->pmb_next = NULL;
	328	pmb->pmb_refcnt = 0;
	329	pmb->pmb_bsize = pmemblk_bsize(pmb->pmb_pool);
	330	pmb->pmb_nblocks = pmemblk_nblock(pmb->pmb_pool);
	331
	332	fio_pmemblk_cache_insert(pmb);
	333	}
	334
	335	pmb->pmb_refcnt += 1;
	336
	337	CACHE_UNLOCK();
	338
	339	return pmb;
	340
	341	error:
	342	if (NULL != pmb) {
	343	if (NULL != pmb->pmb_pool)
	344	pmemblk_close(pmb->pmb_pool);
	345	pmb->pmb_pool = NULL;
	346	pmb->pmb_filename = NULL;
	347	free(pmb);
	348	}
	349	if (NULL != path)
	350	free(path);
	351	CACHE_UNLOCK();
	352	return NULL;
	353
	354	} /* pmb_open() */
	355
	356
	357	static void
	358	pmb_close(
	359	fio_pmemblk_file_t pmb,
	360	const int keep
	361	)
	362	{
	363	CACHE_LOCK();
	364
	365	pmb->pmb_refcnt--;
	366
	367	if (!keep && (0 == pmb->pmb_refcnt)) {
	368	pmemblk_close(pmb->pmb_pool);
	369	pmb->pmb_pool = NULL;
	370	free(pmb->pmb_filename);
	371	pmb->pmb_filename = NULL;
	372	fio_pmemblk_cache_remove(pmb);
	373	free(pmb);
	374	}
	375
	376	CACHE_UNLOCK();
	377
	378	} /* pmb_close() */
	379
	380
	381	static int
	382	pmb_get_flags(
	383	struct thread_data* td,
	384	uint64_t* pflags
	385	)
	386	{
	387	static int thread_warned = 0;
	388	static int odirect_warned = 0;
	389
	390	uint64_t flags = 0;
	391
	392	if (!td->o.use_thread) {
	393	if (!thread_warned) {
	394	thread_warned = 1;
	395	log_err("fio: must set thread=1 for pmemblk engine\n");
	396	}
	397	return 1;
	398	}
	399
	400	if (!td->o.odirect && !odirect_warned) {
	401	odirect_warned = 1;
	402	log_info("fio: direct == 0, but pmemblk is always direct\n");
	403	}
	404
	405	if (td->o.allow_create)
	406	flags \|= PMB_CREATE;
	407
	408	(*pflags) = flags;
	409	return 0;
	410
	411	} /* pmb_get_flags() */
	412
	413
	414	static int
	415	fio_pmemblk_open_file(
	416	struct thread_data* td,
	417	struct fio_file* f)
	418	{
	419	uint64_t flags = 0;
	420	fio_pmemblk_file_t pmb;
	421
	422	if (0 != pmb_get_flags(td, &flags))
	423	return 1;
	424
	425	pmb = pmb_open(f->file_name, flags);
	426	if (NULL == pmb)
	427	return 1;
	428
	429	FIOFILEPMBSET(f, pmb);
	430
	431	return 0;
	432
	433	} /* fio_pmemblk_open_file() */
	434
	435
	436	static int
	437	fio_pmemblk_close_file(
	438	struct thread_data fio_unused* td,
	439	struct fio_file* f
	440	)
	441	{
	442	fio_pmemblk_file_t pmb = FIOFILEPMBGET(f);
	443
	444	if (pmb)
	445	pmb_close(pmb, 0);
	446
	447	FIOFILEPMBSET(f, NULL);
	448
	449	return 0;
	450
	451	} /* fio_pmemblk_close_file() */
	452
	453
	454	static int
	455	fio_pmemblk_get_file_size(
	456	struct thread_data* td,
	457	struct fio_file* f
	458	)
	459	{
	460	uint64_t flags = 0;
	461	fio_pmemblk_file_t pmb = FIOFILEPMBGET(f);
	462
	463	if (fio_file_size_known(f))
	464	return 0;
	465
	466	if (NULL == pmb) {
	467	if (0 != pmb_get_flags(td, &flags))
	468	return 1;
	469	pmb = pmb_open(f->file_name, flags);
	470	if (NULL == pmb)
	471	return 1;
	472	}
	473
	474	f->real_file_size = pmb->pmb_bsize * pmb->pmb_nblocks;
	475
	476	fio_file_set_size_known(f);
	477
	478	if (NULL == FIOFILEPMBGET(f))
	479	pmb_close(pmb, 1);
	480
	481	return 0;
	482
	483	} /* fio_pmemblk_get_file_size() */
	484
	485
	486	static int
	487	fio_pmemblk_queue(
	488	struct thread_data* td,
	489	struct io_u* io_u)
	490	{
	491	struct fio_file* f = io_u->file;
	492	fio_pmemblk_file_t pmb = FIOFILEPMBGET(f);
	493
	494	unsigned long long off;
	495	unsigned long len;
	496	void* buf;
	497	int (blkop)(PMEMblkpool, void, off_t) = (void)pmemblk_write;
	498
	499	fio_ro_check(td, io_u);
	500
	501	switch (io_u->ddir) {
	502	case DDIR_READ:
	503	blkop = pmemblk_read;
	504	/* fall through */
	505	case DDIR_WRITE:
	506	off = io_u->offset;
	507	len = io_u->xfer_buflen;
	508
	509	io_u->error = EINVAL;
	510	if (0 != (off % pmb->pmb_bsize))
	511	break;
	512	if (0 != (len % pmb->pmb_bsize))
	513	break;
	514	if ((off + len) / pmb->pmb_bsize > pmb->pmb_nblocks)
	515	break;
	516
	517	io_u->error = 0;
	518	buf = io_u->xfer_buf;
	519	off /= pmb->pmb_bsize;
	520	len /= pmb->pmb_bsize;
	521	while (0 < len) {
	522	if (0 != blkop(pmb->pmb_pool, buf, off)) {
	523	io_u->error = errno;
	524	break;
	525	}
	526	buf += pmb->pmb_bsize;
	527	off++;
	528	len--;
	529	}
	530	off *= pmb->pmb_bsize;
	531	len *= pmb->pmb_bsize;
	532	io_u->resid = io_u->xfer_buflen - (off - io_u->offset);
	533	break;
	534	case DDIR_SYNC:
	535	case DDIR_DATASYNC:
	536	case DDIR_SYNC_FILE_RANGE:
	537	/* we're always sync'd */
	538	io_u->error = 0;
	539	break;
	540	default:
	541	io_u->error = EINVAL;
	542	break;
	543	}
	544
	545	return FIO_Q_COMPLETED;
	546
	547	} /* fio_pmemblk_queue() */
	548
	549
	550	static int
	551	fio_pmemblk_unlink_file(
	552	struct thread_data* td,
	553	struct fio_file* f
	554	)
	555	{
	556	char* path = NULL;
	557	uint64_t bsize = 0;
	558	uint64_t fsize = 0;
	559
	560	/*
	561	* we need our own unlink in case the user has specified
	562	* the block and file sizes in the path name. we parse
	563	* the file_name to determine the file name we actually used.
	564	*/
	565
	566	pmb_parse_path(f->file_name, &path, &bsize, &fsize);
	567	if (NULL == path)
	568	return 1;
	569
	570	unlink(path);
	571	free(path);
	572
	573	return 0;
	574
	575	} /* fio_pmemblk_unlink_file() */
	576
	577
	578	struct ioengine_ops ioengine = {
	579	.name = "pmemblk",
	580	.version = FIO_IOOPS_VERSION,
	581	.queue = fio_pmemblk_queue,
	582	.open_file = fio_pmemblk_open_file,
	583	.close_file = fio_pmemblk_close_file,
	584	.get_file_size = fio_pmemblk_get_file_size,
	585	.unlink_file = fio_pmemblk_unlink_file,
	586	.flags = FIO_SYNCIO \| FIO_DISKLESSIO \| FIO_NOEXTEND \| FIO_NODISKUTIL,
	587	};
	588
	589
	590	static void
	591	fio_init fio_pmemblk_register(void)
	592	{
	593	register_ioengine(&ioengine);
	594	}
	595
	596
	597	static void
	598	fio_exit fio_pmemblk_unregister(void)
	599	{
	600	unregister_ioengine(&ioengine);
	601	}
	602