write_unprepared_txn_db.h 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105
  1. // Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
  2. // This source code is licensed under both the GPLv2 (found in the
  3. // COPYING file in the root directory) and Apache 2.0 License
  4. // (found in the LICENSE.Apache file in the root directory).
  5. #pragma once
  6. #include "utilities/transactions/write_prepared_txn_db.h"
  7. #include "utilities/transactions/write_unprepared_txn.h"
  8. namespace ROCKSDB_NAMESPACE {
  9. class WriteUnpreparedTxn;
  10. class WriteUnpreparedTxnDB : public WritePreparedTxnDB {
  11. public:
  12. using WritePreparedTxnDB::WritePreparedTxnDB;
  13. Status Initialize(const std::vector<size_t>& compaction_enabled_cf_indices,
  14. const std::vector<ColumnFamilyHandle*>& handles) override;
  15. Transaction* BeginTransaction(const WriteOptions& write_options,
  16. const TransactionOptions& txn_options,
  17. Transaction* old_txn) override;
  18. // Struct to hold ownership of snapshot and read callback for cleanup.
  19. struct IteratorState;
  20. using WritePreparedTxnDB::NewIterator;
  21. Iterator* NewIterator(const ReadOptions& _read_options,
  22. ColumnFamilyHandle* column_family,
  23. WriteUnpreparedTxn* txn);
  24. private:
  25. Status RollbackRecoveredTransaction(const DBImpl::RecoveredTransaction* rtxn);
  26. };
  27. class WriteUnpreparedCommitEntryPreReleaseCallback : public PreReleaseCallback {
  28. // TODO(lth): Reduce code duplication with
  29. // WritePreparedCommitEntryPreReleaseCallback
  30. public:
  31. // includes_data indicates that the commit also writes non-empty
  32. // CommitTimeWriteBatch to memtable, which needs to be committed separately.
  33. WriteUnpreparedCommitEntryPreReleaseCallback(
  34. WritePreparedTxnDB* db, DBImpl* db_impl,
  35. const std::map<SequenceNumber, size_t>& unprep_seqs,
  36. size_t data_batch_cnt = 0, bool publish_seq = true)
  37. : db_(db),
  38. db_impl_(db_impl),
  39. unprep_seqs_(unprep_seqs),
  40. data_batch_cnt_(data_batch_cnt),
  41. includes_data_(data_batch_cnt_ > 0),
  42. publish_seq_(publish_seq) {
  43. assert(unprep_seqs.size() > 0);
  44. }
  45. Status Callback(SequenceNumber commit_seq,
  46. bool is_mem_disabled __attribute__((__unused__)), uint64_t,
  47. size_t /*index*/, size_t /*total*/) override {
  48. const uint64_t last_commit_seq = LIKELY(data_batch_cnt_ <= 1)
  49. ? commit_seq
  50. : commit_seq + data_batch_cnt_ - 1;
  51. // Recall that unprep_seqs maps (un)prepared_seq => prepare_batch_cnt.
  52. for (const auto& s : unprep_seqs_) {
  53. for (size_t i = 0; i < s.second; i++) {
  54. db_->AddCommitted(s.first + i, last_commit_seq);
  55. }
  56. }
  57. if (includes_data_) {
  58. assert(data_batch_cnt_);
  59. // Commit the data that is accompanied with the commit request
  60. for (size_t i = 0; i < data_batch_cnt_; i++) {
  61. // For commit seq of each batch use the commit seq of the last batch.
  62. // This would make debugging easier by having all the batches having
  63. // the same sequence number.
  64. db_->AddCommitted(commit_seq + i, last_commit_seq);
  65. }
  66. }
  67. if (db_impl_->immutable_db_options().two_write_queues && publish_seq_) {
  68. assert(is_mem_disabled); // implies the 2nd queue
  69. // Publish the sequence number. We can do that here assuming the callback
  70. // is invoked only from one write queue, which would guarantee that the
  71. // publish sequence numbers will be in order, i.e., once a seq is
  72. // published all the seq prior to that are also publishable.
  73. db_impl_->SetLastPublishedSequence(last_commit_seq);
  74. }
  75. // else SequenceNumber that is updated as part of the write already does the
  76. // publishing
  77. return Status::OK();
  78. }
  79. private:
  80. WritePreparedTxnDB* db_;
  81. DBImpl* db_impl_;
  82. const std::map<SequenceNumber, size_t>& unprep_seqs_;
  83. size_t data_batch_cnt_;
  84. // Either because it is commit without prepare or it has a
  85. // CommitTimeWriteBatch
  86. bool includes_data_;
  87. // Should the callback also publishes the commit seq number
  88. bool publish_seq_;
  89. };
  90. } // namespace ROCKSDB_NAMESPACE