ceph/src/rocksdb/utilities/transactions/transaction_base.h

   1 // Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
   2 // This source code is licensed under the BSD-style license found in the
   3 // LICENSE file in the root directory of this source tree. An additional grant
   4 // of patent rights can be found in the PATENTS file in the same directory.
   5
   6 #pragma once
   7
   8 #ifndef ROCKSDB_LITE
   9
  10 #include <stack>
  11 #include <string>
  12 #include <vector>
  13
  14 #include "rocksdb/db.h"
  15 #include "rocksdb/slice.h"
  16 #include "rocksdb/snapshot.h"
  17 #include "rocksdb/status.h"
  18 #include "rocksdb/types.h"
  19 #include "rocksdb/utilities/transaction.h"
  20 #include "rocksdb/utilities/transaction_db.h"
  21 #include "rocksdb/utilities/write_batch_with_index.h"
  22 #include "utilities/transactions/transaction_util.h"
  23
  24 namespace rocksdb {
  25
  26 class TransactionBaseImpl : public Transaction {
  27  public:
  28   TransactionBaseImpl(DB* db, const WriteOptions& write_options);
  29
  30   virtual ~TransactionBaseImpl();
  31
  32   // Remove pending operations queued in this transaction.
  33   virtual void Clear();
  34
  35   void Reinitialize(DB* db, const WriteOptions& write_options);
  36
  37   // Called before executing Put, Merge, Delete, and GetForUpdate.  If TryLock
  38   // returns non-OK, the Put/Merge/Delete/GetForUpdate will be failed.
  39   // untracked will be true if called from PutUntracked, DeleteUntracked, or
  40   // MergeUntracked.
  41   virtual Status TryLock(ColumnFamilyHandle* column_family, const Slice& key,
  42                          bool read_only, bool exclusive,
  43                          bool untracked = false) = 0;
  44
  45   void SetSavePoint() override;
  46
  47   Status RollbackToSavePoint() override;
  48
  49   Status Get(const ReadOptions& options, ColumnFamilyHandle* column_family,
  50              const Slice& key, std::string* value) override;
  51
  52   Status Get(const ReadOptions& options, const Slice& key,
  53              std::string* value) override {
  54     return Get(options, db_->DefaultColumnFamily(), key, value);
  55   }
  56
  57   Status GetForUpdate(const ReadOptions& options,
  58                       ColumnFamilyHandle* column_family, const Slice& key,
  59                       std::string* value, bool exclusive) override;
  60
  61   Status GetForUpdate(const ReadOptions& options, const Slice& key,
  62                       std::string* value, bool exclusive) override {
  63     return GetForUpdate(options, db_->DefaultColumnFamily(), key, value,
  64                         exclusive);
  65   }
  66
  67   std::vector<Status> MultiGet(
  68       const ReadOptions& options,
  69       const std::vector<ColumnFamilyHandle*>& column_family,
  70       const std::vector<Slice>& keys,
  71       std::vector<std::string>* values) override;
  72
  73   std::vector<Status> MultiGet(const ReadOptions& options,
  74                                const std::vector<Slice>& keys,
  75                                std::vector<std::string>* values) override {
  76     return MultiGet(options, std::vector<ColumnFamilyHandle*>(
  77                                  keys.size(), db_->DefaultColumnFamily()),
  78                     keys, values);
  79   }
  80
  81   std::vector<Status> MultiGetForUpdate(
  82       const ReadOptions& options,
  83       const std::vector<ColumnFamilyHandle*>& column_family,
  84       const std::vector<Slice>& keys,
  85       std::vector<std::string>* values) override;
  86
  87   std::vector<Status> MultiGetForUpdate(
  88       const ReadOptions& options, const std::vector<Slice>& keys,
  89       std::vector<std::string>* values) override {
  90     return MultiGetForUpdate(options,
  91                              std::vector<ColumnFamilyHandle*>(
  92                                  keys.size(), db_->DefaultColumnFamily()),
  93                              keys, values);
  94   }
  95
  96   Iterator* GetIterator(const ReadOptions& read_options) override;
  97   Iterator* GetIterator(const ReadOptions& read_options,
  98                         ColumnFamilyHandle* column_family) override;
  99
 100   Status Put(ColumnFamilyHandle* column_family, const Slice& key,
 101              const Slice& value) override;
 102   Status Put(const Slice& key, const Slice& value) override {
 103     return Put(nullptr, key, value);
 104   }
 105
 106   Status Put(ColumnFamilyHandle* column_family, const SliceParts& key,
 107              const SliceParts& value) override;
 108   Status Put(const SliceParts& key, const SliceParts& value) override {
 109     return Put(nullptr, key, value);
 110   }
 111
 112   Status Merge(ColumnFamilyHandle* column_family, const Slice& key,
 113                const Slice& value) override;
 114   Status Merge(const Slice& key, const Slice& value) override {
 115     return Merge(nullptr, key, value);
 116   }
 117
 118   Status Delete(ColumnFamilyHandle* column_family, const Slice& key) override;
 119   Status Delete(const Slice& key) override { return Delete(nullptr, key); }
 120   Status Delete(ColumnFamilyHandle* column_family,
 121                 const SliceParts& key) override;
 122   Status Delete(const SliceParts& key) override { return Delete(nullptr, key); }
 123
 124   Status SingleDelete(ColumnFamilyHandle* column_family,
 125                       const Slice& key) override;
 126   Status SingleDelete(const Slice& key) override {
 127     return SingleDelete(nullptr, key);
 128   }
 129   Status SingleDelete(ColumnFamilyHandle* column_family,
 130                       const SliceParts& key) override;
 131   Status SingleDelete(const SliceParts& key) override {
 132     return SingleDelete(nullptr, key);
 133   }
 134
 135   Status PutUntracked(ColumnFamilyHandle* column_family, const Slice& key,
 136                       const Slice& value) override;
 137   Status PutUntracked(const Slice& key, const Slice& value) override {
 138     return PutUntracked(nullptr, key, value);
 139   }
 140
 141   Status PutUntracked(ColumnFamilyHandle* column_family, const SliceParts& key,
 142                       const SliceParts& value) override;
 143   Status PutUntracked(const SliceParts& key, const SliceParts& value) override {
 144     return PutUntracked(nullptr, key, value);
 145   }
 146
 147   Status MergeUntracked(ColumnFamilyHandle* column_family, const Slice& key,
 148                         const Slice& value) override;
 149   Status MergeUntracked(const Slice& key, const Slice& value) override {
 150     return MergeUntracked(nullptr, key, value);
 151   }
 152
 153   Status DeleteUntracked(ColumnFamilyHandle* column_family,
 154                          const Slice& key) override;
 155   Status DeleteUntracked(const Slice& key) override {
 156     return DeleteUntracked(nullptr, key);
 157   }
 158   Status DeleteUntracked(ColumnFamilyHandle* column_family,
 159                          const SliceParts& key) override;
 160   Status DeleteUntracked(const SliceParts& key) override {
 161     return DeleteUntracked(nullptr, key);
 162   }
 163
 164   void PutLogData(const Slice& blob) override;
 165
 166   WriteBatchWithIndex* GetWriteBatch() override;
 167
 168   virtual void SetLockTimeout(int64_t timeout) override { /* Do nothing */
 169   }
 170
 171   const Snapshot* GetSnapshot() const override {
 172     return snapshot_ ? snapshot_.get() : nullptr;
 173   }
 174
 175   void SetSnapshot() override;
 176   void SetSnapshotOnNextOperation(
 177       std::shared_ptr<TransactionNotifier> notifier = nullptr) override;
 178
 179   void ClearSnapshot() override {
 180     snapshot_.reset();
 181     snapshot_needed_ = false;
 182     snapshot_notifier_ = nullptr;
 183   }
 184
 185   void DisableIndexing() override { indexing_enabled_ = false; }
 186
 187   void EnableIndexing() override { indexing_enabled_ = true; }
 188
 189   uint64_t GetElapsedTime() const override;
 190
 191   uint64_t GetNumPuts() const override;
 192
 193   uint64_t GetNumDeletes() const override;
 194
 195   uint64_t GetNumMerges() const override;
 196
 197   uint64_t GetNumKeys() const override;
 198
 199   void UndoGetForUpdate(ColumnFamilyHandle* column_family,
 200                         const Slice& key) override;
 201   void UndoGetForUpdate(const Slice& key) override {
 202     return UndoGetForUpdate(nullptr, key);
 203   };
 204
 205   // Get list of keys in this transaction that must not have any conflicts
 206   // with writes in other transactions.
 207   const TransactionKeyMap& GetTrackedKeys() const { return tracked_keys_; }
 208
 209   WriteOptions* GetWriteOptions() override { return &write_options_; }
 210
 211   void SetWriteOptions(const WriteOptions& write_options) override {
 212     write_options_ = write_options;
 213   }
 214
 215   // Used for memory management for snapshot_
 216   void ReleaseSnapshot(const Snapshot* snapshot, DB* db);
 217
 218   // iterates over the given batch and makes the appropriate inserts.
 219   // used for rebuilding prepared transactions after recovery.
 220   Status RebuildFromWriteBatch(WriteBatch* src_batch) override;
 221
 222   WriteBatch* GetCommitTimeWriteBatch() override;
 223
 224  protected:
 225   // Add a key to the list of tracked keys.
 226   //
 227   // seqno is the earliest seqno this key was involved with this transaction.
 228   // readonly should be set to true if no data was written for this key
 229   void TrackKey(uint32_t cfh_id, const std::string& key, SequenceNumber seqno,
 230                 bool readonly, bool exclusive);
 231
 232   // Helper function to add a key to the given TransactionKeyMap
 233   static void TrackKey(TransactionKeyMap* key_map, uint32_t cfh_id,
 234                        const std::string& key, SequenceNumber seqno,
 235                        bool readonly, bool exclusive);
 236
 237   // Called when UndoGetForUpdate determines that this key can be unlocked.
 238   virtual void UnlockGetForUpdate(ColumnFamilyHandle* column_family,
 239                                   const Slice& key) = 0;
 240
 241   std::unique_ptr<TransactionKeyMap> GetTrackedKeysSinceSavePoint();
 242
 243   // Sets a snapshot if SetSnapshotOnNextOperation() has been called.
 244   void SetSnapshotIfNeeded();
 245
 246   DB* db_;
 247   DBImpl* dbimpl_;
 248
 249   WriteOptions write_options_;
 250
 251   const Comparator* cmp_;
 252
 253   // Stores that time the txn was constructed, in microseconds.
 254   uint64_t start_time_;
 255
 256   // Stores the current snapshot that was set by SetSnapshot or null if
 257   // no snapshot is currently set.
 258   std::shared_ptr<const Snapshot> snapshot_;
 259
 260   // Count of various operations pending in this transaction
 261   uint64_t num_puts_ = 0;
 262   uint64_t num_deletes_ = 0;
 263   uint64_t num_merges_ = 0;
 264
 265   struct SavePoint {
 266     std::shared_ptr<const Snapshot> snapshot_;
 267     bool snapshot_needed_;
 268     std::shared_ptr<TransactionNotifier> snapshot_notifier_;
 269     uint64_t num_puts_;
 270     uint64_t num_deletes_;
 271     uint64_t num_merges_;
 272
 273     // Record all keys tracked since the last savepoint
 274     TransactionKeyMap new_keys_;
 275
 276     SavePoint(std::shared_ptr<const Snapshot> snapshot, bool snapshot_needed,
 277               std::shared_ptr<TransactionNotifier> snapshot_notifier,
 278               uint64_t num_puts, uint64_t num_deletes, uint64_t num_merges)
 279         : snapshot_(snapshot),
 280           snapshot_needed_(snapshot_needed),
 281           snapshot_notifier_(snapshot_notifier),
 282           num_puts_(num_puts),
 283           num_deletes_(num_deletes),
 284           num_merges_(num_merges) {}
 285   };
 286
 287   // Records writes pending in this transaction
 288   WriteBatchWithIndex write_batch_;
 289
 290  private:
 291   // batch to be written at commit time
 292   WriteBatch commit_time_batch_;
 293
 294   // Stack of the Snapshot saved at each save point.  Saved snapshots may be
 295   // nullptr if there was no snapshot at the time SetSavePoint() was called.
 296   std::unique_ptr<std::stack<TransactionBaseImpl::SavePoint>> save_points_;
 297
 298   // Map from column_family_id to map of keys that are involved in this
 299   // transaction.
 300   // Pessimistic Transactions will do conflict checking before adding a key
 301   // by calling TrackKey().
 302   // Optimistic Transactions will wait till commit time to do conflict checking.
 303   TransactionKeyMap tracked_keys_;
 304
 305   // If true, future Put/Merge/Deletes will be indexed in the
 306   // WriteBatchWithIndex.
 307   // If false, future Put/Merge/Deletes will be inserted directly into the
 308   // underlying WriteBatch and not indexed in the WriteBatchWithIndex.
 309   bool indexing_enabled_;
 310
 311   // SetSnapshotOnNextOperation() has been called and the snapshot has not yet
 312   // been reset.
 313   bool snapshot_needed_ = false;
 314
 315   // SetSnapshotOnNextOperation() has been called and the caller would like
 316   // a notification through the TransactionNotifier interface
 317   std::shared_ptr<TransactionNotifier> snapshot_notifier_ = nullptr;
 318
 319   Status TryLock(ColumnFamilyHandle* column_family, const SliceParts& key,
 320                  bool read_only, bool exclusive, bool untracked = false);
 321
 322   WriteBatchBase* GetBatchForWrite();
 323
 324   void SetSnapshotInternal(const Snapshot* snapshot);
 325 };
 326
 327 }  // namespace rocksdb
 328
 329 #endif  // ROCKSDB_LITE