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 5D19F42EF62 for ; Fri, 3 Jul 2020 09:33:30 +0300 (MSK) Received: by smtpng3.m.smailru.net with esmtpa (envelope-from ) id 1jrFGL-0003jC-OX for tarantool-patches@dev.tarantool.org; Fri, 03 Jul 2020 09:33:30 +0300 From: Aleksandr Lyapunov Date: Fri, 3 Jul 2020 09:33:13 +0300 Message-Id: <1593757997-4145-12-git-send-email-alyapunov@tarantool.org> In-Reply-To: <1593757997-4145-1-git-send-email-alyapunov@tarantool.org> References: <1593757997-4145-1-git-send-email-alyapunov@tarantool.org> Subject: [Tarantool-patches] [PATCH 11/15] tx: introduce conflict tracker List-Id: Tarantool development patches List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , To: tarantool-patches@dev.tarantool.org --- src/box/txn.c | 98 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/box/txn.h | 5 +++ 2 files changed, 103 insertions(+) diff --git a/src/box/txn.c b/src/box/txn.c index d34ec3a..79384f0 100644 --- a/src/box/txn.c +++ b/src/box/txn.c @@ -44,6 +44,13 @@ struct tx_manager static struct tx_manager txm; +struct tx_conflict_tracker { + struct txn *wreaker; + struct txn *victim; + struct rlist in_conflict_list; + struct rlist in_conflicted_by_list; +}; + double too_long_threshold; /* Txn cache. */ @@ -196,6 +203,8 @@ txn_new(void) } assert(region_used(®ion) == sizeof(*txn)); txn->region = region; + rlist_create(&txn->conflict_list); + rlist_create(&txn->conflicted_by_list); return txn; } @@ -205,6 +214,20 @@ 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)); + struct txn_stmt *stmt; stailq_foreach_entry(stmt, &txn->stmts, next) txn_stmt_destroy(stmt); @@ -222,6 +245,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); @@ -279,6 +304,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; @@ -558,6 +592,16 @@ 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) { + diag_set(ClientError, ER_TRANSACTION_CONFLICT); + return -1; + } + /* * Perform transaction conflict resolution. Engine == NULL when * we have a bunch of IPROTO_NOP statements. @@ -566,6 +610,17 @@ txn_prepare(struct txn *txn) if (engine_prepare(txn->engine, txn) != 0) return -1; } + + struct tx_conflict_tracker *entry, *next; + rlist_foreach_entry_safe(entry, &txn->conflict_list, + in_conflict_list, next) { + if (entry->victim->status == TXN_INPROGRESS) + entry->victim->status = TXN_CONFLICTED; + rlist_del(&entry->in_conflict_list); + rlist_del(&entry->in_conflicted_by_list); + } + + trigger_clear(&txn->fiber_on_stop); if (!txn_has_flag(txn, TXN_CAN_YIELD)) trigger_clear(&txn->fiber_on_yield); @@ -973,3 +1028,46 @@ void tx_manager_free() { } + +int +txm_cause_conflict(struct txn *wreaker, struct txn *victim) +{ + struct tx_conflict_tracker *tracker = NULL; + struct rlist *r1 = wreaker->conflict_list.next; + struct rlist *r2 = wreaker->conflicted_by_list.next; + while (r1 != &wreaker->conflict_list && + r2 != &wreaker->conflicted_by_list) { + tracker = rlist_entry(r1, struct tx_conflict_tracker, + in_conflict_list); + if (tracker->wreaker == wreaker && tracker->victim == victim) + break; + tracker = rlist_entry(r2, struct tx_conflict_tracker, + in_conflicted_by_list); + if (tracker->wreaker == wreaker && tracker->victim == victim) + 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->wreaker = wreaker; + tracker->victim = victim; + rlist_add(&wreaker->conflict_list, &tracker->in_conflict_list); + rlist_add(&wreaker->conflicted_by_list, &tracker->in_conflicted_by_list); + return 0; +} \ No newline at end of file diff --git a/src/box/txn.h b/src/box/txn.h index cd1665f..92c0116 100644 --- a/src/box/txn.h +++ b/src/box/txn.h @@ -280,6 +280,8 @@ struct txn { uint32_t fk_deferred_count; /** List of savepoints to find savepoint by name. */ struct rlist savepoints; + struct rlist conflict_list; + struct rlist conflicted_by_list; }; static inline bool @@ -647,6 +649,9 @@ tx_manager_init(); void tx_manager_free(); +int +txm_cause_conflict(struct txn *wreaker, struct txn *victim); + #if defined(__cplusplus) } /* extern "C" */ #endif /* defined(__cplusplus) */ -- 2.7.4