obj.c

// SPDX-License-Identifier: BSD-3-Clause
/* Copyright 2014-2023, Intel Corporation */

/*
 * obj.c -- transactional object store implementation
 */
#include <inttypes.h>
#include <limits.h>
#include <wchar.h>
#include <stdbool.h>

#include "valgrind_internal.h"
#include "libpmem.h"
#include "memblock.h"
#include "critnib.h"
#include "list.h"
#include "mmap.h"
#include "obj.h"
#include "ctl_global.h"
#include "ravl.h"

#include "heap_layout.h"
#include "os.h"
#include "os_thread.h"
#include "pmemops.h"
#include "set.h"
#include "sync.h"
#include "tx.h"
#include "sys_util.h"

/*
 * The variable from which the config is directly loaded. The string
 * cannot contain any comments or extraneous white characters.
 */
#define OBJ_CONFIG_ENV_VARIABLE "PMEMOBJ_CONF"

/*
 * The variable that points to a config file from which the config is loaded.
 */
#define OBJ_CONFIG_FILE_ENV_VARIABLE "PMEMOBJ_CONF_FILE"

/*
 * The variable which overwrites a number of lanes available at runtime.
 */
#define OBJ_NLANES_ENV_VARIABLE "PMEMOBJ_NLANES"

#define OBJ_X_VALID_FLAGS PMEMOBJ_F_RELAXED

static const struct pool_attr Obj_create_attr = {
		OBJ_HDR_SIG,
		OBJ_FORMAT_MAJOR,
		OBJ_FORMAT_FEAT_DEFAULT,
		{0}, {0}, {0}, {0}, {0}
};

static const struct pool_attr Obj_open_attr = {
		OBJ_HDR_SIG,
		OBJ_FORMAT_MAJOR,
		OBJ_FORMAT_FEAT_CHECK,
		{0}, {0}, {0}, {0}, {0}
};

static struct critnib *pools_ht; /* hash table used for searching by UUID */
static struct critnib *pools_tree; /* tree used for searching by address */

int _pobj_cache_invalidate;
static os_mutex_t pools_mutex;

__thread struct _pobj_pcache _pobj_cached_pool;

/*
 * pmemobj_direct -- returns the direct pointer of an object
 */
void *
pmemobj_direct(PMEMoid oid)
{
	return pmemobj_direct_inline(oid);
}

/*
 * obj_ctl_init_and_load -- (static) initializes CTL and loads configuration
 *	from env variable and file
 */
static int
obj_ctl_init_and_load(PMEMobjpool *pop)
{
	LOG(3, "pop %p", pop);

	if (pop != NULL && (pop->ctl = ctl_new()) == NULL) {
		LOG(2, "!ctl_new");
		return -1;
	}

	if (pop) {
		tx_ctl_register(pop);
		pmalloc_ctl_register(pop);
		stats_ctl_register(pop);
		debug_ctl_register(pop);
	}

	char *env_config = os_getenv(OBJ_CONFIG_ENV_VARIABLE);
	if (env_config != NULL) {
		if (ctl_load_config_from_string(pop ? pop->ctl : NULL,
				pop, env_config) != 0) {
			LOG(2, "unable to parse config stored in %s "
				"environment variable",
				OBJ_CONFIG_ENV_VARIABLE);
			goto err;
		}
	}

	char *env_config_file = os_getenv(OBJ_CONFIG_FILE_ENV_VARIABLE);
	if (env_config_file != NULL && env_config_file[0] != '\0') {
		if (ctl_load_config_from_file(pop ? pop->ctl : NULL,
				pop, env_config_file) != 0) {
			LOG(2, "unable to parse config stored in %s "
				"file (from %s environment variable)",
				env_config_file,
				OBJ_CONFIG_FILE_ENV_VARIABLE);
			goto err;
		}
	}

	return 0;
err:
	if (pop)
		ctl_delete(pop->ctl);
	return -1;
}

/*
 * obj_pool_init -- (internal) allocate global structs holding all opened pools
 *
 * This is invoked on a first call to pmemobj_open() or pmemobj_create().
 * Memory is released in library destructor.
 *
 * This function needs to be threadsafe.
 */
static void
obj_pool_init(void)
{
	LOG(3, NULL);

	struct critnib *c;

	if (pools_ht == NULL) {
		c = critnib_new();
		if (c == NULL)
			FATAL("!critnib_new for pools_ht");
		if (!util_bool_compare_and_swap64(&pools_ht, NULL, c))
			critnib_delete(c);
	}

	if (pools_tree == NULL) {
		c = critnib_new();
		if (c == NULL)
			FATAL("!critnib_new for pools_tree");
		if (!util_bool_compare_and_swap64(&pools_tree, NULL, c))
			critnib_delete(c);
	}
}

/*
 * pmemobj_oid -- return a PMEMoid based on the virtual address
 *
 * If the address does not belong to any pool OID_NULL is returned.
 */
PMEMoid
pmemobj_oid(const void *addr)
{
	PMEMobjpool *pop = pmemobj_pool_by_ptr(addr);
	if (pop == NULL)
		return OID_NULL;

	PMEMoid oid = {pop->uuid_lo, (uintptr_t)addr - (uintptr_t)pop};
	return oid;
}

/*
 * obj_init -- initialization of obj
 *
 * Called by constructor.
 */
void
obj_init(void)
{
	LOG(3, NULL);

	COMPILE_ERROR_ON(sizeof(struct pmemobjpool) !=
		POOL_HDR_SIZE + POOL_DESC_SIZE);

	COMPILE_ERROR_ON(PMEMOBJ_F_MEM_NODRAIN != PMEM_F_MEM_NODRAIN);

	COMPILE_ERROR_ON(PMEMOBJ_F_MEM_NONTEMPORAL != PMEM_F_MEM_NONTEMPORAL);
	COMPILE_ERROR_ON(PMEMOBJ_F_MEM_TEMPORAL != PMEM_F_MEM_TEMPORAL);

	COMPILE_ERROR_ON(PMEMOBJ_F_MEM_WC != PMEM_F_MEM_WC);
	COMPILE_ERROR_ON(PMEMOBJ_F_MEM_WB != PMEM_F_MEM_WB);

	COMPILE_ERROR_ON(PMEMOBJ_F_MEM_NOFLUSH != PMEM_F_MEM_NOFLUSH);

	os_mutex_init(&pools_mutex);

	/*
	 * Load global config, ignore any issues. They will be caught on the
	 * subsequent call to this function for individual pools.
	 */
	ctl_global_register();
	pmalloc_global_ctl_register();

	if (obj_ctl_init_and_load(NULL))
		FATAL("error: %s", pmemobj_errormsg());

	lane_info_boot();
}

/*
 * obj_fini -- cleanup of obj
 *
 * Called by destructor.
 */
void
obj_fini(void)
{
	LOG(3, NULL);

	if (pools_ht)
		critnib_delete(pools_ht);
	if (pools_tree)
		critnib_delete(pools_tree);
	lane_info_destroy();

	os_mutex_destroy(&pools_mutex);
}

/*
 * obj_drain_empty -- (internal) empty function for drain on non-pmem memory
 */
static void
obj_drain_empty(void)
{
	/* do nothing */
}

/*
 * obj_msync_nofail -- (internal) pmem_msync wrapper that never fails from
 * caller's perspective
 */
static void
obj_msync_nofail(const void *addr, size_t size)
{
	if (pmem_msync(addr, size))
		FATAL("!pmem_msync");
}

/*
 * obj_nopmem_memcpy -- (internal) memcpy followed by an msync
 */
static void *
obj_nopmem_memcpy(void *dest, const void *src, size_t len, unsigned flags)
{
	LOG(15, "dest %p src %p len %zu flags 0x%x", dest, src, len, flags);

	/*
	 * Use pmem_memcpy instead of memcpy, because pmemobj_memcpy is supposed
	 * to guarantee that multiple of 8 byte stores to 8 byte aligned
	 * addresses are fail safe atomic. pmem_memcpy guarantees that, while
	 * libc memcpy does not.
	 */
	pmem_memcpy(dest, src, len, PMEM_F_MEM_NOFLUSH);
	obj_msync_nofail(dest, len);
	return dest;
}

/*
 * obj_nopmem_memmove -- (internal) memmove followed by an msync
 */
static void *
obj_nopmem_memmove(void *dest, const void *src, size_t len, unsigned flags)
{
	LOG(15, "dest %p src %p len %zu flags 0x%x", dest, src, len, flags);

	/* see comment in obj_nopmem_memcpy */
	pmem_memmove(dest, src, len, PMEM_F_MEM_NOFLUSH);
	obj_msync_nofail(dest, len);
	return dest;
}

/*
 * obj_nopmem_memset -- (internal) memset followed by an msync
 */
static void *
obj_nopmem_memset(void *dest, int c, size_t len, unsigned flags)
{
	LOG(15, "dest %p c 0x%02x len %zu flags 0x%x", dest, c, len, flags);

	/* see comment in obj_nopmem_memcpy */
	pmem_memset(dest, c, len, PMEM_F_MEM_NOFLUSH);
	obj_msync_nofail(dest, len);
	return dest;
}

/*
 * XXX - Consider removing obj_norep_*() wrappers to call *_local()
 * functions directly.  Alternatively, always use obj_rep_*(), even
 * if there are no replicas.  Verify the performance penalty.
 */

/*
 * obj_norep_memcpy -- (internal) memcpy w/o replication
 */
static void *
obj_norep_memcpy(void *ctx, void *dest, const void *src, size_t len,
		unsigned flags)
{
	PMEMobjpool *pop = ctx;
	LOG(15, "pop %p dest %p src %p len %zu flags 0x%x", pop, dest, src, len,
			flags);

	return pop->memcpy_local(dest, src, len,
					flags & PMEM_F_MEM_VALID_FLAGS);
}

/*
 * obj_norep_memmove -- (internal) memmove w/o replication
 */
static void *
obj_norep_memmove(void *ctx, void *dest, const void *src, size_t len,
		unsigned flags)
{
	PMEMobjpool *pop = ctx;
	LOG(15, "pop %p dest %p src %p len %zu flags 0x%x", pop, dest, src, len,
			flags);

	return pop->memmove_local(dest, src, len,
					flags & PMEM_F_MEM_VALID_FLAGS);
}

/*
 * obj_norep_memset -- (internal) memset w/o replication
 */
static void *
obj_norep_memset(void *ctx, void *dest, int c, size_t len, unsigned flags)
{
	PMEMobjpool *pop = ctx;
	LOG(15, "pop %p dest %p c 0x%02x len %zu flags 0x%x", pop, dest, c, len,
			flags);

	return pop->memset_local(dest, c, len, flags & PMEM_F_MEM_VALID_FLAGS);
}

/*
 * obj_norep_persist -- (internal) persist w/o replication
 */
static int
obj_norep_persist(void *ctx, const void *addr, size_t len, unsigned flags)
{
	/* suppress unused-parameter errors */
	SUPPRESS_UNUSED(flags);

	PMEMobjpool *pop = ctx;
	LOG(15, "pop %p addr %p len %zu", pop, addr, len);

	pop->persist_local(addr, len);

	return 0;
}

/*
 * obj_norep_flush -- (internal) flush w/o replication
 */
static int
obj_norep_flush(void *ctx, const void *addr, size_t len, unsigned flags)
{
	/* suppress unused-parameter errors */
	SUPPRESS_UNUSED(flags);

	PMEMobjpool *pop = ctx;
	LOG(15, "pop %p addr %p len %zu", pop, addr, len);

	pop->flush_local(addr, len);

	return 0;
}

/*
 * obj_norep_drain -- (internal) drain w/o replication
 */
static void
obj_norep_drain(void *ctx)
{
	PMEMobjpool *pop = ctx;
	LOG(15, "pop %p", pop);

	pop->drain_local();
}

static void obj_pool_cleanup(PMEMobjpool *pop);

/*
 * obj_rep_memcpy -- (internal) memcpy with replication
 */
static void *
obj_rep_memcpy(void *ctx, void *dest, const void *src, size_t len,
		unsigned flags)
{
	PMEMobjpool *pop = ctx;
	LOG(15, "pop %p dest %p src %p len %zu flags 0x%x", pop, dest, src, len,
			flags);

	void *ret = pop->memcpy_local(dest, src, len, flags);

	PMEMobjpool *rep = pop->replica;
	while (rep) {
		void *rdest = (char *)rep + (uintptr_t)dest - (uintptr_t)pop;
		rep->memcpy_local(rdest, src, len,
			flags & PMEM_F_MEM_VALID_FLAGS);
		rep = rep->replica;
	}

	return ret;
}

/*
 * obj_rep_memmove -- (internal) memmove with replication
 */
static void *
obj_rep_memmove(void *ctx, void *dest, const void *src, size_t len,
		unsigned flags)
{
	PMEMobjpool *pop = ctx;
	LOG(15, "pop %p dest %p src %p len %zu flags 0x%x", pop, dest, src, len,
			flags);

	void *ret = pop->memmove_local(dest, src, len, flags);

	PMEMobjpool *rep = pop->replica;
	while (rep) {
		void *rdest = (char *)rep + (uintptr_t)dest - (uintptr_t)pop;
		rep->memmove_local(rdest, src, len,
			flags & PMEM_F_MEM_VALID_FLAGS);
		rep = rep->replica;
	}

	return ret;
}

/*
 * obj_rep_memset -- (internal) memset with replication
 */
static void *
obj_rep_memset(void *ctx, void *dest, int c, size_t len, unsigned flags)
{
	PMEMobjpool *pop = ctx;
	LOG(15, "pop %p dest %p c 0x%02x len %zu flags 0x%x", pop, dest, c, len,
			flags);

	void *ret = pop->memset_local(dest, c, len, flags);

	PMEMobjpool *rep = pop->replica;
	while (rep) {
		void *rdest = (char *)rep + (uintptr_t)dest - (uintptr_t)pop;
		rep->memset_local(rdest, c, len,
			flags & PMEM_F_MEM_VALID_FLAGS);
		rep = rep->replica;
	}

	return ret;
}

/*
 * obj_rep_persist -- (internal) persist with replication
 */
static int
obj_rep_persist(void *ctx, const void *addr, size_t len, unsigned flags)
{
	/* suppress unused-parameter errors */
	SUPPRESS_UNUSED(flags);

	PMEMobjpool *pop = ctx;
	LOG(15, "pop %p addr %p len %zu", pop, addr, len);

	pop->persist_local(addr, len);

	PMEMobjpool *rep = pop->replica;
	while (rep) {
		void *raddr = (char *)rep + (uintptr_t)addr - (uintptr_t)pop;
		rep->memcpy_local(raddr, addr, len, 0);
		rep = rep->replica;
	}

	return 0;
}

/*
 * obj_rep_flush -- (internal) flush with replication
 */
static int
obj_rep_flush(void *ctx, const void *addr, size_t len, unsigned flags)
{
	/* suppress unused-parameter errors */
	SUPPRESS_UNUSED(flags);

	PMEMobjpool *pop = ctx;
	LOG(15, "pop %p addr %p len %zu", pop, addr, len);

	pop->flush_local(addr, len);

	PMEMobjpool *rep = pop->replica;
	while (rep) {
		void *raddr = (char *)rep + (uintptr_t)addr - (uintptr_t)pop;
		rep->memcpy_local(raddr, addr, len, PMEM_F_MEM_NODRAIN);
		rep = rep->replica;
	}

	return 0;
}

/*
 * obj_rep_drain -- (internal) drain with replication
 */
static void
obj_rep_drain(void *ctx)
{
	PMEMobjpool *pop = ctx;
	LOG(15, "pop %p", pop);

	pop->drain_local();

	PMEMobjpool *rep = pop->replica;
	while (rep) {
		rep->drain_local();
		rep = rep->replica;
	}
}

#if VG_MEMCHECK_ENABLED
/*
 * Arbitrary value. When there's more undefined regions than MAX_UNDEFS, it's
 * not worth reporting everything - developer should fix the code.
 */
#define MAX_UNDEFS 1000

/*
 * obj_vg_check_no_undef -- (internal) check whether there are any undefined
 *				regions
 */
static void
obj_vg_check_no_undef(struct pmemobjpool *pop)
{
	LOG(4, "pop %p", pop);

	static struct {
		void *start, *end;
	} undefs[MAX_UNDEFS];
	int num_undefs = 0;

	VALGRIND_DO_DISABLE_ERROR_REPORTING;
	char *addr_start = pop->addr;
	char *addr_end = addr_start + pop->set->poolsize;

	while (addr_start < addr_end) {
		char *noaccess = (char *)VALGRIND_CHECK_MEM_IS_ADDRESSABLE(
					addr_start, addr_end - addr_start);
		if (noaccess == NULL)
			noaccess = addr_end;

		while (addr_start < noaccess) {
			char *undefined =
				(char *)VALGRIND_CHECK_MEM_IS_DEFINED(
					addr_start, noaccess - addr_start);

			if (undefined) {
				addr_start = undefined;

#ifdef VALGRIND_CHECK_MEM_IS_UNDEFINED
				addr_start = (char *)
					VALGRIND_CHECK_MEM_IS_UNDEFINED(
					addr_start, noaccess - addr_start);
				if (addr_start == NULL)
					addr_start = noaccess;
#else
				while (addr_start < noaccess &&
						VALGRIND_CHECK_MEM_IS_DEFINED(
								addr_start, 1))
					addr_start++;
#endif

				if (num_undefs < MAX_UNDEFS) {
					undefs[num_undefs].start = undefined;
					undefs[num_undefs].end = addr_start - 1;
					num_undefs++;
				}
			} else
				addr_start = noaccess;
		}

#ifdef VALGRIND_CHECK_MEM_IS_UNADDRESSABLE
		addr_start = (char *)VALGRIND_CHECK_MEM_IS_UNADDRESSABLE(
				addr_start, addr_end - addr_start);
		if (addr_start == NULL)
			addr_start = addr_end;
#else
		while (addr_start < addr_end &&
				(char *)VALGRIND_CHECK_MEM_IS_ADDRESSABLE(
						addr_start, 1) == addr_start)
			addr_start++;
#endif
	}
	VALGRIND_DO_ENABLE_ERROR_REPORTING;

	if (num_undefs) {
		/*
		 * How to resolve this error:
		 * If it's part of the free space Valgrind should be told about
		 * it by VALGRIND_DO_MAKE_MEM_NOACCESS request. If it's
		 * allocated - initialize it or use VALGRIND_DO_MAKE_MEM_DEFINED
		 * request.
		 */

		VALGRIND_PRINTF("Part of the pool is left in undefined state on"
				" boot. This is pmemobj's bug.\nUndefined"
				" regions: [pool address: %p]\n", pop);
		for (int i = 0; i < num_undefs; ++i)
			VALGRIND_PRINTF("   [%p, %p]\n", undefs[i].start,
					undefs[i].end);
		if (num_undefs == MAX_UNDEFS)
			VALGRIND_PRINTF("   ...\n");

		/* Trigger error. */
		VALGRIND_CHECK_MEM_IS_DEFINED(undefs[0].start, 1);
	}
}

/*
 * obj_vg_boot -- (internal) notify Valgrind about pool objects
 */
static void
obj_vg_boot(struct pmemobjpool *pop)
{
	if (!On_memcheck)
		return;

	LOG(4, "pop %p", pop);

	if (os_getenv("PMEMOBJ_VG_CHECK_UNDEF"))
		obj_vg_check_no_undef(pop);
}

#endif /* VG_MEMCHECK_ENABLED */

/*
 * obj_runtime_init_common -- (internal) runtime initialization
 *
 * Common routine for create/open and check.
 */
static int
obj_runtime_init_common(PMEMobjpool *pop)
{
	LOG(3, "pop %p", pop);

	if ((errno = lane_boot(pop)) != 0) {
		ERR("!lane_boot");
		return errno;
	}

	if ((errno = lane_recover_and_section_boot(pop)) != 0) {
		ERR("!lane_recover_and_section_boot");
		return errno;
	}

	pop->conversion_flags = 0;
	pmemops_persist(&pop->p_ops,
		&pop->conversion_flags, sizeof(pop->conversion_flags));

	return 0;
}

/*
 * obj_runtime_cleanup_common -- (internal) runtime cleanup
 *
 * Common routine for create/open and check
 */
static void
obj_runtime_cleanup_common(PMEMobjpool *pop)
{
	lane_section_cleanup(pop);
	lane_cleanup(pop);
}

/*
 * obj_descr_create -- (internal) create obj pool descriptor
 */
static int
obj_descr_create(PMEMobjpool *pop, const char *layout, size_t poolsize)
{
	LOG(3, "pop %p layout %s poolsize %zu", pop, layout, poolsize);

	ASSERTeq(poolsize % Pagesize, 0);

	/* opaque info lives at the beginning of mapped memory pool */
	void *dscp = (void *)((uintptr_t)pop + sizeof(struct pool_hdr));

	/* create the persistent part of pool's descriptor */
	memset(dscp, 0, OBJ_DSC_P_SIZE);
	if (layout)
		strncpy(pop->layout, layout, PMEMOBJ_MAX_LAYOUT - 1);
	struct pmem_ops *p_ops = &pop->p_ops;

	pop->lanes_offset = OBJ_LANES_OFFSET;
	pop->nlanes = OBJ_NLANES;

	/* zero all lanes */
	lane_init_data(pop);

	pop->heap_offset = pop->lanes_offset +
		pop->nlanes * sizeof(struct lane_layout);
	pop->heap_offset = (pop->heap_offset + Pagesize - 1) & ~(Pagesize - 1);

	size_t heap_size = pop->set->poolsize - pop->heap_offset;

	/* initialize heap prior to storing the checksum */
	errno = palloc_init((char *)pop + pop->heap_offset, heap_size,
		&pop->heap_size, p_ops);
	if (errno != 0) {
		ERR("!palloc_init");
		return -1;
	}

	util_checksum(dscp, OBJ_DSC_P_SIZE, &pop->checksum, 1, 0);

	/*
	 * store the persistent part of pool's descriptor (2kB)
	 *
	 * It's safe to use PMEMOBJ_F_RELAXED flag because the entire
	 * structure is protected by checksum.
	 */
	pmemops_xpersist(p_ops, dscp, OBJ_DSC_P_SIZE, PMEMOBJ_F_RELAXED);

	/* initialize run_id, it will be incremented later */
	pop->run_id = 0;
	pmemops_persist(p_ops, &pop->run_id, sizeof(pop->run_id));

	pop->root_offset = 0;
	pmemops_persist(p_ops, &pop->root_offset, sizeof(pop->root_offset));
	pop->root_size = 0;
	pmemops_persist(p_ops, &pop->root_size, sizeof(pop->root_size));

	pop->conversion_flags = 0;
	pmemops_persist(p_ops, &pop->conversion_flags,
		sizeof(pop->conversion_flags));

	/*
	 * It's safe to use PMEMOBJ_F_RELAXED flag because the reserved
	 * area must be entirely zeroed.
	 */
	pmemops_memset(p_ops, pop->pmem_reserved, 0,
		sizeof(pop->pmem_reserved), PMEMOBJ_F_RELAXED);

	return 0;
}

/*
 * obj_descr_check -- (internal) validate obj pool descriptor
 */
static int
obj_descr_check(PMEMobjpool *pop, const char *layout, size_t poolsize)
{
	LOG(3, "pop %p layout %s poolsize %zu", pop, layout, poolsize);

	void *dscp = (void *)((uintptr_t)pop + sizeof(struct pool_hdr));

	if (!util_checksum(dscp, OBJ_DSC_P_SIZE, &pop->checksum, 0, 0)) {
		ERR("invalid checksum of pool descriptor");
		errno = EINVAL;
		return -1;
	}

	if (layout &&
	    strncmp(pop->layout, layout, PMEMOBJ_MAX_LAYOUT)) {
		ERR("wrong layout (\"%s\"), "
			"pool created with layout \"%s\"",
			layout, pop->layout);
		errno = EINVAL;
		return -1;
	}

	if (pop->heap_offset % Pagesize) {
		ERR("unaligned heap: off %" PRIu64, pop->heap_offset);
		errno = EINVAL;
		return -1;
	}

	return 0;
}

/*
 * obj_replica_init_local -- (internal) initialize runtime part
 *                               of the local replicas
 */
static int
obj_replica_init_local(PMEMobjpool *rep, int is_pmem, size_t resvsize)
{
	LOG(3, "rep %p is_pmem %d resvsize %zu", rep, is_pmem, resvsize);

	/*
	 * Use some of the memory pool area for run-time info.  This
	 * run-time state is never loaded from the file, it is always
	 * created here, so no need to worry about byte-order.
	 */
	rep->is_pmem = is_pmem;

	/*
	 * All replicas, except for master, are ignored as far as valgrind is
	 * concerned. This is to save CPU time and lessen the complexity of
	 * instrumentation.
	 */
	if (!rep->is_master_replica)
		VALGRIND_ADD_TO_GLOBAL_TX_IGNORE(rep, resvsize);

	if (rep->is_pmem) {
		rep->persist_local = pmem_persist;
		rep->flush_local = pmem_flush;
		rep->drain_local = pmem_drain;
		rep->memcpy_local = pmem_memcpy;
		rep->memmove_local = pmem_memmove;
		rep->memset_local = pmem_memset;
	} else {
		rep->persist_local = obj_msync_nofail;
		rep->flush_local = obj_msync_nofail;
		rep->drain_local = obj_drain_empty;
		rep->memcpy_local = obj_nopmem_memcpy;
		rep->memmove_local = obj_nopmem_memmove;
		rep->memset_local = obj_nopmem_memset;
	}

	return 0;
}

/*
 * obj_replica_init -- (internal) initialize runtime part of the replica
 */
static int
obj_replica_init(PMEMobjpool *rep, struct pool_set *set, unsigned repidx)
{
	struct pool_replica *repset = set->replica[repidx];

	if (repidx == 0) {
		/* master replica */
		rep->is_master_replica = 1;

		if (set->nreplicas > 1) {
			rep->p_ops.persist = obj_rep_persist;
			rep->p_ops.flush = obj_rep_flush;
			rep->p_ops.drain = obj_rep_drain;
			rep->p_ops.memcpy = obj_rep_memcpy;
			rep->p_ops.memmove = obj_rep_memmove;
			rep->p_ops.memset = obj_rep_memset;
		} else {
			rep->p_ops.persist = obj_norep_persist;
			rep->p_ops.flush = obj_norep_flush;
			rep->p_ops.drain = obj_norep_drain;
			rep->p_ops.memcpy = obj_norep_memcpy;
			rep->p_ops.memmove = obj_norep_memmove;
			rep->p_ops.memset = obj_norep_memset;
		}
		rep->p_ops.base = rep;
	} else {
		/* non-master replicas */
		rep->is_master_replica = 0;

		rep->p_ops.persist = NULL;
		rep->p_ops.flush = NULL;
		rep->p_ops.drain = NULL;
		rep->p_ops.memcpy = NULL;
		rep->p_ops.memmove = NULL;
		rep->p_ops.memset = NULL;

		rep->p_ops.base = NULL;
	}

	rep->is_dev_dax = set->replica[repidx]->part[0].is_dev_dax;

	int ret;
	ret = obj_replica_init_local(rep, repset->is_pmem,
		set->resvsize);

	if (ret)
		return ret;

	return 0;
}

/*
 * obj_replica_fini -- (internal) deinitialize replica
 */
static void
obj_replica_fini(struct pool_replica *repset)
{
	LOG(3, "repset %p", repset);
}

/*
 * obj_runtime_init -- (internal) initialize runtime part of the pool header
 */
static int
obj_runtime_init(PMEMobjpool *pop, int rdonly, int boot, unsigned nlanes)
{
	LOG(3, "pop %p rdonly %d boot %d", pop, rdonly, boot);
	struct pmem_ops *p_ops = &pop->p_ops;

	/* run_id is made unique by incrementing the previous value */
	pop->run_id += 2;
	if (pop->run_id == 0)
		pop->run_id += 2;
	pmemops_persist(p_ops, &pop->run_id, sizeof(pop->run_id));

	/*
	 * Use some of the memory pool area for run-time info.  This
	 * run-time state is never loaded from the file, it is always
	 * created here, so no need to worry about byte-order.
	 */
	pop->rdonly = rdonly;

	pop->uuid_lo = pmemobj_get_uuid_lo(pop);

	pop->lanes_desc.runtime_nlanes = nlanes;

	pop->tx_params = tx_params_new();
	if (pop->tx_params == NULL)
		goto err_tx_params;

	pop->stats = stats_new(pop);
	if (pop->stats == NULL)
		goto err_stat;

	pop->user_data = NULL;

	VALGRIND_REMOVE_PMEM_MAPPING(&pop->mutex_head,
		sizeof(pop->mutex_head));
	VALGRIND_REMOVE_PMEM_MAPPING(&pop->rwlock_head,
		sizeof(pop->rwlock_head));
	VALGRIND_REMOVE_PMEM_MAPPING(&pop->cond_head,
		sizeof(pop->cond_head));
	pop->mutex_head = NULL;
	pop->rwlock_head = NULL;
	pop->cond_head = NULL;

	if (boot) {
		if ((errno = obj_runtime_init_common(pop)) != 0)
			goto err_boot;

#if VG_MEMCHECK_ENABLED
		if (On_memcheck) {
			/* mark unused part of the pool as not accessible */
			void *end = palloc_heap_end(&pop->heap);
			VALGRIND_DO_MAKE_MEM_NOACCESS(end,
				(char *)pop + pop->set->poolsize - (char *)end);
		}
#endif

		obj_pool_init();

		if ((errno = critnib_insert(pools_ht, pop->uuid_lo, pop))) {
			ERR("!critnib_insert to pools_ht");
			goto err_critnib_insert;
		}

		if ((errno = critnib_insert(pools_tree, (uint64_t)pop, pop))) {
			ERR("!critnib_insert to pools_tree");
			goto err_tree_insert;
		}
	}

	if (obj_ctl_init_and_load(pop) != 0) {
		errno = EINVAL;
		goto err_ctl;
	}

	util_mutex_init(&pop->ulog_user_buffers.lock);
	pop->ulog_user_buffers.map = ravl_new_sized(
		operation_user_buffer_range_cmp,
		sizeof(struct user_buffer_def));
	if (pop->ulog_user_buffers.map == NULL) {
		ERR("!ravl_new_sized");
		goto err_user_buffers_map;
	}
	pop->ulog_user_buffers.verify = 0;

	/*
	 * If possible, turn off all permissions on the pool header page.
	 *
	 * The prototype PMFS doesn't allow this when large pages are in
	 * use. It is not considered an error if this fails.
	 */
	RANGE_NONE(pop->addr, sizeof(struct pool_hdr), pop->is_dev_dax);

	return 0;

err_user_buffers_map:
	util_mutex_destroy(&pop->ulog_user_buffers.lock);
	ctl_delete(pop->ctl);
err_ctl:;
	void *n = critnib_remove(pools_tree, (uint64_t)pop);
	ASSERTne(n, NULL);
err_tree_insert:
	critnib_remove(pools_ht, pop->uuid_lo);
err_critnib_insert:
	obj_runtime_cleanup_common(pop);
err_boot:
	stats_delete(pop, pop->stats);
err_stat:
	tx_params_delete(pop->tx_params);
err_tx_params:

	return -1;
}

/*
 * obj_get_nlanes -- get a number of lanes available at runtime. If the value
 * provided with the PMEMOBJ_NLANES environment variable is greater than 0 and
 * smaller than OBJ_NLANES constant it returns PMEMOBJ_NLANES. Otherwise it
 * returns OBJ_NLANES.
 */
static unsigned
obj_get_nlanes(void)
{
	LOG(3, NULL);

	char *env_nlanes = os_getenv(OBJ_NLANES_ENV_VARIABLE);
	if (env_nlanes) {
		int nlanes = atoi(env_nlanes);
		if (nlanes <= 0) {
			ERR("%s variable must be a positive integer",
					OBJ_NLANES_ENV_VARIABLE);
			errno = EINVAL;
			goto no_valid_env;
		}

		return (unsigned)(OBJ_NLANES < nlanes ? OBJ_NLANES : nlanes);
	}

no_valid_env:
	return OBJ_NLANES;
}

/*
 * pmemobj_createU -- create a transactional memory pool (set)
 */
static inline
PMEMobjpool *
pmemobj_createU(const char *path, const char *layout,
		size_t poolsize, mode_t mode)
{
	LOG(3, "path %s layout %s poolsize %zu mode %o",
			path, layout, poolsize, mode);

	PMEMobjpool *pop;
	struct pool_set *set;

	/* check length of layout */
	if (layout && (strlen(layout) >= PMEMOBJ_MAX_LAYOUT)) {
		ERR("Layout too long");
		errno = EINVAL;
		return NULL;
	}

	os_mutex_lock(&pools_mutex);

	/*
	 * The runtime number of lanes is equal to the total number of lanes
	 * available in the pool or the value provided with PMEMOBJ_NLANES
	 * environment variable whichever is lower.
	 */
	unsigned runtime_nlanes = obj_get_nlanes();

	struct pool_attr adj_pool_attr = Obj_create_attr;

	/* force set SDS feature */
	if (SDS_at_create)
		adj_pool_attr.features.incompat |= POOL_FEAT_SDS;
	else
		adj_pool_attr.features.incompat &= ~POOL_FEAT_SDS;

	if (util_pool_create(&set, path, poolsize, PMEMOBJ_MIN_POOL,
			PMEMOBJ_MIN_PART, &adj_pool_attr, &runtime_nlanes,
			REPLICAS_ENABLED) != 0) {
		LOG(2, "cannot create pool or pool set");
		os_mutex_unlock(&pools_mutex);
		return NULL;
	}

	ASSERT(set->nreplicas > 0);

	/* pop is master replica from now on */
	pop = set->replica[0]->part[0].addr;

	for (unsigned r = 0; r < set->nreplicas; r++) {
		struct pool_replica *repset = set->replica[r];
		PMEMobjpool *rep = repset->part[0].addr;

		size_t rt_size = (uintptr_t)(rep + 1) - (uintptr_t)&rep->addr;
		VALGRIND_REMOVE_PMEM_MAPPING(&rep->addr, rt_size);

		memset(&rep->addr, 0, rt_size);

		rep->addr = rep;
		rep->replica = NULL;

		/* initialize replica runtime - is_pmem, funcs, ... */
		if (obj_replica_init(rep, set, r) != 0) {
			ERR("initialization of replica #%u failed", r);
			goto err;
		}

		/* link replicas */
		if (r < set->nreplicas - 1)
			rep->replica = set->replica[r + 1]->part[0].addr;
	}

	pop->set = set;

	/* create pool descriptor */
	if (obj_descr_create(pop, layout, set->poolsize) != 0) {
		LOG(2, "creation of pool descriptor failed");
		goto err;
	}

	/* initialize runtime parts - lanes, obj stores, ... */
	if (obj_runtime_init(pop, 0, 1 /* boot */,
					runtime_nlanes) != 0) {
		ERR("pool initialization failed");
		goto err;
	}

	if (util_poolset_chmod(set, mode))
		goto err;

	util_poolset_fdclose(set);

	LOG(3, "pop %p", pop);
	os_mutex_unlock(&pools_mutex);

	return pop;

err:
	LOG(4, "error clean up");
	int oerrno = errno;
	util_poolset_close(set, DELETE_CREATED_PARTS);
	os_mutex_unlock(&pools_mutex);
	errno = oerrno;
	return NULL;
}

/*
 * pmemobj_create -- create a transactional memory pool (set)
 */
PMEMobjpool *
pmemobj_create(const char *path, const char *layout,
		size_t poolsize, mode_t mode)
{
	PMEMOBJ_API_START();

	PMEMobjpool *pop = pmemobj_createU(path, layout, poolsize, mode);

	PMEMOBJ_API_END();
	return pop;
}

/*
 * obj_check_basic_local -- (internal) basic pool consistency check
 *                              of a local replica
 */
static int
obj_check_basic_local(PMEMobjpool *pop, size_t mapped_size)
{
	LOG(3, "pop %p mapped_size %zu", pop, mapped_size);

	int consistent = 1;

	if (pop->run_id % 2) {
		ERR("invalid run_id %" PRIu64, pop->run_id);
		consistent = 0;
	}

	if ((errno = lane_check(pop)) != 0) {
		LOG(2, "!lane_check");
		consistent = 0;
	}

	/* pop->heap_size can still be 0 at this point */
	size_t heap_size = mapped_size - pop->heap_offset;
	errno = palloc_heap_check((char *)pop + pop->heap_offset,
		heap_size);
	if (errno != 0) {
		LOG(2, "!heap_check");
		consistent = 0;
	}

	return consistent;
}

/*
 * obj_check_basic -- (internal) basic pool consistency check
 *
 * Used to check if all the replicas are consistent prior to pool recovery.
 */
static int
obj_check_basic(PMEMobjpool *pop, size_t mapped_size)
{
	LOG(3, "pop %p mapped_size %zu", pop, mapped_size);

	return obj_check_basic_local(pop, mapped_size);
}

/*
 * obj_pool_close -- (internal) close the pool set
 */
static void
obj_pool_close(struct pool_set *set)
{
	int oerrno = errno;
	util_poolset_close(set, DO_NOT_DELETE_PARTS);
	errno = oerrno;
}

/*
 * obj_pool_open -- (internal) open the given pool
 */
static int
obj_pool_open(struct pool_set **set, const char *path, unsigned flags,
	unsigned *nlanes)
{
	if (util_pool_open(set, path, PMEMOBJ_MIN_PART, &Obj_open_attr,
				nlanes, NULL, flags) != 0) {
		LOG(2, "cannot open pool or pool set");
		return -1;
	}

	ASSERT((*set)->nreplicas > 0);

	/* read-only mode is not supported in libpmemobj */
	if ((*set)->rdonly) {
		ERR("read-only mode is not supported");
		errno = EINVAL;
		goto err_rdonly;
	}

	return 0;
err_rdonly:
	obj_pool_close(*set);
	return -1;
}

/*
 * obj_replicas_init -- (internal) initialize all replicas
 */
static int
obj_replicas_init(struct pool_set *set)
{
	unsigned r;
	for (r = 0; r < set->nreplicas; r++) {
		struct pool_replica *repset = set->replica[r];
		PMEMobjpool *rep = repset->part[0].addr;

		size_t rt_size = (uintptr_t)(rep + 1) - (uintptr_t)&rep->addr;

		VALGRIND_REMOVE_PMEM_MAPPING(&rep->addr, rt_size);

		memset(&rep->addr, 0, rt_size);

		rep->addr = rep;
		rep->replica = NULL;

		/* initialize replica runtime - is_pmem, funcs, ... */
		if (obj_replica_init(rep, set, r) != 0) {
			ERR("initialization of replica #%u failed", r);
			goto err;
		}

		/* link replicas */
		if (r < set->nreplicas - 1)
			rep->replica = set->replica[r + 1]->part[0].addr;
	}

	return 0;
err:
	for (unsigned p = 0; p < r; p++)
		obj_replica_fini(set->replica[p]);

	return -1;
}

/*
 * obj_replicas_fini -- (internal) deinitialize all replicas
 */
static void
obj_replicas_fini(struct pool_set *set)
{
	int oerrno = errno;
	for (unsigned r = 0; r < set->nreplicas; r++)
		obj_replica_fini(set->replica[r]);
	errno = oerrno;
}

/*
 * obj_replicas_check_basic -- (internal) perform basic consistency check
 * for all replicas
 */
static int
obj_replicas_check_basic(PMEMobjpool *pop)
{
	PMEMobjpool *rep;
	for (unsigned r = 0; r < pop->set->nreplicas; r++) {
		rep = pop->set->replica[r]->part[0].addr;
		if (obj_check_basic(rep, pop->set->poolsize) == 0) {
			ERR("inconsistent replica #%u", r);
			return -1;
		}
	}

	/* copy lanes */
	void *src = (void *)((uintptr_t)pop + pop->lanes_offset);
	size_t len = pop->nlanes * sizeof(struct lane_layout);

	for (unsigned r = 1; r < pop->set->nreplicas; r++) {
		rep = pop->set->replica[r]->part[0].addr;
		void *dst = (void *)((uintptr_t)rep + pop->lanes_offset);
		rep->memcpy_local(dst, src, len, 0);
	}

	return 0;
}

/*
 * obj_open_common -- open a transactional memory pool (set)
 *
 * This routine takes flags and does all the work
 * (flag POOL_OPEN_COW - internal calls can map a read-only pool if required).
 */
static PMEMobjpool *
obj_open_common(const char *path, const char *layout, unsigned flags, int boot)
{
	LOG(3, "path %s layout %s flags 0x%x", path, layout, flags);

	PMEMobjpool *pop = NULL;
	struct pool_set *set;

	os_mutex_lock(&pools_mutex);

	/*
	 * The runtime number of lanes is equal to the total number of lanes
	 * available in the pool or the value provided with PMEMOBJ_NLANES
	 * environment variable whichever is lower.
	 */
	unsigned runtime_nlanes = obj_get_nlanes();
	if (obj_pool_open(&set, path, flags, &runtime_nlanes)) {
		os_mutex_unlock(&pools_mutex);
		return NULL;
	}

	/* pop is master replica from now on */
	pop = set->replica[0]->part[0].addr;

	if (obj_replicas_init(set))
		goto replicas_init;

	for (unsigned r = 0; r < set->nreplicas; r++) {
		struct pool_replica *repset = set->replica[r];
		PMEMobjpool *rep = repset->part[0].addr;
		/* check descriptor */
		if (obj_descr_check(rep, layout, set->poolsize) != 0) {
			LOG(2, "descriptor check of replica #%u failed", r);
			goto err_descr_check;
		}
	}

	pop->set = set;

	if (boot) {
		/* check consistency of 'master' replica */
		if (obj_check_basic(pop, pop->set->poolsize) == 0) {
			goto err_check_basic;
		}
	}

	if (set->nreplicas > 1) {
		if (obj_replicas_check_basic(pop))
			goto err_replicas_check_basic;
	}

	/*
	 * before runtime initialization lanes are unavailable
	 */
	pop->lanes_desc.runtime_nlanes = 0;

#if VG_MEMCHECK_ENABLED
	pop->vg_boot = boot;
#endif
	/* initialize runtime parts - lanes, obj stores, ... */
	if (obj_runtime_init(pop, 0, boot, runtime_nlanes) != 0) {
		ERR("pool initialization failed");
		goto err_runtime_init;
	}

#if VG_MEMCHECK_ENABLED
	if (boot)
		obj_vg_boot(pop);
#endif

	util_poolset_fdclose(set);
	os_mutex_unlock(&pools_mutex);

	LOG(3, "pop %p", pop);

	return pop;

err_runtime_init:
err_replicas_check_basic:
err_check_basic:
err_descr_check:
	obj_replicas_fini(set);
replicas_init:
	obj_pool_close(set);
	os_mutex_unlock(&pools_mutex);
	return NULL;
}

/*
 * pmemobj_openU -- open a transactional memory pool
 */
static inline
PMEMobjpool *
pmemobj_openU(const char *path, const char *layout)
{
	LOG(3, "path %s layout %s", path, layout);

	return obj_open_common(path, layout,
			COW_at_open ? POOL_OPEN_COW : 0, 1);
}

/*
 * pmemobj_open -- open a transactional memory pool
 */
PMEMobjpool *
pmemobj_open(const char *path, const char *layout)
{
	PMEMOBJ_API_START();

	PMEMobjpool *pop = pmemobj_openU(path, layout);

	PMEMOBJ_API_END();
	return pop;
}

/*
 * obj_pool_lock_cleanup -- (internal) Destroy any locks or condition
 *	variables that were allocated at run time
 */
static void
obj_pool_lock_cleanup(PMEMobjpool *pop)
{
	LOG(3, "pop %p", pop);

	PMEMmutex_internal *nextm;
	for (PMEMmutex_internal *m = pop->mutex_head; m != NULL; m = nextm) {
		nextm = m->PMEMmutex_next;
		LOG(4, "mutex %p *mutex %p", &m->PMEMmutex_lock,
			m->PMEMmutex_bsd_mutex_p);
		os_mutex_destroy(&m->PMEMmutex_lock);
		m->PMEMmutex_next = NULL;
		m->PMEMmutex_bsd_mutex_p = NULL;
	}
	pop->mutex_head = NULL;

	PMEMrwlock_internal *nextr;
	for (PMEMrwlock_internal *r = pop->rwlock_head; r != NULL; r = nextr) {
		nextr = r->PMEMrwlock_next;
		LOG(4, "rwlock %p *rwlock %p", &r->PMEMrwlock_lock,
			r->PMEMrwlock_bsd_rwlock_p);
		os_rwlock_destroy(&r->PMEMrwlock_lock);
		r->PMEMrwlock_next = NULL;
		r->PMEMrwlock_bsd_rwlock_p = NULL;
	}
	pop->rwlock_head = NULL;

	PMEMcond_internal *nextc;
	for (PMEMcond_internal *c = pop->cond_head; c != NULL; c = nextc) {
		nextc = c->PMEMcond_next;
		LOG(4, "cond %p *cond %p", &c->PMEMcond_cond,
			c->PMEMcond_bsd_cond_p);
		os_cond_destroy(&c->PMEMcond_cond);
		c->PMEMcond_next = NULL;
		c->PMEMcond_bsd_cond_p = NULL;
	}
	pop->cond_head = NULL;
}
/*
 * obj_pool_cleanup -- (internal) cleanup the pool and unmap
 */
static void
obj_pool_cleanup(PMEMobjpool *pop)
{
	LOG(3, "pop %p", pop);

	ravl_delete(pop->ulog_user_buffers.map);
	util_mutex_destroy(&pop->ulog_user_buffers.lock);

	stats_delete(pop, pop->stats);
	tx_params_delete(pop->tx_params);
	ctl_delete(pop->ctl);

	obj_pool_lock_cleanup(pop);

	lane_section_cleanup(pop);
	lane_cleanup(pop);

	/* unmap all the replicas */
	util_poolset_close(pop->set, DO_NOT_DELETE_PARTS);
}

/*
 * pmemobj_close -- close a transactional memory pool
 */
void
pmemobj_close(PMEMobjpool *pop)
{
	LOG(3, "pop %p", pop);
	PMEMOBJ_API_START();

	os_mutex_lock(&pools_mutex);

	if (critnib_remove(pools_ht, pop->uuid_lo) != pop) {
		ERR("critnib_remove for pools_ht");
	}

	if (critnib_remove(pools_tree, (uint64_t)pop) != pop)
		ERR("critnib_remove for pools_tree");

	if (_pobj_cached_pool.pop == pop) {
		_pobj_cached_pool.pop = NULL;
		_pobj_cached_pool.uuid_lo = 0;
	}

	VALGRIND_HG_DRD_DISABLE_CHECKING(&_pobj_cache_invalidate,
		sizeof(_pobj_cache_invalidate));
	_pobj_cache_invalidate++;

	obj_pool_cleanup(pop);
	os_mutex_unlock(&pools_mutex);
	PMEMOBJ_API_END();
}

/*
 * pmemobj_checkU -- transactional memory pool consistency check
 */
static inline
int
pmemobj_checkU(const char *path, const char *layout)
{
	LOG(3, "path %s layout %s", path, layout);

	PMEMobjpool *pop = obj_open_common(path, layout, POOL_OPEN_COW, 0);
	if (pop == NULL)
		return -1;	/* errno set by obj_open_common() */

	int consistent = 1;

	/*
	 * For replicated pools, basic consistency check is performed
	 * in obj_open_common().
	 */
	if (pop->replica == NULL)
		consistent = obj_check_basic(pop, pop->set->poolsize);

	if (consistent && (errno = obj_runtime_init_common(pop)) != 0) {
		LOG(3, "!obj_boot");
		consistent = 0;
	}

	if (consistent) {
		obj_pool_cleanup(pop);
	} else {
		stats_delete(pop, pop->stats);
		tx_params_delete(pop->tx_params);
		ctl_delete(pop->ctl);

		/* unmap all the replicas */
		util_poolset_close(pop->set, DO_NOT_DELETE_PARTS);
	}

	if (consistent)
		LOG(4, "pool consistency check OK");

	return consistent;
}

/*
 * pmemobj_check -- transactional memory pool consistency check
 */
int
pmemobj_check(const char *path, const char *layout)
{
	PMEMOBJ_API_START();

	int ret = pmemobj_checkU(path, layout);

	PMEMOBJ_API_END();
	return ret;
}

/*
 * pmemobj_pool_by_oid -- returns the pool handle associated with the oid
 */
PMEMobjpool *
pmemobj_pool_by_oid(PMEMoid oid)
{
	LOG(3, "oid.off 0x%016" PRIx64, oid.off);

	/* XXX this is a temporary fix, to be fixed properly later */
	if (pools_ht == NULL)
		return NULL;

	return critnib_get(pools_ht, oid.pool_uuid_lo);
}

/*
 * pmemobj_pool_by_ptr -- returns the pool handle associated with the address
 */
PMEMobjpool *
pmemobj_pool_by_ptr(const void *addr)
{
	LOG(3, "addr %p", addr);

	/* fast path for transactions */
	PMEMobjpool *pop = tx_get_pop();

	if ((pop != NULL) && OBJ_PTR_FROM_POOL(pop, addr))
		return pop;

	/* XXX this is a temporary fix, to be fixed properly later */
	if (pools_tree == NULL)
		return NULL;

	pop = critnib_find_le(pools_tree, (uint64_t)addr);
	if (pop == NULL)
		return NULL;

	size_t pool_size = pop->heap_offset + pop->heap_size;
	if ((char *)addr >= (char *)pop + pool_size)
		return NULL;

	return pop;
}

/*
 * pmemobj_set_user_data -- sets volatile pointer to the user data for specified
 * pool
 */
void
pmemobj_set_user_data(PMEMobjpool *pop, void *data)
{
	LOG(3, "pop %p data %p", pop, data);

	pop->user_data = data;
}

/*
 * pmemobj_get_user_data -- gets volatile pointer to the user data associated
 * with the specified pool
 */
void *
pmemobj_get_user_data(PMEMobjpool *pop)
{
	LOG(3, "pop %p", pop);

	return pop->user_data;
}

/* arguments for constructor_alloc */
struct constr_args {
	int zero_init;
	pmemobj_constr constructor;
	void *arg;
};

/*
 * constructor_alloc -- (internal) constructor for obj_alloc_construct
 */
static int
constructor_alloc(void *ctx, void *ptr, size_t usable_size, void *arg)
{
	PMEMobjpool *pop = ctx;
	LOG(3, "pop %p ptr %p arg %p", pop, ptr, arg);
	struct pmem_ops *p_ops = &pop->p_ops;

	ASSERTne(ptr, NULL);
	ASSERTne(arg, NULL);

	struct constr_args *carg = arg;

	if (carg->zero_init)
		pmemops_memset(p_ops, ptr, 0, usable_size, 0);

	int ret = 0;
	if (carg->constructor)
		ret = carg->constructor(pop, ptr, carg->arg);

	return ret;
}

/*
 * obj_alloc_construct -- (internal) allocates a new object with constructor
 */
static int
obj_alloc_construct(PMEMobjpool *pop, PMEMoid *oidp, size_t size,
	type_num_t type_num, uint64_t flags,
	pmemobj_constr constructor, void *arg)
{
	if (size > PMEMOBJ_MAX_ALLOC_SIZE) {
		ERR("requested size too large");
		errno = ENOMEM;
		return -1;
	}

	struct constr_args carg;

	carg.zero_init = flags & POBJ_FLAG_ZERO;
	carg.constructor = constructor;
	carg.arg = arg;

	struct operation_context *ctx = pmalloc_operation_hold(pop);

	if (oidp)
		operation_add_entry(ctx, &oidp->pool_uuid_lo, pop->uuid_lo,
				ULOG_OPERATION_SET);

	int ret = palloc_operation(&pop->heap, 0,
			oidp != NULL ? &oidp->off : NULL, size,
			constructor_alloc, &carg, type_num, 0,
			CLASS_ID_FROM_FLAG(flags), ARENA_ID_FROM_FLAG(flags),
			ctx);

	pmalloc_operation_release(pop);

	return ret;
}

/*
 * pmemobj_alloc -- allocates a new object
 */
int
pmemobj_alloc(PMEMobjpool *pop, PMEMoid *oidp, size_t size,
	uint64_t type_num, pmemobj_constr constructor, void *arg)
{
	LOG(3, "pop %p oidp %p size %zu type_num %llx constructor %p arg %p",
		pop, oidp, size, (unsigned long long)type_num,
		constructor, arg);

	/* log notice message if used inside a transaction */
	_POBJ_DEBUG_NOTICE_IN_TX();

	if (size == 0) {
		ERR("allocation with size 0");
		errno = EINVAL;
		return -1;
	}

	PMEMOBJ_API_START();
	int ret = obj_alloc_construct(pop, oidp, size, type_num,
			0, constructor, arg);

	PMEMOBJ_API_END();
	return ret;
}

/*
 * pmemobj_xalloc -- allocates with flags
 */
int
pmemobj_xalloc(PMEMobjpool *pop, PMEMoid *oidp, size_t size,
	uint64_t type_num, uint64_t flags,
	pmemobj_constr constructor, void *arg)
{
	LOG(3, "pop %p oidp %p size %zu type_num %llx flags %llx "
		"constructor %p arg %p",
		pop, oidp, size, (unsigned long long)type_num,
		(unsigned long long)flags,
		constructor, arg);

	/* log notice message if used inside a transaction */
	_POBJ_DEBUG_NOTICE_IN_TX();

	if (size == 0) {
		ERR("allocation with size 0");
		errno = EINVAL;
		return -1;
	}

	if (flags & ~POBJ_TX_XALLOC_VALID_FLAGS) {
		ERR("unknown flags 0x%" PRIx64,
				flags & ~POBJ_TX_XALLOC_VALID_FLAGS);
		errno = EINVAL;
		return -1;
	}

	PMEMOBJ_API_START();
	int ret = obj_alloc_construct(pop, oidp, size, type_num,
			flags, constructor, arg);

	PMEMOBJ_API_END();
	return ret;
}

/* arguments for constructor_realloc and constructor_zrealloc */
struct carg_realloc {
	void *ptr;
	size_t old_size;
	size_t new_size;
	int zero_init;
	type_num_t user_type;
	pmemobj_constr constructor;
	void *arg;
};

/*
 * pmemobj_zalloc -- allocates a new zeroed object
 */
int
pmemobj_zalloc(PMEMobjpool *pop, PMEMoid *oidp, size_t size,
		uint64_t type_num)
{
	LOG(3, "pop %p oidp %p size %zu type_num %llx",
			pop, oidp, size, (unsigned long long)type_num);

	/* log notice message if used inside a transaction */
	_POBJ_DEBUG_NOTICE_IN_TX();

	if (size == 0) {
		ERR("allocation with size 0");
		errno = EINVAL;
		return -1;
	}

	PMEMOBJ_API_START();
	int ret = obj_alloc_construct(pop, oidp, size, type_num, POBJ_FLAG_ZERO,
		NULL, NULL);

	PMEMOBJ_API_END();
	return ret;
}

/*
 * obj_free -- (internal) free an object
 */
static void
obj_free(PMEMobjpool *pop, PMEMoid *oidp)
{
	ASSERTne(oidp, NULL);

	struct operation_context *ctx = pmalloc_operation_hold(pop);

	operation_add_entry(ctx, &oidp->pool_uuid_lo, 0, ULOG_OPERATION_SET);

	palloc_operation(&pop->heap, oidp->off, &oidp->off, 0, NULL, NULL,
			0, 0, 0, 0, ctx);

	pmalloc_operation_release(pop);
}

/*
 * constructor_realloc -- (internal) constructor for pmemobj_realloc
 */
static int
constructor_realloc(void *ctx, void *ptr, size_t usable_size, void *arg)
{
	PMEMobjpool *pop = ctx;
	LOG(3, "pop %p ptr %p arg %p", pop, ptr, arg);
	struct pmem_ops *p_ops = &pop->p_ops;

	ASSERTne(ptr, NULL);
	ASSERTne(arg, NULL);

	struct carg_realloc *carg = arg;

	if (!carg->zero_init)
		return 0;

	if (usable_size > carg->old_size) {
		size_t grow_len = usable_size - carg->old_size;
		void *new_data_ptr = (void *)((uintptr_t)ptr + carg->old_size);

		pmemops_memset(p_ops, new_data_ptr, 0, grow_len, 0);
	}

	return 0;
}

/*
 * obj_realloc_common -- (internal) common routine for resizing
 *                          existing objects
 */
static int
obj_realloc_common(PMEMobjpool *pop,
	PMEMoid *oidp, size_t size, type_num_t type_num, int zero_init)
{
	/* if OID is NULL just allocate memory */
	if (OBJ_OID_IS_NULL(*oidp)) {
		/* if size is 0 - do nothing */
		if (size == 0)
			return 0;

		return obj_alloc_construct(pop, oidp, size, type_num,
				POBJ_FLAG_ZERO, NULL, NULL);
	}

	if (size > PMEMOBJ_MAX_ALLOC_SIZE) {
		ERR("requested size too large");
		errno = ENOMEM;
		return -1;
	}

	/* if size is 0 just free */
	if (size == 0) {
		obj_free(pop, oidp);
		return 0;
	}

	struct carg_realloc carg;
	carg.ptr = OBJ_OFF_TO_PTR(pop, oidp->off);
	carg.new_size = size;
	carg.old_size = pmemobj_alloc_usable_size(*oidp);
	carg.user_type = type_num;
	carg.constructor = NULL;
	carg.arg = NULL;
	carg.zero_init = zero_init;

	struct operation_context *ctx = pmalloc_operation_hold(pop);

	int ret = palloc_operation(&pop->heap, oidp->off, &oidp->off,
			size, constructor_realloc, &carg, type_num,
			0, 0, 0, ctx);

	pmalloc_operation_release(pop);

	return ret;
}

/*
 * constructor_zrealloc_root -- (internal) constructor for pmemobj_root
 */
static int
constructor_zrealloc_root(void *ctx, void *ptr, size_t usable_size, void *arg)
{
	PMEMobjpool *pop = ctx;
	LOG(3, "pop %p ptr %p arg %p", pop, ptr, arg);

	ASSERTne(ptr, NULL);
	ASSERTne(arg, NULL);

	VALGRIND_ADD_TO_TX(ptr, usable_size);

	struct carg_realloc *carg = arg;

	constructor_realloc(pop, ptr, usable_size, arg);
	int ret = 0;
	if (carg->constructor)
		ret = carg->constructor(pop, ptr, carg->arg);

	VALGRIND_REMOVE_FROM_TX(ptr, usable_size);

	return ret;
}

/*
 * pmemobj_realloc -- resizes an existing object
 */
int
pmemobj_realloc(PMEMobjpool *pop, PMEMoid *oidp, size_t size,
		uint64_t type_num)
{
	ASSERTne(oidp, NULL);

	LOG(3, "pop %p oid.off 0x%016" PRIx64 " size %zu type_num %" PRIu64,
		pop, oidp->off, size, type_num);

	PMEMOBJ_API_START();
	/* log notice message if used inside a transaction */
	_POBJ_DEBUG_NOTICE_IN_TX();
	ASSERT(OBJ_OID_IS_VALID(pop, *oidp));

	int ret = obj_realloc_common(pop, oidp, size, (type_num_t)type_num, 0);

	PMEMOBJ_API_END();
	return ret;
}

/*
 * pmemobj_zrealloc -- resizes an existing object, any new space is zeroed.
 */
int
pmemobj_zrealloc(PMEMobjpool *pop, PMEMoid *oidp, size_t size,
		uint64_t type_num)
{
	ASSERTne(oidp, NULL);

	LOG(3, "pop %p oid.off 0x%016" PRIx64 " size %zu type_num %" PRIu64,
		pop, oidp->off, size, type_num);

	PMEMOBJ_API_START();

	/* log notice message if used inside a transaction */
	_POBJ_DEBUG_NOTICE_IN_TX();
	ASSERT(OBJ_OID_IS_VALID(pop, *oidp));

	int ret = obj_realloc_common(pop, oidp, size, (type_num_t)type_num, 1);

	PMEMOBJ_API_END();
	return ret;
}

/* arguments for constructor_strdup */
struct carg_strdup {
	size_t size;
	const char *s;
};

/*
 * constructor_strdup -- (internal) constructor of pmemobj_strdup
 */
static int
constructor_strdup(PMEMobjpool *pop, void *ptr, void *arg)
{
	LOG(3, "pop %p ptr %p arg %p", pop, ptr, arg);

	ASSERTne(ptr, NULL);
	ASSERTne(arg, NULL);

	struct carg_strdup *carg = arg;

	/* copy string */
	pmemops_memcpy(&pop->p_ops, ptr, carg->s, carg->size, 0);

	return 0;
}

/*
 * pmemobj_strdup -- allocates a new object with duplicate of the string s.
 */
int
pmemobj_strdup(PMEMobjpool *pop, PMEMoid *oidp, const char *s,
		uint64_t type_num)
{
	LOG(3, "pop %p oidp %p string %s type_num %" PRIu64,
	    pop, oidp, s, type_num);

	/* log notice message if used inside a transaction */
	_POBJ_DEBUG_NOTICE_IN_TX();

	if (NULL == s) {
		errno = EINVAL;
		return -1;
	}

	PMEMOBJ_API_START();
	struct carg_strdup carg;
	carg.size = (strlen(s) + 1) * sizeof(char);
	carg.s = s;

	int ret = obj_alloc_construct(pop, oidp, carg.size,
		(type_num_t)type_num, 0, constructor_strdup, &carg);

	PMEMOBJ_API_END();
	return ret;
}

/* arguments for constructor_wcsdup */
struct carg_wcsdup {
	size_t size;
	const wchar_t *s;
};

/*
 * constructor_wcsdup -- (internal) constructor of pmemobj_wcsdup
 */
static int
constructor_wcsdup(PMEMobjpool *pop, void *ptr, void *arg)
{
	LOG(3, "pop %p ptr %p arg %p", pop, ptr, arg);

	ASSERTne(ptr, NULL);
	ASSERTne(arg, NULL);

	struct carg_wcsdup *carg = arg;

	/* copy string */
	pmemops_memcpy(&pop->p_ops, ptr, carg->s, carg->size, 0);

	return 0;
}

/*
 * pmemobj_wcsdup -- allocates a new object with duplicate of the wide character
 * string s.
 */
int
pmemobj_wcsdup(PMEMobjpool *pop, PMEMoid *oidp, const wchar_t *s,
	uint64_t type_num)
{
	LOG(3, "pop %p oidp %p string %S type_num %" PRIu64,
		    pop, oidp, s, type_num);

	/* log notice message if used inside a transaction */
	_POBJ_DEBUG_NOTICE_IN_TX();

	if (NULL == s) {
		errno = EINVAL;
		return -1;
	}

	PMEMOBJ_API_START();
	struct carg_wcsdup carg;
	carg.size = (wcslen(s) + 1) * sizeof(wchar_t);
	carg.s = s;

	int ret = obj_alloc_construct(pop, oidp, carg.size,
		(type_num_t)type_num, 0, constructor_wcsdup, &carg);

	PMEMOBJ_API_END();
	return ret;
}

/*
 * pmemobj_free -- frees an existing object
 */
void
pmemobj_free(PMEMoid *oidp)
{
	ASSERTne(oidp, NULL);

	LOG(3, "oid.off 0x%016" PRIx64, oidp->off);

	/* log notice message if used inside a transaction */
	_POBJ_DEBUG_NOTICE_IN_TX();

	if (oidp->off == 0)
		return;

	PMEMOBJ_API_START();
	PMEMobjpool *pop = pmemobj_pool_by_oid(*oidp);

	ASSERTne(pop, NULL);
	ASSERT(OBJ_OID_IS_VALID(pop, *oidp));

	obj_free(pop, oidp);
	PMEMOBJ_API_END();
}

/*
 * pmemobj_alloc_usable_size -- returns usable size of object
 */
size_t
pmemobj_alloc_usable_size(PMEMoid oid)
{
	LOG(3, "oid.off 0x%016" PRIx64, oid.off);

	if (oid.off == 0)
		return 0;

	PMEMobjpool *pop = pmemobj_pool_by_oid(oid);

	ASSERTne(pop, NULL);
	ASSERT(OBJ_OID_IS_VALID(pop, oid));

	return (palloc_usable_size(&pop->heap, oid.off));
}

/*
 * pmemobj_memcpy_persist -- pmemobj version of memcpy
 */
void *
pmemobj_memcpy_persist(PMEMobjpool *pop, void *dest, const void *src,
	size_t len)
{
	LOG(15, "pop %p dest %p src %p len %zu", pop, dest, src, len);
	PMEMOBJ_API_START();

	void *ptr = pmemops_memcpy(&pop->p_ops, dest, src, len, 0);

	PMEMOBJ_API_END();
	return ptr;
}

/*
 * pmemobj_memset_persist -- pmemobj version of memset
 */
void *
pmemobj_memset_persist(PMEMobjpool *pop, void *dest, int c, size_t len)
{
	LOG(15, "pop %p dest %p c 0x%02x len %zu", pop, dest, c, len);
	PMEMOBJ_API_START();

	void *ptr = pmemops_memset(&pop->p_ops, dest, c, len, 0);

	PMEMOBJ_API_END();
	return ptr;
}

/*
 * pmemobj_memcpy -- pmemobj version of memcpy
 */
void *
pmemobj_memcpy(PMEMobjpool *pop, void *dest, const void *src, size_t len,
		unsigned flags)
{
	LOG(15, "pop %p dest %p src %p len %zu flags 0x%x", pop, dest, src, len,
			flags);

	PMEMOBJ_API_START();

	void *ptr = pmemops_memcpy(&pop->p_ops, dest, src, len, flags);

	PMEMOBJ_API_END();
	return ptr;
}

/*
 * pmemobj_memmove -- pmemobj version of memmove
 */
void *
pmemobj_memmove(PMEMobjpool *pop, void *dest, const void *src, size_t len,
		unsigned flags)
{
	LOG(15, "pop %p dest %p src %p len %zu flags 0x%x", pop, dest, src, len,
			flags);

	PMEMOBJ_API_START();

	void *ptr = pmemops_memmove(&pop->p_ops, dest, src, len, flags);

	PMEMOBJ_API_END();
	return ptr;
}

/*
 * pmemobj_memset -- pmemobj version of memset
 */
void *
pmemobj_memset(PMEMobjpool *pop, void *dest, int c, size_t len, unsigned flags)
{
	LOG(15, "pop %p dest %p c 0x%02x len %zu flags 0x%x", pop, dest, c, len,
			flags);

	PMEMOBJ_API_START();

	void *ptr = pmemops_memset(&pop->p_ops, dest, c, len, flags);

	PMEMOBJ_API_END();
	return ptr;
}

/*
 * pmemobj_persist -- pmemobj version of pmem_persist
 */
void
pmemobj_persist(PMEMobjpool *pop, const void *addr, size_t len)
{
	LOG(15, "pop %p addr %p len %zu", pop, addr, len);

	pmemops_persist(&pop->p_ops, addr, len);
}

/*
 * pmemobj_flush -- pmemobj version of pmem_flush
 */
void
pmemobj_flush(PMEMobjpool *pop, const void *addr, size_t len)
{
	LOG(15, "pop %p addr %p len %zu", pop, addr, len);

	pmemops_flush(&pop->p_ops, addr, len);
}

/*
 * pmemobj_xpersist -- pmemobj version of pmem_persist with additional flags
 * argument
 */
int
pmemobj_xpersist(PMEMobjpool *pop, const void *addr, size_t len, unsigned flags)
{
	LOG(15, "pop %p addr %p len %zu", pop, addr, len);

	if (flags & ~OBJ_X_VALID_FLAGS) {
		errno = EINVAL;
		ERR("invalid flags 0x%x", flags);
		return -1;
	}

	return pmemops_xpersist(&pop->p_ops, addr, len, flags);
}

/*
 * pmemobj_xflush -- pmemobj version of pmem_flush with additional flags
 * argument
 */
int
pmemobj_xflush(PMEMobjpool *pop, const void *addr, size_t len, unsigned flags)
{
	LOG(15, "pop %p addr %p len %zu", pop, addr, len);

	if (flags & ~OBJ_X_VALID_FLAGS) {
		errno = EINVAL;
		ERR("invalid flags 0x%x", flags);
		return -1;
	}

	return pmemops_xflush(&pop->p_ops, addr, len, flags);
}

/*
 * pmemobj_drain -- pmemobj version of pmem_drain
 */
void
pmemobj_drain(PMEMobjpool *pop)
{
	LOG(15, "pop %p", pop);

	pmemops_drain(&pop->p_ops);
}

/*
 * pmemobj_type_num -- returns type number of object
 */
uint64_t
pmemobj_type_num(PMEMoid oid)
{
	LOG(3, "oid.off 0x%016" PRIx64, oid.off);

	ASSERT(!OID_IS_NULL(oid));

	PMEMobjpool *pop = pmemobj_pool_by_oid(oid);

	ASSERTne(pop, NULL);
	ASSERT(OBJ_OID_IS_VALID(pop, oid));

	return palloc_extra(&pop->heap, oid.off);
}

/* arguments for constructor_alloc_root */
struct carg_root {
	size_t size;
	pmemobj_constr constructor;
	void *arg;
};

/*
 * obj_realloc_root -- (internal) reallocate root object
 */
static int
obj_alloc_root(PMEMobjpool *pop, size_t size,
	pmemobj_constr constructor, void *arg)
{
	LOG(3, "pop %p size %zu", pop, size);

	struct carg_realloc carg;

	carg.ptr = OBJ_OFF_TO_PTR(pop, pop->root_offset);
	carg.old_size = pop->root_size;
	carg.new_size = size;
	carg.user_type = POBJ_ROOT_TYPE_NUM;
	carg.constructor = constructor;
	carg.zero_init = 1;
	carg.arg = arg;

	struct operation_context *ctx = pmalloc_operation_hold(pop);

	operation_add_entry(ctx, &pop->root_size, size, ULOG_OPERATION_SET);

	int ret = palloc_operation(&pop->heap, pop->root_offset,
			&pop->root_offset, size,
			constructor_zrealloc_root, &carg,
			POBJ_ROOT_TYPE_NUM, OBJ_INTERNAL_OBJECT_MASK,
			0, 0, ctx);

	pmalloc_operation_release(pop);

	return ret;
}

/*
 * pmemobj_root_size -- returns size of the root object
 */
size_t
pmemobj_root_size(PMEMobjpool *pop)
{
	LOG(3, "pop %p", pop);

	if (pop->root_offset && pop->root_size) {
		return pop->root_size;
	} else
		return 0;
}

/*
 * pmemobj_root_construct -- returns root object
 */
PMEMoid
pmemobj_root_construct(PMEMobjpool *pop, size_t size,
	pmemobj_constr constructor, void *arg)
{
	LOG(3, "pop %p size %zu constructor %p args %p", pop, size, constructor,
		arg);

	if (size > PMEMOBJ_MAX_ALLOC_SIZE) {
		ERR("requested size too large");
		errno = ENOMEM;
		return OID_NULL;
	}

	if (size == 0 && pop->root_offset == 0) {
		ERR("requested size cannot equals zero");
		errno = EINVAL;
		return OID_NULL;
	}

	PMEMOBJ_API_START();

	PMEMoid root;

	pmemobj_mutex_lock_nofail(pop, &pop->rootlock);

	if (size > pop->root_size &&
		obj_alloc_root(pop, size, constructor, arg)) {
		pmemobj_mutex_unlock_nofail(pop, &pop->rootlock);
		LOG(2, "obj_realloc_root failed");
		PMEMOBJ_API_END();
		return OID_NULL;
	}

	root.pool_uuid_lo = pop->uuid_lo;
	root.off = pop->root_offset;

	pmemobj_mutex_unlock_nofail(pop, &pop->rootlock);

	PMEMOBJ_API_END();
	return root;
}

/*
 * pmemobj_root -- returns root object
 */
PMEMoid
pmemobj_root(PMEMobjpool *pop, size_t size)
{
	LOG(3, "pop %p size %zu", pop, size);

	PMEMOBJ_API_START();
	PMEMoid oid = pmemobj_root_construct(pop, size, NULL, NULL);
	PMEMOBJ_API_END();
	return oid;
}

/*
 * pmemobj_first - returns first object of specified type
 */
PMEMoid
pmemobj_first(PMEMobjpool *pop)
{
	LOG(3, "pop %p", pop);

	PMEMoid ret = {0, 0};

	uint64_t off = palloc_first(&pop->heap);
	if (off != 0) {
		ret.off = off;
		ret.pool_uuid_lo = pop->uuid_lo;

		if (palloc_flags(&pop->heap, off) & OBJ_INTERNAL_OBJECT_MASK) {
			return pmemobj_next(ret);
		}
	}

	return ret;
}

/*
 * pmemobj_next - returns next object of specified type
 */
PMEMoid
pmemobj_next(PMEMoid oid)
{
	LOG(3, "oid.off 0x%016" PRIx64, oid.off);

	PMEMoid curr = oid;
	if (curr.off == 0)
		return OID_NULL;

	PMEMobjpool *pop = pmemobj_pool_by_oid(curr);
	ASSERTne(pop, NULL);

	do {
		ASSERT(OBJ_OID_IS_VALID(pop, curr));
		uint64_t next_off = palloc_next(&pop->heap, curr.off);

		if (next_off == 0)
			return OID_NULL;

		/* next object exists */
		curr.off = next_off;

	} while (palloc_flags(&pop->heap, curr.off) & OBJ_INTERNAL_OBJECT_MASK);

	return curr;
}

/*
 * pmemobj_reserve -- reserves a single object
 */
PMEMoid
pmemobj_reserve(PMEMobjpool *pop, struct pobj_action *act,
	size_t size, uint64_t type_num)
{
	LOG(3, "pop %p act %p size %zu type_num %llx",
		pop, act, size,
		(unsigned long long)type_num);

	PMEMOBJ_API_START();
	PMEMoid oid = OID_NULL;

	if (palloc_reserve(&pop->heap, size, NULL, NULL, type_num,
		0, 0, 0, act) != 0) {
		PMEMOBJ_API_END();
		return oid;
	}

	oid.off = act->heap.offset;
	oid.pool_uuid_lo = pop->uuid_lo;

	PMEMOBJ_API_END();
	return oid;
}

/*
 * pmemobj_xreserve -- reserves a single object
 */
PMEMoid
pmemobj_xreserve(PMEMobjpool *pop, struct pobj_action *act,
	size_t size, uint64_t type_num, uint64_t flags)
{
	LOG(3, "pop %p act %p size %zu type_num %llx flags %llx",
		pop, act, size,
		(unsigned long long)type_num, (unsigned long long)flags);

	PMEMoid oid = OID_NULL;

	if (flags & ~POBJ_ACTION_XRESERVE_VALID_FLAGS) {
		ERR("unknown flags 0x%" PRIx64,
				flags & ~POBJ_ACTION_XRESERVE_VALID_FLAGS);
		errno = EINVAL;
		return oid;
	}

	PMEMOBJ_API_START();
	struct constr_args carg;

	carg.zero_init = flags & POBJ_FLAG_ZERO;
	carg.constructor = NULL;
	carg.arg = NULL;

	if (palloc_reserve(&pop->heap, size, constructor_alloc, &carg,
		type_num, 0, CLASS_ID_FROM_FLAG(flags),
		ARENA_ID_FROM_FLAG(flags), act) != 0) {
		PMEMOBJ_API_END();
		return oid;
	}

	oid.off = act->heap.offset;
	oid.pool_uuid_lo = pop->uuid_lo;

	PMEMOBJ_API_END();
	return oid;
}

/*
 * pmemobj_set_value -- creates an action to set a value
 */
void
pmemobj_set_value(PMEMobjpool *pop, struct pobj_action *act,
	uint64_t *ptr, uint64_t value)
{
	palloc_set_value(&pop->heap, act, ptr, value);
}

/*
 * pmemobj_defer_free -- creates a deferred free action
 */
void
pmemobj_defer_free(PMEMobjpool *pop, PMEMoid oid, struct pobj_action *act)
{
	ASSERT(!OID_IS_NULL(oid));
	palloc_defer_free(&pop->heap, oid.off, act);
}

/*
 * pmemobj_publish -- publishes a collection of actions
 */
int
pmemobj_publish(PMEMobjpool *pop, struct pobj_action *actv, size_t actvcnt)
{
	PMEMOBJ_API_START();
	struct operation_context *ctx = pmalloc_operation_hold(pop);

	size_t entries_size = actvcnt * sizeof(struct ulog_entry_val);

	if (operation_reserve(ctx, entries_size) != 0) {
		PMEMOBJ_API_END();
		return -1;
	}

	palloc_publish(&pop->heap, actv, actvcnt, ctx);

	pmalloc_operation_release(pop);

	PMEMOBJ_API_END();
	return 0;
}

/*
 * pmemobj_cancel -- cancels collection of actions
 */
void
pmemobj_cancel(PMEMobjpool *pop, struct pobj_action *actv, size_t actvcnt)
{
	PMEMOBJ_API_START();
	palloc_cancel(&pop->heap, actv, actvcnt);
	PMEMOBJ_API_END();
}

/*
 * pmemobj_defrag -- reallocates provided PMEMoids so that the underlying memory
 *	is efficiently arranged.
 */
int
pmemobj_defrag(PMEMobjpool *pop, PMEMoid **oidv, size_t oidcnt,
	struct pobj_defrag_result *result)
{
	PMEMOBJ_API_START();

	if (result) {
		result->relocated = 0;
		result->total = 0;
	}

	uint64_t **objv = Malloc(sizeof(uint64_t *) * oidcnt);
	if (objv == NULL)
		return -1;

	int ret = 0;

	size_t j = 0;
	for (size_t i = 0; i < oidcnt; ++i) {
		if (OID_IS_NULL(*oidv[i]))
			continue;
		if (oidv[i]->pool_uuid_lo != pop->uuid_lo) {
			ret = -1;
			ERR("Not all PMEMoids belong to the provided pool");
			goto out;
		}
		objv[j++] = &oidv[i]->off;
	}

	struct operation_context *ctx = pmalloc_operation_hold(pop);

	ret = palloc_defrag(&pop->heap, objv, j, ctx, result);

	pmalloc_operation_release(pop);

out:
	Free(objv);

	PMEMOBJ_API_END();
	return ret;
}

/*
 * pmemobj_list_insert -- adds object to a list
 */
int
pmemobj_list_insert(PMEMobjpool *pop, size_t pe_offset, void *head,
		    PMEMoid dest, int before, PMEMoid oid)
{
	LOG(3, "pop %p pe_offset %zu head %p dest.off 0x%016" PRIx64
	    " before %d oid.off 0x%016" PRIx64,
	    pop, pe_offset, head, dest.off, before, oid.off);
	PMEMOBJ_API_START();

	/* log notice message if used inside a transaction */
	_POBJ_DEBUG_NOTICE_IN_TX();
	ASSERT(OBJ_OID_IS_VALID(pop, oid));
	ASSERT(OBJ_OID_IS_VALID(pop, dest));

	ASSERT(pe_offset <= pmemobj_alloc_usable_size(dest)
			- sizeof(struct list_entry));
	ASSERT(pe_offset <= pmemobj_alloc_usable_size(oid)
			- sizeof(struct list_entry));

	int ret = list_insert(pop, (ssize_t)pe_offset, head, dest, before, oid);

	PMEMOBJ_API_END();
	return ret;
}

/*
 * pmemobj_list_insert_new -- adds new object to a list
 */
PMEMoid
pmemobj_list_insert_new(PMEMobjpool *pop, size_t pe_offset, void *head,
			PMEMoid dest, int before, size_t size,
			uint64_t type_num,
			pmemobj_constr constructor, void *arg)
{
	LOG(3, "pop %p pe_offset %zu head %p dest.off 0x%016" PRIx64
	    " before %d size %zu type_num %" PRIu64,
	    pop, pe_offset, head, dest.off, before, size, type_num);

	/* log notice message if used inside a transaction */
	_POBJ_DEBUG_NOTICE_IN_TX();
	ASSERT(OBJ_OID_IS_VALID(pop, dest));

	ASSERT(pe_offset <= pmemobj_alloc_usable_size(dest)
			- sizeof(struct list_entry));
	ASSERT(pe_offset <= size - sizeof(struct list_entry));

	if (size > PMEMOBJ_MAX_ALLOC_SIZE) {
		ERR("requested size too large");
		errno = ENOMEM;
		return OID_NULL;
	}

	PMEMOBJ_API_START();
	struct constr_args carg;

	carg.constructor = constructor;
	carg.arg = arg;
	carg.zero_init = 0;

	PMEMoid retoid = OID_NULL;
	list_insert_new_user(pop, pe_offset, head, dest, before, size, type_num,
			constructor_alloc, &carg, &retoid);

	PMEMOBJ_API_END();
	return retoid;
}

/*
 * pmemobj_list_remove -- removes object from a list
 */
int
pmemobj_list_remove(PMEMobjpool *pop, size_t pe_offset, void *head,
		    PMEMoid oid, int free)
{
	LOG(3, "pop %p pe_offset %zu head %p oid.off 0x%016" PRIx64 " free %d",
	    pop, pe_offset, head, oid.off, free);
	PMEMOBJ_API_START();

	/* log notice message if used inside a transaction */
	_POBJ_DEBUG_NOTICE_IN_TX();
	ASSERT(OBJ_OID_IS_VALID(pop, oid));

	ASSERT(pe_offset <= pmemobj_alloc_usable_size(oid)
			- sizeof(struct list_entry));

	int ret;
	if (free)
		ret = list_remove_free_user(pop, pe_offset, head, &oid);
	else
		ret = list_remove(pop, (ssize_t)pe_offset, head, oid);

	PMEMOBJ_API_END();
	return ret;
}

/*
 * pmemobj_list_move -- moves object between lists
 */
int
pmemobj_list_move(PMEMobjpool *pop, size_t pe_old_offset, void *head_old,
			size_t pe_new_offset, void *head_new,
			PMEMoid dest, int before, PMEMoid oid)
{
	LOG(3, "pop %p pe_old_offset %zu pe_new_offset %zu"
	    " head_old %p head_new %p dest.off 0x%016" PRIx64
	    " before %d oid.off 0x%016" PRIx64 "",
	    pop, pe_old_offset, pe_new_offset,
	    head_old, head_new, dest.off, before, oid.off);
	PMEMOBJ_API_START();

	/* log notice message if used inside a transaction */
	_POBJ_DEBUG_NOTICE_IN_TX();

	ASSERT(OBJ_OID_IS_VALID(pop, oid));
	ASSERT(OBJ_OID_IS_VALID(pop, dest));

	ASSERT(pe_old_offset <= pmemobj_alloc_usable_size(oid)
			- sizeof(struct list_entry));
	ASSERT(pe_new_offset <= pmemobj_alloc_usable_size(oid)
			- sizeof(struct list_entry));
	ASSERT(pe_old_offset <= pmemobj_alloc_usable_size(dest)
			- sizeof(struct list_entry));
	ASSERT(pe_new_offset <= pmemobj_alloc_usable_size(dest)
			- sizeof(struct list_entry));

	int ret = list_move(pop, pe_old_offset, head_old,
				pe_new_offset, head_new,
				dest, before, oid);

	PMEMOBJ_API_END();
	return ret;
}

/*
 * pmemobj_ctl_getU -- programmatically executes a read ctl query
 */
static inline
int
pmemobj_ctl_getU(PMEMobjpool *pop, const char *name, void *arg)
{
	LOG(3, "pop %p name %s arg %p", pop, name, arg);
	return ctl_query(pop == NULL ? NULL : pop->ctl, pop,
			CTL_QUERY_PROGRAMMATIC, name, CTL_QUERY_READ, arg);
}

/*
 * pmemobj_ctl_setU -- programmatically executes a write ctl query
 */
static inline
int
pmemobj_ctl_setU(PMEMobjpool *pop, const char *name, void *arg)
{
	LOG(3, "pop %p name %s arg %p", pop, name, arg);
	return ctl_query(pop == NULL ? NULL : pop->ctl, pop,
		CTL_QUERY_PROGRAMMATIC, name, CTL_QUERY_WRITE, arg);
}

/*
 * pmemobj_ctl_execU -- programmatically executes a runnable ctl query
 */
static inline
int
pmemobj_ctl_execU(PMEMobjpool *pop, const char *name, void *arg)
{
	LOG(3, "pop %p name %s arg %p", pop, name, arg);
	return ctl_query(pop == NULL ? NULL : pop->ctl, pop,
		CTL_QUERY_PROGRAMMATIC, name, CTL_QUERY_RUNNABLE, arg);
}

/*
 * pmemobj_ctl_get -- programmatically executes a read ctl query
 */
int
pmemobj_ctl_get(PMEMobjpool *pop, const char *name, void *arg)
{
	return pmemobj_ctl_getU(pop, name, arg);
}

/*
 * pmemobj_ctl_set -- programmatically executes a write ctl query
 */
int
pmemobj_ctl_set(PMEMobjpool *pop, const char *name, void *arg)
{
	PMEMOBJ_API_START();

	int ret = pmemobj_ctl_setU(pop, name, arg);

	PMEMOBJ_API_END();
	return ret;
}

/*
 * pmemobj_ctl_exec -- programmatically executes a runnable ctl query
 */
int
pmemobj_ctl_exec(PMEMobjpool *pop, const char *name, void *arg)
{
	PMEMOBJ_API_START();

	int ret =  pmemobj_ctl_execU(pop, name, arg);

	PMEMOBJ_API_END();
	return ret;
}

/*
 * _pobj_debug_notice -- logs notice message if used inside a transaction
 */
void
_pobj_debug_notice(const char *api_name, const char *file, int line)
{
	/* suppress unused-parameter errors */
	SUPPRESS_UNUSED(api_name, file, line);

#ifdef DEBUG
	if (pmemobj_tx_stage() != TX_STAGE_NONE) {
		if (file)
			LOG(4, "Notice: non-transactional API"
				" used inside a transaction (%s in %s:%d)",
				api_name, file, line);
		else
			LOG(4, "Notice: non-transactional API"
				" used inside a transaction (%s)", api_name);
	}
#endif /* DEBUG */
}

#if VG_PMEMCHECK_ENABLED
/*
 * pobj_emit_log -- logs library and function names to pmemcheck store log
 */
void
pobj_emit_log(const char *func, int order)
{
	util_emit_log("libpmemobj", func, order);
}
#endif

#if FAULT_INJECTION
void
pmemobj_inject_fault_at(enum pmem_allocation_type type, int nth,
							const char *at)
{
	core_inject_fault_at(type, nth, at);
}

int
pmemobj_fault_injection_enabled(void)
{
	return core_fault_injection_enabled();
}
#endif