db_stress: Reinstate Transaction::Rollback() calls before destruction (#11656)

Summary:
https://github.com/facebook/rocksdb/issues/11653 broke some crash tests.
Apparently these Rollbacks are needed for pessimistic transaction cases. (I'm still not sure if the API makes any sense with regard to safe usage. It's certainly not documented. Will consider in follow-up PRs.)

Pull Request resolved: https://github.com/facebook/rocksdb/pull/11656

Test Plan: manual crash test runs, crash_test_with_multiops_wc_txn and crash_test_with_multiops_wp_txn

Reviewed By: cbi42

Differential Revision: D47906280

Pulled By: pdillinger

fbshipit-source-id: d058a01b6dbb47a4f08d199e335364168304f81b

db_stress_tool/multi_ops_txns_stress.cc

@@ -572,7 +572,7 @@ Status MultiOpsTxnsStressTest::PrimaryKeyUpdateTxn(ThreadState* thread,
   assert(txn);
   txn->SetSnapshotOnNextOperation(/*notifier=*/nullptr);
 
-  const Defer cleanup([new_a, &s, thread, this]() {
+  const Defer cleanup([new_a, &s, thread, this, &txn]() {
     if (s.ok()) {
       // Two gets, one for existing pk, one for locking potential new pk.
       thread->stats.AddGets(/*ngets=*/2, /*nfounds=*/1);
@@ -594,6 +594,7 @@ Status MultiOpsTxnsStressTest::PrimaryKeyUpdateTxn(ThreadState* thread,
     }
     auto& key_gen = key_gen_for_a_[thread->tid];
     key_gen->UndoAllocation(new_a);
+    txn->Rollback().PermitUncheckedError();
   });
 
   ReadOptions ropts;
@@ -692,7 +693,7 @@ Status MultiOpsTxnsStressTest::SecondaryKeyUpdateTxn(ThreadState* thread,
 
   Iterator* it = nullptr;
   long iterations = 0;
-  const Defer cleanup([new_c, &s, thread, &it, this, &iterations]() {
+  const Defer cleanup([new_c, &s, thread, &txn, &it, this, &iterations]() {
     delete it;
     if (s.ok()) {
       thread->stats.AddIterations(iterations);
@@ -717,6 +718,7 @@ Status MultiOpsTxnsStressTest::SecondaryKeyUpdateTxn(ThreadState* thread,
     }
     auto& key_gen = key_gen_for_c_[thread->tid];
     key_gen->UndoAllocation(new_c);
+    txn->Rollback().PermitUncheckedError();
   });
 
   // TODO (yanqin) try SetSnapshotOnNextOperation(). We currently need to take
@@ -887,7 +889,7 @@ Status MultiOpsTxnsStressTest::UpdatePrimaryIndexValueTxn(ThreadState* thread,
 
   assert(txn);
 
-  const Defer cleanup([&s, thread]() {
+  const Defer cleanup([&s, thread, &txn]() {
     if (s.ok()) {
       thread->stats.AddGets(/*ngets=*/1, /*nfounds=*/1);
       thread->stats.AddBytesForWrites(
@@ -904,6 +906,7 @@ Status MultiOpsTxnsStressTest::UpdatePrimaryIndexValueTxn(ThreadState* thread,
     } else {
       thread->stats.AddErrors(1);
     }
+    txn->Rollback().PermitUncheckedError();
   });
   ReadOptions ropts;
   ropts.rate_limiter_priority =
@@ -967,7 +970,7 @@ Status MultiOpsTxnsStressTest::PointLookupTxn(ThreadState* thread,
 
   assert(txn);
 
-  const Defer cleanup([&s, thread]() {
+  const Defer cleanup([&s, thread, &txn]() {
     if (s.ok()) {
       thread->stats.AddGets(/*ngets=*/1, /*nfounds=*/1);
       return;
@@ -976,6 +979,7 @@ Status MultiOpsTxnsStressTest::PointLookupTxn(ThreadState* thread,
     } else {
       thread->stats.AddErrors(1);
     }
+    txn->Rollback().PermitUncheckedError();
   });
 
   std::shared_ptr<const Snapshot> snapshot;
@@ -1010,12 +1014,13 @@ Status MultiOpsTxnsStressTest::RangeScanTxn(ThreadState* thread,
 
   assert(txn);
 
-  const Defer cleanup([&s, thread]() {
+  const Defer cleanup([&s, thread, &txn]() {
     if (s.ok()) {
       thread->stats.AddIterations(1);
       return;
     }
     thread->stats.AddErrors(1);
+    txn->Rollback().PermitUncheckedError();
   });
 
   std::shared_ptr<const Snapshot> snapshot;

db_stress_tool/no_batched_ops_stress.cc

@@ -865,6 +865,9 @@ class NonBatchedOpsStressTest : public StressTest {
     if (readoptionscopy.snapshot) {
       db_->ReleaseSnapshot(readoptionscopy.snapshot);
     }
+    if (use_txn) {
+      txn->Rollback().PermitUncheckedError();
+    }
     return statuses;
   }
 

include/rocksdb/utilities/transaction.h

@@ -140,9 +140,6 @@ class Transaction {
   Transaction(const Transaction&) = delete;
   void operator=(const Transaction&) = delete;
 
-  // The transaction is safely discarded on destruction, though must be
-  // discarded before the DB is closed or destroyed. (Calling Rollback()
-  // is not necessary before destruction.)
   virtual ~Transaction() {}
 
   // If a transaction has a snapshot set, the transaction will ensure that
@@ -263,6 +260,7 @@ class Transaction {
       std::shared_ptr<const Snapshot>* snapshot = nullptr);
 
   // Discard all batched writes in this transaction.
+  // FIXME: what happens if this isn't called before destruction?
   virtual Status Rollback() = 0;
 
   // Records the state of the transaction for future calls to

utilities/transactions/pessimistic_transaction.h

@@ -308,4 +308,3 @@ class WriteCommittedTxn : public PessimisticTransaction {
 };
 
 }  // namespace ROCKSDB_NAMESPACE
-

utilities/transactions/write_prepared_txn.h

@@ -114,4 +114,3 @@ class WritePreparedTxn : public PessimisticTransaction {
 };
 
 }  // namespace ROCKSDB_NAMESPACE
-