Adding upstream version 1.7.3.upstream/1.7.3upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

2 files changed, 974 insertions, 0 deletions

diff --git a/ext/lmdb-safe/lmdb-safe.cc b/ext/lmdb-safe/lmdb-safe.cc
new file mode 100644
index 0000000..7a023fd
--- /dev/null
+++ b/ext/lmdb-safe/lmdb-safe.cc
@@ -0,0 +1,364 @@
+#include "lmdb-safe.hh"
+#include <fcntl.h>
+#include <mutex>
+#include <memory>
+#include <sys/stat.h>
+#include <string.h>
+#include <map>
+
+using std::string;
+using std::runtime_error;
+using std::tuple;
+using std::weak_ptr;
+
+static string MDBError(int rc)
+{
+  return mdb_strerror(rc);
+}
+
+MDBDbi::MDBDbi(MDB_env* env, MDB_txn* txn, const string_view dbname, int flags)
+{
+  // A transaction that uses this function must finish (either commit or abort) before any other transaction in the process may use this function.
+  
+  int rc = mdb_dbi_open(txn, dbname.empty() ? 0 : &dbname[0], flags, &d_dbi);
+  if(rc)
+    throw std::runtime_error("Unable to open named database: " + MDBError(rc));
+  
+  // Database names are keys in the unnamed database, and may be read but not written.
+}
+
+MDBEnv::MDBEnv(const char* fname, int flags, int mode)
+{
+  mdb_env_create(&d_env);
+  uint64_t mapsizeMB = (sizeof(long)==4) ? 100 : 16000;
+  // on 32 bit platforms, there is just no room for more
+  if(mdb_env_set_mapsize(d_env, mapsizeMB * 1048576))
+    throw std::runtime_error("setting map size");
+    /*
+Various other options may also need to be set before opening the handle, e.g. mdb_env_set_mapsize(), mdb_env_set_maxreaders(), mdb_env_set_maxdbs(),
+    */
+
+  mdb_env_set_maxdbs(d_env, 128);
+
+  // we need MDB_NOTLS since we rely on its semantics
+  if(int rc=mdb_env_open(d_env, fname, flags | MDB_NOTLS, mode)) {
+    // If this function fails, mdb_env_close() must be called to discard the MDB_env handle.
+    mdb_env_close(d_env);
+    throw std::runtime_error("Unable to open database file "+std::string(fname)+": " + MDBError(rc));
+  }
+
+  if ((flags & MDB_RDONLY) == 0) {
+    // Check for stale readers to prevent unbridled database growth.
+    // Only do this when in RW mode since it affects the file.
+    mdb_reader_check(d_env, nullptr);
+  }
+}
+
+void MDBEnv::incROTX()
+{
+  std::lock_guard<std::mutex> l(d_countmutex);
+  ++d_ROtransactionsOut[std::this_thread::get_id()];
+}
+
+void MDBEnv::decROTX()
+{
+  std::lock_guard<std::mutex> l(d_countmutex);
+  --d_ROtransactionsOut[std::this_thread::get_id()];
+}
+
+void MDBEnv::incRWTX()
+{
+  std::lock_guard<std::mutex> l(d_countmutex);
+  ++d_RWtransactionsOut[std::this_thread::get_id()];
+}
+
+void MDBEnv::decRWTX()
+{
+  std::lock_guard<std::mutex> l(d_countmutex);
+  --d_RWtransactionsOut[std::this_thread::get_id()];
+}
+
+int MDBEnv::getRWTX()
+{
+  std::lock_guard<std::mutex> l(d_countmutex);
+  return d_RWtransactionsOut[std::this_thread::get_id()];
+}
+int MDBEnv::getROTX()
+{
+  std::lock_guard<std::mutex> l(d_countmutex);
+  return d_ROtransactionsOut[std::this_thread::get_id()];
+}
+
+
+std::shared_ptr<MDBEnv> getMDBEnv(const char* fname, int flags, int mode)
+{
+  struct Value
+  {
+    weak_ptr<MDBEnv> wp;
+    int flags;
+  };
+  
+  static std::map<tuple<dev_t, ino_t>, Value> s_envs;
+  static std::mutex mut;
+  
+  struct stat statbuf;
+  if(stat(fname, &statbuf)) {
+    if(errno != ENOENT)
+      throw std::runtime_error("Unable to stat prospective mdb database: "+string(strerror(errno)));
+    else {
+      std::lock_guard<std::mutex> l(mut);
+      auto fresh = std::make_shared<MDBEnv>(fname, flags, mode);
+      if(stat(fname, &statbuf))
+        throw std::runtime_error("Unable to stat prospective mdb database: "+string(strerror(errno)));
+      auto key = std::tie(statbuf.st_dev, statbuf.st_ino);
+      s_envs[key] = {fresh, flags};
+      return fresh;
+    }
+  }
+
+  std::lock_guard<std::mutex> l(mut);
+  auto key = std::tie(statbuf.st_dev, statbuf.st_ino);
+  auto iter = s_envs.find(key);
+  if(iter != s_envs.end()) {
+    auto sp = iter->second.wp.lock();
+    if(sp) {
+      if(iter->second.flags != flags)
+        throw std::runtime_error("Can't open mdb with differing flags");
+
+      return sp;
+    }
+    else {
+      s_envs.erase(iter); // useful if make_shared fails
+    }
+  }
+
+  auto fresh = std::make_shared<MDBEnv>(fname, flags, mode);
+  s_envs[key] = {fresh, flags};
+  
+  return fresh;
+}
+
+
+MDBDbi MDBEnv::openDB(const string_view dbname, int flags)
+{
+  unsigned int envflags;
+  mdb_env_get_flags(d_env, &envflags);
+  /*
+    This function must not be called from multiple concurrent transactions in the same process. A transaction that uses this function must finish (either commit or abort) before any other transaction in the process may use this function.
+  */
+  std::lock_guard<std::mutex> l(d_openmut);
+  
+  if(!(envflags & MDB_RDONLY)) {
+    auto rwt = getRWTransaction();
+    MDBDbi ret = rwt->openDB(dbname, flags);
+    rwt->commit();
+    return ret;
+  }
+
+  MDBDbi ret;
+  {
+    auto rwt = getROTransaction(); 
+    ret = rwt->openDB(dbname, flags);
+  }
+  return ret;
+}
+
+MDBRWTransactionImpl::MDBRWTransactionImpl(MDBEnv *parent, MDB_txn *txn):
+  MDBROTransactionImpl(parent, txn)
+
+{
+
+}
+
+MDB_txn *MDBRWTransactionImpl::openRWTransaction(MDBEnv *env, MDB_txn *parent, int flags)
+{
+  MDB_txn *result;
+  if(env->getROTX() || env->getRWTX())
+    throw std::runtime_error("Duplicate RW transaction");
+
+  if(int rc=mdb_txn_begin(env->d_env, parent, flags, &result))
+    throw std::runtime_error("Unable to start RW transaction: "+std::string(mdb_strerror(rc)));
+
+  env->incRWTX();
+  return result;
+}
+
+MDBRWTransactionImpl::MDBRWTransactionImpl(MDBEnv* parent, int flags):
+  MDBRWTransactionImpl(parent, openRWTransaction(parent, nullptr, flags))
+{
+}
+
+MDBRWTransactionImpl::~MDBRWTransactionImpl()
+{
+  abort();
+}
+
+void MDBRWTransactionImpl::commit()
+{
+  closeRORWCursors();
+  if (!d_txn) {
+    return;
+  }
+
+  if(int rc = mdb_txn_commit(d_txn)) {
+    throw std::runtime_error("committing: " + std::string(mdb_strerror(rc)));
+  }
+  environment().decRWTX();
+  d_txn = nullptr;
+}
+
+void MDBRWTransactionImpl::abort()
+{
+  closeRORWCursors();
+  if (!d_txn) {
+    return;
+  }
+
+  mdb_txn_abort(d_txn);
+  // prevent the RO destructor from cleaning up the transaction itself
+  environment().decRWTX();
+  d_txn = nullptr;
+}
+
+MDBROTransactionImpl::MDBROTransactionImpl(MDBEnv *parent, MDB_txn *txn):
+  d_parent(parent),
+  d_cursors(),
+  d_txn(txn)
+{
+
+}
+
+MDB_txn *MDBROTransactionImpl::openROTransaction(MDBEnv *env, MDB_txn *parent, int flags)
+{
+  if(env->getRWTX())
+    throw std::runtime_error("Duplicate RO transaction");
+  
+  /*
+    A transaction and its cursors must only be used by a single thread, and a thread may only have a single transaction at a time. If MDB_NOTLS is in use, this does not apply to read-only transactions. */
+  MDB_txn *result = nullptr;
+
+  if(int rc=mdb_txn_begin(env->d_env, parent, MDB_RDONLY | flags, &result))
+    throw std::runtime_error("Unable to start RO transaction: "+string(mdb_strerror(rc)));
+
+  env->incROTX();
+
+  return result;
+}
+
+void MDBROTransactionImpl::closeROCursors()
+{
+  // we need to move the vector away to ensure that the cursors don’t mess with our iteration.
+  std::vector<MDBROCursor*> buf;
+  std::swap(d_cursors, buf);
+  for (auto &cursor: buf) {
+    cursor->close();
+  }
+}
+
+MDBROTransactionImpl::MDBROTransactionImpl(MDBEnv *parent, int flags):
+    MDBROTransactionImpl(parent, openROTransaction(parent, nullptr, flags))
+{
+
+}
+
+MDBROTransactionImpl::~MDBROTransactionImpl()
+{
+  // this is safe because C++ will not call overrides of virtual methods in destructors.
+  commit();
+}
+
+void MDBROTransactionImpl::abort()
+{
+  closeROCursors();
+  // if d_txn is non-nullptr here, either the transaction object was invalidated earlier (e.g. by moving from it), or it is an RW transaction which has already cleaned up the d_txn pointer (with an abort).
+  if (d_txn) {
+    d_parent->decROTX();
+    mdb_txn_abort(d_txn); // this appears to work better than abort for r/o database opening
+    d_txn = nullptr;
+  }
+}
+
+void MDBROTransactionImpl::commit()
+{
+  closeROCursors();
+  // if d_txn is non-nullptr here, either the transaction object was invalidated earlier (e.g. by moving from it), or it is an RW transaction which has already cleaned up the d_txn pointer (with an abort).
+  if (d_txn) {
+    d_parent->decROTX();
+    mdb_txn_commit(d_txn); // this appears to work better than abort for r/o database opening
+    d_txn = nullptr;
+  }
+}
+
+
+
+void MDBRWTransactionImpl::clear(MDB_dbi dbi)
+{
+  if(int rc = mdb_drop(d_txn, dbi, 0)) {
+    throw runtime_error("Error clearing database: " + MDBError(rc));
+  }
+}
+
+MDBRWCursor MDBRWTransactionImpl::getRWCursor(const MDBDbi& dbi)
+{
+  MDB_cursor *cursor;
+  int rc= mdb_cursor_open(d_txn, dbi, &cursor);
+  if(rc) {
+    throw std::runtime_error("Error creating RO cursor: "+std::string(mdb_strerror(rc)));
+  }
+  return MDBRWCursor(d_rw_cursors, cursor);
+}
+
+MDBRWCursor MDBRWTransactionImpl::getCursor(const MDBDbi &dbi)
+{
+  return getRWCursor(dbi);
+}
+
+MDBRWTransaction MDBRWTransactionImpl::getRWTransaction()
+{
+  MDB_txn *txn;
+  if (int rc = mdb_txn_begin(environment(), *this, 0, &txn)) {
+    throw std::runtime_error(std::string("failed to start child transaction: ")+mdb_strerror(rc));
+  }
+  // we need to increase the counter here because commit/abort on the child transaction will decrease it
+  environment().incRWTX();
+  return MDBRWTransaction(new MDBRWTransactionImpl(&environment(), txn));
+}
+
+MDBROTransaction MDBRWTransactionImpl::getROTransaction()
+{
+  return getRWTransaction();
+}
+
+MDBROTransaction MDBEnv::getROTransaction()
+{
+  return MDBROTransaction(new MDBROTransactionImpl(this));
+}
+MDBRWTransaction MDBEnv::getRWTransaction()
+{
+  return MDBRWTransaction(new MDBRWTransactionImpl(this));
+}
+
+
+void MDBRWTransactionImpl::closeRWCursors()
+{
+  decltype(d_rw_cursors) buf;
+  std::swap(d_rw_cursors, buf);
+  for (auto &cursor: buf) {
+    cursor->close();
+  }
+}
+
+MDBROCursor MDBROTransactionImpl::getCursor(const MDBDbi& dbi)
+{
+  return getROCursor(dbi);
+}
+
+MDBROCursor MDBROTransactionImpl::getROCursor(const MDBDbi &dbi)
+{
+  MDB_cursor *cursor;
+  int rc= mdb_cursor_open(d_txn, dbi, &cursor);
+  if(rc) {
+    throw std::runtime_error("Error creating RO cursor: "+std::string(mdb_strerror(rc)));
+  }
+  return MDBROCursor(d_cursors, cursor);
+}
diff --git a/ext/lmdb-safe/lmdb-safe.hh b/ext/lmdb-safe/lmdb-safe.hh
new file mode 100644
index 0000000..6844909
--- /dev/null
+++ b/ext/lmdb-safe/lmdb-safe.hh
@@ -0,0 +1,610 @@
+#pragma once
+#include <lmdb.h>
+#include <iostream>
+#include <fstream>
+#include <set>
+#include <map>
+#include <thread>
+#include <memory>
+#include <string>
+#include <string.h>
+#include <mutex>
+#include <vector>
+#include <algorithm>
+
+using std::string_view;
+
+/* open issues:
+ *
+ * - missing convenience functions (string_view, string)
+ */ 
+
+/*
+The error strategy. Anything that "should never happen" turns into an exception. But things like 'duplicate entry' or 'no such key' are for you to deal with.
+ */
+
+/*
+  Thread safety: we are as safe as lmdb. You can talk to MDBEnv from as many threads as you want 
+*/
+
+/** MDBDbi is our only 'value type' object, as 1) a dbi is actually an integer
+    and 2) per LMDB documentation, we never close it. */
+class MDBDbi
+{
+public:
+  MDBDbi(): d_dbi(-1)
+  {
+  }
+  explicit MDBDbi(MDB_env* env, MDB_txn* txn, string_view dbname, int flags);  
+
+  operator const MDB_dbi&() const
+  {
+    return d_dbi;
+  }
+  
+  MDB_dbi d_dbi;
+};
+
+class MDBRWTransactionImpl;
+class MDBROTransactionImpl;
+
+using MDBROTransaction = std::unique_ptr<MDBROTransactionImpl>;
+using MDBRWTransaction = std::unique_ptr<MDBRWTransactionImpl>;
+
+class MDBEnv
+{
+public:
+  MDBEnv(const char* fname, int flags, int mode);
+
+  ~MDBEnv()
+  {
+    //    Only a single thread may call this function. All transactions, databases, and cursors must already be closed before calling this function
+    mdb_env_close(d_env);
+    // but, elsewhere, docs say database handles do not need to be closed?
+  }
+
+  MDBDbi openDB(const string_view dbname, int flags);
+  
+  MDBRWTransaction getRWTransaction();
+  MDBROTransaction getROTransaction();
+
+  operator MDB_env*& ()
+  {
+    return d_env;
+  }
+  MDB_env* d_env;
+
+  int getRWTX();
+  void incRWTX();
+  void decRWTX();
+  int getROTX();
+  void incROTX();
+  void decROTX();
+private:
+  std::mutex d_openmut;
+  std::mutex d_countmutex;
+  std::map<std::thread::id, int> d_RWtransactionsOut;
+  std::map<std::thread::id, int> d_ROtransactionsOut;
+};
+
+std::shared_ptr<MDBEnv> getMDBEnv(const char* fname, int flags, int mode);
+
+
+
+struct MDBOutVal
+{
+  operator MDB_val&()
+  {
+    return d_mdbval;
+  }
+
+  template <class T,
+          typename std::enable_if<std::is_arithmetic<T>::value,
+                                  T>::type* = nullptr> const
+  T get()
+  {
+    T ret;
+    if(d_mdbval.mv_size != sizeof(T))
+      throw std::runtime_error("MDB data has wrong length for type");
+    
+    memcpy(&ret, d_mdbval.mv_data, sizeof(T));
+    return ret;
+  }
+
+  template <class T,
+            typename std::enable_if<std::is_class<T>::value,T>::type* = nullptr>
+  T get() const;
+
+  template<class T>
+  T get_struct() const
+  {
+    T ret;
+    if(d_mdbval.mv_size != sizeof(T))
+      throw std::runtime_error("MDB data has wrong length for type");
+    
+    memcpy(&ret, d_mdbval.mv_data, sizeof(T));
+    return ret;
+  }
+
+  template<class T>
+  const T* get_struct_ptr() const
+  {
+    if(d_mdbval.mv_size != sizeof(T))
+      throw std::runtime_error("MDB data has wrong length for type");
+    
+    return reinterpret_cast<const T*>(d_mdbval.mv_data);
+  }
+  
+  
+  MDB_val d_mdbval;
+};
+
+template<> inline std::string MDBOutVal::get<std::string>() const
+{
+  return std::string((char*)d_mdbval.mv_data, d_mdbval.mv_size);
+}
+
+template<> inline string_view MDBOutVal::get<string_view>() const
+{
+  return string_view((char*)d_mdbval.mv_data, d_mdbval.mv_size);
+}
+
+class MDBInVal
+{
+public:
+  MDBInVal(const MDBOutVal& rhs): d_mdbval(rhs.d_mdbval)
+  {
+  }
+
+  template <class T,
+            typename std::enable_if<std::is_arithmetic<T>::value,
+                                    T>::type* = nullptr>
+  MDBInVal(T i) 
+  {
+    memcpy(&d_memory[0], &i, sizeof(i));
+    d_mdbval.mv_size = sizeof(T);
+    d_mdbval.mv_data = d_memory;;
+  }
+
+  MDBInVal(const char* s)
+  {
+    d_mdbval.mv_size = strlen(s);
+    d_mdbval.mv_data = (void*)s;
+  }
+  
+  MDBInVal(const string_view& v) 
+  {
+    d_mdbval.mv_size = v.size();
+    d_mdbval.mv_data = (void*)&v[0];
+  }
+
+  MDBInVal(const std::string& v) 
+  {
+    d_mdbval.mv_size = v.size();
+    d_mdbval.mv_data = (void*)&v[0];
+  }
+
+  
+  template<typename T>
+  static MDBInVal fromStruct(const T& t)
+  {
+    MDBInVal ret;
+    ret.d_mdbval.mv_size = sizeof(T);
+    ret.d_mdbval.mv_data = (void*)&t;
+    return ret;
+  }
+  
+  operator MDB_val&()
+  {
+    return d_mdbval;
+  }
+  MDB_val d_mdbval;
+private:
+  MDBInVal(){}
+  char d_memory[sizeof(double)];
+
+};
+
+
+
+
+class MDBROCursor;
+
+class MDBROTransactionImpl
+{
+protected:
+  MDBROTransactionImpl(MDBEnv *parent, MDB_txn *txn);
+
+private:
+  static MDB_txn *openROTransaction(MDBEnv *env, MDB_txn *parent, int flags=0);
+
+  MDBEnv* d_parent;
+  std::vector<MDBROCursor*> d_cursors;
+
+protected:
+  MDB_txn* d_txn;
+
+  void closeROCursors();
+
+public:
+  explicit MDBROTransactionImpl(MDBEnv* parent, int flags=0);
+
+  MDBROTransactionImpl(const MDBROTransactionImpl& src) = delete;
+  MDBROTransactionImpl &operator=(const MDBROTransactionImpl& src) = delete;
+
+  // The move constructor/operator cannot be made safe due to Object Slicing with MDBRWTransaction.
+  MDBROTransactionImpl(MDBROTransactionImpl&& rhs) = delete;
+  MDBROTransactionImpl &operator=(MDBROTransactionImpl &&rhs) = delete;
+
+  virtual ~MDBROTransactionImpl();
+
+  virtual void abort();
+  virtual void commit();
+
+  int get(MDB_dbi dbi, const MDBInVal& key, MDBOutVal& val)
+  {
+    if(!d_txn)
+      throw std::runtime_error("Attempt to use a closed RO transaction for get");
+
+    int rc = mdb_get(d_txn, dbi, const_cast<MDB_val*>(&key.d_mdbval),
+                     const_cast<MDB_val*>(&val.d_mdbval));
+    if(rc && rc != MDB_NOTFOUND)
+      throw std::runtime_error("getting data: " + std::string(mdb_strerror(rc)));
+    
+    return rc;
+  }
+
+  int get(MDB_dbi dbi, const MDBInVal& key, string_view& val)
+  {
+    MDBOutVal out;
+    int rc = get(dbi, key, out);
+    if(!rc)
+      val = out.get<string_view>();
+    return rc;
+  }
+
+  
+  // this is something you can do, readonly
+  MDBDbi openDB(string_view dbname, int flags)
+  {
+    return MDBDbi( d_parent->d_env, d_txn, dbname, flags);
+  }
+
+  MDBROCursor getCursor(const MDBDbi&);
+  MDBROCursor getROCursor(const MDBDbi&);
+    
+  operator MDB_txn*()
+  {
+    return d_txn;
+  }
+
+  inline operator bool() const {
+    return d_txn;
+  }
+
+  inline MDBEnv &environment()
+  {
+    return *d_parent;
+  }
+};
+
+/* 
+   A cursor in a read-only transaction must be closed explicitly, before or after its transaction ends. It can be reused with mdb_cursor_renew() before finally closing it. 
+
+   "If the parent transaction commits, the cursor must not be used again."
+*/
+
+template<class Transaction, class T>
+class MDBGenCursor
+{
+private:
+  std::vector<T*> *d_registry;
+  MDB_cursor* d_cursor{nullptr};
+
+public:
+  MDBGenCursor():
+    d_registry(nullptr),
+    d_cursor(nullptr)
+  {
+
+  }
+
+  MDBGenCursor(std::vector<T*> &registry, MDB_cursor *cursor):
+    d_registry(&registry),
+    d_cursor(cursor)
+  {
+    registry.emplace_back(static_cast<T*>(this));
+  }
+
+private:
+  void move_from(MDBGenCursor *src)
+  {
+    if (!d_registry) {
+      return;
+    }
+
+    auto iter = std::find(d_registry->begin(),
+                          d_registry->end(),
+                          src);
+    if (iter != d_registry->end()) {
+      *iter = static_cast<T*>(this);
+    } else {
+      d_registry->emplace_back(static_cast<T*>(this));
+    }
+  }
+
+public:
+  MDBGenCursor(const MDBGenCursor &src) = delete;
+
+  MDBGenCursor(MDBGenCursor &&src) noexcept:
+    d_registry(src.d_registry),
+    d_cursor(src.d_cursor)
+  {
+    move_from(&src);
+    src.d_registry = nullptr;
+    src.d_cursor = nullptr;
+  }
+
+  MDBGenCursor &operator=(const MDBGenCursor &src) = delete;
+
+  MDBGenCursor &operator=(MDBGenCursor &&src) noexcept
+  {
+    d_registry = src.d_registry;
+    d_cursor = src.d_cursor;
+    move_from(&src);
+    src.d_registry = nullptr;
+    src.d_cursor = nullptr;
+    return *this;
+  }
+
+  ~MDBGenCursor()
+  {
+    close();
+  }
+
+public:
+  int get(MDBOutVal& key, MDBOutVal& data, MDB_cursor_op op)
+  {
+    int rc = mdb_cursor_get(d_cursor, &key.d_mdbval, &data.d_mdbval, op);
+    if(rc && rc != MDB_NOTFOUND)
+       throw std::runtime_error("Unable to get from cursor: " + std::string(mdb_strerror(rc)));
+    return rc;
+  }
+
+  int find(const MDBInVal& in, MDBOutVal& key, MDBOutVal& data)
+  {
+    key.d_mdbval = in.d_mdbval;
+    int rc=mdb_cursor_get(d_cursor, const_cast<MDB_val*>(&key.d_mdbval), &data.d_mdbval, MDB_SET);
+    if(rc && rc != MDB_NOTFOUND)
+       throw std::runtime_error("Unable to find from cursor: " + std::string(mdb_strerror(rc)));
+    return rc;
+  }
+  
+  int lower_bound(const MDBInVal& in, MDBOutVal& key, MDBOutVal& data)
+  {
+    key.d_mdbval = in.d_mdbval;
+
+    int rc = mdb_cursor_get(d_cursor, const_cast<MDB_val*>(&key.d_mdbval), &data.d_mdbval, MDB_SET_RANGE);
+    if(rc && rc != MDB_NOTFOUND)
+       throw std::runtime_error("Unable to lower_bound from cursor: " + std::string(mdb_strerror(rc)));
+    return rc;
+  }
+
+  
+  int nextprev(MDBOutVal& key, MDBOutVal& data, MDB_cursor_op op)
+  {
+    int rc = mdb_cursor_get(d_cursor, const_cast<MDB_val*>(&key.d_mdbval), &data.d_mdbval, op);
+    if(rc && rc != MDB_NOTFOUND)
+       throw std::runtime_error("Unable to prevnext from cursor: " + std::string(mdb_strerror(rc)));
+    return rc;
+  }
+
+  int next(MDBOutVal& key, MDBOutVal& data)
+  {
+    return nextprev(key, data, MDB_NEXT);
+  }
+
+  int prev(MDBOutVal& key, MDBOutVal& data)
+  {
+    return nextprev(key, data, MDB_PREV);
+  }
+
+  int currentlast(MDBOutVal& key, MDBOutVal& data, MDB_cursor_op op)
+  {
+    int rc = mdb_cursor_get(d_cursor, const_cast<MDB_val*>(&key.d_mdbval), &data.d_mdbval, op);
+    if(rc && rc != MDB_NOTFOUND)
+       throw std::runtime_error("Unable to next from cursor: " + std::string(mdb_strerror(rc)));
+    return rc;
+  }
+
+  int current(MDBOutVal& key, MDBOutVal& data)
+  {
+    return currentlast(key, data, MDB_GET_CURRENT);
+  }
+  int last(MDBOutVal& key, MDBOutVal& data)
+  {
+    return currentlast(key, data, MDB_LAST);
+  }
+  int first(MDBOutVal& key, MDBOutVal& data)
+  {
+    return currentlast(key, data, MDB_FIRST);
+  }
+
+  operator MDB_cursor*()
+  {
+    return d_cursor;
+  }
+
+  operator bool() const
+  {
+    return d_cursor;
+  }
+
+  void close()
+  {
+    if (d_registry) {
+      auto iter = std::find(d_registry->begin(),
+                            d_registry->end(),
+                            static_cast<T*>(this));
+      if (iter != d_registry->end()) {
+        d_registry->erase(iter);
+      }
+      d_registry = nullptr;
+    }
+    if (d_cursor) {
+      mdb_cursor_close(d_cursor);
+      d_cursor = nullptr;
+    }
+  }
+};
+
+class MDBROCursor : public MDBGenCursor<MDBROTransactionImpl, MDBROCursor>
+{
+public:
+  MDBROCursor() = default;
+  using MDBGenCursor<MDBROTransactionImpl, MDBROCursor>::MDBGenCursor;
+  MDBROCursor(const MDBROCursor &src) = delete;
+  MDBROCursor(MDBROCursor &&src) = default;
+  MDBROCursor &operator=(const MDBROCursor &src) = delete;
+  MDBROCursor &operator=(MDBROCursor &&src) = default;
+  ~MDBROCursor() = default;
+
+};
+
+class MDBRWCursor;
+
+class MDBRWTransactionImpl: public MDBROTransactionImpl
+{
+protected:
+  MDBRWTransactionImpl(MDBEnv* parent, MDB_txn* txn);
+
+private:
+  static MDB_txn *openRWTransaction(MDBEnv* env, MDB_txn *parent, int flags);
+
+private:
+  std::vector<MDBRWCursor*> d_rw_cursors;
+
+  void closeRWCursors();
+  inline void closeRORWCursors() {
+    closeROCursors();
+    closeRWCursors();
+  }
+
+public:
+  explicit MDBRWTransactionImpl(MDBEnv* parent, int flags=0);
+
+  MDBRWTransactionImpl(const MDBRWTransactionImpl& rhs) = delete;
+  MDBRWTransactionImpl(MDBRWTransactionImpl&& rhs) = delete;
+  MDBRWTransactionImpl &operator=(const MDBRWTransactionImpl& rhs) = delete;
+  MDBRWTransactionImpl &operator=(MDBRWTransactionImpl&& rhs) = delete;
+
+  ~MDBRWTransactionImpl() override;
+  
+  void commit() override;
+  void abort() override;
+
+  void clear(MDB_dbi dbi);
+  
+  void put(MDB_dbi dbi, const MDBInVal& key, const MDBInVal& val, int flags=0)
+  {
+    if(!d_txn)
+      throw std::runtime_error("Attempt to use a closed RW transaction for put");
+    int rc;
+    if((rc=mdb_put(d_txn, dbi,
+                   const_cast<MDB_val*>(&key.d_mdbval),
+                   const_cast<MDB_val*>(&val.d_mdbval), flags)))
+      throw std::runtime_error("putting data: " + std::string(mdb_strerror(rc)));
+  }
+
+
+  int del(MDBDbi& dbi, const MDBInVal& key, const MDBInVal& val)
+  {
+    int rc;
+    rc=mdb_del(d_txn, dbi, (MDB_val*)&key.d_mdbval, (MDB_val*)&val.d_mdbval);
+    if(rc && rc != MDB_NOTFOUND)
+      throw std::runtime_error("deleting data: " + std::string(mdb_strerror(rc)));
+    return rc;
+  }
+
+  int del(MDBDbi& dbi, const MDBInVal& key)
+  {
+    int rc;
+    rc=mdb_del(d_txn, dbi, (MDB_val*)&key.d_mdbval, 0);
+    if(rc && rc != MDB_NOTFOUND)
+      throw std::runtime_error("deleting data: " + std::string(mdb_strerror(rc)));
+    return rc;
+  }
+
+ 
+  int get(MDBDbi& dbi, const MDBInVal& key, MDBOutVal& val)
+  {
+    if(!d_txn)
+      throw std::runtime_error("Attempt to use a closed RW transaction for get");
+
+    int rc = mdb_get(d_txn, dbi, const_cast<MDB_val*>(&key.d_mdbval),
+                     const_cast<MDB_val*>(&val.d_mdbval));
+    if(rc && rc != MDB_NOTFOUND)
+      throw std::runtime_error("getting data: " + std::string(mdb_strerror(rc)));
+    return rc;
+  }
+
+  int get(MDBDbi& dbi, const MDBInVal& key, string_view& val)
+  {
+    MDBOutVal out;
+    int rc = get(dbi, key, out);
+    if(!rc)
+      val = out.get<string_view>();
+    return rc;
+  }
+  
+  MDBDbi openDB(string_view dbname, int flags)
+  {
+    return MDBDbi(environment().d_env, d_txn, dbname, flags);
+  }
+
+  MDBRWCursor getRWCursor(const MDBDbi&);
+  MDBRWCursor getCursor(const MDBDbi&);
+
+  MDBRWTransaction getRWTransaction();
+  MDBROTransaction getROTransaction();
+};
+
+/* "A cursor in a write-transaction can be closed before its transaction ends, and will otherwise be closed when its transaction ends" 
+   This is a problem for us since it may means we are closing the cursor twice, which is bad
+*/
+class MDBRWCursor : public MDBGenCursor<MDBRWTransactionImpl, MDBRWCursor>
+{
+public:
+  MDBRWCursor() = default;
+  using MDBGenCursor<MDBRWTransactionImpl, MDBRWCursor>::MDBGenCursor;
+  MDBRWCursor(const MDBRWCursor &src) = delete;
+  MDBRWCursor(MDBRWCursor &&src) = default;
+  MDBRWCursor &operator=(const MDBRWCursor &src) = delete;
+  MDBRWCursor &operator=(MDBRWCursor &&src) = default;
+  ~MDBRWCursor() = default;
+
+  void put(const MDBOutVal& key, const MDBInVal& data)
+  {
+    int rc = mdb_cursor_put(*this,
+                            const_cast<MDB_val*>(&key.d_mdbval),
+                            const_cast<MDB_val*>(&data.d_mdbval), MDB_CURRENT);
+    if(rc)
+      throw std::runtime_error("mdb_cursor_put: " + std::string(mdb_strerror(rc)));
+  }
+
+  
+  int put(const MDBOutVal& key, const MDBOutVal& data, int flags=0)
+  {
+    // XXX check errors
+    return mdb_cursor_put(*this,
+                          const_cast<MDB_val*>(&key.d_mdbval),
+                          const_cast<MDB_val*>(&data.d_mdbval), flags);
+  }
+
+  int del(int flags=0)
+  {
+    return mdb_cursor_del(*this, flags);
+  }
+
+};
+