From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from smtpng3.m.smailru.net (smtpng3.m.smailru.net [94.100.177.149]) (using TLSv1.2 with cipher ECDHE-RSA-AES256-GCM-SHA384 (256/256 bits)) (No client certificate requested) by dev.tarantool.org (Postfix) with ESMTPS id 6752D44532B for ; Wed, 15 Jul 2020 16:55:43 +0300 (MSK) Received: by smtpng3.m.smailru.net with esmtpa (envelope-from ) id 1jvhss-0002ht-Vz for tarantool-patches@dev.tarantool.org; Wed, 15 Jul 2020 16:55:43 +0300 From: Aleksandr Lyapunov Date: Wed, 15 Jul 2020 16:55:33 +0300 Message-Id: <1594821336-14468-11-git-send-email-alyapunov@tarantool.org> In-Reply-To: <1594821336-14468-1-git-send-email-alyapunov@tarantool.org> References: <1594821336-14468-1-git-send-email-alyapunov@tarantool.org> Subject: [Tarantool-patches] [PATCH v3 10/13] tmx: introduce conflict tracker List-Id: Tarantool development patches List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , To: tarantool-patches@dev.tarantool.org There are situations when we have to track that if some TX is committed then some others must be aborted due to conflict. The common case is that one r/w TX have read some value while the second is about to overwrite the value; is the second is committed, the first must be aborted. Thus we have to store many-to-many TX relations between breaker TX and victim TX. The patch implements that. Part of #4897 --- src/box/txn.c | 156 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++- src/box/txn.h | 32 ++++++++++++ 2 files changed, 187 insertions(+), 1 deletion(-) diff --git a/src/box/txn.c b/src/box/txn.c index 7a15986..ba81b58 100644 --- a/src/box/txn.c +++ b/src/box/txn.c @@ -42,6 +42,12 @@ struct tx_manager { /** Last prepare-sequence-number that was assigned to prepared TX. */ int64_t last_psn; + /** + * List of all transactions that are in a read view. + * New transactions are added to the tail of this list, + * so the list is ordered by rv_psn. + */ + struct rlist read_view_txs; }; /** That's a definition, see declaration for description. */ @@ -50,6 +56,21 @@ bool tx_manager_use_mvcc_engine = false; /** The one and only instance of tx_manager. */ static struct tx_manager txm; +/** + * Record that links two transactions, breaker and victim. + * See txm_cause_conflict for details. + */ +struct tx_conflict_tracker { + /** TX that aborts victim on commit. */ + struct txn *breaker; + /** TX that aborts will be aborted on breaker's commit. */ + struct txn *victim; + /** Link in breaker->conflict_list. */ + struct rlist in_conflict_list; + /** Link in victim->conflicted_by_list. */ + struct rlist in_conflicted_by_list; +}; + double too_long_threshold; /* Txn cache. */ @@ -209,6 +230,9 @@ txn_new(void) } assert(region_used(®ion) == sizeof(*txn)); txn->region = region; + rlist_create(&txn->conflict_list); + rlist_create(&txn->conflicted_by_list); + rlist_create(&txn->in_read_view_txs); return txn; } @@ -218,6 +242,22 @@ txn_new(void) inline static void txn_free(struct txn *txn) { + struct tx_conflict_tracker *entry, *next; + rlist_foreach_entry_safe(entry, &txn->conflict_list, + in_conflict_list, next) { + rlist_del(&entry->in_conflict_list); + rlist_del(&entry->in_conflicted_by_list); + } + rlist_foreach_entry_safe(entry, &txn->conflicted_by_list, + in_conflicted_by_list, next) { + rlist_del(&entry->in_conflict_list); + rlist_del(&entry->in_conflicted_by_list); + } + assert(rlist_empty(&txn->conflict_list)); + assert(rlist_empty(&txn->conflicted_by_list)); + + rlist_del(&txn->in_read_view_txs); + struct txn_stmt *stmt; stailq_foreach_entry(stmt, &txn->stmts, next) txn_stmt_destroy(stmt); @@ -235,6 +275,8 @@ txn_begin(void) struct txn *txn = txn_new(); if (txn == NULL) return NULL; + assert(rlist_empty(&txn->conflict_list)); + assert(rlist_empty(&txn->conflicted_by_list)); /* Initialize members explicitly to save time on memset() */ stailq_create(&txn->stmts); @@ -245,6 +287,7 @@ txn_begin(void) txn->in_sub_stmt = 0; txn->id = ++tsn; txn->psn = 0; + txn->rv_psn = 0; txn->status = TXN_INPROGRESS; txn->signature = TXN_SIGNATURE_ROLLBACK; txn->engine = NULL; @@ -293,6 +336,15 @@ txn_begin_stmt(struct txn *txn, struct space *space) diag_set(ClientError, ER_SUB_STMT_MAX); return -1; } + + /* + * A conflict have happened; there is no reason to continue the TX. + */ + if (txn->status == TXN_CONFLICTED) { + diag_set(ClientError, ER_TRANSACTION_CONFLICT); + return -1; + } + struct txn_stmt *stmt = txn_stmt_new(&txn->region); if (stmt == NULL) return -1; @@ -647,6 +699,32 @@ txn_journal_entry_new(struct txn *txn) return req; } +/** + * Handle conflict when @breaker transaction is prepared. + * The conflict is happened if @victim have read something that @breaker + * overwrites. + * If @victim is read-only or haven't made any changes, it should be send + * to read view, in which is will not see @breaker. + * Otherwise @vistim must be marked as conflicted. + */ +static void +txn_handle_conflict(struct txn *breaker, struct txn *victim) +{ + assert(breaker->psn != 0); + if (!victim->status != TXN_INPROGRESS) { + /* Was conflicted by somebody else. */ + return; + } + if (stailq_empty(&victim->stmts)) { + /* Send to read view. */ + victim->rv_psn = breaker->psn; + rlist_add_tail(&txm.read_view_txs, &victim->in_read_view_txs); + } else { + /* Mark as conflicted. */ + victim->status = TXN_CONFLICTED; + } +} + /* * Prepare a transaction using engines. */ @@ -670,6 +748,17 @@ txn_prepare(struct txn *txn) diag_set(ClientError, ER_FOREIGN_KEY_CONSTRAINT); return -1; } + + /* + * Somebody else has written some value that we have read. + * The transaction is not possible. + */ + if (txn->status == TXN_CONFLICTED || txn->status == TXN_IN_READ_VIEW) { + diag_set(ClientError, ER_TRANSACTION_CONFLICT); + return -1; + } + assert(txn->status == TXN_INPROGRESS); + /* * Perform transaction conflict resolution. Engine == NULL when * we have a bunch of IPROTO_NOP statements. @@ -678,6 +767,26 @@ txn_prepare(struct txn *txn) if (engine_prepare(txn->engine, txn) != 0) return -1; } + + struct tx_conflict_tracker *entry, *next; + /* Handle conflicts. */ + rlist_foreach_entry_safe(entry, &txn->conflict_list, + in_conflict_list, next) { + assert(entry->breaker == txn); + txn_handle_conflict(txn, entry->victim); + rlist_del(&entry->in_conflict_list); + rlist_del(&entry->in_conflicted_by_list); + } + /* Just free conflict list - we don't need it anymore. */ + rlist_foreach_entry_safe(entry, &txn->conflicted_by_list, + in_conflicted_by_list, next) { + assert(entry->victim == txn); + rlist_del(&entry->in_conflict_list); + rlist_del(&entry->in_conflicted_by_list); + } + assert(rlist_empty(&txn->conflict_list)); + assert(rlist_empty(&txn->conflicted_by_list)); + trigger_clear(&txn->fiber_on_stop); if (!txn_has_flag(txn, TXN_CAN_YIELD)) trigger_clear(&txn->fiber_on_yield); @@ -1184,10 +1293,55 @@ txn_on_yield(struct trigger *trigger, void *event) void tx_manager_init() { - (void)txm; + rlist_create(&txm.read_view_txs); } void tx_manager_free() { } + +int +txm_cause_conflict(struct txn *breaker, struct txn *victim) +{ + struct tx_conflict_tracker *tracker = NULL; + struct rlist *r1 = breaker->conflict_list.next; + struct rlist *r2 = victim->conflicted_by_list.next; + while (r1 != &breaker->conflict_list && + r2 != &victim->conflicted_by_list) { + tracker = rlist_entry(r1, struct tx_conflict_tracker, + in_conflict_list); + assert(tracker->breaker == breaker); + if (tracker->victim == victim) + break; + tracker = rlist_entry(r2, struct tx_conflict_tracker, + in_conflicted_by_list); + assert(tracker->victim == victim); + if (tracker->breaker == breaker) + break; + tracker = NULL; + r1 = r1->next; + r2 = r2->next; + } + if (tracker != NULL) { + /* Move to the beginning of a list + * for a case of subsequent lookups */ + rlist_del(&tracker->in_conflict_list); + rlist_del(&tracker->in_conflicted_by_list); + } else { + size_t size; + tracker = region_alloc_object(&victim->region, + struct tx_conflict_tracker, + &size); + if (tracker == NULL) { + diag_set(OutOfMemory, size, "tx region", + "conflict_tracker"); + return -1; + } + tracker->breaker = breaker; + tracker->victim = victim; + } + rlist_add(&breaker->conflict_list, &tracker->in_conflict_list); + rlist_add(&victim->conflicted_by_list, &tracker->in_conflicted_by_list); + return 0; +} diff --git a/src/box/txn.h b/src/box/txn.h index b5b94fd..03ccb76 100644 --- a/src/box/txn.h +++ b/src/box/txn.h @@ -275,6 +275,11 @@ struct txn { * Transactions are committed in that order. */ int64_t psn; + /** + * Read view of that TX. The TX can see only changes with ps < rv_psn. + * Is nonzero if and only if status = TXN_IN_READ_VIEW. + */ + int64_t rv_psn; /** Status of the TX */ enum txn_status status; /** List of statements in a transaction. */ @@ -336,6 +341,22 @@ struct txn { uint32_t fk_deferred_count; /** List of savepoints to find savepoint by name. */ struct rlist savepoints; + /** + * List of tx_conflict_tracker records where .breaker is the current + * transaction and .victim is the transactions that must be aborted + * if the current transaction is committed. + */ + struct rlist conflict_list; + /** + * List of tx_conflict_tracker records where .victim is the current + * transaction and .breaker is the transactions that, if committed, + * will abort the current transaction. + */ + struct rlist conflicted_by_list; + /** + * Link in tx_manager::read_view_txs. + */ + struct rlist in_read_view_txs; }; static inline bool @@ -717,6 +738,17 @@ tx_manager_init(); void tx_manager_free(); +/** + * Notify TX manager that if transaction @breaker is committed then the + * transactions @victim must be aborted due to conflict. + * For example: there's two rw transaction in progress, one have read + * some value while the second is about to overwrite it. If the second + * is committed first, the first must be aborted. + * @return 0 on success, -1 on memory error. + */ +int +txm_cause_conflict(struct txn *breaker, struct txn *victim); + #if defined(__cplusplus) } /* extern "C" */ #endif /* defined(__cplusplus) */ -- 2.7.4