From: Georgy Kirichenko <georgy@tarantool.org> To: tarantool-patches@freelists.org Cc: Georgy Kirichenko <georgy@tarantool.org> Subject: [tarantool-patches] [PATCH v3 05/14] txn: get rid of autocommit from a txn structure Date: Sun, 9 Jun 2019 23:44:34 +0300 [thread overview] Message-ID: <e6181c23019464896f03515ac7efdd63375d2d09.1560112747.git.georgy@tarantool.org> (raw) In-Reply-To: <cover.1560112747.git.georgy@tarantool.org> Move transaction auto start and auto commit behavior to the box level. From now a transaction won't start and commit automatically without txn_begin/txn_commit invocations. This is a part of a bigger transaction refactoring in order to implement detachable transactions and a parallel applier. Prerequisites: #1254 --- src/box/applier.cc | 35 +++++++++++++--- src/box/box.cc | 94 +++++++++++++++++++++++++++++++++--------- src/box/index.cc | 10 ++--- src/box/memtx_engine.c | 10 ++++- src/box/memtx_space.c | 8 ++-- src/box/sql.c | 2 +- src/box/txn.c | 46 +++++++-------------- src/box/txn.h | 16 +++---- src/box/vinyl.c | 12 ++---- src/box/vy_scheduler.c | 6 +-- 10 files changed, 148 insertions(+), 91 deletions(-) diff --git a/src/box/applier.cc b/src/box/applier.cc index 373e1feb9..e3203a4c8 100644 --- a/src/box/applier.cc +++ b/src/box/applier.cc @@ -172,11 +172,26 @@ applier_writer_f(va_list ap) static int apply_initial_join_row(struct xrow_header *row) { + struct txn *txn = txn_begin(); + if (txn == NULL) + return -1; struct request request; xrow_decode_dml(row, &request, dml_request_key_map(row->type)); - struct space *space = space_cache_find_xc(request.space_id); + struct space *space = space_cache_find(request.space_id); + if (space == NULL) + goto rollback; /* no access checks here - applier always works with admin privs */ - return space_apply_initial_join_row(space, &request); + if (space_apply_initial_join_row(space, &request)) + goto rollback; + int rc; + rc = txn_commit(txn); + if (rc < 0) + return -1; + fiber_gc(); + return rc; +rollback: + txn_rollback(); + return -1; } /** @@ -189,8 +204,8 @@ static int process_nop(struct request *request) { assert(request->type == IPROTO_NOP); - struct txn *txn = txn_begin_stmt(NULL); - if (txn == NULL) + struct txn *txn = in_txn(); + if (txn_begin_stmt(txn, NULL) == NULL) return -1; return txn_commit_stmt(txn, request); } @@ -403,8 +418,16 @@ applier_join(struct applier *applier) applier->last_row_time = ev_monotonic_now(loop()); if (iproto_type_is_dml(row.type)) { vclock_follow_xrow(&replicaset.vclock, &row); - if (apply_row(&row) != 0) + struct txn *txn = txn_begin(); + if (txn == NULL) + diag_raise(); + if (apply_row(&row) != 0) { + txn_rollback(); + diag_raise(); + } + if (txn_commit(txn) != 0) diag_raise(); + fiber_gc(); if (++row_count % 100000 == 0) say_info("%.1fM rows received", row_count / 1e6); } else if (row.type == IPROTO_OK) { @@ -555,7 +578,7 @@ applier_apply_tx(struct stailq *rows) * conflict safely access failed xrow object and allocate * IPROTO_NOP on gc. */ - struct txn *txn = txn_begin(false); + struct txn *txn = txn_begin(); struct applier_tx_row *item; if (txn == NULL) diag_raise(); diff --git a/src/box/box.cc b/src/box/box.cc index 57419ee01..7f23716e5 100644 --- a/src/box/box.cc +++ b/src/box/box.cc @@ -169,34 +169,62 @@ int box_process_rw(struct request *request, struct space *space, struct tuple **result) { + struct tuple *tuple = NULL; + struct txn *txn = in_txn(); + bool is_autocommit = txn == NULL; + if (is_autocommit && (txn = txn_begin()) == NULL) + return -1; assert(iproto_type_is_dml(request->type)); rmean_collect(rmean_box, request->type, 1); if (access_check_space(space, PRIV_W) != 0) - return -1; - struct txn *txn = txn_begin_stmt(space); - if (txn == NULL) - return -1; - struct tuple *tuple; + goto fail; + if (txn_begin_stmt(txn, space) == NULL) + goto fail; if (space_execute_dml(space, txn, request, &tuple) != 0) { - txn_rollback_stmt(); - return -1; + txn_rollback_stmt(txn); + goto fail; + } + if (result != NULL) + *result = tuple; + + if (result == NULL || tuple == NULL) { + if (txn_commit_stmt(txn, request) != 0) + goto fail; + if (is_autocommit) { + if (txn_commit(txn) != 0) + return -1; + fiber_gc(); + } + return 0; } - if (result == NULL) - return txn_commit_stmt(txn, request); - *result = tuple; - if (tuple == NULL) - return txn_commit_stmt(txn, request); /* * Pin the tuple locally before the commit, * otherwise it may go away during yield in * when WAL is written in autocommit mode. */ tuple_ref(tuple); - int rc = txn_commit_stmt(txn, request); - if (rc == 0) - tuple_bless(tuple); + + if (txn_commit_stmt(txn, request)) { + /* Unref tuple and rollback if autocommit. */ + tuple_unref(tuple); + goto fail; + } + if (is_autocommit) { + if (txn_commit(txn) != 0) { + /* Unref tuple and exit. */ + tuple_unref(tuple); + return -1; + } + fiber_gc(); + } + tuple_bless(tuple); tuple_unref(tuple); - return rc; + return 0; + +fail: + if (is_autocommit) + txn_rollback(); + return -1; } void @@ -299,10 +327,20 @@ apply_wal_row(struct xstream *stream, struct xrow_header *row) xrow_decode_dml_xc(row, &request, dml_request_key_map(row->type)); if (request.type != IPROTO_NOP) { struct space *space = space_cache_find_xc(request.space_id); + struct txn *txn = txn_begin(); + if (txn == NULL) { + say_error("error applying row: %s", request_str(&request)); + diag_raise(); + } if (box_process_rw(&request, space, NULL) != 0) { say_error("error applying row: %s", request_str(&request)); + txn_rollback(); diag_raise(); } + if (txn_commit(txn) != 0) { + diag_raise(); + } + fiber_gc(); } struct wal_stream *xstream = container_of(stream, struct wal_stream, base); @@ -1055,7 +1093,7 @@ box_select(uint32_t space_id, uint32_t index_id, struct iterator *it = index_create_iterator(index, type, key, part_count); if (it == NULL) { - txn_rollback_stmt(); + txn_rollback_stmt(txn); return -1; } @@ -1080,7 +1118,7 @@ box_select(uint32_t space_id, uint32_t index_id, if (rc != 0) { port_destroy(port); - txn_rollback_stmt(); + txn_rollback_stmt(txn); return -1; } txn_commit_ro_stmt(txn); @@ -1313,9 +1351,17 @@ box_sequence_reset(uint32_t seq_id) static inline void box_register_replica(uint32_t id, const struct tt_uuid *uuid) { + struct txn *txn = txn_begin(); + if (txn == NULL) + diag_raise(); if (boxk(IPROTO_INSERT, BOX_CLUSTER_ID, "[%u%s]", - (unsigned) id, tt_uuid_str(uuid)) != 0) + (unsigned) id, tt_uuid_str(uuid)) != 0) { + txn_rollback(); diag_raise(); + } + if (txn_commit(txn) != 0) + diag_raise(); + fiber_gc(); assert(replica_by_uuid(uuid)->id == id); } @@ -1636,10 +1682,18 @@ box_set_replicaset_uuid(const struct tt_uuid *replicaset_uuid) uu = *replicaset_uuid; else tt_uuid_create(&uu); + struct txn *txn = txn_begin(); + if (txn == NULL) + diag_raise(); /* Save replica set UUID in _schema */ if (boxk(IPROTO_INSERT, BOX_SCHEMA_ID, "[%s%s]", "cluster", - tt_uuid_str(&uu))) + tt_uuid_str(&uu))) { + txn_rollback(); + diag_raise(); + } + if (txn_commit(txn) != 0) diag_raise(); + fiber_gc(); } void diff --git a/src/box/index.cc b/src/box/index.cc index 4a444e5d0..7f26c9bc2 100644 --- a/src/box/index.cc +++ b/src/box/index.cc @@ -240,7 +240,7 @@ box_index_get(uint32_t space_id, uint32_t index_id, const char *key, if (txn_begin_ro_stmt(space, &txn) != 0) return -1; if (index_get(index, key, part_count, result) != 0) { - txn_rollback_stmt(); + txn_rollback_stmt(txn); return -1; } txn_commit_ro_stmt(txn); @@ -274,7 +274,7 @@ box_index_min(uint32_t space_id, uint32_t index_id, const char *key, if (txn_begin_ro_stmt(space, &txn) != 0) return -1; if (index_min(index, key, part_count, result) != 0) { - txn_rollback_stmt(); + txn_rollback_stmt(txn); return -1; } txn_commit_ro_stmt(txn); @@ -306,7 +306,7 @@ box_index_max(uint32_t space_id, uint32_t index_id, const char *key, if (txn_begin_ro_stmt(space, &txn) != 0) return -1; if (index_max(index, key, part_count, result) != 0) { - txn_rollback_stmt(); + txn_rollback_stmt(txn); return -1; } txn_commit_ro_stmt(txn); @@ -340,7 +340,7 @@ box_index_count(uint32_t space_id, uint32_t index_id, int type, return -1; ssize_t count = index_count(index, itype, key, part_count); if (count < 0) { - txn_rollback_stmt(); + txn_rollback_stmt(txn); return -1; } txn_commit_ro_stmt(txn); @@ -377,7 +377,7 @@ box_index_iterator(uint32_t space_id, uint32_t index_id, int type, struct iterator *it = index_create_iterator(index, itype, key, part_count); if (it == NULL) { - txn_rollback_stmt(); + txn_rollback_stmt(txn); return NULL; } txn_commit_ro_stmt(txn); diff --git a/src/box/memtx_engine.c b/src/box/memtx_engine.c index f4312484a..149215b87 100644 --- a/src/box/memtx_engine.c +++ b/src/box/memtx_engine.c @@ -272,16 +272,22 @@ memtx_engine_recover_snapshot_row(struct memtx_engine *memtx, diag_set(ClientError, ER_CROSS_ENGINE_TRANSACTION); return -1; } + struct txn *txn = txn_begin(); + if (txn == NULL) + return -1; /* no access checks here - applier always works with admin privs */ - if (space_apply_initial_join_row(space, &request) != 0) + if (space_apply_initial_join_row(space, &request) != 0) { + txn_rollback(); return -1; + } + int rc = txn_commit(txn); /* * Don't let gc pool grow too much. Yet to * it before reading the next row, to make * sure it's not freed along here. */ fiber_gc(); - return 0; + return rc; } /** Called at start to tell memtx to recover to a given LSN. */ diff --git a/src/box/memtx_space.c b/src/box/memtx_space.c index 1d209033c..1e630783b 100644 --- a/src/box/memtx_space.c +++ b/src/box/memtx_space.c @@ -310,10 +310,10 @@ memtx_space_apply_initial_join_row(struct space *space, struct request *request) return -1; } request->header->replica_id = 0; - struct txn *txn = txn_begin_stmt(space); - if (txn == NULL) + struct txn *txn = in_txn(); + struct txn_stmt *stmt = txn_begin_stmt(txn, space); + if (stmt == NULL) return -1; - struct txn_stmt *stmt = txn_current_stmt(txn); stmt->new_tuple = memtx_tuple_new(space->format, request->tuple, request->tuple_end); if (stmt->new_tuple == NULL) @@ -326,7 +326,7 @@ memtx_space_apply_initial_join_row(struct space *space, struct request *request) rollback: say_error("rollback: %s", diag_last_error(diag_get())->errmsg); - txn_rollback_stmt(); + txn_rollback_stmt(txn); return -1; } diff --git a/src/box/sql.c b/src/box/sql.c index fbfa59992..1bdbdb815 100644 --- a/src/box/sql.c +++ b/src/box/sql.c @@ -915,7 +915,7 @@ cursor_seek(BtCursor *pCur, int *pRes) part_count); if (it == NULL) { if (txn != NULL) - txn_rollback_stmt(); + txn_rollback_stmt(txn); pCur->eState = CURSOR_INVALID; return SQL_TARANTOOL_ITERATOR_FAIL; } diff --git a/src/box/txn.c b/src/box/txn.c index 1d8271e51..21f34e526 100644 --- a/src/box/txn.c +++ b/src/box/txn.c @@ -174,7 +174,7 @@ txn_free(struct txn *txn) } struct txn * -txn_begin(bool is_autocommit) +txn_begin() { static int64_t tsn = 0; assert(! in_txn()); @@ -187,7 +187,6 @@ txn_begin(bool is_autocommit) txn->n_new_rows = 0; txn->n_local_rows = 0; txn->n_applier_rows = 0; - txn->is_autocommit = is_autocommit; txn->has_triggers = false; txn->is_aborted = false; txn->in_sub_stmt = 0; @@ -220,26 +219,20 @@ txn_begin_in_engine(struct engine *engine, struct txn *txn) return 0; } -struct txn * -txn_begin_stmt(struct space *space) +struct txn_stmt * +txn_begin_stmt(struct txn *txn, struct space *space) { - struct txn *txn = in_txn(); - if (txn == NULL) { - txn = txn_begin(true); - if (txn == NULL) - return NULL; - } else if (txn->in_sub_stmt > TXN_SUB_STMT_MAX) { + assert(txn == in_txn()); + assert(txn != NULL); + if (txn->in_sub_stmt > TXN_SUB_STMT_MAX) { diag_set(ClientError, ER_SUB_STMT_MAX); return NULL; } struct txn_stmt *stmt = txn_stmt_new(txn); - if (stmt == NULL) { - if (txn->is_autocommit && txn->in_sub_stmt == 0) - txn_rollback(); + if (stmt == NULL) return NULL; - } if (space == NULL) - return txn; + return stmt; if (trigger_run(&space->on_stmt_begin, txn) != 0) goto fail; @@ -252,9 +245,9 @@ txn_begin_stmt(struct space *space) if (engine_begin_statement(engine, txn) != 0) goto fail; - return txn; + return stmt; fail: - txn_rollback_stmt(); + txn_rollback_stmt(txn); return NULL; } @@ -272,8 +265,7 @@ txn_is_distributed(struct txn *txn) } /** - * End a statement. In autocommit mode, end - * the current transaction as well. + * End a statement. */ int txn_commit_stmt(struct txn *txn, struct request *request) @@ -316,14 +308,9 @@ txn_commit_stmt(struct txn *txn, struct request *request) goto fail; } --txn->in_sub_stmt; - if (txn->is_autocommit && txn->in_sub_stmt == 0) { - int rc = txn_commit(txn); - fiber_gc(); - return rc; - } return 0; fail: - txn_rollback_stmt(); + txn_rollback_stmt(txn); return -1; } @@ -359,7 +346,7 @@ txn_write_to_wal(struct txn *txn) if (res < 0) { /* Cascading rollback. */ - txn_rollback(); /* Perform our part of cascading rollback. */ + txn_rollback(txn); /* Perform our part of cascading rollback. */ /* * Move fiber to end of event loop to avoid * execution of any new requests before all @@ -441,14 +428,11 @@ fail: } void -txn_rollback_stmt() +txn_rollback_stmt(struct txn *txn) { - struct txn *txn = in_txn(); if (txn == NULL || txn->in_sub_stmt == 0) return; txn->in_sub_stmt--; - if (txn->is_autocommit && txn->in_sub_stmt == 0) - return txn_rollback(); txn_rollback_to_svp(txn, txn->sub_stmt_begin[txn->in_sub_stmt]); } @@ -519,7 +503,7 @@ box_txn_begin() diag_set(ClientError, ER_ACTIVE_TRANSACTION); return -1; } - if (txn_begin(false) == NULL) + if (txn_begin() == NULL) return -1; return 0; } diff --git a/src/box/txn.h b/src/box/txn.h index 196a7dcc1..ff2018d57 100644 --- a/src/box/txn.h +++ b/src/box/txn.h @@ -162,11 +162,6 @@ struct txn { * already assigned LSN. */ int n_applier_rows; - /** - * True if this transaction is running in autocommit mode - * (statement end causes an automatic transaction commit). - */ - bool is_autocommit; /** * True if the transaction was aborted so should be * rolled back at commit. @@ -214,7 +209,7 @@ in_txn() * @pre no transaction is active */ struct txn * -txn_begin(bool is_autocommit); +txn_begin(); /** * Commit a transaction. @@ -271,11 +266,10 @@ txn_on_rollback(struct txn *txn, struct trigger *trigger) } /** - * Start a new statement. If no current transaction, - * start a new transaction with autocommit = true. + * Start a new statement. */ -struct txn * -txn_begin_stmt(struct space *space); +struct txn_stmt * +txn_begin_stmt(struct txn *txn, struct space *space); int txn_begin_in_engine(struct engine *engine, struct txn *txn); @@ -334,7 +328,7 @@ txn_commit_stmt(struct txn *txn, struct request *request); * rolls back the entire transaction. */ void -txn_rollback_stmt(); +txn_rollback_stmt(struct txn *txn); /** * Raise an error if this is a multi-statement transaction: DDL diff --git a/src/box/vinyl.c b/src/box/vinyl.c index e37631064..c61f9477e 100644 --- a/src/box/vinyl.c +++ b/src/box/vinyl.c @@ -2414,10 +2414,8 @@ vinyl_engine_begin(struct engine *engine, struct txn *txn) txn->engine_tx = vy_tx_begin(env->xm); if (txn->engine_tx == NULL) return -1; - if (!txn->is_autocommit) { - trigger_create(&txn->fiber_on_stop, txn_on_stop, NULL, NULL); - trigger_add(&fiber()->on_stop, &txn->fiber_on_stop); - } + trigger_create(&txn->fiber_on_stop, txn_on_stop, NULL, NULL); + trigger_add(&fiber()->on_stop, &txn->fiber_on_stop); return 0; } @@ -2487,8 +2485,7 @@ vinyl_engine_commit(struct engine *engine, struct txn *txn) vy_regulator_check_dump_watermark(&env->regulator); txn->engine_tx = NULL; - if (!txn->is_autocommit) - trigger_clear(&txn->fiber_on_stop); + trigger_clear(&txn->fiber_on_stop); } static void @@ -2502,8 +2499,7 @@ vinyl_engine_rollback(struct engine *engine, struct txn *txn) vy_tx_rollback(tx); txn->engine_tx = NULL; - if (!txn->is_autocommit) - trigger_clear(&txn->fiber_on_stop); + trigger_clear(&txn->fiber_on_stop); } static int diff --git a/src/box/vy_scheduler.c b/src/box/vy_scheduler.c index fabb4bb48..9d5f18e32 100644 --- a/src/box/vy_scheduler.c +++ b/src/box/vy_scheduler.c @@ -840,14 +840,14 @@ vy_deferred_delete_process_one(struct space *deferred_delete_space, tuple_unref(delete); - struct txn *txn = txn_begin_stmt(deferred_delete_space); - if (txn == NULL) + struct txn *txn = in_txn(); + if (txn_begin_stmt(txn, deferred_delete_space) == NULL) return -1; struct tuple *unused; if (space_execute_dml(deferred_delete_space, txn, &request, &unused) != 0) { - txn_rollback_stmt(); + txn_rollback_stmt(txn); return -1; } return txn_commit_stmt(txn, &request); -- 2.21.0
next prev parent reply other threads:[~2019-06-09 20:44 UTC|newest] Thread overview: 42+ messages / expand[flat|nested] mbox.gz Atom feed top 2019-06-09 20:44 [tarantool-patches] [PATCH v3 00/14] Parallel applier Georgy Kirichenko 2019-06-09 20:44 ` [tarantool-patches] [PATCH v3 01/14] txn: Fire a trigger after a transaction finalization Georgy Kirichenko 2019-06-09 21:59 ` [tarantool-patches] " Konstantin Osipov 2019-06-11 11:42 ` [tarantool-patches] " Vladimir Davydov 2019-06-09 20:44 ` [tarantool-patches] [PATCH v3 02/14] ddl: synchronize privileges cache with actual data state Georgy Kirichenko 2019-06-11 13:13 ` Vladimir Davydov 2019-06-09 20:44 ` [tarantool-patches] [PATCH v3 03/14] txn: transaction memory allocation Georgy Kirichenko 2019-06-09 20:44 ` [tarantool-patches] [PATCH v3 04/14] ddl: place alter structures onto a txn memory region Georgy Kirichenko 2019-06-11 14:14 ` Vladimir Davydov 2019-06-09 20:44 ` Georgy Kirichenko [this message] 2019-06-13 14:11 ` [tarantool-patches] [PATCH v3 05/14] txn: get rid of autocommit from a txn structure Vladimir Davydov 2019-06-16 16:20 ` [tarantool-patches] " Konstantin Osipov 2019-06-16 16:14 ` Konstantin Osipov 2019-06-09 20:44 ` [tarantool-patches] [PATCH v3 06/14] txn: get rid of fiber_gc from txn_rollback Georgy Kirichenko 2019-06-13 14:12 ` Vladimir Davydov 2019-06-13 19:28 ` Георгий Кириченко 2019-06-14 9:21 ` Vladimir Davydov 2019-06-16 16:38 ` [tarantool-patches] " Konstantin Osipov 2019-06-09 20:44 ` [tarantool-patches] [PATCH v3 07/14] wal: remove fiber from a journal_entry structure Georgy Kirichenko 2019-06-13 14:17 ` Vladimir Davydov 2019-06-13 19:33 ` Георгий Кириченко 2019-06-14 8:05 ` Vladimir Davydov 2019-06-09 20:44 ` [tarantool-patches] [PATCH v3 08/14] wal: enable asyncronous wal writes Georgy Kirichenko 2019-06-13 14:21 ` Vladimir Davydov 2019-06-09 20:44 ` [tarantool-patches] [PATCH v3 09/14] wal: a dedicated wal scheduling fiber Georgy Kirichenko 2019-06-13 14:24 ` Vladimir Davydov 2019-06-13 19:36 ` Георгий Кириченко 2019-06-14 9:20 ` Vladimir Davydov 2019-06-09 20:44 ` [tarantool-patches] [PATCH v3 10/14] core: latch_unlock_external routine Georgy Kirichenko 2019-06-13 14:27 ` Vladimir Davydov 2019-06-13 19:38 ` Георгий Кириченко 2019-06-14 8:10 ` Vladimir Davydov 2019-06-14 9:18 ` Vladimir Davydov 2019-06-09 20:44 ` [tarantool-patches] [PATCH v3 11/14] txn: introduce asynchronous txn commit Georgy Kirichenko 2019-06-13 14:34 ` Vladimir Davydov 2019-06-13 19:45 ` Георгий Кириченко 2019-06-14 7:58 ` Vladimir Davydov 2019-06-09 20:44 ` [tarantool-patches] [PATCH v3 12/14] txn: handle fiber stop event at transaction level Georgy Kirichenko 2019-06-13 14:36 ` Vladimir Davydov 2019-06-09 20:44 ` [tarantool-patches] [PATCH v3 13/14] applier: apply transaction in parallel Georgy Kirichenko 2019-06-13 15:17 ` Vladimir Davydov 2019-06-09 20:44 ` [tarantool-patches] [PATCH v3 14/14] test: fix flaky test Georgy Kirichenko
Reply instructions: You may reply publicly to this message via plain-text email using any one of the following methods: * Save the following mbox file, import it into your mail client, and reply-to-all from there: mbox Avoid top-posting and favor interleaved quoting: https://en.wikipedia.org/wiki/Posting_style#Interleaved_style * Reply using the --to, --cc, and --in-reply-to switches of git-send-email(1): git send-email \ --in-reply-to=e6181c23019464896f03515ac7efdd63375d2d09.1560112747.git.georgy@tarantool.org \ --to=georgy@tarantool.org \ --cc=tarantool-patches@freelists.org \ --subject='Re: [tarantool-patches] [PATCH v3 05/14] txn: get rid of autocommit from a txn structure' \ /path/to/YOUR_REPLY https://kernel.org/pub/software/scm/git/docs/git-send-email.html * If your mail client supports setting the In-Reply-To header via mailto: links, try the mailto: link
This is a public inbox, see mirroring instructions for how to clone and mirror all data and code used for this inbox