obj: introduce pmemobj_tx_abort_on_failure

This function can be used to disable or enable aborting tx
in case of an error (for example when allocation fails).

This patch also introduces pmemobj_tx_is_abort_on_failure
which allows to query current setting of a transaction.

doc/libpmemobj/pmemobj_tx_begin.3.md

@@ -37,7 +37,10 @@ date: pmemobj API version 2.3
 **pmemobj_tx_log_intents_max_size**(),
 
 **pmemobj_tx_set_user_data**(),
-**pmemobj_tx_get_user_data**()
+**pmemobj_tx_get_user_data**(),
+
+**pmemobj_tx_set_abort_on_failure**(),
+**pmemobj_tx_get_abort_on_failure**()
 - transactional object manipulation
 
 # SYNOPSIS #
@@ -72,6 +75,9 @@ size_t pmemobj_tx_log_intents_max_size(size_t nintents);
 
 void pmemobj_tx_set_user_data(void *data);
 void *pmemobj_tx_get_user_data(void);
+
+void pmemobj_tx_set_abort_on_failure(int enable);
+int pmemobj_tx_get_abort_on_failure(void);
 ```
 
 # DESCRIPTION #
@@ -412,6 +418,19 @@ If **pmemobj_tx_set_user_data**() was not called for a current transaction,
 during **TX_STAGE_WORK** or **TX_STAGE_ONABORT** or **TX_STAGE_ONCOMMIT** or
 **TX_STAGE_FINALLY**.
 
+**pmemobj_tx_set_abort_on_failure**() enables or disables automatic transaction
+abort in case **pmemobj_tx_add_rage**, **pmemobj_tx_alloc**, **pmemobj_tx_free**,
+**pmemobj_tx_lock**, **pmemobj_tx_strdup**, **pmemobj_tx_xpublish** or
+**pmemobj_tx_log_append_buffer** fails. If called with **enable** == 0
+*_NO_ABORT flag will be implicitly passed to all of those functions. If
+**pmemobj_tx_set_abort_on_failure** is not called (or called with **enable** == 1),
+the transaction will abort if any of the previously listed functions fails (unless
+*_NO_ABORT flag was passed expliciltly). This functions affects the current transaction
+and all inner transactions. It does not affect any outer transactions.
+**pmemobj_tx_get_abort_on_failure**() function returns 1 if transaction automatically
+aborts on failure, 0 otherwise. Both **pmemobj_tx_set_abort_on_failure**() and
+**pmemobj_tx_get_abort_on_failure**() must be called during **TX_STAGE_WORK**.
+
 # RETURN VALUE #
 
 The **pmemobj_tx_stage**() function returns the stage of the current transaction

src/include/libpmemobj/tx_base.h

@@ -424,6 +424,25 @@ void pmemobj_tx_set_user_data(void *data);
  */
 void *pmemobj_tx_get_user_data(void);
 
+/*
+ * Enables or disables automatic transaction abort in case of an error.
+ * If called with enable == 0 _NO_ABORT flag will be implicitly passed to
+ * functions which accept such flag.
+ *
+ * This setting is inherited by inner transactions. Aborting on failure is
+ * the default behaviour.
+ *
+ * This function must be called during TX_STAGE_WORK.
+ */
+void pmemobj_tx_set_abort_on_failure(int enable);
+
+/*
+ * Returns 1 if transaction automatically aborts on failure, 0 otherwise.
+ *
+ * This function must be called during TX_STAGE_WORK.
+ */
+int pmemobj_tx_get_abort_on_failure(void);
+
 #ifdef __cplusplus
 }
 #endif

src/libpmemobj/libpmemobj.def

@@ -96,6 +96,8 @@ EXPORTS
 	pmemobj_tx_log_intents_max_size
 	pmemobj_tx_set_user_data
 	pmemobj_tx_get_user_data
+	pmemobj_tx_set_abort_on_failure
+	pmemobj_tx_get_abort_on_failure
 	pmemobj_memcpy
 	pmemobj_memcpy_persist
 	pmemobj_memmove

src/libpmemobj/libpmemobj.link.in

@@ -87,6 +87,8 @@ LIBPMEMOBJ_1.0 {
 		pmemobj_tx_log_intents_max_size;
 		pmemobj_tx_set_user_data;
 		pmemobj_tx_get_user_data;
+		pmemobj_tx_set_abort_on_failure;
+		pmemobj_tx_get_abort_on_failure;
 		pmemobj_memcpy;
 		pmemobj_memcpy_persist;
 		pmemobj_memmove;

src/libpmemobj/tx.c

@@ -20,6 +20,7 @@
 struct tx_data {
 	PMDK_SLIST_ENTRY(tx_data) tx_entry;
 	jmp_buf env;
+	int abort_on_failure;
 };
 
 struct tx {
@@ -135,7 +136,10 @@ obj_tx_abort(int errnum, int user);
 static inline int
 obj_tx_fail_err(int errnum, uint64_t flags)
 {
-	if ((flags & POBJ_FLAG_TX_NO_ABORT) == 0)
+	struct tx *tx = get_tx();
+	struct tx_data *txd = PMDK_SLIST_FIRST(&tx->tx_entries);
+
+	if ((flags & POBJ_FLAG_TX_NO_ABORT) == 0 && txd->abort_on_failure)
 		obj_tx_abort(errnum, 0);
 	errno = errnum;
 	return errnum;
@@ -148,7 +152,10 @@ obj_tx_fail_err(int errnum, uint64_t flags)
 static inline PMEMoid
 obj_tx_fail_null(int errnum, uint64_t flags)
 {
-	if ((flags & POBJ_FLAG_TX_NO_ABORT) == 0)
+	struct tx *tx = get_tx();
+	struct tx_data *txd = PMDK_SLIST_FIRST(&tx->tx_entries);
+
+	if ((flags & POBJ_FLAG_TX_NO_ABORT) == 0 && txd->abort_on_failure)
 		obj_tx_abort(errnum, 0);
 	errno = errnum;
 	return OID_NULL;
@@ -719,13 +726,19 @@ pmemobj_tx_begin(PMEMobjpool *pop, jmp_buf env, ...)
 	int err = 0;
 	struct tx *tx = get_tx();
 
+	int abort_on_failure = 1;
+
 	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 parent transaction */
+		struct tx_data *txd = PMDK_SLIST_FIRST(&tx->tx_entries);
+		abort_on_failure = txd->abort_on_failure;
+
 		VALGRIND_START_TX;
 	} else if (tx->stage == TX_STAGE_NONE) {
 		VALGRIND_START_TX;
@@ -764,6 +777,8 @@ pmemobj_tx_begin(PMEMobjpool *pop, jmp_buf env, ...)
 	else
 		memset(txd->env, 0, sizeof(jmp_buf));
 
+	txd->abort_on_failure = abort_on_failure;
+
 	PMDK_SLIST_INSERT_HEAD(&tx->tx_entries, txd, tx_entry);
 
 	tx->stage = TX_STAGE_WORK;
@@ -2053,6 +2068,44 @@ pmemobj_tx_get_user_data(void)
 	return tx->user_data;
 }
 
+/*
+ * pmemobj_tx_set_abort_on_failure -- enables or disables automatic transaction
+ * abort in case of an error
+ */
+void
+pmemobj_tx_set_abort_on_failure(int enable)
+{
+	LOG(3, "enable %d", enable);
+
+	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->abort_on_failure = enable;
+}
+
+/*
+ * pmemobj_tx_get_abort_on_failure -- returns 1 if transaction automatically
+ * aborts on failure, 0 otherwise
+ */
+int
+pmemobj_tx_get_abort_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->abort_on_failure;
+}
+
 /*
  * CTL_READ_HANDLER(size) -- gets the cache size transaction parameter
  */

src/test/obj_tx_add_range/obj_tx_add_range.c

@@ -1,5 +1,5 @@
 // SPDX-License-Identifier: BSD-3-Clause
-/* Copyright 2015-2019, Intel Corporation */
+/* Copyright 2015-2020, Intel Corporation */
 
 /*
  * obj_tx_add_range.c -- unit test for pmemobj_tx_add_range
@@ -1053,6 +1053,15 @@ do_tx_add_range_wrong_uuid(PMEMobjpool *pop)
 	} TX_END
 
 	UT_ASSERTeq(errno, EINVAL);
+
+	TX_BEGIN(pop) {
+		pmemobj_tx_set_abort_on_failure(0);
+		pmemobj_tx_add_range(oid, 0, 0);
+	} TX_ONABORT {
+		UT_ASSERT(0);
+	} TX_END
+
+	UT_ASSERTeq(errno, EINVAL);
 }
 
 int

src/test/obj_tx_add_range_direct/obj_tx_add_range_direct.c

@@ -1,5 +1,5 @@
 // SPDX-License-Identifier: BSD-3-Clause
-/* Copyright 2015-2019, Intel Corporation */
+/* Copyright 2015-2020, Intel Corporation */
 
 /*
  * obj_tx_add_range_direct.c -- unit test for pmemobj_tx_add_range_direct
@@ -754,6 +754,20 @@ do_tx_add_range_too_large(PMEMobjpool *pop)
 		UT_ASSERT(0);
 	} TX_END
 
+	errno = 0;
+	ret = 0;
+
+	TX_BEGIN(pop) {
+		pmemobj_tx_set_abort_on_failure(0);
+		ret = pmemobj_tx_add_range_direct(pmemobj_direct(obj.oid),
+				PMEMOBJ_MAX_ALLOC_SIZE + 1);
+	} TX_ONCOMMIT {
+		UT_ASSERTeq(errno, EINVAL);
+		UT_ASSERTeq(ret, EINVAL);
+	} TX_ONABORT {
+		UT_ASSERT(0);
+	} TX_END
+
 	errno = 0;
 }
 

src/test/obj_tx_alloc/obj_tx_alloc.c

@@ -547,6 +547,18 @@ do_tx_xalloc_zerolen(PMEMobjpool *pop)
 
 	UT_ASSERT(TOID_IS_NULL(obj));
 
+	/* alloc 0 with pmemobj_tx_set_abort_on_failure called */
+	TX_BEGIN(pop) {
+		pmemobj_tx_set_abort_on_failure(0);
+		TOID_ASSIGN(obj, pmemobj_tx_alloc(0, TYPE_XABORT));
+	} TX_ONCOMMIT {
+		TOID_ASSIGN(obj, OID_NULL);
+	} TX_ONABORT {
+		UT_ASSERT(0); /* should not get to this point */
+	} TX_END
+
+	UT_ASSERT(TOID_IS_NULL(obj));
+
 	TOID(struct object) first;
 	TOID_ASSIGN(first, POBJ_FIRST_TYPE_NUM(pop, TYPE_XABORT));
 	UT_ASSERT(TOID_IS_NULL(first));

src/test/obj_tx_free/obj_tx_free.c

@@ -1,5 +1,5 @@
 // SPDX-License-Identifier: BSD-3-Clause
-/* Copyright 2015-2019, Intel Corporation */
+/* Copyright 2015-2020, Intel Corporation */
 
 /*
  * obj_tx_free.c -- unit test for pmemobj_tx_free
@@ -87,6 +87,103 @@ do_tx_free_wrong_uuid(PMEMobjpool *pop)
 	UT_ASSERT(!TOID_IS_NULL(obj));
 }
 
+/*
+ * do_tx_free_wrong_uuid_abort_on_failure -- try to free object with
+ * invalid uuid in a transaction where pmemobj_tx_set_abort_on_failure
+ * was called.
+ */
+static void
+do_tx_free_wrong_uuid_abort_on_failure(PMEMobjpool *pop)
+{
+	volatile int ret = 0;
+	PMEMoid oid = do_tx_alloc(pop, TYPE_FREE_WRONG_UUID);
+	oid.pool_uuid_lo = ~oid.pool_uuid_lo;
+
+	/* pmemobj_tx_set_abort_on_failure is called */
+	TX_BEGIN(pop) {
+		pmemobj_tx_set_abort_on_failure(0);
+
+		UT_ASSERT(!pmemobj_tx_get_abort_on_failure());
+		ret = pmemobj_tx_free(oid);
+	} TX_ONCOMMIT {
+		UT_ASSERTeq(ret, EINVAL);
+	} TX_ONABORT {
+		UT_ASSERT(0); /* should not get to this point */
+	} TX_END
+
+	/* pmemobj_tx_set_abort_on_failure is called in outer tx */
+	TX_BEGIN(pop) {
+		pmemobj_tx_set_abort_on_failure(0);
+		TX_BEGIN(pop) {
+			UT_ASSERT(!pmemobj_tx_get_abort_on_failure());
+			ret = pmemobj_tx_free(oid);
+		} TX_ONCOMMIT {
+			UT_ASSERTeq(ret, EINVAL);
+		} TX_ONABORT {
+			UT_ASSERT(0); /* should not get to this point */
+		} TX_END
+		ret = pmemobj_tx_free(oid);
+	} TX_ONCOMMIT {
+		UT_ASSERTeq(ret, EINVAL);
+	} TX_ONABORT {
+		UT_ASSERT(0); /* should not get to this point */
+	} TX_END
+
+	/* pmemobj_tx_set_abort_on_failure is called in neighbour tx */
+	TX_BEGIN(pop) {
+		TX_BEGIN(pop) {
+			pmemobj_tx_set_abort_on_failure(0);
+			ret = pmemobj_tx_free(oid);
+		} TX_ONCOMMIT {
+			UT_ASSERTeq(ret, EINVAL);
+		} TX_ONABORT {
+			UT_ASSERT(0); /* should not get to this point */
+		} TX_END
+
+		TX_BEGIN(pop) {
+			UT_ASSERT(pmemobj_tx_get_abort_on_failure());
+		} TX_ONCOMMIT {
+			UT_ASSERTeq(ret, EINVAL);
+		} TX_ONABORT {
+			UT_ASSERT(0); /* should not get to this point */
+		} TX_END
+	} TX_ONCOMMIT {
+		UT_ASSERTeq(ret, EINVAL);
+	} TX_ONABORT {
+		UT_ASSERT(0); /* should not get to this point */
+	} TX_END
+
+	/* pmemobj_tx_set_abort_on_failure is called in neighbour tx */
+	TX_BEGIN(pop) {
+		pmemobj_tx_set_abort_on_failure(0);
+		TX_BEGIN(pop) {
+			pmemobj_tx_set_abort_on_failure(1);
+			UT_ASSERT(pmemobj_tx_get_abort_on_failure());
+		} TX_ONCOMMIT {
+			UT_ASSERTeq(ret, EINVAL);
+		} TX_ONABORT {
+			UT_ASSERT(0); /* should not get to this point */
+		} TX_END
+
+		TX_BEGIN(pop) {
+			UT_ASSERT(!pmemobj_tx_get_abort_on_failure());
+			ret = pmemobj_tx_free(oid);
+		} TX_ONCOMMIT {
+			UT_ASSERTeq(ret, EINVAL);
+		} TX_ONABORT {
+			UT_ASSERT(0); /* should not get to this point */
+		} TX_END
+	} TX_ONCOMMIT {
+		UT_ASSERTeq(ret, EINVAL);
+	} TX_ONABORT {
+		UT_ASSERT(0); /* should not get to this point */
+	} TX_END
+
+	TOID(struct object) obj;
+	TOID_ASSIGN(obj, POBJ_FIRST_TYPE_NUM(pop, TYPE_FREE_WRONG_UUID));
+	UT_ASSERT(!TOID_IS_NULL(obj));
+}
+
 /*
  * do_tx_free_null_oid -- call pmemobj_tx_free with OID_NULL
  */
@@ -366,6 +463,8 @@ main(int argc, char *argv[])
 
 	do_tx_free_wrong_uuid(pop);
 	VALGRIND_WRITE_STATS;
+	do_tx_free_wrong_uuid_abort_on_failure(pop);
+	VALGRIND_WRITE_STATS;
 	do_tx_free_null_oid(pop);
 	VALGRIND_WRITE_STATS;
 	do_tx_free_commit(pop);