ldb key_value: Add batch_mode option

When performing a join the overhead of the sub transactions protecting
key value operations becomes significant.  This commit adds a new
"batch_mode" option that disables the sub transactions around key value
operations.

The operation level index cache is also disabled, which means the
overall transaction level index cache can become inconsistent if an
operation fails. To protect against this and other possible on disk
inconsistencies, if any operation fails during a batch_mode
transaction the commit will fail and transaction will be rolled back.

Signed-off-by: Gary Lockyer <gary@catalyst.net.nz>
Reviewed-by: Andrew Bartlett <abartlet@samba.org>

2 files changed, 76 insertions, 0 deletions

diff --git a/lib/ldb/ldb_key_value/ldb_kv.c b/lib/ldb/ldb_key_value/ldb_kv.c
index 56316eb8975..f768fb5e1e4 100644
--- a/lib/ldb/ldb_key_value/ldb_kv.c
+++ b/lib/ldb/ldb_key_value/ldb_kv.c
@@ -471,11 +471,16 @@ static bool ldb_kv_single_valued(const struct ldb_schema_attribute *a,
 
 /*
  * Starts a sub transaction if they are supported by the backend
+ * and the ldb connection has not been opened in batch mode.
  */
 static int ldb_kv_sub_transaction_start(struct ldb_kv_private *ldb_kv)
 {
 	int ret = LDB_SUCCESS;
 
+	if (ldb_kv->batch_mode) {
+		return ret;
+	}
+
 	ret = ldb_kv->kv_ops->begin_nested_write(ldb_kv);
 	if (ret == LDB_SUCCESS) {
 		ret = ldb_kv_index_sub_transaction_start(ldb_kv);
@@ -485,11 +490,16 @@ static int ldb_kv_sub_transaction_start(struct ldb_kv_private *ldb_kv)
 
 /*
  * Commits a sub transaction if they are supported by the backend
+ * and the ldb connection has not been opened in batch mode.
  */
 static int ldb_kv_sub_transaction_commit(struct ldb_kv_private *ldb_kv)
 {
 	int ret = LDB_SUCCESS;
 
+	if (ldb_kv->batch_mode) {
+		return ret;
+	}
+
 	ret = ldb_kv_index_sub_transaction_commit(ldb_kv);
 	if (ret != LDB_SUCCESS) {
 		return ret;
@@ -500,11 +510,16 @@ static int ldb_kv_sub_transaction_commit(struct ldb_kv_private *ldb_kv)
 
 /*
  * Cancels a sub transaction if they are supported by the backend
+ * and the ldb connection has not been opened in batch mode.
  */
 static int ldb_kv_sub_transaction_cancel(struct ldb_kv_private *ldb_kv)
 {
 	int ret = LDB_SUCCESS;
 
+	if (ldb_kv->batch_mode) {
+		return ret;
+	}
+
 	ret = ldb_kv_index_sub_transaction_cancel(ldb_kv);
 	if (ret != LDB_SUCCESS) {
 		struct ldb_context *ldb = ldb_module_get_ctx(ldb_kv->module);
@@ -691,6 +706,7 @@ static int ldb_kv_add(struct ldb_kv_context *ctx)
 				__location__
 				": Unable to roll back sub transaction");
 		}
+		ldb_kv->operation_failed = true;
 		return ret;
 	}
 	ret = ldb_kv_sub_transaction_commit(ldb_kv);
@@ -816,6 +832,9 @@ static int ldb_kv_delete(struct ldb_kv_context *ctx)
 				__location__
 				": Unable to roll back sub transaction");
 		}
+		if (ret != LDB_ERR_NO_SUCH_OBJECT) {
+			ldb_kv->operation_failed = true;
+		}
 		return ret;
 	}
 	ret = ldb_kv_sub_transaction_commit(ldb_kv);
@@ -1380,6 +1399,9 @@ static int ldb_kv_modify(struct ldb_kv_context *ctx)
 				__location__
 				": Unable to roll back sub transaction");
 		}
+		if (ret != LDB_ERR_NO_SUCH_OBJECT) {
+			ldb_kv->operation_failed = true;
+		}
 		return ret;
 	}
 	ret = ldb_kv_sub_transaction_commit(ldb_kv);
@@ -1545,6 +1567,7 @@ static int ldb_kv_rename(struct ldb_kv_context *ctx)
 				": Unable to roll back sub transaction");
 		}
 		talloc_free(msg);
+		ldb_kv->operation_failed = true;
 		return ret;
 	}
 	ret = ldb_kv_sub_transaction_commit(ldb_kv);
@@ -1585,6 +1608,7 @@ static int ldb_kv_start_trans(struct ldb_module *module)
 		ldb_kv->index_transaction_cache_size);
 
 	ldb_kv->reindex_failed = false;
+	ldb_kv->operation_failed = false;
 
 	return LDB_SUCCESS;
 }
@@ -1681,6 +1705,32 @@ static int ldb_kv_end_trans(struct ldb_module *module)
 	struct ldb_kv_private *ldb_kv =
 	    talloc_get_type(data, struct ldb_kv_private);
 
+	/*
+	 * If in batch mode and there has been an operation failure
+	 * rollback the transaction rather than committing it to avoid
+	 * any possible corruption
+	 */
+	if (ldb_kv->batch_mode && ldb_kv->operation_failed) {
+		ret = ldb_kv_del_trans( module);
+		if (ret != LDB_SUCCESS) {
+			ldb_debug_set(ldb_module_get_ctx(module),
+				      LDB_DEBUG_FATAL,
+				      "An operation failed during a batch mode "
+				      "transaction. The transaction could not"
+				      "be rolled back, ldb_kv_del_trans "
+				      "returned (%s, %s)",
+				      ldb_kv->kv_ops->errorstr(ldb_kv),
+				      ldb_strerror(ret));
+		} else {
+			ldb_debug_set(ldb_module_get_ctx(module),
+				      LDB_DEBUG_FATAL,
+				      "An operation failed during a batch mode "
+				      "transaction, the transaction was "
+				      "rolled back");
+		}
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
 	if (!ldb_kv->prepared_commit) {
 		ret = ldb_kv_prepare_commit(module);
 		if (ret != LDB_SUCCESS) {
@@ -2226,6 +2276,19 @@ int ldb_kv_init_store(struct ldb_kv_private *ldb_kv,
 			}
 		}
 	}
+	/*
+	 * Set batch mode operation.
+	 * This disables the nested sub transactions, and increases the
+	 * chance of index corruption.  If using this mode the transaction
+	 * commit will be aborted if any operation fails.
+	 */
+	{
+		const char *batch_mode = ldb_options_find(
+			ldb, options, "batch_mode");
+		if (batch_mode != NULL) {
+			ldb_kv->batch_mode = true;
+		}
+	}
 
 	return LDB_SUCCESS;
 }
diff --git a/lib/ldb/ldb_key_value/ldb_kv.h b/lib/ldb/ldb_key_value/ldb_kv.h
index 7ebda13692e..e627644ba34 100644
--- a/lib/ldb/ldb_key_value/ldb_kv.h
+++ b/lib/ldb/ldb_key_value/ldb_kv.h
@@ -98,6 +98,19 @@ struct ldb_kv_private {
 	 * the contents of this cache are copied to idxptr
 	 */
 	struct ldb_kv_idxptr *nested_idx_ptr;
+	/*
+	 * If batch mode is set the sub transactions and index caching
+	 * wrapping individual operations is disabled.
+	 * This is to improve the performance of large batch operations
+	 * i.e. domain joins.
+	 */
+	bool batch_mode;
+	/*
+	 * Has an operation failed, if true and we're in batch_mode
+	 * the transaction commit will fail.
+	 */
+	bool operation_failed;
+
 	bool prepared_commit;
 	int read_lock_count;