sin/samba-mirror
1
0
Fork 0
mirror of https://github.com/samba-team/samba.git synced 2024-12-23 17:34:34 +03:00
Code

ldb key_value: Add batch_mode option

Browse Source 
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>
This commit is contained in:
Gary Lockyer 2019-07-02 12:30:44 +12:00 committed by Andrew Bartlett
parent 7d17dbd10e
commit 652258768a
2 changed files with 76 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
lib/ldb/ldb_key_value/ldb_kv.c
View File

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

13
lib/ldb/ldb_key_value/ldb_kv.h
View File

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