[reseek bugfix] testcode to reproduct it

CMakeLists.txt

@@ -1171,5 +1171,6 @@ if(WITH_TOOLS)
 
   add_subdirectory(tools)
   #add_subdirectory(terark-tools/terark-test)
+  add_subdirectory(terark-tools/trx-test)
   add_subdirectory(terark-tools/batch-write-bench)
 endif()

db/db_iter.cc

@@ -572,6 +572,8 @@ bool DBIter::FindNextUserEntryInternal(bool skipping, bool prefix_check) {
                                             kValueTypeForSeek));
       }
       iter_->Seek(last_key);
+      TEST_SYNC_POINT_CALLBACK("DBIter::FindNextUserEntryInternal::Reseek",
+                               &last_key);
       RecordTick(statistics_, NUMBER_OF_RESEEKS_IN_ITERATION);
     } else {
       iter_->Next();

terark-tools/trx-test/CMakeLists.txt

+Cmake_minimum_required(VERSION 2.8)
+project(trx_test)
+
+set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++14 -DROCKSDB_PLATFORM_POSIX -D_DEBUG=1 -DDebugBreak=__builtin_trap -O0 -g3 -march=haswell")
+
+set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS}")
+
+add_executable(trx_test trx_test.cc)
+
+target_link_libraries(trx_test terarkdb gtest)

terark-tools/trx-test/trx_test.cc

+#include "trx_test.hpp"
+
+namespace rocksdb {
+uint32_t key_prefix = 8;
+void CreateUnCommittedData() {
+  Options options;
+  options.create_if_missing = true;
+  options.max_write_buffer_number = 2;
+  options.write_buffer_size = 4 * 1024;
+  options.blob_size = -1;
+  options.level0_file_num_compaction_trigger = 2;
+  options.merge_operator = MergeOperators::CreateFromStringId("stringappend");
+  auto env = Env::Default();
+  options.env = env;
+  options.two_write_queues = true;
+  std::string dbname = "trxdb";
+
+  DestroyDB(dbname, options);
+  TransactionDBOptions txn_db_options;
+  txn_db_options.write_policy = TxnDBWritePolicy::WRITE_PREPARED;
+  Status s;
+  TransactionDB *db = nullptr;
+  s = TransactionDB::Open(options, txn_db_options, dbname, &db);
+
+  TransactionOptions txn_options;
+  // Header (12 bytes) + NOOP (1 byte) + 2 * 8 bytes for data.
+  txn_options.max_write_batch_size = 100;
+  std::random_device rd;
+  std::mt19937_64 mt(rd());
+  std::uniform_int_distribution<uint64_t> gen(1, 16);
+  std::uniform_int_distribution<uint64_t> genval(199, 12382344);
+
+  uint32_t key_prefix_storage = __builtin_bswap32(key_prefix);
+
+  std::vector<Transaction *> trxs(100, nullptr);
+  for (size_t i = 0; i < trxs.size(); i++) {
+    trxs[i] = db->BeginTransaction(WriteOptions(), txn_options);
+    assert(trxs[i] != nullptr);
+    s = trxs[i]->SetName("myxid-" + std::to_string(i));
+  }
+
+  for (size_t i = 0; i < trxs.size(); i++) {
+    char key_buf[12] = {0};
+    memcpy(&key_buf[0], &key_prefix_storage, sizeof(int));
+    uint64_t randkey = gen(mt);
+    memcpy(&key_buf[0] + 4, &randkey, sizeof(uint64_t));
+    const rocksdb::Slice key =
+        rocksdb::Slice(reinterpret_cast<char *>(key_buf), sizeof(key_buf));
+
+    char value_buf[16] = {0};
+    uint64_t randval = genval(mt);
+    memcpy(&value_buf[0], &randval, sizeof(uint64_t));
+    randval = genval(mt);
+    memcpy(&value_buf[0], &randval, sizeof(uint64_t));
+    randval = genval(mt);
+    const rocksdb::Slice value =
+        rocksdb::Slice(reinterpret_cast<char *>(value_buf), sizeof(value_buf));
+
+    trxs[i]->GetWriteBatch()->GetWriteBatch()->Put(key, value);
+    trxs[i]->Prepare();
+
+    if (i==0) {
+      trxs[i]->Commit();
+    }
+  }
+
+  FlushOptions fo;
+  fo.wait = true;
+  db->Flush(fo);
+
+  abort();
+
+  using namespace std::chrono_literals;
+  std::cout << "sleep wait kill" << std::endl;
+  std::this_thread::sleep_for(60s);
+
+  assert(true);
+}
+void OpenAndSeek() {
+  rocksdb::SyncPoint::GetInstance()->SetCallBack(
+      "DBIter::FindNextUserEntryInternal::Reseek", [&](void *arg) {
+        std::string *lastkey = reinterpret_cast<std::string *>(arg);
+        std::cout << "Reseek: " << Slice(*lastkey).ToString(true) << std::endl;
+      });
+  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
+
+  Options options;
+  options.create_if_missing = false;
+  options.max_write_buffer_number = 2;
+  options.write_buffer_size = 4 * 1024;
+  options.level0_file_num_compaction_trigger = 2;
+  options.merge_operator = MergeOperators::CreateFromStringId("stringappend");
+  auto env = Env::Default();
+  options.env = env;
+  options.two_write_queues = true;
+  std::string dbname = "trxdb";
+
+  TransactionDBOptions txn_db_options;
+  txn_db_options.write_policy = TxnDBWritePolicy::WRITE_PREPARED;
+  TransactionDB *db = nullptr;
+  auto s = TransactionDB::Open(options, txn_db_options, dbname, &db);
+  assert(s.ok());
+
+  char seek_key_buf[4];
+  uint32_t key_prefix_storage = __builtin_bswap32(key_prefix);
+  memcpy(&seek_key_buf[0], &key_prefix_storage, sizeof(int));
+  const rocksdb::Slice seek_key(reinterpret_cast<char *>(&seek_key_buf[0]),
+                                sizeof(int));
+  rocksdb::ReadOptions read_options;
+  read_options.total_order_seek = true;
+  auto it = db->NewIterator(read_options, db->DefaultColumnFamily());
+  for (it->Seek(seek_key); it->Valid(); it->Next()) {
+  }
+}
+}  // namespace rocksdb
+
+int main(int argc, char **argv) {
+  google::ParseCommandLineFlags(&argc, &argv, true);
+  if (FLAGS_open_and_seek) {
+    rocksdb::OpenAndSeek();
+  } else {
+    rocksdb::CreateUnCommittedData();
+  }
+
+  return 0;
+}

terark-tools/trx-test/trx_test.hpp

+#include <sys/epoll.h>
+
+#include <cctype>
+#include <chrono>
+#include <cinttypes>
+#include <cstdlib>
+#include <fstream>
+#include <functional>
+#include <iomanip>
+#include <iostream>
+#include <iterator>
+#include <memory>
+#include <thread>
+#include <utility>
+#include <vector>
+#define GUJIA_HAS_EPOLL
+namespace gujia {
+typedef struct epoll_event Event;
+}
+
+#include <rocksdb/compaction_filter.h>
+#include <rocksdb/convenience.h>
+#include <rocksdb/db.h>
+#include <rocksdb/experimental.h>
+#include <rocksdb/memtablerep.h>
+#include <rocksdb/merge_operator.h>
+#include <rocksdb/slice.h>
+#include <rocksdb/slice_transform.h>
+#include <rocksdb/sst_file_writer.h>
+#include <rocksdb/table.h>
+#include <rocksdb/utilities/optimistic_transaction_db.h>
+#include <rocksdb/utilities/transaction_db.h>
+#include <rocksdb/utilities/write_batch_with_index.h>
+#include <table/get_context.h>
+#include <table/iterator_wrapper.h>
+#include <table/table_builder.h>
+#include <table/table_reader.h>
+#include <util/coding.h>
+#include <util/filename.h>
+#include <util/gflags_compat.h>
+#include <utilities/merge_operators.h>
+
+DEFINE_bool(open_and_seek,false,"open and seek");
\ No newline at end of file

utilities/transactions/write_prepared_txn.cc

@@ -102,6 +102,7 @@ Status WritePreparedTxn::PrepareInternal() {
   assert(!s.ok() || seq_used != kMaxSequenceNumber);
   auto prepare_seq = seq_used;
   SetId(prepare_seq);
+  TEST_SYNC_POINT_CALLBACK("WritePreparedTxn::PrepareInternal::End", nullptr);
   return s;
 }