tx.c

// SPDX-License-Identifier: BSD-3-Clause
/* Copyright 2015-2020, Intel Corporation */

/*
 * tx.c -- transactions implementation
 */

#include <inttypes.h>
#include <wchar.h>

#include "queue.h"
#include "ravl.h"
#include "obj.h"
#include "out.h"
#include "pmalloc.h"
#include "tx.h"
#include "valgrind_internal.h"
#include "memops.h"

struct tx_data {
	PMDK_SLIST_ENTRY(tx_data) tx_entry;
	jmp_buf env;
	enum pobj_tx_action_on_failure action_on_failure;
};

struct tx {
	PMEMobjpool *pop;
	enum pobj_tx_stage stage;
	int last_errnum;
	struct lane *lane;
	PMDK_SLIST_HEAD(txl, tx_lock_data) tx_locks;
	PMDK_SLIST_HEAD(txd, tx_data) tx_entries;

	struct ravl *ranges;

	VEC(, struct pobj_action) actions;
	VEC(, struct user_buffer_def) redo_userbufs;
	size_t redo_userbufs_capacity;

	pmemobj_tx_callback stage_callback;
	void *stage_callback_arg;

	int first_snapshot;

	void *user_data;
};

/*
 * get_tx -- (internal) returns current transaction
 *
 * This function should be used only in high-level functions.
 */
static struct tx *
get_tx()
{
	static __thread struct tx tx;
	return &tx;
}

struct tx_lock_data {
	union {
		PMEMmutex *mutex;
		PMEMrwlock *rwlock;
	} lock;
	enum pobj_tx_param lock_type;
	PMDK_SLIST_ENTRY(tx_lock_data) tx_lock;
};

struct tx_alloc_args {
	uint64_t flags;
	const void *copy_ptr;
	size_t copy_size;
};

#define COPY_ARGS(flags, copy_ptr, copy_size)\
(struct tx_alloc_args){flags, copy_ptr, copy_size}

#define ALLOC_ARGS(flags)\
(struct tx_alloc_args){flags, NULL, 0}

struct tx_range_def {
	uint64_t offset;
	uint64_t size;
	uint64_t flags;
};

/*
 * tx_range_def_cmp -- compares two snapshot ranges
 */
static int
tx_range_def_cmp(const void *lhs, const void *rhs)
{
	const struct tx_range_def *l = lhs;
	const struct tx_range_def *r = rhs;

	if (l->offset > r->offset)
		return 1;
	else if (l->offset < r->offset)
		return -1;

	return 0;
}

/*
 * tx_params_new -- creates a new transactional parameters instance and fills it
 *	with default values.
 */
struct tx_parameters *
tx_params_new(void)
{
	struct tx_parameters *tx_params = Malloc(sizeof(*tx_params));
	if (tx_params == NULL)
		return NULL;

	tx_params->cache_size = TX_DEFAULT_RANGE_CACHE_SIZE;

	return tx_params;
}

/*
 * tx_params_delete -- deletes transactional parameters instance
 */
void
tx_params_delete(struct tx_parameters *tx_params)
{
	Free(tx_params);
}

static void
obj_tx_abort(int errnum, int user);

/*
 * obj_tx_fail_err -- (internal) pmemobj_tx_abort variant that returns
 * error code
 */
static inline int
obj_tx_fail_err(int errnum, uint64_t flags)
{
	if ((flags & POBJ_FLAG_TX_NO_ABORT) == 0)
		obj_tx_abort(errnum, 0);
	errno = errnum;
	return errnum;
}

/*
 * obj_tx_fail_null -- (internal) pmemobj_tx_abort variant that returns
 * null PMEMoid
 */
static inline PMEMoid
obj_tx_fail_null(int errnum, uint64_t flags)
{
	if ((flags & POBJ_FLAG_TX_NO_ABORT) == 0)
		obj_tx_abort(errnum, 0);
	errno = errnum;
	return OID_NULL;
}

/* ASSERT_IN_TX -- checks whether there's open transaction */
#define ASSERT_IN_TX(tx) do {\
	if ((tx)->stage == TX_STAGE_NONE)\
		FATAL("%s called outside of transaction", __func__);\
} while (0)

/* ASSERT_TX_STAGE_WORK -- checks whether current transaction stage is WORK */
#define ASSERT_TX_STAGE_WORK(tx) do {\
	if ((tx)->stage != TX_STAGE_WORK)\
		FATAL("%s called in invalid stage %d", __func__, (tx)->stage);\
} while (0)

/*
 * tx_action_reserve -- (internal) reserve space for the given number of actions
 */
static int
tx_action_reserve(struct tx *tx, size_t n)
{
	size_t entries_size = (VEC_SIZE(&tx->actions) + n) *
		sizeof(struct ulog_entry_val);

	/* take the provided user buffers into account when reserving */
	entries_size -= MIN(tx->redo_userbufs_capacity, entries_size);

	if (operation_reserve(tx->lane->external, entries_size) != 0)
		return -1;

	return 0;
}

/*
 * tx_action_add -- (internal) reserve space and add a new tx action
 */
static struct pobj_action *
tx_action_add(struct tx *tx)
{
	if (tx_action_reserve(tx, 1) != 0)
		return NULL;

	VEC_INC_BACK(&tx->actions);

	return &VEC_BACK(&tx->actions);
}

/*
 * tx_action_remove -- (internal) remove last tx action
 */
static void
tx_action_remove(struct tx *tx)
{
	VEC_POP_BACK(&tx->actions);
}

/*
 * constructor_tx_alloc -- (internal) constructor for normal alloc
 */
static int
constructor_tx_alloc(void *ctx, void *ptr, size_t usable_size, void *arg)
{
	LOG(5, NULL);

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

	struct tx_alloc_args *args = arg;

	/* do not report changes to the new object */
	VALGRIND_ADD_TO_TX(ptr, usable_size);

	if (args->flags & POBJ_FLAG_ZERO)
		memset(ptr, 0, usable_size);

	if (args->copy_ptr && args->copy_size != 0) {
		memcpy(ptr, args->copy_ptr, args->copy_size);
	}

	return 0;
}

struct tx_range_data {
	void *begin;
	void *end;
	PMDK_SLIST_ENTRY(tx_range_data) tx_range;
};

PMDK_SLIST_HEAD(txr, tx_range_data);

/*
 * tx_remove_range -- (internal) removes specified range from ranges list
 */
static void
tx_remove_range(struct txr *tx_ranges, void *begin, void *end)
{
	struct tx_range_data *txr = PMDK_SLIST_FIRST(tx_ranges);

	while (txr) {
		if (begin >= txr->end || end < txr->begin) {
			txr = PMDK_SLIST_NEXT(txr, tx_range);
			continue;
		}

		LOG(4, "detected PMEM lock in undo log; "
			"range %p-%p, lock %p-%p",
			txr->begin, txr->end, begin, end);

		/* split the range into new ones */
		if (begin > txr->begin) {
			struct tx_range_data *txrn = Malloc(sizeof(*txrn));
			if (txrn == NULL)
				/* we can't do it any other way */
				FATAL("!Malloc");

			txrn->begin = txr->begin;
			txrn->end = begin;
			LOG(4, "range split; %p-%p", txrn->begin, txrn->end);
			PMDK_SLIST_INSERT_HEAD(tx_ranges, txrn, tx_range);
		}

		if (end < txr->end) {
			struct tx_range_data *txrn = Malloc(sizeof(*txrn));
			if (txrn == NULL)
				/* we can't do it any other way */
				FATAL("!Malloc");

			txrn->begin = end;
			txrn->end = txr->end;
			LOG(4, "range split; %p-%p", txrn->begin, txrn->end);
			PMDK_SLIST_INSERT_HEAD(tx_ranges, txrn, tx_range);
		}

		struct tx_range_data *next = PMDK_SLIST_NEXT(txr, tx_range);
		/* remove the original range from the list */
		PMDK_SLIST_REMOVE(tx_ranges, txr, tx_range_data, tx_range);
		Free(txr);

		txr = next;
	}
}
/*
 * tx_restore_range -- (internal) restore a single range from undo log
 *
 * If the snapshot contains any PMEM locks that are held by the current
 * transaction, they won't be overwritten with the saved data to avoid changing
 * their state.  Those locks will be released in tx_end().
 */
static void
tx_restore_range(PMEMobjpool *pop, struct tx *tx, struct ulog_entry_buf *range)
{
	COMPILE_ERROR_ON(sizeof(PMEMmutex) != _POBJ_CL_SIZE);
	COMPILE_ERROR_ON(sizeof(PMEMrwlock) != _POBJ_CL_SIZE);
	COMPILE_ERROR_ON(sizeof(PMEMcond) != _POBJ_CL_SIZE);

	struct txr tx_ranges;
	PMDK_SLIST_INIT(&tx_ranges);

	struct tx_range_data *txr;
	txr = Malloc(sizeof(*txr));
	if (txr == NULL) {
		/* we can't do it any other way */
		FATAL("!Malloc");
	}

	uint64_t range_offset = ulog_entry_offset(&range->base);

	txr->begin = OBJ_OFF_TO_PTR(pop, range_offset);
	txr->end = (char *)txr->begin + range->size;
	PMDK_SLIST_INSERT_HEAD(&tx_ranges, txr, tx_range);

	struct tx_lock_data *txl;

	/* check if there are any locks within given memory range */
	PMDK_SLIST_FOREACH(txl, &tx->tx_locks, tx_lock) {
		void *lock_begin = txl->lock.mutex;
		/* all PMEM locks have the same size */
		void *lock_end = (char *)lock_begin + _POBJ_CL_SIZE;

		tx_remove_range(&tx_ranges, lock_begin, lock_end);
	}

	ASSERT(!PMDK_SLIST_EMPTY(&tx_ranges));

	void *dst_ptr = OBJ_OFF_TO_PTR(pop, range_offset);

	while (!PMDK_SLIST_EMPTY(&tx_ranges)) {
		txr = PMDK_SLIST_FIRST(&tx_ranges);
		PMDK_SLIST_REMOVE_HEAD(&tx_ranges, tx_range);
		/* restore partial range data from snapshot */
		ASSERT((char *)txr->begin >= (char *)dst_ptr);
		uint8_t *src = &range->data[
				(char *)txr->begin - (char *)dst_ptr];
		ASSERT((char *)txr->end >= (char *)txr->begin);
		size_t size = (size_t)((char *)txr->end - (char *)txr->begin);
		pmemops_memcpy(&pop->p_ops, txr->begin, src, size, 0);
		Free(txr);
	}
}

/*
 * tx_undo_entry_apply -- applies modifications of a single ulog entry
 */
static int
tx_undo_entry_apply(struct ulog_entry_base *e, void *arg,
	const struct pmem_ops *p_ops)
{
	struct ulog_entry_buf *eb;

	switch (ulog_entry_type(e)) {
		case ULOG_OPERATION_BUF_CPY:
			eb = (struct ulog_entry_buf *)e;

			tx_restore_range(p_ops->base, get_tx(), eb);
		break;
		case ULOG_OPERATION_AND:
		case ULOG_OPERATION_OR:
		case ULOG_OPERATION_SET:
		case ULOG_OPERATION_BUF_SET:
		default:
			ASSERT(0);
	}

	return 0;
}

/*
 * tx_abort_set -- (internal) abort all set operations
 */
static void
tx_abort_set(PMEMobjpool *pop, struct lane *lane)
{
	LOG(7, NULL);

	ulog_foreach_entry((struct ulog *)&lane->layout->undo,
		tx_undo_entry_apply, NULL, &pop->p_ops);
	operation_finish(lane->undo, ULOG_INC_FIRST_GEN_NUM);
}

/*
 * tx_flush_range -- (internal) flush one range
 */
static void
tx_flush_range(void *data, void *ctx)
{
	PMEMobjpool *pop = ctx;
	struct tx_range_def *range = data;
	if (!(range->flags & POBJ_FLAG_NO_FLUSH)) {
		pmemops_xflush(&pop->p_ops, OBJ_OFF_TO_PTR(pop, range->offset),
				range->size, PMEMOBJ_F_RELAXED);
	}
	VALGRIND_REMOVE_FROM_TX(OBJ_OFF_TO_PTR(pop, range->offset),
		range->size);
}

/*
 * tx_clean_range -- (internal) clean one range
 */
static void
tx_clean_range(void *data, void *ctx)
{
	PMEMobjpool *pop = ctx;
	struct tx_range_def *range = data;
	VALGRIND_REMOVE_FROM_TX(OBJ_OFF_TO_PTR(pop, range->offset),
		range->size);
	VALGRIND_SET_CLEAN(OBJ_OFF_TO_PTR(pop, range->offset), range->size);
}

/*
 * tx_pre_commit -- (internal) do pre-commit operations
 */
static void
tx_pre_commit(struct tx *tx)
{
	LOG(5, NULL);

	/* Flush all regions and destroy the whole tree. */
	ravl_delete_cb(tx->ranges, tx_flush_range, tx->pop);
	tx->ranges = NULL;
}

/*
 * tx_abort -- (internal) abort all allocated objects
 */
static void
tx_abort(PMEMobjpool *pop, struct lane *lane)
{
	LOG(7, NULL);

	struct tx *tx = get_tx();

	tx_abort_set(pop, lane);

	ravl_delete_cb(tx->ranges, tx_clean_range, pop);
	palloc_cancel(&pop->heap,
		VEC_ARR(&tx->actions), VEC_SIZE(&tx->actions));
	tx->ranges = NULL;
}

/*
 * tx_get_pop -- returns the current transaction's pool handle, NULL if not
 * within a transaction.
 */
PMEMobjpool *
tx_get_pop(void)
{
	return get_tx()->pop;
}

/*
 * add_to_tx_and_lock -- (internal) add lock to the transaction and acquire it
 */
static int
add_to_tx_and_lock(struct tx *tx, enum pobj_tx_param type, void *lock)
{
	LOG(15, NULL);

	int retval = 0;
	struct tx_lock_data *txl;
	/* check if the lock is already on the list */
	PMDK_SLIST_FOREACH(txl, &tx->tx_locks, tx_lock) {
		if (memcmp(&txl->lock, &lock, sizeof(lock)) == 0)
			return 0;
	}

	txl = Malloc(sizeof(*txl));
	if (txl == NULL)
		return ENOMEM;

	txl->lock_type = type;
	switch (txl->lock_type) {
		case TX_PARAM_MUTEX:
			txl->lock.mutex = lock;
			retval = pmemobj_mutex_lock(tx->pop,
				txl->lock.mutex);
			if (retval) {
				ERR("!pmemobj_mutex_lock");
				goto err;
			}
			break;
		case TX_PARAM_RWLOCK:
			txl->lock.rwlock = lock;
			retval = pmemobj_rwlock_wrlock(tx->pop,
				txl->lock.rwlock);
			if (retval) {
				ERR("!pmemobj_rwlock_wrlock");
				goto err;
			}
			break;
		default:
			ERR("Unrecognized lock type");
			ASSERT(0);
			break;
	}

	PMDK_SLIST_INSERT_HEAD(&tx->tx_locks, txl, tx_lock);
	return 0;

err:
	errno = retval;
	Free(txl);

	return retval;
}

/*
 * release_and_free_tx_locks -- (internal) release and remove all locks from the
 *				transaction
 */
static void
release_and_free_tx_locks(struct tx *tx)
{
	LOG(15, NULL);

	while (!PMDK_SLIST_EMPTY(&tx->tx_locks)) {
		struct tx_lock_data *tx_lock = PMDK_SLIST_FIRST(&tx->tx_locks);
		PMDK_SLIST_REMOVE_HEAD(&tx->tx_locks, tx_lock);
		switch (tx_lock->lock_type) {
			case TX_PARAM_MUTEX:
				pmemobj_mutex_unlock(tx->pop,
					tx_lock->lock.mutex);
				break;
			case TX_PARAM_RWLOCK:
				pmemobj_rwlock_unlock(tx->pop,
					tx_lock->lock.rwlock);
				break;
			default:
				ERR("Unrecognized lock type");
				ASSERT(0);
				break;
		}
		Free(tx_lock);
	}
}

/*
 * tx_lane_ranges_insert_def -- (internal) allocates and inserts a new range
 *	definition into the ranges tree
 */
static int
tx_lane_ranges_insert_def(PMEMobjpool *pop, struct tx *tx,
	const struct tx_range_def *rdef)
{
	LOG(3, "rdef->offset %"PRIu64" rdef->size %"PRIu64,
		rdef->offset, rdef->size);

	int ret = ravl_emplace_copy(tx->ranges, rdef);
	if (ret && errno == EEXIST)
		FATAL("invalid state of ranges tree");
	return ret;
}

/*
 * tx_alloc_common -- (internal) common function for alloc and zalloc
 */
static PMEMoid
tx_alloc_common(struct tx *tx, size_t size, type_num_t type_num,
		palloc_constr constructor, struct tx_alloc_args args)
{
	LOG(3, NULL);

	if (size > PMEMOBJ_MAX_ALLOC_SIZE) {
		ERR("requested size too large");
		return obj_tx_fail_null(ENOMEM, args.flags);
	}

	PMEMobjpool *pop = tx->pop;

	struct pobj_action *action = tx_action_add(tx);
	if (action == NULL)
		return obj_tx_fail_null(ENOMEM, args.flags);

	if (palloc_reserve(&pop->heap, size, constructor, &args, type_num, 0,
		CLASS_ID_FROM_FLAG(args.flags),
		ARENA_ID_FROM_FLAG(args.flags), action) != 0)
		goto err_oom;

	/* allocate object to undo log */
	PMEMoid retoid = OID_NULL;
	retoid.off = action->heap.offset;
	retoid.pool_uuid_lo = pop->uuid_lo;
	size = palloc_usable_size(&pop->heap, retoid.off);

	const struct tx_range_def r = {retoid.off, size, args.flags};
	if (tx_lane_ranges_insert_def(pop, tx, &r) != 0)
		goto err_oom;

	return retoid;

err_oom:
	tx_action_remove(tx);
	ERR("out of memory");
	return obj_tx_fail_null(ENOMEM, args.flags);
}

/*
 * tx_realloc_common -- (internal) common function for tx realloc
 */
static PMEMoid
tx_realloc_common(struct tx *tx, PMEMoid oid, size_t size, uint64_t type_num,
	palloc_constr constructor_alloc,
	palloc_constr constructor_realloc,
	uint64_t flags)
{
	LOG(3, NULL);

	if (size > PMEMOBJ_MAX_ALLOC_SIZE) {
		ERR("requested size too large");
		return obj_tx_fail_null(ENOMEM, flags);
	}

	/* if oid is NULL just alloc */
	if (OBJ_OID_IS_NULL(oid))
		return tx_alloc_common(tx, size, (type_num_t)type_num,
				constructor_alloc, ALLOC_ARGS(flags));

	ASSERT(OBJ_OID_IS_VALID(tx->pop, oid));

	/* if size is 0 just free */
	if (size == 0) {
		if (pmemobj_tx_free(oid)) {
			ERR("pmemobj_tx_free failed");
			return oid;
		} else {
			return OID_NULL;
		}
	}

	/* oid is not NULL and size is not 0 so do realloc by alloc and free */
	void *ptr = OBJ_OFF_TO_PTR(tx->pop, oid.off);
	size_t old_size = palloc_usable_size(&tx->pop->heap, oid.off);

	size_t copy_size = old_size < size ? old_size : size;

	PMEMoid new_obj = tx_alloc_common(tx, size, (type_num_t)type_num,
			constructor_realloc, COPY_ARGS(flags, ptr, copy_size));

	if (!OBJ_OID_IS_NULL(new_obj)) {
		if (pmemobj_tx_free(oid)) {
			ERR("pmemobj_tx_free failed");
			VEC_POP_BACK(&tx->actions);
			return OID_NULL;
		}
	}

	return new_obj;
}

/*
 * tx_construct_user_buffer -- add user buffer to the ulog
 */
static int
tx_construct_user_buffer(struct tx *tx, void *addr, size_t size,
		enum pobj_log_type type, int outer_tx, uint64_t flags)
{
	if (tx->pop != pmemobj_pool_by_ptr(addr)) {
		ERR("Buffer from a different pool");
		goto err;
	}

	/*
	 * We want to extend a log of a specified type, but if it is
	 * an outer transaction and the first user buffer we need to
	 * free all logs except the first at the beginning.
	 */
	struct operation_context *ctx = type == TX_LOG_TYPE_INTENT ?
		tx->lane->external : tx->lane->undo;

	if (outer_tx && !operation_get_any_user_buffer(ctx))
		operation_free_logs(ctx, ULOG_ANY_USER_BUFFER);

	struct user_buffer_def userbuf = {addr, size};
	if (operation_user_buffer_verify_align(ctx, &userbuf) != 0)
		goto err;

	if (type == TX_LOG_TYPE_INTENT) {
		/*
		 * Redo log context is not used until transaction commit and
		 * cannot be used until then, and so the user buffers have to
		 * be stored and added the operation at commit time.
		 * This is because atomic operations can executed independently
		 * in the same lane as a running transaction.
		 */
		if (VEC_PUSH_BACK(&tx->redo_userbufs, userbuf) != 0)
			goto err;
		tx->redo_userbufs_capacity +=
			userbuf.size - TX_INTENT_LOG_BUFFER_OVERHEAD;
	} else {
		operation_add_user_buffer(ctx, &userbuf);
	}

	return 0;

err:
	return obj_tx_fail_err(EINVAL, flags);
}

/*
 * pmemobj_tx_begin -- initializes new transaction
 */
int
pmemobj_tx_begin(PMEMobjpool *pop, jmp_buf env, ...)
{
	LOG(3, NULL);

	int err = 0;
	struct tx *tx = get_tx();

	enum pobj_tx_action_on_failure action_on_failure = POBJ_TX_ABORT;

	if (tx->stage == TX_STAGE_WORK) {
		ASSERTne(tx->lane, NULL);
		if (tx->pop != pop) {
			ERR("nested transaction for different pool");
			return obj_tx_fail_err(EINVAL, 0);
		}

		/* inherits this value from the parent transaction */
		struct tx_data *txd = PMDK_SLIST_FIRST(&tx->tx_entries);
		action_on_failure = txd->action_on_failure;

		VALGRIND_START_TX;
	} else if (tx->stage == TX_STAGE_NONE) {
		VALGRIND_START_TX;

		lane_hold(pop, &tx->lane);
		operation_start(tx->lane->undo);

		VEC_INIT(&tx->actions);
		VEC_INIT(&tx->redo_userbufs);
		tx->redo_userbufs_capacity = 0;
		PMDK_SLIST_INIT(&tx->tx_entries);
		PMDK_SLIST_INIT(&tx->tx_locks);

		tx->ranges = ravl_new_sized(tx_range_def_cmp,
			sizeof(struct tx_range_def));

		tx->pop = pop;

		tx->first_snapshot = 1;

		tx->user_data = NULL;
	} else {
		FATAL("Invalid stage %d to begin new transaction", tx->stage);
	}

	struct tx_data *txd = Malloc(sizeof(*txd));
	if (txd == NULL) {
		err = errno;
		ERR("!Malloc");
		goto err_abort;
	}

	tx->last_errnum = 0;
	if (env != NULL)
		memcpy(txd->env, env, sizeof(jmp_buf));
	else
		memset(txd->env, 0, sizeof(jmp_buf));

	txd->action_on_failure = action_on_failure;

	PMDK_SLIST_INSERT_HEAD(&tx->tx_entries, txd, tx_entry);

	tx->stage = TX_STAGE_WORK;

	/* handle locks */
	va_list argp;
	va_start(argp, env);
	enum pobj_tx_param param_type;

	while ((param_type = va_arg(argp, enum pobj_tx_param)) !=
			TX_PARAM_NONE) {
		if (param_type == TX_PARAM_CB) {
			pmemobj_tx_callback cb =
					va_arg(argp, pmemobj_tx_callback);
			void *arg = va_arg(argp, void *);

			if (tx->stage_callback &&
					(tx->stage_callback != cb ||
					tx->stage_callback_arg != arg)) {
				FATAL("transaction callback is already set, "
					"old %p new %p old_arg %p new_arg %p",
					tx->stage_callback, cb,
					tx->stage_callback_arg, arg);
			}

			tx->stage_callback = cb;
			tx->stage_callback_arg = arg;
		} else {
			err = add_to_tx_and_lock(tx, param_type,
				va_arg(argp, void *));
			if (err) {
				va_end(argp);
				goto err_abort;
			}
		}
	}
	va_end(argp);

	ASSERT(err == 0);
	return 0;

err_abort:
	if (tx->stage == TX_STAGE_WORK)
		obj_tx_abort(err, 0);
	else
		tx->stage = TX_STAGE_ONABORT;
	return err;
}

/*
 * tx_abort_on_failure_flag -- (internal) return 0 or POBJ_FLAG_TX_NO_ABORT
 * based on transaction setting
 */
static uint64_t
tx_abort_on_failure_flag(struct tx *tx)
{
	struct tx_data *txd = PMDK_SLIST_FIRST(&tx->tx_entries);

	if (txd->action_on_failure == POBJ_RETURN_ERROR)
		return POBJ_FLAG_TX_NO_ABORT;
	return 0;
}

/*
 * pmemobj_tx_xlock -- get lane from pool and add lock to transaction,
 * with no_abort option
 */
int
pmemobj_tx_xlock(enum pobj_tx_param type, void *lockp, uint64_t flags)
{
	struct tx *tx = get_tx();
	ASSERT_IN_TX(tx);
	ASSERT_TX_STAGE_WORK(tx);

	flags |= tx_abort_on_failure_flag(tx);

	if (flags & ~POBJ_XLOCK_VALID_FLAGS) {
		ERR("unknown flags 0x%" PRIx64,
				flags & ~POBJ_XLOCK_VALID_FLAGS);
		return obj_tx_fail_err(EINVAL, flags);
	}

	int ret = add_to_tx_and_lock(tx, type, lockp);
	if (ret)
		return obj_tx_fail_err(ret, flags);
	return 0;
}

/*
 * pmemobj_tx_lock -- get lane from pool and add lock to transaction.
 */
int
pmemobj_tx_lock(enum pobj_tx_param type, void *lockp)
{
	return pmemobj_tx_xlock(type, lockp, POBJ_XLOCK_NO_ABORT);
}

/*
 * obj_tx_callback -- (internal) executes callback associated with current stage
 */
static void
obj_tx_callback(struct tx *tx)
{
	if (!tx->stage_callback)
		return;

	struct tx_data *txd = PMDK_SLIST_FIRST(&tx->tx_entries);

	/* is this the outermost transaction? */
	if (PMDK_SLIST_NEXT(txd, tx_entry) == NULL)
		tx->stage_callback(tx->pop, tx->stage, tx->stage_callback_arg);
}

/*
 * pmemobj_tx_stage -- returns current transaction stage
 */
enum pobj_tx_stage
pmemobj_tx_stage(void)
{
	LOG(3, NULL);

	return get_tx()->stage;
}

/*
 * obj_tx_abort -- aborts current transaction
 */
static void
obj_tx_abort(int errnum, int user)
{
	LOG(3, NULL);
	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);
	ASSERT_TX_STAGE_WORK(tx);

	ASSERT(tx->lane != NULL);

	if (errnum == 0)
		errnum = ECANCELED;

	tx->stage = TX_STAGE_ONABORT;
	struct tx_data *txd = PMDK_SLIST_FIRST(&tx->tx_entries);

	if (PMDK_SLIST_NEXT(txd, tx_entry) == NULL) {
		/* this is the outermost transaction */

		/* process the undo log */
		tx_abort(tx->pop, tx->lane);

		lane_release(tx->pop);
		tx->lane = NULL;
	}

	tx->last_errnum = errnum;
	errno = errnum;
	if (user)
		ERR("!explicit transaction abort");

	/* ONABORT */
	obj_tx_callback(tx);

	if (!util_is_zeroed(txd->env, sizeof(jmp_buf)))
		longjmp(txd->env, errnum);
}

/*
 * pmemobj_tx_abort -- aborts current transaction
 *
 * Note: this function should not be called from inside of pmemobj.
 */
void
pmemobj_tx_abort(int errnum)
{
	PMEMOBJ_API_START();
	obj_tx_abort(errnum, 1);
	PMEMOBJ_API_END();
}

/*
 * pmemobj_tx_errno -- returns last transaction error code
 */
int
pmemobj_tx_errno(void)
{
	LOG(3, NULL);

	return get_tx()->last_errnum;
}

static void
tx_post_commit(struct tx *tx)
{
	operation_finish(tx->lane->undo, 0);
}

/*
 * pmemobj_tx_commit -- commits current transaction
 */
void
pmemobj_tx_commit(void)
{
	LOG(3, NULL);

	PMEMOBJ_API_START();
	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);
	ASSERT_TX_STAGE_WORK(tx);

	/* WORK */
	obj_tx_callback(tx);

	ASSERT(tx->lane != NULL);

	struct tx_data *txd = PMDK_SLIST_FIRST(&tx->tx_entries);

	if (PMDK_SLIST_NEXT(txd, tx_entry) == NULL) {
		/* this is the outermost transaction */

		PMEMobjpool *pop = tx->pop;

		/* pre-commit phase */
		tx_pre_commit(tx);

		pmemops_drain(&pop->p_ops);

		operation_start(tx->lane->external);

		struct user_buffer_def *userbuf;
		VEC_FOREACH_BY_PTR(userbuf, &tx->redo_userbufs)
			operation_add_user_buffer(tx->lane->external, userbuf);

		palloc_publish(&pop->heap, VEC_ARR(&tx->actions),
			VEC_SIZE(&tx->actions), tx->lane->external);

		tx_post_commit(tx);

		lane_release(pop);

		tx->lane = NULL;
	}

	tx->stage = TX_STAGE_ONCOMMIT;

	/* ONCOMMIT */
	obj_tx_callback(tx);
	PMEMOBJ_API_END();
}

/*
 * pmemobj_tx_end -- ends current transaction
 */
int
pmemobj_tx_end(void)
{
	LOG(3, NULL);

	struct tx *tx = get_tx();

	if (tx->stage == TX_STAGE_WORK)
		FATAL("pmemobj_tx_end called without pmemobj_tx_commit");

	if (tx->pop == NULL)
		FATAL("pmemobj_tx_end called without pmemobj_tx_begin");

	if (tx->stage_callback &&
			(tx->stage == TX_STAGE_ONCOMMIT ||
			tx->stage == TX_STAGE_ONABORT)) {
		tx->stage = TX_STAGE_FINALLY;
		obj_tx_callback(tx);
	}

	struct tx_data *txd = PMDK_SLIST_FIRST(&tx->tx_entries);
	PMDK_SLIST_REMOVE_HEAD(&tx->tx_entries, tx_entry);

	Free(txd);

	VALGRIND_END_TX;

	if (PMDK_SLIST_EMPTY(&tx->tx_entries)) {
		ASSERTeq(tx->lane, NULL);

		release_and_free_tx_locks(tx);
		tx->pop = NULL;
		tx->stage = TX_STAGE_NONE;
		VEC_DELETE(&tx->actions);
		VEC_DELETE(&tx->redo_userbufs);

		if (tx->stage_callback) {
			pmemobj_tx_callback cb = tx->stage_callback;
			void *arg = tx->stage_callback_arg;

			tx->stage_callback = NULL;
			tx->stage_callback_arg = NULL;

			cb(tx->pop, TX_STAGE_NONE, arg);
		}
	} else {
		/* resume the next transaction */
		tx->stage = TX_STAGE_WORK;

		/* abort called within inner transaction, waterfall the error */
		if (tx->last_errnum)
			obj_tx_abort(tx->last_errnum, 0);
	}

	return tx->last_errnum;
}

/*
 * pmemobj_tx_process -- processes current transaction stage
 */
void
pmemobj_tx_process(void)
{
	LOG(5, NULL);
	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);

	switch (tx->stage) {
	case TX_STAGE_NONE:
		break;
	case TX_STAGE_WORK:
		pmemobj_tx_commit();
		break;
	case TX_STAGE_ONABORT:
	case TX_STAGE_ONCOMMIT:
		tx->stage = TX_STAGE_FINALLY;
		obj_tx_callback(tx);
		break;
	case TX_STAGE_FINALLY:
		tx->stage = TX_STAGE_NONE;
		break;
	default:
		ASSERT(0);
	}
}

/*
 * vg_verify_initialized -- when executed under Valgrind verifies that
 *   the buffer has been initialized; explicit check at snapshotting time,
 *   because Valgrind may find it much later when it's impossible to tell
 *   for which snapshot it triggered
 */
static void
vg_verify_initialized(PMEMobjpool *pop, const struct tx_range_def *def)
{
#if VG_MEMCHECK_ENABLED
	if (!On_valgrind)
		return;

	VALGRIND_DO_DISABLE_ERROR_REPORTING;
	char *start = (char *)pop + def->offset;
	char *uninit = (char *)VALGRIND_CHECK_MEM_IS_DEFINED(start, def->size);
	if (uninit) {
		VALGRIND_PRINTF(
			"Snapshotting uninitialized data in range <%p,%p> (<offset:0x%lx,size:0x%lx>)\n",
			start, start + def->size, def->offset, def->size);

		if (uninit != start)
			VALGRIND_PRINTF("Uninitialized data starts at: %p\n",
					uninit);

		VALGRIND_DO_ENABLE_ERROR_REPORTING;
		VALGRIND_CHECK_MEM_IS_DEFINED(start, def->size);
	} else {
		VALGRIND_DO_ENABLE_ERROR_REPORTING;
	}
#endif
}

/*
 * pmemobj_tx_add_snapshot -- (internal) creates a variably sized snapshot
 */
static int
pmemobj_tx_add_snapshot(struct tx *tx, struct tx_range_def *snapshot)
{
	/*
	 * Depending on the size of the block, either allocate an
	 * entire new object or use cache.
	 */
	void *ptr = OBJ_OFF_TO_PTR(tx->pop, snapshot->offset);

	VALGRIND_ADD_TO_TX(ptr, snapshot->size);

	/* do nothing */
	if (snapshot->flags & POBJ_XADD_NO_SNAPSHOT)
		return 0;

	if (!(snapshot->flags & POBJ_XADD_ASSUME_INITIALIZED))
		vg_verify_initialized(tx->pop, snapshot);

	/*
	 * If we are creating the first snapshot, setup a redo log action to
	 * increment counter in the undo log, so that the log becomes
	 * invalid once the redo log is processed.
	 */
	if (tx->first_snapshot) {
		struct pobj_action *action = tx_action_add(tx);
		if (action == NULL)
			return -1;

		uint64_t *n = &tx->lane->layout->undo.gen_num;
		palloc_set_value(&tx->pop->heap, action,
			n, *n + 1);

		tx->first_snapshot = 0;
	}

	return operation_add_buffer(tx->lane->undo, ptr, ptr, snapshot->size,
		ULOG_OPERATION_BUF_CPY);
}

/*
 * pmemobj_tx_merge_flags -- (internal) common code for merging flags between
 * two ranges to ensure resultant behavior is correct
 */
static void
pmemobj_tx_merge_flags(struct tx_range_def *dest, struct tx_range_def *merged)
{
	/*
	 * POBJ_XADD_NO_FLUSH should only be set in merged range if set in
	 * both ranges
	 */
	if ((dest->flags & POBJ_XADD_NO_FLUSH) &&
				!(merged->flags & POBJ_XADD_NO_FLUSH)) {
		dest->flags = dest->flags & (~POBJ_XADD_NO_FLUSH);
	}
}

/*
 * pmemobj_tx_add_common -- (internal) common code for adding persistent memory
 * into the transaction
 */
static int
pmemobj_tx_add_common(struct tx *tx, struct tx_range_def *args)
{
	LOG(15, NULL);

	if (args->size > PMEMOBJ_MAX_ALLOC_SIZE) {
		ERR("snapshot size too large");
		return obj_tx_fail_err(EINVAL, args->flags);
	}

	if (args->offset < tx->pop->heap_offset ||
		(args->offset + args->size) >
		(tx->pop->heap_offset + tx->pop->heap_size)) {
		ERR("object outside of heap");
		return obj_tx_fail_err(EINVAL, args->flags);
	}

	int ret = 0;

	/*
	 * Search existing ranges backwards starting from the end of the
	 * snapshot.
	 */
	struct tx_range_def r = *args;
	struct tx_range_def search = {0, 0, 0};
	/*
	 * If the range is directly adjacent to an existing one,
	 * they can be merged, so search for less or equal elements.
	 */
	enum ravl_predicate p = RAVL_PREDICATE_LESS_EQUAL;
	struct ravl_node *nprev = NULL;
	while (r.size != 0) {
		search.offset = r.offset + r.size;
		struct ravl_node *n = ravl_find(tx->ranges, &search, p);
		/*
		 * We have to skip searching for LESS_EQUAL because
		 * the snapshot we would find is the one that was just
		 * created.
		 */
		p = RAVL_PREDICATE_LESS;

		struct tx_range_def *f = n ? ravl_data(n) : NULL;

		size_t fend = f == NULL ? 0: f->offset + f->size;
		size_t rend = r.offset + r.size;
		if (fend == 0 || fend < r.offset) {
			/*
			 * If found no range or the found range is not
			 * overlapping or adjacent on the left side, we can just
			 * create the entire r.offset + r.size snapshot.
			 *
			 * Snapshot:
			 *	--+-
			 * Existing ranges:
			 *	---- (no ranges)
			 * or	+--- (no overlap)
			 * or	---+ (adjacent on on right side)
			 */
			if (nprev != NULL) {
				/*
				 * But, if we have an existing adjacent snapshot
				 * on the right side, we can just extend it to
				 * include the desired range.
				 */
				struct tx_range_def *fprev = ravl_data(nprev);
				ASSERTeq(rend, fprev->offset);
				fprev->offset -= r.size;
				fprev->size += r.size;
			} else {
				/*
				 * If we don't have anything adjacent, create
				 * a new range in the tree.
				 */
				ret = tx_lane_ranges_insert_def(tx->pop,
					tx, &r);
				if (ret != 0)
					break;
			}
			ret = pmemobj_tx_add_snapshot(tx, &r);
			break;
		} else if (fend <= rend) {
			/*
			 * If found range has its end inside of the desired
			 * snapshot range, we can extend the found range by the
			 * size leftover on the left side.
			 *
			 * Snapshot:
			 *	--+++--
			 * Existing ranges:
			 *	+++---- (overlap on left)
			 * or	---+--- (found snapshot is inside)
			 * or	---+-++ (inside, and adjacent on the right)
			 * or	+++++-- (desired snapshot is inside)
			 *
			 */
			struct tx_range_def snapshot = *args;
			snapshot.offset = fend;
			/* the side not yet covered by an existing snapshot */
			snapshot.size = rend - fend;

			/* the number of bytes intersecting in both ranges */
			size_t intersection = fend - MAX(f->offset, r.offset);
			r.size -= intersection + snapshot.size;
			f->size += snapshot.size;
			pmemobj_tx_merge_flags(f, args);

			if (snapshot.size != 0) {
				ret = pmemobj_tx_add_snapshot(tx, &snapshot);
				if (ret != 0)
					break;
			}

			/*
			 * If there's a snapshot adjacent on right side, merge
			 * the two ranges together.
			 */
			if (nprev != NULL) {
				struct tx_range_def *fprev = ravl_data(nprev);
				ASSERTeq(rend, fprev->offset);
				f->size += fprev->size;
				pmemobj_tx_merge_flags(f, fprev);
				ravl_remove(tx->ranges, nprev);
			}
		} else if (fend >= r.offset) {
			/*
			 * If found range has its end extending beyond the
			 * desired snapshot.
			 *
			 * Snapshot:
			 *	--+++--
			 * Existing ranges:
			 *	-----++ (adjacent on the right)
			 * or	----++- (overlapping on the right)
			 * or	----+++ (overlapping and adjacent on the right)
			 * or	--+++++ (desired snapshot is inside)
			 *
			 * Notice that we cannot create a snapshot based solely
			 * on this information without risking overwriting an
			 * existing one. We have to continue iterating, but we
			 * keep the information about adjacent snapshots in the
			 * nprev variable.
			 */
			size_t overlap = rend - MAX(f->offset, r.offset);
			r.size -= overlap;
			pmemobj_tx_merge_flags(f, args);
		} else {
			ASSERT(0);
		}

		nprev = n;
	}

	if (ret != 0) {
		ERR("out of memory");
		return obj_tx_fail_err(ENOMEM, args->flags);
	}

	return 0;
}

/*
 * pmemobj_tx_add_range_direct -- adds persistent memory range into the
 *					transaction
 */
int
pmemobj_tx_add_range_direct(const void *ptr, size_t size)
{
	LOG(3, NULL);

	PMEMOBJ_API_START();
	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);
	ASSERT_TX_STAGE_WORK(tx);

	int ret;

	uint64_t flags = tx_abort_on_failure_flag(tx);

	if (!OBJ_PTR_FROM_POOL(tx->pop, ptr)) {
		ERR("object outside of pool");
		ret = obj_tx_fail_err(EINVAL, flags);
		PMEMOBJ_API_END();
		return ret;
	}

	struct tx_range_def args = {
		.offset = (uint64_t)((char *)ptr - (char *)tx->pop),
		.size = size,
		.flags = flags,
	};

	ret = pmemobj_tx_add_common(tx, &args);

	PMEMOBJ_API_END();
	return ret;
}

/*
 * pmemobj_tx_xadd_range_direct -- adds persistent memory range into the
 *					transaction
 */
int
pmemobj_tx_xadd_range_direct(const void *ptr, size_t size, uint64_t flags)
{
	LOG(3, NULL);

	PMEMOBJ_API_START();
	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);
	ASSERT_TX_STAGE_WORK(tx);

	int ret;

	flags |= tx_abort_on_failure_flag(tx);

	if (flags & ~POBJ_XADD_VALID_FLAGS) {
		ERR("unknown flags 0x%" PRIx64, flags
			& ~POBJ_XADD_VALID_FLAGS);
		ret = obj_tx_fail_err(EINVAL, flags);
		PMEMOBJ_API_END();
		return ret;
	}

	if (!OBJ_PTR_FROM_POOL(tx->pop, ptr)) {
		ERR("object outside of pool");
		ret = obj_tx_fail_err(EINVAL, flags);
		PMEMOBJ_API_END();
		return ret;
	}

	struct tx_range_def args = {
		.offset = (uint64_t)((char *)ptr - (char *)tx->pop),
		.size = size,
		.flags = flags,
	};

	ret = pmemobj_tx_add_common(tx, &args);

	PMEMOBJ_API_END();
	return ret;
}

/*
 * pmemobj_tx_add_range -- adds persistent memory range into the transaction
 */
int
pmemobj_tx_add_range(PMEMoid oid, uint64_t hoff, size_t size)
{
	LOG(3, NULL);

	PMEMOBJ_API_START();
	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);
	ASSERT_TX_STAGE_WORK(tx);

	int ret;

	uint64_t flags = tx_abort_on_failure_flag(tx);

	if (oid.pool_uuid_lo != tx->pop->uuid_lo) {
		ERR("invalid pool uuid");
		ret = obj_tx_fail_err(EINVAL, flags);
		PMEMOBJ_API_END();
		return ret;
	}
	ASSERT(OBJ_OID_IS_VALID(tx->pop, oid));

	struct tx_range_def args = {
		.offset = oid.off + hoff,
		.size = size,
		.flags = flags,
	};

	ret = pmemobj_tx_add_common(tx, &args);

	PMEMOBJ_API_END();
	return ret;
}

/*
 * pmemobj_tx_xadd_range -- adds persistent memory range into the transaction
 */
int
pmemobj_tx_xadd_range(PMEMoid oid, uint64_t hoff, size_t size, uint64_t flags)
{
	LOG(3, NULL);

	PMEMOBJ_API_START();
	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);
	ASSERT_TX_STAGE_WORK(tx);

	int ret;

	flags |= tx_abort_on_failure_flag(tx);

	if (flags & ~POBJ_XADD_VALID_FLAGS) {
		ERR("unknown flags 0x%" PRIx64, flags
			& ~POBJ_XADD_VALID_FLAGS);
		ret = obj_tx_fail_err(EINVAL, flags);
		PMEMOBJ_API_END();
		return ret;
	}

	if (oid.pool_uuid_lo != tx->pop->uuid_lo) {
		ERR("invalid pool uuid");
		ret = obj_tx_fail_err(EINVAL, flags);
		PMEMOBJ_API_END();
		return ret;
	}
	ASSERT(OBJ_OID_IS_VALID(tx->pop, oid));

	struct tx_range_def args = {
		.offset = oid.off + hoff,
		.size = size,
		.flags = flags,
	};

	ret = pmemobj_tx_add_common(tx, &args);

	PMEMOBJ_API_END();
	return ret;
}

/*
 * pmemobj_tx_alloc -- allocates a new object
 */
PMEMoid
pmemobj_tx_alloc(size_t size, uint64_t type_num)
{
	LOG(3, NULL);

	PMEMOBJ_API_START();
	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);
	ASSERT_TX_STAGE_WORK(tx);

	uint64_t flags = tx_abort_on_failure_flag(tx);

	PMEMoid oid;
	if (size == 0) {
		ERR("allocation with size 0");
		oid = obj_tx_fail_null(EINVAL, flags);
		PMEMOBJ_API_END();
		return oid;
	}

	oid = tx_alloc_common(tx, size, (type_num_t)type_num,
			constructor_tx_alloc, ALLOC_ARGS(flags));

	PMEMOBJ_API_END();
	return oid;
}

/*
 * pmemobj_tx_zalloc -- allocates a new zeroed object
 */
PMEMoid
pmemobj_tx_zalloc(size_t size, uint64_t type_num)
{
	LOG(3, NULL);
	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);
	ASSERT_TX_STAGE_WORK(tx);

	uint64_t flags = POBJ_FLAG_ZERO;
	flags |= tx_abort_on_failure_flag(tx);

	PMEMOBJ_API_START();
	PMEMoid oid;
	if (size == 0) {
		ERR("allocation with size 0");
		oid = obj_tx_fail_null(EINVAL, flags);
		PMEMOBJ_API_END();
		return oid;
	}

	oid = tx_alloc_common(tx, size, (type_num_t)type_num,
			constructor_tx_alloc, ALLOC_ARGS(flags));

	PMEMOBJ_API_END();
	return oid;
}

/*
 * pmemobj_tx_xalloc -- allocates a new object
 */
PMEMoid
pmemobj_tx_xalloc(size_t size, uint64_t type_num, uint64_t flags)
{
	LOG(3, NULL);
	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);
	ASSERT_TX_STAGE_WORK(tx);

	flags |= tx_abort_on_failure_flag(tx);

	PMEMOBJ_API_START();

	PMEMoid oid;
	if (size == 0) {
		ERR("allocation with size 0");
		oid = obj_tx_fail_null(EINVAL, flags);
		PMEMOBJ_API_END();
		return oid;
	}

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

	oid = tx_alloc_common(tx, size, (type_num_t)type_num,
			constructor_tx_alloc, ALLOC_ARGS(flags));

	PMEMOBJ_API_END();
	return oid;
}

/*
 * pmemobj_tx_realloc -- resizes an existing object
 */
PMEMoid
pmemobj_tx_realloc(PMEMoid oid, size_t size, uint64_t type_num)
{
	LOG(3, NULL);
	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);
	ASSERT_TX_STAGE_WORK(tx);

	PMEMOBJ_API_START();
	PMEMoid ret = tx_realloc_common(tx, oid, size, type_num,
			constructor_tx_alloc, constructor_tx_alloc, 0);
	PMEMOBJ_API_END();
	return ret;
}

/*
 * pmemobj_zrealloc -- resizes an existing object, any new space is zeroed.
 */
PMEMoid
pmemobj_tx_zrealloc(PMEMoid oid, size_t size, uint64_t type_num)
{
	LOG(3, NULL);
	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);
	ASSERT_TX_STAGE_WORK(tx);

	PMEMOBJ_API_START();
	PMEMoid ret = tx_realloc_common(tx, oid, size, type_num,
			constructor_tx_alloc, constructor_tx_alloc,
			POBJ_FLAG_ZERO);
	PMEMOBJ_API_END();
	return ret;
}

/*
 * pmemobj_tx_xstrdup -- allocates a new object with duplicate of the string s.
 */
PMEMoid
pmemobj_tx_xstrdup(const char *s, uint64_t type_num, uint64_t flags)
{
	LOG(3, NULL);

	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);
	ASSERT_TX_STAGE_WORK(tx);

	flags |= tx_abort_on_failure_flag(tx);

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

	PMEMOBJ_API_START();
	PMEMoid oid;
	if (NULL == s) {
		ERR("cannot duplicate NULL string");
		oid = obj_tx_fail_null(EINVAL, flags);
		PMEMOBJ_API_END();
		return oid;
	}

	size_t len = strlen(s);

	if (len == 0) {
		oid = tx_alloc_common(tx, sizeof(char), (type_num_t)type_num,
				constructor_tx_alloc,
			ALLOC_ARGS(POBJ_XALLOC_ZERO));
		PMEMOBJ_API_END();
		return oid;
	}

	size_t size = (len + 1) * sizeof(char);

	oid = tx_alloc_common(tx, size, (type_num_t)type_num,
			constructor_tx_alloc, COPY_ARGS(flags, s, size));

	PMEMOBJ_API_END();
	return oid;
}

/*
 * pmemobj_tx_strdup -- allocates a new object with duplicate of the string s.
 */
PMEMoid
pmemobj_tx_strdup(const char *s, uint64_t type_num)
{
	return pmemobj_tx_xstrdup(s, type_num, 0);
}
/*
 * pmemobj_tx_xwcsdup -- allocates a new object with duplicate of the wide
 * character string s.
 */
PMEMoid
pmemobj_tx_xwcsdup(const wchar_t *s, uint64_t type_num, uint64_t flags)
{
	LOG(3, NULL);

	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);
	ASSERT_TX_STAGE_WORK(tx);

	flags |= tx_abort_on_failure_flag(tx);

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

	PMEMOBJ_API_START();
	PMEMoid oid;
	if (NULL == s) {
		ERR("cannot duplicate NULL string");
		oid = obj_tx_fail_null(EINVAL, flags);
		PMEMOBJ_API_END();
		return oid;
	}

	size_t len = wcslen(s);

	if (len == 0) {
		oid = tx_alloc_common(tx, sizeof(wchar_t),
				(type_num_t)type_num, constructor_tx_alloc,
				ALLOC_ARGS(POBJ_XALLOC_ZERO));
		PMEMOBJ_API_END();
		return oid;
	}

	size_t size = (len + 1) * sizeof(wchar_t);

	oid = tx_alloc_common(tx, size, (type_num_t)type_num,
			constructor_tx_alloc, COPY_ARGS(flags, s, size));

	PMEMOBJ_API_END();
	return oid;
}

/*
 * pmemobj_tx_wcsdup -- allocates a new object with duplicate of the wide
 * character string s.
 */
PMEMoid
pmemobj_tx_wcsdup(const wchar_t *s, uint64_t type_num)
{
	return pmemobj_tx_xwcsdup(s, type_num, 0);
}

/*
 * pmemobj_tx_xfree -- frees an existing object, with no_abort option
 */
int
pmemobj_tx_xfree(PMEMoid oid, uint64_t flags)
{
	LOG(3, NULL);

	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);
	ASSERT_TX_STAGE_WORK(tx);

	flags |= tx_abort_on_failure_flag(tx);

	if (flags & ~POBJ_XFREE_VALID_FLAGS) {
		ERR("unknown flags 0x%" PRIx64,
				flags & ~POBJ_XFREE_VALID_FLAGS);
		return obj_tx_fail_err(EINVAL, flags);
	}

	if (OBJ_OID_IS_NULL(oid))
		return 0;

	PMEMobjpool *pop = tx->pop;

	if (pop->uuid_lo != oid.pool_uuid_lo) {
		ERR("invalid pool uuid");
		return obj_tx_fail_err(EINVAL, flags);
	}

	ASSERT(OBJ_OID_IS_VALID(pop, oid));

	PMEMOBJ_API_START();

	struct pobj_action *action;

	struct tx_range_def range = {oid.off, 0, 0};
	struct ravl_node *n = ravl_find(tx->ranges, &range,
		RAVL_PREDICATE_EQUAL);

	/*
	 * If attempting to free an object allocated within the same
	 * transaction, simply cancel the alloc and remove it from the actions.
	 */
	if (n != NULL) {
		VEC_FOREACH_BY_PTR(action, &tx->actions) {
			if (action->type == POBJ_ACTION_TYPE_HEAP &&
				action->heap.offset == oid.off) {
				struct tx_range_def *r = ravl_data(n);
				void *ptr = OBJ_OFF_TO_PTR(pop, r->offset);
				VALGRIND_SET_CLEAN(ptr, r->size);
				VALGRIND_REMOVE_FROM_TX(ptr, r->size);
				ravl_remove(tx->ranges, n);
				palloc_cancel(&pop->heap, action, 1);
				VEC_ERASE_BY_PTR(&tx->actions, action);
				PMEMOBJ_API_END();
				return 0;
			}
		}
	}

	action = tx_action_add(tx);
	if (action == NULL) {
		int ret = obj_tx_fail_err(errno, flags);
		PMEMOBJ_API_END();
		return ret;
	}

	palloc_defer_free(&pop->heap, oid.off, action);

	PMEMOBJ_API_END();
	return 0;
}

/*
 * pmemobj_tx_free -- frees an existing object
 */
int
pmemobj_tx_free(PMEMoid oid)
{
	return pmemobj_tx_xfree(oid, 0);
}

/*
 * pmemobj_tx_xpublish -- publishes actions inside of a transaction,
 * with no_abort option
 */
int
pmemobj_tx_xpublish(struct pobj_action *actv, size_t actvcnt, uint64_t flags)
{
	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);
	ASSERT_TX_STAGE_WORK(tx);

	flags |= tx_abort_on_failure_flag(tx);

	if (flags & ~POBJ_XPUBLISH_VALID_FLAGS) {
		ERR("unknown flags 0x%" PRIx64,
				flags & ~POBJ_XPUBLISH_VALID_FLAGS);
		return obj_tx_fail_err(EINVAL, flags);
	}

	PMEMOBJ_API_START();

	if (tx_action_reserve(tx, actvcnt) != 0) {
		int ret = obj_tx_fail_err(ENOMEM, flags);
		PMEMOBJ_API_END();
		return ret;
	}

	for (size_t i = 0; i < actvcnt; ++i) {
		VEC_PUSH_BACK(&tx->actions, actv[i]);
	}

	PMEMOBJ_API_END();
	return 0;
}

/*
 * pmemobj_tx_publish -- publishes actions inside of a transaction
 */
int
pmemobj_tx_publish(struct pobj_action *actv, size_t actvcnt)
{
	return pmemobj_tx_xpublish(actv, actvcnt, 0);
}

/*
 * pmemobj_tx_xlog_append_buffer -- append user allocated buffer to the ulog
 */
int
pmemobj_tx_xlog_append_buffer(enum pobj_log_type type, void *addr, size_t size,
		uint64_t flags)
{
	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);
	ASSERT_TX_STAGE_WORK(tx);

	flags |= tx_abort_on_failure_flag(tx);

	if (flags & ~POBJ_XLOG_APPEND_BUFFER_VALID_FLAGS) {
		ERR("unknown flags 0x%" PRIx64,
				flags & ~POBJ_XLOG_APPEND_BUFFER_VALID_FLAGS);
		return obj_tx_fail_err(EINVAL, flags);
	}

	PMEMOBJ_API_START();
	int err;

	struct tx_data *td = PMDK_SLIST_FIRST(&tx->tx_entries);
	err = tx_construct_user_buffer(tx, addr, size, type,
			PMDK_SLIST_NEXT(td, tx_entry) == NULL, flags);

	PMEMOBJ_API_END();
	return err;
}

/*
 * pmemobj_tx_log_append_buffer -- append user allocated buffer to the ulog
 */
int
pmemobj_tx_log_append_buffer(enum pobj_log_type type, void *addr, size_t size)
{
	return pmemobj_tx_xlog_append_buffer(type, addr, size, 0);
}

/*
 * pmemobj_tx_log_auto_alloc -- enable/disable automatic ulog allocation
 */
int
pmemobj_tx_log_auto_alloc(enum pobj_log_type type, int on_off)
{
	struct tx *tx = get_tx();
	ASSERT_TX_STAGE_WORK(tx);

	struct operation_context *ctx = type == TX_LOG_TYPE_INTENT ?
		tx->lane->external : tx->lane->undo;

	operation_set_auto_reserve(ctx, on_off);

	return 0;
}

/*
 * pmemobj_tx_log_snapshots_max_size -- calculates the maximum
 * size of a buffer which will be able to hold nsizes snapshots,
 * each of size from sizes array
 */
size_t
pmemobj_tx_log_snapshots_max_size(size_t *sizes, size_t nsizes)
{
	LOG(3, NULL);

	/* each buffer has its header */
	size_t result = TX_SNAPSHOT_LOG_BUFFER_OVERHEAD;
	for (size_t i = 0; i < nsizes; ++i) {
		/* check for overflow */
		if (sizes[i] + TX_SNAPSHOT_LOG_ENTRY_OVERHEAD +
				TX_SNAPSHOT_LOG_ENTRY_ALIGNMENT < sizes[i])
			goto err_overflow;
		/* each entry has its header */
		size_t size =
			ALIGN_UP(sizes[i] + TX_SNAPSHOT_LOG_ENTRY_OVERHEAD,
				TX_SNAPSHOT_LOG_ENTRY_ALIGNMENT);
		/* check for overflow */
		if (result + size < result)
			goto err_overflow;
		/* sum up */
		result += size;
	}

	/*
	 * if the result is bigger than a single allocation it must be divided
	 * into multiple allocations where each of them will have its own buffer
	 * header and entry header
	 */
	size_t allocs_overhead = (result / PMEMOBJ_MAX_ALLOC_SIZE) *
	    (TX_SNAPSHOT_LOG_BUFFER_OVERHEAD + TX_SNAPSHOT_LOG_ENTRY_OVERHEAD);
	/* check for overflow */
	if (result + allocs_overhead < result)
		goto err_overflow;
	result += allocs_overhead;

	/* SIZE_MAX is a special value */
	if (result == SIZE_MAX)
		goto err_overflow;

	return result;

err_overflow:
	errno = ERANGE;
	return SIZE_MAX;
}

/*
 * pmemobj_tx_log_intents_max_size -- calculates the maximum size of a buffer
 * which will be able to hold nintents
 */
size_t
pmemobj_tx_log_intents_max_size(size_t nintents)
{
	LOG(3, NULL);

	/* check for overflow */
	if (nintents > SIZE_MAX / TX_INTENT_LOG_ENTRY_OVERHEAD)
		goto err_overflow;
	/* each entry has its header */
	size_t entries_overhead = nintents * TX_INTENT_LOG_ENTRY_OVERHEAD;
	/* check for overflow */
	if (entries_overhead + TX_INTENT_LOG_BUFFER_ALIGNMENT
			< entries_overhead)
		goto err_overflow;
	/* the whole buffer is aligned */
	size_t result =
		ALIGN_UP(entries_overhead, TX_INTENT_LOG_BUFFER_ALIGNMENT);

	/* check for overflow */
	if (result + TX_INTENT_LOG_BUFFER_OVERHEAD < result)
		goto err_overflow;
	/* add a buffer overhead */
	result += TX_INTENT_LOG_BUFFER_OVERHEAD;

	/*
	 * if the result is bigger than a single allocation it must be divided
	 * into multiple allocations where each of them will have its own buffer
	 * header and entry header
	 */
	size_t allocs_overhead = (result / PMEMOBJ_MAX_ALLOC_SIZE) *
	    (TX_INTENT_LOG_BUFFER_OVERHEAD + TX_INTENT_LOG_ENTRY_OVERHEAD);
	/* check for overflow */
	if (result + allocs_overhead < result)
		goto err_overflow;
	result += allocs_overhead;

	/* SIZE_MAX is a special value */
	if (result == SIZE_MAX)
		goto err_overflow;

	return result;

err_overflow:
	errno = ERANGE;
	return SIZE_MAX;
}

/*
 * pmemobj_tx_set_user_data -- sets volatile pointer to the user data for the
 * current transaction
 */
void
pmemobj_tx_set_user_data(void *data)
{
	LOG(3, "data %p", data);

	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);

	tx->user_data = data;
}

/*
 * pmemobj_tx_get_user_data -- gets volatile pointer to the user data associated
 * with the current transaction
 */
void *
pmemobj_tx_get_user_data(void)
{
	LOG(3, NULL);

	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);

	return tx->user_data;
}

/*
 * pmemobj_tx_set_action_on_failure -- enables or disables automatic transaction
 * abort in case of an error
 */
void
pmemobj_tx_set_action_on_failure(enum pobj_tx_action_on_failure action)
{
	LOG(3, "enable %d", action);

	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);
	ASSERT_TX_STAGE_WORK(tx);

	struct tx_data *txd = PMDK_SLIST_FIRST(&tx->tx_entries);

	txd->action_on_failure = action;
}

/*
 * pmemobj_tx_get_action_on_failure -- returns enum specifying action on failure
 * for the current transaction.
 */
enum pobj_tx_action_on_failure
pmemobj_tx_get_action_on_failure(void)
{
	LOG(3, NULL);

	struct tx *tx = get_tx();

	ASSERT_IN_TX(tx);
	ASSERT_TX_STAGE_WORK(tx);

	struct tx_data *txd = PMDK_SLIST_FIRST(&tx->tx_entries);

	return txd->action_on_failure;
}

/*
 * CTL_READ_HANDLER(size) -- gets the cache size transaction parameter
 */
static int
CTL_READ_HANDLER(size)(void *ctx,
	enum ctl_query_source source, void *arg, struct ctl_indexes *indexes)
{
	PMEMobjpool *pop = ctx;

	ssize_t *arg_out = arg;

	*arg_out = (ssize_t)pop->tx_params->cache_size;

	return 0;
}

/*
 * CTL_WRITE_HANDLER(size) -- sets the cache size transaction parameter
 */
static int
CTL_WRITE_HANDLER(size)(void *ctx,
	enum ctl_query_source source, void *arg, struct ctl_indexes *indexes)
{
	PMEMobjpool *pop = ctx;

	ssize_t arg_in = *(int *)arg;

	if (arg_in < 0 || arg_in > (ssize_t)PMEMOBJ_MAX_ALLOC_SIZE) {
		errno = EINVAL;
		ERR("invalid cache size, must be between 0 and max alloc size");
		return -1;
	}

	size_t argu = (size_t)arg_in;

	pop->tx_params->cache_size = argu;

	return 0;
}

static const struct ctl_argument CTL_ARG(size) = CTL_ARG_LONG_LONG;

/*
 * CTL_READ_HANDLER(threshold) -- gets the cache threshold transaction parameter
 */
static int
CTL_READ_HANDLER(threshold)(void *ctx,
	enum ctl_query_source source, void *arg, struct ctl_indexes *indexes)
{
	LOG(1, "tx.cache.threshold parameter is deprecated");

	return 0;
}

/*
 * CTL_WRITE_HANDLER(threshold) -- deprecated
 */
static int
CTL_WRITE_HANDLER(threshold)(void *ctx,
	enum ctl_query_source source, void *arg, struct ctl_indexes *indexes)
{
	LOG(1, "tx.cache.threshold parameter is deprecated");

	return 0;
}

static const struct ctl_argument CTL_ARG(threshold) = CTL_ARG_LONG_LONG;

static const struct ctl_node CTL_NODE(cache)[] = {
	CTL_LEAF_RW(size),
	CTL_LEAF_RW(threshold),

	CTL_NODE_END
};

/*
 * CTL_READ_HANDLER(skip_expensive_checks) -- returns "skip_expensive_checks"
 * var from pool ctl
 */
static int
CTL_READ_HANDLER(skip_expensive_checks)(void *ctx,
	enum ctl_query_source source, void *arg, struct ctl_indexes *indexes)
{
	PMEMobjpool *pop = ctx;

	int *arg_out = arg;

	*arg_out = pop->tx_debug_skip_expensive_checks;

	return 0;
}

/*
 * CTL_WRITE_HANDLER(skip_expensive_checks) -- stores "skip_expensive_checks"
 * var in pool ctl
 */
static int
CTL_WRITE_HANDLER(skip_expensive_checks)(void *ctx,
	enum ctl_query_source source, void *arg, struct ctl_indexes *indexes)
{
	PMEMobjpool *pop = ctx;

	int arg_in = *(int *)arg;

	pop->tx_debug_skip_expensive_checks = arg_in;
	return 0;
}

static const struct ctl_argument CTL_ARG(skip_expensive_checks) =
		CTL_ARG_BOOLEAN;

/*
 * CTL_READ_HANDLER(verify_user_buffers) -- returns "ulog_user_buffers.verify"
 * variable from the pool
 */
static int
CTL_READ_HANDLER(verify_user_buffers)(void *ctx,
	enum ctl_query_source source, void *arg, struct ctl_indexes *indexes)
{
	PMEMobjpool *pop = ctx;

	int *arg_out = arg;

	*arg_out = pop->ulog_user_buffers.verify;

	return 0;
}

/*
 * CTL_WRITE_HANDLER(verify_user_buffers) -- sets "ulog_user_buffers.verify"
 * variable in the pool
 */
static int
CTL_WRITE_HANDLER(verify_user_buffers)(void *ctx,
	enum ctl_query_source source, void *arg, struct ctl_indexes *indexes)
{
	PMEMobjpool *pop = ctx;

	int arg_in = *(int *)arg;

	pop->ulog_user_buffers.verify = arg_in;
	return 0;
}

static const struct ctl_argument CTL_ARG(verify_user_buffers) =
		CTL_ARG_BOOLEAN;

static const struct ctl_node CTL_NODE(debug)[] = {
	CTL_LEAF_RW(skip_expensive_checks),
	CTL_LEAF_RW(verify_user_buffers),

	CTL_NODE_END
};

/*
 * CTL_READ_HANDLER(queue_depth) -- returns the depth of the post commit queue
 */
static int
CTL_READ_HANDLER(queue_depth)(void *ctx, enum ctl_query_source source,
	void *arg, struct ctl_indexes *indexes)
{
	return 0;
}

/*
 * CTL_WRITE_HANDLER(queue_depth) -- sets the depth of the post commit queue
 */
static int
CTL_WRITE_HANDLER(queue_depth)(void *ctx, enum ctl_query_source source,
	void *arg, struct ctl_indexes *indexes)
{
	return 0;
}

static const struct ctl_argument CTL_ARG(queue_depth) = CTL_ARG_INT;

/*
 * CTL_READ_HANDLER(worker) -- launches the post commit worker thread function
 */
static int
CTL_READ_HANDLER(worker)(void *ctx, enum ctl_query_source source,
	void *arg, struct ctl_indexes *indexes)
{
	return 0;
}

/*
 * CTL_READ_HANDLER(stop) -- stops all post commit workers
 */
static int
CTL_READ_HANDLER(stop)(void *ctx, enum ctl_query_source source,
	void *arg, struct ctl_indexes *indexes)
{
	return 0;
}

static const struct ctl_node CTL_NODE(post_commit)[] = {
	CTL_LEAF_RW(queue_depth),
	CTL_LEAF_RO(worker),
	CTL_LEAF_RO(stop),

	CTL_NODE_END
};

static const struct ctl_node CTL_NODE(tx)[] = {
	CTL_CHILD(debug),
	CTL_CHILD(cache),
	CTL_CHILD(post_commit),

	CTL_NODE_END
};

/*
 * tx_ctl_register -- registers ctl nodes for "tx" module
 */
void
tx_ctl_register(PMEMobjpool *pop)
{
	CTL_REGISTER_MODULE(pop->ctl, tx);
}