From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from [87.239.111.99] (localhost [127.0.0.1]) by dev.tarantool.org (Postfix) with ESMTP id BC4F76FC86; Wed, 24 Mar 2021 15:25:55 +0300 (MSK) DKIM-Filter: OpenDKIM Filter v2.11.0 dev.tarantool.org BC4F76FC86 DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/simple; d=tarantool.org; s=dev; t=1616588755; bh=jm/qKjQ0f8pXBOQvWMSYsx02nN422sNkUlcxeM7Ig0E=; h=To:Date:In-Reply-To:References:Subject:List-Id:List-Unsubscribe: List-Archive:List-Post:List-Help:List-Subscribe:From:Reply-To:Cc: From; b=eNUsaoBOI2qaCgKV90ssTBHUO/WaKsVE6jvZYOboNIxNnvzLi40KcszCKTjpWRjAH eVvWWADxlR4nkAHNy+6EGsOhP4RfeWECsxL/bX49XwHzdTZMZ7E8Whct0MVOuf2SFt D4HaU3PqsVJu4SJUGcNWdg4kwe72HRw5pwWZD1IQ= Received: from smtp36.i.mail.ru (smtp36.i.mail.ru [94.100.177.96]) (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 AFFE26BD34 for ; Wed, 24 Mar 2021 15:24:23 +0300 (MSK) DKIM-Filter: OpenDKIM Filter v2.11.0 dev.tarantool.org AFFE26BD34 Received: by smtp36.i.mail.ru with esmtpa (envelope-from ) id 1lP2Yg-0004oK-Ti; Wed, 24 Mar 2021 15:24:23 +0300 To: v.shpilevoy@tarantool.org, gorcunov@gmail.com Date: Wed, 24 Mar 2021 15:24:13 +0300 Message-Id: X-Mailer: git-send-email 2.24.3 (Apple Git-128) In-Reply-To: References: MIME-Version: 1.0 Content-Transfer-Encoding: 8bit X-7564579A: 646B95376F6C166E X-77F55803: 4F1203BC0FB41BD95D6E7CC48CB1F5F1DDD90A25A8FA528D0BFD61B598B81272182A05F5380850408821BF0C7740566522C68B6BDCBA90D8AB4F63C86FCD234C96875625773F2424 X-7FA49CB5: FF5795518A3D127A4AD6D5ED66289B5278DA827A17800CE7B7733D0215A2F71AEA1F7E6F0F101C67BD4B6F7A4D31EC0BCC500DACC3FED6E28638F802B75D45FF8AA50765F7900637F63E14183F8C6AF98638F802B75D45FF914D58D5BE9E6BC131B5C99E7648C95C5DD32608FC869F5DD641A689F0811F826E651694AB830EBFA471835C12D1D9774AD6D5ED66289B5259CC434672EE6371117882F4460429724CE54428C33FAD30A8DF7F3B2552694AC26CFBAC0749D213D2E47CDBA5A9658359CC434672EE6371117882F4460429728AD0CFFFB425014E868A13BD56FB6657D81D268191BDAD3DC09775C1D3CA48CF85267523C8841599BA3038C0950A5D36C8A9BA7A39EFB766EC990983EF5C0329BA3038C0950A5D36D5E8D9A59859A8B64AF3B6C4CE722FA576E601842F6C81A1F004C906525384307823802FF610243DF43C7A68FF6260569E8FC8737B5C2249EC8D19AE6D49635B3BBE47FD9DD3FB59A8DF7F3B2552694A2BEBFE083D3B9BA73A03B725D353964BB11811A4A51E3B096D1867E19FE14079BA9C0B312567BB23089D37D7C0E48F6CA18204E546F3947CC0F9454058DFE53C262FEC7FBD7D1F5BC8A9BA7A39EFB7666BA297DBC24807EA089D37D7C0E48F6C8AA50765F79006377AD8C0A31FFD7ED5EFF80C71ABB335746BA297DBC24807EA27F269C8F02392CD20465B3A5AADEC6827F269C8F02392CD5571747095F342E88FB05168BE4CE3AF X-B7AD71C0: AC4F5C86D027EB782CDD5689AFBDA7A2BBE337FB72E92315FF39D8DB89857825EFA8BF88FCBFD63CE0852D54D1EC5181C00AE116A61B0410B4BB7AE275D8542B X-C1DE0DAB: C20DE7B7AB408E4181F030C43753B8183A4AFAF3EA6BDC44C234C8B12C006B7A5F433A0BE06BC4658F124B06F154E80B8160CF060A956096B1881A6453793CE9C32612AADDFBE061C61BE10805914D3804EBA3D8E7E5B87ABF8C51168CD8EBDB63AF70AF8205D7DCDC48ACC2A39D04F89CDFB48F4795C241BDAD6C7F3747799A X-C8649E89: 4E36BF7865823D7055A7F0CF078B5EC49A30900B95165D34A503FBFE8BE8FC49CAD5E47C6F7254406A44857E58A6F6AA0E725EA6F4B575A0B467E9083D168C881D7E09C32AA3244CAC23010B5B6BC7A9B7149B3868A05E8D39C99C45E8D137E9927AC6DF5659F194 X-D57D3AED: 3ZO7eAau8CL7WIMRKs4sN3D3tLDjz0dLbV79QFUyzQ2Ujvy7cMT6pYYqY16iZVKkSc3dCLJ7zSJH7+u4VD18S7Vl4ZUrpaVfd2+vE6kuoey4m4VkSEu530nj6fImhcD4MUrOEAnl0W826KZ9Q+tr5ycPtXkTV4k65bRjmOUUP8cvGozZ33TWg5HZplvhhXbhDGzqmQDTd6OAevLeAnq3Ra9uf7zvY2zzsIhlcp/Y7m53TZgf2aB4JOg4gkr2biojbL9S8ysBdXjhXhGERY7yL8MNHt76gtwu X-Mailru-Sender: 583F1D7ACE8F49BDD2846D59FC20E9F86C523D0B3819F67ECAB92D41A1FB43BCB2586E328ED17518424AE0EB1F3D1D21E2978F233C3FAE6EE63DB1732555E4A8EE80603BA4A5B0BC112434F685709FCF0DA7A0AF5A3A8387 X-Mras: Ok Subject: [Tarantool-patches] [PATCH v2 3/7] applier: extract plain tx application from applier_apply_tx() X-BeenThere: tarantool-patches@dev.tarantool.org X-Mailman-Version: 2.1.34 Precedence: list List-Id: Tarantool development patches List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , From: Serge Petrenko via Tarantool-patches Reply-To: Serge Petrenko Cc: tarantool-patches@dev.tarantool.org Errors-To: tarantool-patches-bounces@dev.tarantool.org Sender: "Tarantool-patches" The new routine, called apply_plain_tx(), may be used not only by applier_apply_tx(), but also by final join, once we make it transactional, and recovery, once it's also turned transactional. Also, while we're at it. Remove excess fiber_gc() call from applier_subscribe loop. Let's better make sure fiber_gc() is called on any return from applier_apply_tx(). Prerequisite #5874 Part of #5566 --- src/box/applier.cc | 188 ++++++++++++++++++++++----------------------- 1 file changed, 93 insertions(+), 95 deletions(-) diff --git a/src/box/applier.cc b/src/box/applier.cc index 65afa5e98..07e557a51 100644 --- a/src/box/applier.cc +++ b/src/box/applier.cc @@ -905,6 +905,90 @@ applier_handle_raft(struct applier *applier, struct xrow_header *row) return box_raft_process(&req, applier->instance_id); } +static inline int +apply_plain_tx(struct stailq *rows, bool skip_conflict, bool use_triggers) +{ + /** + * Explicitly begin the transaction so that we can + * control fiber->gc life cycle and, in case of apply + * conflict safely access failed xrow object and allocate + * IPROTO_NOP on gc. + */ + struct txn *txn = txn_begin(); + struct applier_tx_row *item; + if (txn == NULL) + return -1; + + stailq_foreach_entry(item, rows, next) { + struct xrow_header *row = &item->row; + int res = apply_row(row); + if (res != 0 && skip_conflict) { + struct error *e = diag_last_error(diag_get()); + /* + * In case of ER_TUPLE_FOUND error and enabled + * replication_skip_conflict configuration + * option, skip applying the foreign row and + * replace it with NOP in the local write ahead + * log. + */ + if (e->type == &type_ClientError && + box_error_code(e) == ER_TUPLE_FOUND && + replication_skip_conflict) { + diag_clear(diag_get()); + row->type = IPROTO_NOP; + row->bodycnt = 0; + res = apply_row(row); + } + } + if (res != 0) + goto fail; + } + + /* + * We are going to commit so it's a high time to check if + * the current transaction has non-local effects. + */ + if (txn_is_distributed(txn)) { + /* + * A transaction mixes remote and local rows. + * Local rows must be replicated back, which + * doesn't make sense since the master likely has + * new changes which local rows may overwrite. + * Raise an error. + */ + diag_set(ClientError, ER_UNSUPPORTED, "Replication", + "distributed transactions"); + goto fail; + } + + if (use_triggers) { + /* We are ready to submit txn to wal. */ + struct trigger *on_rollback, *on_wal_write; + size_t size; + on_rollback = region_alloc_object(&txn->region, typeof(*on_rollback), + &size); + on_wal_write = region_alloc_object(&txn->region, typeof(*on_wal_write), + &size); + if (on_rollback == NULL || on_wal_write == NULL) { + diag_set(OutOfMemory, size, "region_alloc_object", + "on_rollback/on_wal_write"); + goto fail; + } + + trigger_create(on_rollback, applier_txn_rollback_cb, NULL, NULL); + txn_on_rollback(txn, on_rollback); + + trigger_create(on_wal_write, applier_txn_wal_write_cb, NULL, NULL); + txn_on_wal_write(txn, on_wal_write); + } + + return txn_commit_try_async(txn); +fail: + txn_rollback(txn); + return -1; +} + + /** * Apply all rows in the rows queue as a single transaction. * @@ -931,6 +1015,7 @@ applier_apply_tx(struct applier *applier, struct stailq *rows) struct xrow_header *last_row; last_row = &stailq_last_entry(rows, struct applier_tx_row, next)->row; struct replica *replica = replica_by_id(first_row->replica_id); + int rc = 0; /* * In a full mesh topology, the same set of changes * may arrive via two concurrently running appliers. @@ -942,8 +1027,7 @@ applier_apply_tx(struct applier *applier, struct stailq *rows) latch_lock(latch); if (vclock_get(&replicaset.applier.vclock, last_row->replica_id) >= last_row->lsn) { - latch_unlock(latch); - return 0; + goto no_write; } else if (vclock_get(&replicaset.applier.vclock, first_row->replica_id) >= first_row->lsn) { /* @@ -974,103 +1058,18 @@ applier_apply_tx(struct applier *applier, struct stailq *rows) assert(first_row == last_row); if (apply_synchro_row(first_row) != 0) diag_raise(); - goto success; - } - - /** - * Explicitly begin the transaction so that we can - * control fiber->gc life cycle and, in case of apply - * conflict safely access failed xrow object and allocate - * IPROTO_NOP on gc. - */ - struct txn *txn; - txn = txn_begin(); - struct applier_tx_row *item; - if (txn == NULL) { - latch_unlock(latch); - return -1; - } - stailq_foreach_entry(item, rows, next) { - struct xrow_header *row = &item->row; - int res = apply_row(row); - if (res != 0) { - struct error *e = diag_last_error(diag_get()); - /* - * In case of ER_TUPLE_FOUND error and enabled - * replication_skip_conflict configuration - * option, skip applying the foreign row and - * replace it with NOP in the local write ahead - * log. - */ - if (e->type == &type_ClientError && - box_error_code(e) == ER_TUPLE_FOUND && - replication_skip_conflict) { - diag_clear(diag_get()); - row->type = IPROTO_NOP; - row->bodycnt = 0; - res = apply_row(row); - } - } - if (res != 0) - goto rollback; - } - /* - * We are going to commit so it's a high time to check if - * the current transaction has non-local effects. - */ - if (txn_is_distributed(txn)) { - /* - * A transaction mixes remote and local rows. - * Local rows must be replicated back, which - * doesn't make sense since the master likely has - * new changes which local rows may overwrite. - * Raise an error. - */ - diag_set(ClientError, ER_UNSUPPORTED, - "Replication", "distributed transactions"); - goto rollback; + goto written; } - /* We are ready to submit txn to wal. */ - struct trigger *on_rollback, *on_wal_write; - size_t size; - on_rollback = region_alloc_object(&txn->region, typeof(*on_rollback), - &size); - on_wal_write = region_alloc_object(&txn->region, typeof(*on_wal_write), - &size); - if (on_rollback == NULL || on_wal_write == NULL) { - diag_set(OutOfMemory, size, "region_alloc_object", - "on_rollback/on_wal_write"); - goto rollback; + if ((rc = apply_plain_tx(rows, true, true)) == 0) { +written: + vclock_follow(&replicaset.applier.vclock, last_row->replica_id, + last_row->lsn); } - - trigger_create(on_rollback, applier_txn_rollback_cb, NULL, NULL); - txn_on_rollback(txn, on_rollback); - - trigger_create(on_wal_write, applier_txn_wal_write_cb, NULL, NULL); - txn_on_wal_write(txn, on_wal_write); - - if (txn_commit_try_async(txn) < 0) - goto fail; - -success: - /* - * The transaction was sent to journal so promote vclock. - * - * Use the lsn of the last row to guard from 1.10 - * instances, which send every single tx row as a separate - * transaction. - */ - vclock_follow(&replicaset.applier.vclock, last_row->replica_id, - last_row->lsn); - latch_unlock(latch); - return 0; -rollback: - txn_rollback(txn); -fail: +no_write: latch_unlock(latch); fiber_gc(); - return -1; + return rc; } /** @@ -1280,7 +1279,6 @@ applier_subscribe(struct applier *applier) if (ibuf_used(ibuf) == 0) ibuf_reset(ibuf); - fiber_gc(); } } -- 2.24.3 (Apple Git-128)