Транзакции MongoDB против двухфазной фиксации

В этой статье мы собираемся исследовать разницу в производительности между использованием MongoDB Transactions или Two-phase совершить простой перевод на банковский счет. Тесты проводились на локальном рабочем столе и с ограниченным набором учетных записей, чтобы убедиться, что у нас могут возникать периодические конфликты записи.

Ссылки

Узнать больше о MongoDB в следующих местах.

Код

Мы собираемся попытаться перевести сумму денег между двумя аккаунтами и откатиться, если это не удастся. Есть две реализации. Первый основан на использовании two-phase совершать без использования multi-document транзакции. Вторая реализация использует новый multi-document поддержка транзакций в MongoDB 4.0.x или выше. Для простоты мы представляем код, используя mongo shell синтаксис.

Для фактического бенчмаркинга мы использовали платформу kotlin/java, которая находится в разработке, которую можно найти в /mongodb-schema-simulator.

Мультидокументная транзакция

Для multi-document транзакционный подход мы повторяем транзакции в случае сбоя, чтобы убедиться, что они прошли. Ниже приведен пример кода того, как это можно сделать в mongo shell.

var db = db.getSisterDB("bank");
var session = db.getMongo().startSession();
var accounts = session.getDatabase("bank").accounts;
var transactions = session.getDatabase("bank").transactions;

// Retries a transaction commit
function retryUnknownTransactionCommit(session) {
  while(true) {
    try {
      // Attempt to commit the transaction
      session.commitTransaction();
      break;
    } catch (err) {
      if (err.errorLabels != null 
        && err.errorLabels.includes("UnknownTransactionCommitResult")) {
          // Keep retrying the transaction
          continue;
        }

      // The transaction cannot be retried, 
      // return the exception
      return err;
    }
  }
}

function executeTransaction(session, from, to, amount) {
  while (true) {
    try {
      // Start a transaction on the current session
      session.startTransaction({ 
        readConcern: { level: "snapshot" }, writeConcern: { w: "local" } 
      });

      // Debit the `from` account
      var result = accounts.updateOne(
        { name: from, amount: { $gte: amount } }, 
        { $inc: { amount: -amount } });

      // If we could not debit the account, abort the
      // transaction and throw an exception
      if (result.modifiedCount == 0) {
        session.abortTransaction();
        throw Error("failed to debit the account [" + from + "]");
      }

      // Credit the `from` account
      result = accounts.updateOne(
        { name: to }, 
        { $inc: { amount: amount } });

      // If we could not credit the account, abort the
      // transaction and throw an exception
      if (result.modifiedCount == 0) {
        session.abortTransaction();
        throw Error("failed to credit the account [" + to + "]");
      }

      // Insert a record of the transaction
      transactions.insertOne(
        { from: from, to: to, amount: amount, on: new Date() });

      // Attempt to commit the transaction
      session.commitTransaction();
      // Transaction was committed successfully break the while loop
      break;
    } catch (err) {
      // If we have no error labels rethrow the error
      if (err.errorLabels == null) {
        throw err;
      }

      // Our error contains UnknownTransactionCommitResult label
      if (err.errorLabels.includes("UnknownTransactionCommitResult")) {
        // Retry the transaction commit
        var exception = retryUnknownTransactionCommit(session, err);
        // No error, commit as successful, break the while loop
        if (exception == null) break;
        // Error has no errorLabels, rethrow the error
        if (exception.errorLabels == null) throw exception;

        // Error labels include TransientTransactionError label
        // Start while loop again, creating a new transaction
        if (err.errorLabels.includes("TransientTransactionError")) {
          continue;
        }
        
        // Rethrow the error
        throw exception;
      }

      // Error labels include TransientTransactionError label
      // Start while loop again, creating a new transaction
      if (err.errorLabels.includes("TransientTransactionError")) {
        continue;
      }

      // Rethrow the error
      throw err;
    }
  }
}  

executeTransaction(session, "Peter", "Joe", 100);

Давайте разберем код на грубые шаги. Сначала давайте посмотрим на executeTransaction метод.

function executeTransaction(session, from, to, amount) {
  while (true) {
    try {
      // Start a transaction on the current session
      session.startTransaction({ 
        readConcern: { level: "snapshot" }, writeConcern: { w: "local" } 
      });

      // Debit the `from` account
      var result = accounts.updateOne(
        { name: from, amount: { $gte: amount } }, 
        { $inc: { amount: -amount } });

      // If we could not debit the account, abort the
      // transaction and throw an exception
      if (result.modifiedCount == 0) {
        session.abortTransaction();
        throw Error("failed to debit the account [" + from + "]");
      }

      // Credit the `from` account
      result = accounts.updateOne(
        { name: to }, 
        { $inc: { amount: amount } });

      // If we could not credit the account, abort the
      // transaction and throw an exception
      if (result.modifiedCount == 0) {
        session.abortTransaction();
        throw Error("failed to credit the account [" + to + "]");
      }

      // Insert a record of the transaction
      transactions.insertOne(
        { from: from, to: to, amount: amount, on: new Date() });

      // Attempt to commit the transaction
      session.commitTransaction();
      // Transaction was committed successfully break the while loop
      break;
    } catch (err) {
      // If we have no error labels rethrow the error
      if (err.errorLabels == null) {
        throw err;
      }

      // Error labels include TransientTransactionError label
      // Start while loop again, creating a new transaction
      if (err.errorLabels.includes("TransientTransactionError")) {
        continue;
      }

      // Our error contains the UnknownTransactionCommitResult label
      if (err.errorLabels.includes("UnknownTransactionCommitResult")) {
        // Retry the transaction commit
        var exception = retryUnknownTransactionCommit(session);
        // No error, commit as successful, break the while loop
        if (exception == null) break;
        // Error has no errorLabels, rethrow the error
        if (exception.errorLabels == null) throw exception;

        // Error labels include TransientTransactionError label
        // Start while loop again, creating a new transaction
        if (err.errorLabels.includes("TransientTransactionError")) {
          continue;
        }
        
        // Rethrow the error
        throw exception;
      }

      // Rethrow the error
      throw err;
    }
  }
}  

Сначала мы создаем новый transaction а затем применяем операции по переводу денег с одного счета на другой. Первое заявление debits в from учетная запись.

var result = accounts.updateOne(
  { name: from, amount: { $gte: amount } }, 
  { $inc: { amount: -amount } });
if (result.modifiedCount == 0) {
  session.abortTransaction();
  throw Error("failed to debit the account [" + from + "]");
}

Если дебетование не удается, мы прерываем транзакцию и выдаем ошибку, сигнализирующую о том, что debit операция не удалась. Далее мы credit в to учетная запись.

result = accounts.updateOne(
  { name: to }, 
  { $inc: { amount: amount } });
if (result.modifiedCount == 0) {
  session.abortTransaction();
  throw Error("failed to credit the account [" + to + "]");
}

Если credit терпит неудачу, мы прерываем транзакцию и выдаем ошибку, чтобы сигнализировать credit операция не удалась. Наконец мы record перевод.

transactions.insertOne(
    { from: from, to: to, amount: amount, on: new Date() });

После того, как мы настроили все операции, мы пытаемся зафиксировать transaction.

session.commitTransaction();

Если транзакция не проходит, начинается самое интересное. Давайте посмотрим на exception умение обращаться.

} catch (err) {
  // If we have no error labels rethrow the error
  if (err.errorLabels == null) {
    throw err;
  }

  // Error labels include TransientTransactionError label
  // Start while loop again, creating a new transaction
  if (err.errorLabels.includes("TransientTransactionError")) {
    continue;
  }

  // Our error contains the UnknownTransactionCommitResult label
  if (err.errorLabels.includes("UnknownTransactionCommitResult")) {
    // Retry the transaction commit
    var exception = retryUnknownTransactionCommit(session);
    // No error, commit as successful, break the while loop
    if (exception == null) break;
    // Error has no errorLabels, rethrow the error
    if (exception.errorLabels == null) throw exception;

    // Error labels include TransientTransactionError label
    // Start while loop again, creating a new transaction
    if (err.errorLabels.includes("TransientTransactionError")) {
      continue;
    }
    
    // Rethrow the error
    throw exception;
  }

  // Rethrow the error
  throw err;
}

Если у нас есть объект ошибки без errorLabels мы перебрасываем его, поскольку транзакцию нельзя повторить. Однако если у нас есть errorLabels нам нужно их осмотреть.

Если этикетка TransientTransactionError присутствует, мы не можем повторить текущую транзакцию, поэтому мы continue цикл while, заставляющий создавать новую транзакцию.

Однако если errorLabels содержит метку UnknownTransactionCommitResult мы можем повторить текущую транзакцию. Мы делаем это, звоня в retryUnknownTransactionCommit работать с текущим сеансом. Рассмотрим функцию подробнее.

// Retries a transaction commit
function retryUnknownTransactionCommit(session) {
  while(true) {
    try {
      // Attempt to commit the transaction
      session.commitTransaction();
      break;
    } catch (err) {
      if (err.errorLabels != null 
        && err.errorLabels.includes("UnknownTransactionCommitResult")) {
          // Keep retrying the transaction
          continue;
        }

      // The transaction cannot be retried, 
      // return the exception
      return err;
    }
  }
}

В то время как session.commitTransaction() вызов возвращает UnknownTransactionCommitResult мы продолжаем повторять транзакцию. Если фиксация транзакции прошла успешно, мы выходим из цикла, возвращая null. Если возвращаемая ошибка отличается от UnknownTransactionCommitResult мы возвращаем ошибку.

Возвращаясь к тому моменту, когда мы называем retryUnknownTransactionCommit функции мы видим следующую логику.

// Retry the transaction commit
var exception = retryUnknownTransactionCommit(session);
// No error, commit as successful, break the while loop
if (exception == null) break;
// Error has no errorLabels, rethrow the error
if (exception.errorLabels == null) throw exception;

// Error labels include TransientTransactionError label
// Start while loop again, creating a new transaction
if (err.errorLabels.includes("TransientTransactionError")) {
  continue;
}

// Rethrow the error
throw exception;

Если возвращенный exception является null мы ломаем while цикл, так как наша транзакция была успешно зафиксирована. Если exception не включает errorLabels мы повторно выбрасываем исключение. С другой стороны, если исключение содержит errorLabels и ярлыки включают ярлык TransientTransactionError мы не можем повторить текущую транзакцию, поэтому мы continue цикл while, заставляющий создавать новую транзакцию.

Двухэтапная фиксация

Two Phase Подход commit использует другой подход, использующий двойную бухгалтерию для обеспечения согласованных переводов учетных записей. Давайте посмотрим на автономный пример ниже, который демонстрирует, как шаблон может быть реализован с использованием mongo shell.

var db = db.getSisterDB("bank");
db.dropDatabase();
var accounts = db.accounts;
var transactions = db.transactions;

accounts.insertOne({ _id: 1, name: "Joe Moneylender", balance: 1000, pendingTransactions:[] });
accounts.insertOne({ _id: 2, name: "Peter Bum", balance: 1000, pendingTransactions:[] });

function cancel(id) {
  transactions.updateOne(
    { _id: id }, 
    { $set: { state: "canceled" } }
  );
}

function rollback(from, to, amount, id) {
  // Reverse debit
  accounts.updateOne({
      name: from, 
      pendingTransactions: { $in: [id] }
    }, {
      $inc: { balance: amount }, 
      $pull: { pendingTransactions: id }
    });  

  // Reverse credit
  accounts.updateOne({
    name: to, 
    pendingTransactions: { $in: [id] }
  }, {
    $inc: { balance: -amount }, 
    $pull: { pendingTransactions: id }
  });  

  cancel(id);
}

function cleanup(from, to, id) {
  // Remove the transaction ids
  accounts.updateOne(
    { name: from }, 
    { $pull: { pendingTransactions: id } });
  
  // Remove the transaction ids
  accounts.updateOne(
    { name: to }, 
    { $pull: { pendingTransactions: id } });

  // Update transaction to committed
  transactions.updateOne(
    { _id: id }, 
    { $set: { state: "done" } });
}

function executeTransaction(from, to, amount) {
  var transactionId = ObjectId();

  transactions.insert({
    _id: transactionId, 
    source: from, 
    destination: to, 
    amount: amount, 
    state: "initial"
  });

  var result = transactions.updateOne(
    { _id: transactionId }, 
    { $set: { state: "pending" } }
  );

  if (result.modifiedCount == 0) {
    cancel(transactionId);
    throw Error("Failed to move transaction " + transactionId + " to pending");
  }

  // Set up pending debit
  result = accounts.updateOne({
    name: from, 
    pendingTransactions: { $ne: transactionId }, 
    balance: { $gte: amount }
  }, {
    $inc: { balance: -amount }, 
    $push: { pendingTransactions: transactionId }
  });

  if (result.modifiedCount == 0) {
    rollback(from, to, amount, transactionId);
    throw Error("Failed to debit " + from + " account");
  }

  // Setup pending credit
  result = accounts.updateOne({
    name: to, 
    pendingTransactions: { $ne: transactionId }
  }, {
    $inc: { balance: amount }, 
    $push: { pendingTransactions: transactionId }
  });

  if (result.modifiedCount == 0) {
    rollback(from, to, amount, transactionId);
    throw Error("Failed to credit " + to + " account");
  }

  // Update transaction to committed
  result = transactions.updateOne(
    { _id: transactionId }, 
    { $set: { state: "committed" } }
  );

  if (result.modifiedCount == 0) {
    rollback(from, to, amount, transactionId);
    throw Error("Failed to move transaction " + transactionId + " to committed");
  }

  // Attempt cleanup
  cleanup(from, to, transactionId);
}

executeTransaction("Joe Moneylender", "Peter Bum", 100);

Разберем функцию executeTransaction шаг за шагом и обсудите, что происходит на каждом шаге и как исправить ошибку.

transactions.insert({
  _id: transactionId, 
  source: from, 
  destination: to, 
  amount: amount, 
  state: "initial"
});

Первый шаг вставляет новый документ в transactions коллекция, содержащая информацию о переводе, который мы собираемся выполнить. Состояние transaction установлен на initial сигнализируя, что мы только что начали процесс.

var result = transactions.updateOne(
  { _id: transactionId }, 
  { $set: { state: "pending" } }
);

if (result.modifiedCount == 0) {
  cancel();
  throw Error("Failed to move transaction " + transactionId + " to pending");
}

Затем мы пытаемся перевернуть транзакцию в состояние pending. Если не получится(result.modifiedCount == 0) мы пытаемся отменить транзакцию, вызвав функцию cancel. Давайте посмотрим, что cancel функция делает.

function cancel(id) {
  transactions.updateOne(
    { _id: id }, 
    { $set: { state: "canceled" } }
  );
}

Функция в основном пытается установить state сделки в canceled. После возвращения из cancel мы выбрасываем исключение, сигнализирующее вызывающей стороне executeTransaction функция, которая не удалась.

Однако, если нам удастся установить transaction состояние pending мы можем начать процесс применения transaction.

result = accounts.updateOne({
  name: from, 
  pendingTransactions: { $ne: transactionId }, 
  balance: { $gte: amount }
}, {
  $inc: { balance: -amount }, 
  $push: { pendingTransactions: transactionId }
});

if (result.modifiedCount == 0) {
  rollback(from, to, amount, transactionId);
  throw Error("Failed to debit " + from + " account");
}

Мы ищем from счет, гарантируя, что pendingTransactions массив не содержит transactionId и что счет balance является greater or equal к сумме, которую мы собираемся списать. Если документ совпадает, мы собираемся debit счет balance по amount и нажмите transactionId к pendingTransactions множество.

Если ни один документ не был изменен, мы знаем update учетной записи не удалось, и нам нужно позвонить rollback функция для отмены транзакции, прежде чем мы создадим исключение, сигнализирующее приложению, что передача не удалась. Давайте посмотрим на rollback функция.

function rollback(from, to, amount, id) {
  // Reverse debit
  accounts.updateOne({
      name: from, 
      pendingTransactions: { $in: [id] }
    }, {
      $inc: { balance: amount }, 
      $pull: { pendingTransactions: id }
    });  

  // Reverse credit
  accounts.updateOne({
    name: to, 
    pendingTransactions: { $in: [id] }
  }, {
    $inc: { balance: -amount }, 
    $pull: { pendingTransactions: id }
  });  

  cancel(id);
}

Чтобы откатить транзакцию, нам нужно отменить ее на from а также to учетные записи. Сначала мы должны удалить транзакцию из from счет, возвращающий зарезервированный amount к balance.

accounts.updateOne({
    name: from, 
    pendingTransactions: { $in: [id] }
  }, {
    $inc: { balance: amount }, 
    $pull: { pendingTransactions: id }
  });  

Мы обновим учетную запись, если она содержит transaction путем сопоставления на name и если pendingTransactions массив содержит transaction id. Если документ совпадает, мы добавим amount к balance и удалить transaction id от pendingTransaction. Далее нам нужно обратить transaction на to аккаунт тоже.

accounts.updateOne({
  name: to, 
  pendingTransactions: { $in: [id] }
}, {
  $inc: { balance: -amount }, 
  $pull: { pendingTransactions: id }
});  

Единственное отличие от from заключается в том, что мы будем вычитать amount от balance счета. Наконец, мы вызываем cancel способ установить транзакцию state к canceled. Возвращаясь к executeTransaction Функция позволяет взглянуть на следующий оператор.

result = accounts.updateOne({
  name: to, 
  pendingTransactions: { $ne: transactionId }
}, {
  $inc: { balance: amount }, 
  $push: { pendingTransactions: transactionId }
});

if (result.modifiedCount == 0) {
  rollback(from, to, amount, transactionId);
  throw Error("Failed to credit " + to + " account");
}

Так же, как и в случае применения transactionId к from счет мы обеспечиваем account уже не содержит transactionId в pendingTransaction. Если его нет в pendingTransaction мы добавляем amount к balance и нажмите transactionId к pendingTransactions множество.

Если документ не обновляется, мы вызываем rollback работать, как мы делали ранее, а затем генерировать исключение, чтобы сообщить приложению, что транзакция не удалась.

Наконец, мы собираемся перевернуть состояние transaction к committed.

result = transactions.updateOne(
  { _id: transactionId }, 
  { $set: { state: "committed" } }
);

if (result.modifiedCount == 0) {
  rollback(from, to, amount, transactionId);
  throw Error("Failed to move transaction " + transactionId + " to committed");
}

Если это не удается, мы вызываем rollback Функция отмены транзакции. Наконец, мы вызываем cleanup функция. Давайте посмотрим, что делает функция.

function cleanup(from, to, id) {
  // Remove the transaction ids
  accounts.updateOne(
    { name: from }, 
    { $pull: { pendingTransactions: id } });
  
  // Remove the transaction ids
  accounts.updateOne(
    { name: to }, 
    { $pull: { pendingTransactions: id } });

  // Update transaction to committed
  transactions.updateOne(
    { _id: id }, 
    { $set: { state: "done" } });
}

Первое обновление удалит transactionId от from учетная запись. Второе обновление сделает то же самое для to учетная запись. Наконец, последнее обновление установит транзакцию state к done завершение перевода между двумя учетными записями.

Прогон и анализ производительности

Давайте запустим два сравнительных теста, чтобы посмотреть на два конкретных сценария трафика, примененных к обоим transaction подход, а также Two-Phase подход.

Первый сценарий — это тот, где у нас есть single thread выполнение переноса счета каждые millisecond за 35 seconds.

Во втором сценарии мы запускаем одну и ту же передачу каждый millisecond но используя five threads за 35 seconds.

Мы используем инструмент моделирования схемы в /mongodb-schema-simulator для создания нагрузки и записи результатов.

Снимаем мерки с 5 to 35 seconds чтобы избежать начального периода cache warmup на MongoDB так же как Java JIT warmup.

Один поток

Для сценария с одним потоком мы получаем следующие результаты.

На графике выше показаны результаты transaction подход.

На графике выше показаны результаты Two-Phase подход. Давайте возьмем ключевые цифры и поместим их в таблицу для облегчения сравнения.

я> mean это geometric mean. Среднее геометрическое — это среднее или среднее значение, которое указывает на центральную тенденцию или типичное значение набора чисел.
я> min минимальное значение, найденное в наборе.
я> max максимальное значение, найденное в наборе.
я> pth percentile это percentage данных, которые меньше значения. А 95 percentile из 100 означало бы 95% значений в наборе меньше, чем 100.

Давайте расшифруем цифры.

1. mean для транзакционного подхода ~2x ниже.
2. min для транзакционного подхода ~2x ниже.
3. max для транзакционного подхода ~2x ниже.
4. 95 percentile для транзакционного подхода ~2x ниже.
5. 99 percentile для транзакционного подхода ~2x ниже.

Глядя на это, мы видим, что поддержка транзакций примерно в два раза быстрее, чем Two-Phase совершить подход. Это имеет смысл, так как количество операций, которые нам нужно выполнить с MongoDB закончить Two-Phase commit, больше, чем нужно для transaction подход.

Учитывая это, мы можем заключить, что transaction подход превосходит Two-Phase подход. Но придержите лошадей. Давайте посмотрим, что произойдет, когда мы заставим transaction столкновений за счет увеличения нагрузки.

Несколько потоков

Для сценария с несколькими потоками мы получаем следующие результаты.

На первом графике показан перенос аккаунта с использованием MongoDB 4.0.x подход к сопровождению сделок.

На втором графике показан перенос аккаунта с использованием Two-Phase совершить подход. Давайте возьмем основные числа и сравним их.

Давайте расшифруем цифры.

1. mean очень похож между двумя подходами.
2. min для транзакционного подхода ниже.
3. max для транзакционного подхода ниже.
4. 95 percentile выше для транзакционного подхода.
5. 99 percentile очень похож между двумя подходами.

Мы видим, что Two-Phase commit имеют в целом лучшую общую характеристику производительности из-за 95 percentile говоря 95% операций заняло 19.98 ms или меньше по сравнению с 30.62 ms для транзакционного подхода.

Причина Two-Phase подход в настоящее время конкурирует с transaction подход, обусловлен transaction подход write conflicts из-за более высокой одновременной нагрузки, что вынуждает тест повторять транзакции до тех пор, пока они не будут успешными.

Вывод

При первоначальном осмотре может показаться, что использование transactions это по-прежнему жизнеспособный путь вперед, поскольку разница между этими двумя подходами невелика. Однако, прежде чем мы примем решение, мы должны принять во внимание несколько дополнительных факторов.

1. Транзакции работают только для replicasets по состоянию на MongoDB 4.0.x.
2. maximum время выполнения транзакции составляет минуту (все, что превышает минуту, прерывается).
3. Блокировка документов для транзакций может привести к узким местам производительности для документов, которые получают много записей, поскольку все транзакции сериализуются и должны ждать своей очереди для обработки.

Один раз MongoDB поддерживает sharded транзакции ваши repliaset транзакции могут оказаться распределенными транзакциями, включающими несколько shards. Это приводит к значительному снижению производительности, а также к новым и очень сложным режимам отказа, любой из которых может привести к сбою транзакции. В этом случае весьма вероятно, Two-Phase Подход фиксации значительно превзойдет подход транзакции.