Adding transactional_splinterdb into YCSB

core/core_workload.cc

@@ -76,6 +76,9 @@ const string CoreWorkload::INSERT_START_DEFAULT = "0";
 const string CoreWorkload::RECORD_COUNT_PROPERTY = "recordcount";
 const string CoreWorkload::OPERATION_COUNT_PROPERTY = "operationcount";
 
+const string CoreWorkload::OPS_PER_TRANSACTION_PROPERTY = "opspertransaction";
+const string CoreWorkload::OPS_PER_TRANSACTION_DEFAULT = "1";
+
 void CoreWorkload::InitLoadWorkload(const utils::Properties &p, unsigned int nthreads, unsigned int this_thread, BatchedCounterGenerator *key_generator) {
   table_name_ = p.GetProperty(TABLENAME_PROPERTY,TABLENAME_DEFAULT);
 
@@ -178,6 +181,8 @@ void CoreWorkload::InitRunWorkload(const utils::Properties &p, unsigned int nthr
   }
 
   //batch_size_ = 1;
+
+  ops_per_transaction_ = std::stoi(p.GetProperty(OPS_PER_TRANSACTION_PROPERTY, OPS_PER_TRANSACTION_DEFAULT));
 }
 
 ycsbc::Generator<uint64_t> *CoreWorkload::GetFieldLenGenerator(

core/core_workload.h

@@ -141,6 +141,9 @@ class CoreWorkload {
   static const std::string RECORD_COUNT_PROPERTY;
   static const std::string OPERATION_COUNT_PROPERTY;
 
+  static const std::string OPS_PER_TRANSACTION_PROPERTY;
+  static const std::string OPS_PER_TRANSACTION_DEFAULT;
+
   ///
   /// Initialize the scenario.
   /// Called once, in the main client thread, before any operations are started.
@@ -164,6 +167,7 @@ class CoreWorkload {
 
   bool read_all_fields() const { return read_all_fields_; }
   bool write_all_fields() const { return write_all_fields_; }
+  int ops_per_transaction() const { return ops_per_transaction_; }
 
   CoreWorkload() :
       generator_(),
@@ -214,7 +218,7 @@ class CoreWorkload {
   bool ordered_inserts_;
   size_t record_count_;
   int zero_padding_;
-
+  int ops_per_transaction_;
   std::uniform_int_distribution<char> uniform_letter_dist_;
 };
 

core/db.h

@@ -9,27 +9,32 @@
 #ifndef YCSB_C_DB_H_
 #define YCSB_C_DB_H_
 
-#include <vector>
 #include <string>
+#include <vector>
 
 namespace ycsbc {
 
+class Transaction;
+
 class DB {
- public:
+public:
   typedef std::pair<std::string, std::string> KVPair;
   static const int kOK = 0;
   static const int kErrorNoData = 1;
   static const int kErrorConflict = 2;
+  static const int kErrorNotSupport = 3;
   ///
   /// Initializes any state for accessing this DB.
-  /// Called once per DB client (thread); there is a single DB instance globally.
+  /// Called once per DB client (thread); there is a single DB instance
+  /// globally.
   ///
-  virtual void Init() { }
+  virtual void Init() {}
   ///
   /// Clears any state for accessing this DB.
-  /// Called once per DB client (thread); there is a single DB instance globally.
+  /// Called once per DB client (thread); there is a single DB instance
+  /// globally.
   ///
-  virtual void Close() { }
+  virtual void Close() {}
   ///
   /// Reads a record from the database.
   /// Field/value pairs from the result are stored in a vector.
@@ -89,10 +94,93 @@ class DB {
   /// @return Zero on success, a non-zero error code on error.
   ///
   virtual int Delete(const std::string &table, const std::string &key) = 0;
-
-  virtual ~DB() { }
+
+  virtual ~DB() {}
+
+  virtual void Begin(Transaction **txn) {}
+
+  virtual int Commit(Transaction **txn) { return 0; }
+
+  ///
+  /// Reads a record from the database.
+  /// Field/value pairs from the result are stored in a vector.
+  ///
+  /// @param txn The current transaction.
+  /// @param table The name of the table.
+  /// @param key The key of the record to read.
+  /// @param fields The list of fields to read, or NULL for all of them.
+  /// @param result A vector of field/value pairs for the result.
+  /// @return Zero on success, or a non-zero error code on error/record-miss.
+  ///
+  virtual int Read(Transaction *txn, const std::string &table,
+                   const std::string &key,
+                   const std::vector<std::string> *fields,
+                   std::vector<KVPair> &result) {
+    return Read(table, key, fields, result);
+  }
+  ///
+  /// Performs a range scan for a set of records in the database.
+  /// Field/value pairs from the result are stored in a vector.
+  ///
+  /// @param txn The current transaction.
+  /// @param table The name of the table.
+  /// @param key The key of the first record to read.
+  /// @param record_count The number of records to read.
+  /// @param fields The list of fields to read, or NULL for all of them.
+  /// @param result A vector of vector, where each vector contains field/value
+  ///        pairs for one record
+  /// @return Zero on success, or a non-zero error code on error.
+  ///
+  virtual int Scan(Transaction *txn, const std::string &table,
+                   const std::string &key, int record_count,
+                   const std::vector<std::string> *fields,
+                   std::vector<std::vector<KVPair>> &result) {
+    return Scan(table, key, record_count, fields, result);
+  }
+
+  ///
+  /// Updates a record in the database.
+  /// Field/value pairs in the specified vector are written to the record,
+  /// overwriting any existing values with the same field names.
+  ///
+  /// @param txn The current transaction.
+  /// @param table The name of the table.
+  /// @param key The key of the record to write.
+  /// @param values A vector of field/value pairs to update in the record.
+  /// @return Zero on success, a non-zero error code on error.
+  ///
+  virtual int Update(Transaction *txn, const std::string &table,
+                     const std::string &key, std::vector<KVPair> &values) {
+    return Update(table, key, values);
+  }
+  ///
+  /// Inserts a record into the database.
+  /// Field/value pairs in the specified vector are written into the record.
+  ///
+  /// @param txn The current transaction.
+  /// @param table The name of the table.
+  /// @param key The key of the record to insert.
+  /// @param values A vector of field/value pairs to insert in the record.
+  /// @return Zero on success, a non-zero error code on error.
+  ///
+  virtual int Insert(Transaction *txn, const std::string &table,
+                     const std::string &key, std::vector<KVPair> &values) {
+    return Insert(table, key, values);
+  }
+  ///
+  /// Deletes a record from the database.
+  ///
+  /// @param txn The current transaction.
+  /// @param table The name of the table.
+  /// @param key The key of the record to delete.
+  /// @return Zero on success, a non-zero error code on error.
+  ///
+  virtual int Delete(Transaction *txn, const std::string &table,
+                     const std::string &key) {
+    return Delete(table, key);
+  }
 };
 
-} // ycsbc
+} // namespace ycsbc
 
 #endif // YCSB_C_DB_H_

core/transaction.h

@@ -0,0 +1,49 @@
+#ifndef YCSB_C_TRANSACTION_H_
+#define YCSB_C_TRANSACTION_H_
+
+#include <vector>
+
+#include "core_workload.h"
+#include "db.h"
+
+namespace ycsbc {
+
+struct TransactionOperation {
+  enum Operation op;
+  std::string table;
+  std::string key;
+  int len;
+  std::vector<DB::KVPair> values;
+};
+
+class Transaction {
+public:
+  Transaction() : next_op(0), is_aborted(false){};
+
+  virtual ~Transaction(){};
+
+  void ReadyToRecordOperations(unsigned long size) {
+    ops.resize(size);
+    next_op = 0;
+  };
+
+  unsigned long GetTransactionOperationsSize() { return ops.size(); };
+
+  TransactionOperation &GetNextOperation() { return ops[next_op++]; }
+
+  TransactionOperation &GetOperation(unsigned long i) { return ops[i]; }
+
+  void SetAborted(bool aborted) { is_aborted = aborted; };
+
+  bool IsAborted() { return is_aborted; };
+
+protected:
+  std::vector<TransactionOperation> ops;
+  unsigned long next_op;
+
+  bool is_aborted;
+};
+
+} // namespace ycsbc
+
+#endif // YCSB_C_TRANSACTION_H_

db/db_factory.cc

@@ -15,7 +15,9 @@
 #include "db/tbb_rand_db.h"
 #include "db/tbb_scan_db.h"
 #include "db/splinter_db.h"
+#include "db/transactional_splinter_db.h"
 #include "db/rocks_db.h"
+#include "db/transaction_rocks_db.h"
 
 using namespace std;
 using ycsbc::DB;
@@ -33,8 +35,14 @@ DB* DBFactory::CreateDB(utils::Properties &props, bool preloaded) {
     return new RedisDB(props["host"].c_str(), port, slaves);
   } else if (props["dbname"] == "rocksdb") {
     return new RocksDB(props, preloaded);
+  } else if (props["dbname"] == "transaction_rocksdb") {
+    return new TransactionRocksDB(props, preloaded);
+  } else if (props["dbname"] == "optimistic_transaction_rocksdb") {
+    return new OptimisticTransactionRocksDB(props, preloaded);
   } else if (props["dbname"] == "splinterdb") {
     return new SplinterDB(props, preloaded);
+  } else if (props["dbname"] == "transactional_splinterdb") {
+    return new TransactionalSplinterDB(props, preloaded);
   } else if (props["dbname"] == "tbb_rand") {
     assert(!preloaded);
     return new TbbRandDB;