# Синхронная репликация ## Введение Транзакции делятся на **локальные**, то есть инициированны на самом узле, и **удаленные** (реплицированные) - то есть полученные с другой ноды. Они могут быть как **синхронными**, так и **асинхронными**. Транзакция состоит из операций над спейсом. Атрибут "синхронный спейс" определяет, синхронной ли будет транзакция или нет. Более того, транзакция может оперировать над синхронными и асинхронными спейсами одновременно. Такая транзакция называется _композитной_. С точки зрения работы движка, это частный случай синхронной транзакции. Асинхронные транзакции просто пишутся в WAL-журнал по мере их поступления. Синхронные же транзакции отслеживаются через очередь `txn_limbo`, в которую добавляются транзакции через оберточную структуру `txn_limbo_entry`. ```language:c struct txn_limbo_entry { struct rlist in_queue; // <-+ entry struct txn *txn; // | int64_t lsn; // | int ack_count; // | bool is_commit; // | bool is_rollback; // | }; // | // | struct txn_limbo { // | struct rlist queue; // <-+ queue uint32_t owner_id; struct vclock vclock; struct vclock promote_term_map; uint64_t promote_greatest_term; int64_t confirmed_lsn; }; ``` Одновременно только **один** узел в кластере должен быть владельцем очереди, что отслеживается через `txn_limbo::owner_id`. Все элементы очереди упорядоченны по возрастанию номера `txn_libo_entry::lsn`. Если транзакция локальная, то LSN изначально выставляется в `-1`, а затем присваивается WAL-движком, перед записью транзакции в журнал. Если же транзакция удаленная, то она приходит вместе с уже назначенным LSN (поскольку отсылается она из WAL-журнала реплики). Cуществуют следующие виды пакетов для манипуляцией синхронной репликацией: 1) `PROMOTE` и `DEMOTE` - для управления владельцем очереди 2) `CONFIRM` и `ROLLBACK` - для управления синхронной транзакцией ## Управление очередью синхронных транзакций Чтобы использовать синхронную репликацию, помимо самого синхронного спейса, администратор должен сделать узел владельцем очереди. Для этого существует команда `box.ctl.promote()`. Для простоты можно считать, что она вызывается на пустом узле, то есть данные еще не писались. Во время вызова `box.ctl.promote()` записывается следующий пакет ``` language:c struct synchro_request req = { .type = IPROTO_RAFT_PROMOTE, .replica_id = limbo->owner_id, .lsn = lsn, .term = term, }; ``` Поскольку очередь была пуста, то `limbo->owner_id = 0` и `lsn = 0`. Именно эти данные осядут в журнале. Затем отработает ``` lanugage:c static void txn_limbo_read_promote(struct txn_limbo *limbo, uint32_t replica_id, int64_t lsn) { txn_limbo_read_confirm(limbo, lsn); txn_limbo_read_rollback(limbo, lsn + 1); limbo->owner_id = replica_id; box_update_ro_summary(); limbo->confirmed_lsn = 0; } ``` В результате чего владельцем очереди станет сам узел `limbo->owner_id = replica_id`. Например ```language:bash s = box.schema.space.create("sync", {is_sync = true}) s:create_index('primary', { type = 'hash' }) box.ctl.promote() --- ... tarantool> s:insert{3} --- - [3] ... ``` Чтобы отдать владение очередью какому-то другому узлу, надо вызвать команду `box.ctl.demote()`. Она действует аналогично `PROMOTE`, за исключением того, что `limbo::owner_id` выставляется в `0`. Как только владение очередью сбрасывается, любая попытка локальной записи синхронной репликации будет выдавать ошибку, покуда узел не вернет себе статус владельца очереди. ``` language:bash tarantool> s:insert{2} --- - error: The synchronous transaction queue doesn't belong to any instance ... ``` Обе команды (`promote` и `demote`) вычищают очередь по окончанию своей работы, выставляя `confirmed_lsn` в `0`. ## Путешествие синхронных транзакций Основная идея синхронной репликации - записать данные только после того, как они были успешно отреплецированы на `n` узлах. Фактически, инициатор транзакции должен выслать данные репликам, дождаться от них подтверждения, и только потом считать транзакцию завершенной (commit). ### Протокол Qsync Протокол синхронной репликации базируется на двух специальных записях: `CONFIRM` и `ROLLBACK`. Запись `CONFIRM` подразумевает, что транзакция была успешно отреплицирована, в то время как `ROLLBACK` указывает, что транзакция не смогла отреплицироваться и не должна приниматься во внимание. Поскольку в Tarantool журнал только дописывается, то именно `ROLLBACK` запись обеспечивает откат неудачной транзакции (после ее исполнения транзакция выкидывается из памяти, хотя и остается в журнале до следующей операции сборки мусора, это отдельный механизм, который мы не рассматриваем). ### Локальные транзакции Под локальными транзакциями мы будем понимать те, которые инициируются узлом. Такой узел должен быть владельцем очереди, т.е. предварительно вызван `box.ctl.promote()`. Как только транзакция создана и аллоцирована, она добавляется в конец очереди. В этот момент ее `lsn = -1`. Тут же вызывается `journal_write`, который записывает данную транзакцию в локальный WAL, причем назначает записи `lsn`. Как указывалось выше, синхронные транзакции сами по себе бесмысленны, поэтому они должны быть отреплицированы на `n` узлов, причем сам узел инициатор транзакции тоже является частью кластера. Таким образом, запись в представляет собой последовательность ``` language:c txn_commit(struct txn *txn) { limbo_entry = txn_limbo_append(&txn_limbo, origin_id, txn); journal_write(req); lsn = req->rows[req->n_rows - 1]->lsn; txn_limbo_assign_local_lsn(&txn_limbo, limbo_entry, lsn); txn_limbo_ack(&txn_limbo, txn_limbo.owner_id, lsn); txn_limbo_wait_complete(&txn_limbo, limbo_entry) } ``` После записи в WAL номер `LSN` копируется в `txn_limbo_entry::lsn`, тут же инкрементится `txn_limbo_entry::ack_count` и начинается ожидание, пока оставшиеся `n-1` узлы в кластере подтвердят репликацию. Как только узлы подтвердят репликацию, такой элемент удаляется из очереди. Подтверждение высылается `relay` потоком, которому реплика отсылает свой последний записанный `LSN`. ``` language:c void txn_limbo_ack(struct txn_limbo *limbo, uint32_t replica_id, int64_t lsn) { if (rlist_empty(&limbo->queue)) return; struct txn_limbo_entry *e; int64_t confirm_lsn = -1; int64_t prev_lsn = vclock_get(&limbo->vclock, replica_id); rlist_foreach_entry(e, &limbo->queue, in_queue) { if (e->lsn > lsn) break; /* * Sync transactions need to collect acks. Async * transactions are automatically committed right * after all the previous sync transactions are. */ if (!txn_has_flag(e->txn, TXN_WAIT_ACK)) { continue; } else if (e->lsn <= prev_lsn) { continue; } else if (++e->ack_count < replication_synchro_quorum) { continue; } else { confirm_lsn = e->lsn; } } if (confirm_lsn == -1 || confirm_lsn <= limbo->confirmed_lsn) return; txn_limbo_write_confirm(limbo, confirm_lsn); txn_limbo_read_confirm(limbo, confirm_lsn); } ```