From: bert hubert <bert.hubert@netherlabs.nl>
Date: Tue, 5 Feb 2019 16:37:40 +0000 (+0100)
Subject: speed up pdnsutil list-zone significantly, plus load up the lmdbbackend files
X-Git-Tag: auth-4.2.0-beta1~37^2~3
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=6a40fc00256a97e0999542163454138184167e11;p=thirdparty%2Fpdns.git

speed up pdnsutil list-zone significantly, plus load up the lmdbbackend files
---

diff --git a/modules/lmdbbackend/Makefile.am b/modules/lmdbbackend/Makefile.am
new file mode 100644
index 0000000000..6ac36e9c3d
--- /dev/null
+++ b/modules/lmdbbackend/Makefile.am
@@ -0,0 +1,9 @@
+AM_CPPFLAGS += $(LMDB_CFLAGS)
+
+pkglib_LTLIBRARIES = liblmdbbackend.la
+
+EXTRA_DIST = OBJECTFILES OBJECTLIBS
+
+liblmdbbackend_la_SOURCES = lmdbbackend.cc lmdbbackend.hh lmdb-safe.hh lmdb-safe.cc lmdb-typed.hh lmdb-typed.cc
+liblmdbbackend_la_LDFLAGS = -module -avoid-version
+liblmdbbackend_la_LIBADD = $(LMDB_LIBS) $(BOOST_SERIALIZATION_LIBS)
\ No newline at end of file
diff --git a/modules/lmdbbackend/OBJECTFILES b/modules/lmdbbackend/OBJECTFILES
new file mode 100644
index 0000000000..02c8d578e5
--- /dev/null
+++ b/modules/lmdbbackend/OBJECTFILES
@@ -0,0 +1 @@
+lmdbbackend.lo lmdb-safe.lo lmdb-typed.lo
diff --git a/modules/lmdbbackend/OBJECTLIBS b/modules/lmdbbackend/OBJECTLIBS
new file mode 100644
index 0000000000..72a315ed28
--- /dev/null
+++ b/modules/lmdbbackend/OBJECTLIBS
@@ -0,0 +1 @@
+$(LMDB_LIBS) $(BOOST_SERIALIZATION_LIBS)
diff --git a/modules/lmdbbackend/lmdb-safe.cc b/modules/lmdbbackend/lmdb-safe.cc
new file mode 100644
index 0000000000..eea586455b
--- /dev/null
+++ b/modules/lmdbbackend/lmdb-safe.cc
@@ -0,0 +1,235 @@
+#include "lmdb-safe.hh"
+#include <fcntl.h>
+#include <mutex>
+#include <memory>
+#include <sys/stat.h>
+#include <string.h>
+#include <map>
+
+using namespace std;
+
+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);   
+  if(mdb_env_set_mapsize(d_env, 16ULL*4096*244140ULL)) // 4GB
+    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));
+  }
+}
+
+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;
+}
+
+MDBRWTransaction::MDBRWTransaction(MDBEnv* parent, int flags) : d_parent(parent)
+{
+  if(d_parent->getROTX() || d_parent->getRWTX())
+    throw std::runtime_error("Duplicate RW transaction");
+
+  for(int tries =0 ; tries < 3; ++tries) { // it might happen twice, who knows
+    if(int rc=mdb_txn_begin(d_parent->d_env, 0, flags, &d_txn)) {
+      if(rc == MDB_MAP_RESIZED && tries < 2) {
+        // "If the mapsize is increased by another process (..) mdb_txn_begin() will return MDB_MAP_RESIZED.
+        // call mdb_env_set_mapsize with a size of zero to adopt the new size."
+        mdb_env_set_mapsize(d_parent->d_env, 0);
+        continue;
+      }
+      throw std::runtime_error("Unable to start RW transaction: "+std::string(mdb_strerror(rc)));
+    }
+    break;
+  }
+  d_parent->incRWTX();
+}
+
+MDBROTransaction::MDBROTransaction(MDBEnv* parent, int flags) : d_parent(parent)
+{
+  if(d_parent->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. */
+  
+  for(int tries =0 ; tries < 3; ++tries) { // it might happen twice, who knows
+    if(int rc=mdb_txn_begin(d_parent->d_env, 0, MDB_RDONLY | flags, &d_txn)) {
+      if(rc == MDB_MAP_RESIZED && tries < 2) {
+        // "If the mapsize is increased by another process (..) mdb_txn_begin() will return MDB_MAP_RESIZED.
+        // call mdb_env_set_mapsize with a size of zero to adopt the new size."
+        mdb_env_set_mapsize(d_parent->d_env, 0);
+        continue;
+      }
+
+      throw std::runtime_error("Unable to start RO transaction: "+string(mdb_strerror(rc)));
+    }
+    break;
+  }
+  d_parent->incROTX();
+}
+
+
+
+void MDBRWTransaction::clear(MDB_dbi dbi)
+{
+  if(int rc = mdb_drop(d_txn, dbi, 0)) {
+    throw runtime_error("Error clearing database: " + MDBError(rc));
+  }
+}
+
+MDBRWCursor MDBRWTransaction::getCursor(const MDBDbi& dbi)
+{
+  return MDBRWCursor(this, dbi);
+}
+
+MDBROTransaction MDBEnv::getROTransaction()
+{
+  return MDBROTransaction(this);
+}
+MDBRWTransaction MDBEnv::getRWTransaction()
+{
+  return MDBRWTransaction(this);
+}
+
+
+void MDBRWTransaction::closeCursors()
+{
+  for(auto& c : d_cursors)
+    c->close();
+  d_cursors.clear();
+}
+
+MDBROCursor MDBROTransaction::getCursor(const MDBDbi& dbi)
+{
+  return MDBROCursor(this, dbi);
+}
+
+
diff --git a/modules/lmdbbackend/lmdb-safe.hh b/modules/lmdbbackend/lmdb-safe.hh
new file mode 100644
index 0000000000..de04d32085
--- /dev/null
+++ b/modules/lmdbbackend/lmdb-safe.hh
@@ -0,0 +1,613 @@
+#pragma once
+#include <lmdb.h>
+#include <iostream>
+#include <fstream>
+#include <set>
+#include <map>
+#include <thread>
+#include <memory>
+#include <string>
+#include <string.h>
+#include <mutex>
+
+// apple compiler somehow has string_view even in c++11!
+#if __cplusplus < 201703L && !defined(__APPLE__)
+#include <boost/version.hpp>
+#if BOOST_VERSION > 105400
+#include <boost/utility/string_view.hpp>
+using boost::string_view;
+#else
+#include <boost/utility/string_ref.hpp>
+using string_view = boost::string_ref;
+#endif
+#else // C++17
+using std::string_view;
+#endif
+
+
+/* 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 MDBRWTransaction;
+class MDBROTransaction;
+
+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;
+  }
+  
+  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 MDBROTransaction
+{
+public:
+  explicit MDBROTransaction(MDBEnv* parent, int flags=0);
+
+  MDBROTransaction(MDBROTransaction&& rhs)
+  {
+    d_parent = rhs.d_parent;
+    d_txn = rhs.d_txn;
+    rhs.d_parent = 0;
+    rhs.d_txn = 0;
+  }
+
+  void reset()
+  {
+    // this does not free cursors
+    mdb_txn_reset(d_txn);
+    d_parent->decROTX();
+  }
+
+  void renew()
+  {
+    if(d_parent->getROTX())
+      throw std::runtime_error("Duplicate RO transaction");
+    if(int rc = mdb_txn_renew(d_txn))
+      throw std::runtime_error("Renewing RO transaction: "+std::string(mdb_strerror(rc)));
+    d_parent->incROTX();
+  }
+  
+
+  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&);
+    
+  ~MDBROTransaction()
+  {
+    if(d_txn) {
+      d_parent->decROTX();
+      mdb_txn_commit(d_txn); // this appears to work better than abort for r/o database opening
+    }
+  }
+
+  operator MDB_txn*&()
+  {
+    return d_txn;
+  }
+  
+  MDBEnv* d_parent;
+  MDB_txn* d_txn;
+};
+
+/* 
+   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 MDBGenCursor
+{
+public:
+  MDBGenCursor(Transaction *t) : d_parent(t)
+  {}
+  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;
+  }
+
+  MDB_cursor* d_cursor;
+  Transaction* d_parent;
+};
+
+class MDBROCursor : public MDBGenCursor<MDBROTransaction>
+{
+public:
+  MDBROCursor(MDBROTransaction* parent, const MDB_dbi& dbi) : MDBGenCursor<MDBROTransaction>(parent)
+  {
+    int rc= mdb_cursor_open(d_parent->d_txn, dbi, &d_cursor);
+    if(rc) {
+      throw std::runtime_error("Error creating RO cursor: "+std::string(mdb_strerror(rc)));
+    }
+  }
+  MDBROCursor(MDBROCursor&& rhs) : MDBGenCursor<MDBROTransaction>(rhs.d_parent)
+  {
+    d_cursor = rhs.d_cursor;
+    rhs.d_cursor=0;
+  }
+
+  void close()
+  {
+    mdb_cursor_close(d_cursor);
+    d_cursor=0;
+  }
+  
+  ~MDBROCursor()
+  {
+    if(d_cursor)
+      mdb_cursor_close(d_cursor);
+  }
+
+};
+
+class MDBRWCursor;
+
+class MDBRWTransaction
+{
+public:
+  explicit MDBRWTransaction(MDBEnv* parent, int flags=0);
+
+  MDBRWTransaction(MDBRWTransaction&& rhs)
+  {
+    d_parent = rhs.d_parent;
+    d_txn = rhs.d_txn;
+    rhs.d_parent = 0;
+    rhs.d_txn = 0;
+  }
+
+  MDBRWTransaction& operator=(MDBRWTransaction&& rhs)
+  {
+    if(d_txn)
+      abort();
+
+    d_parent = rhs.d_parent;
+    d_txn = rhs.d_txn;
+    rhs.d_parent = 0;
+    rhs.d_txn = 0;
+    
+    return *this;
+  }
+
+  ~MDBRWTransaction()
+  {
+    if(d_txn) {
+      d_parent->decRWTX();
+      closeCursors();
+      mdb_txn_abort(d_txn); // XXX check response?
+    }
+  }
+  void closeCursors();
+  
+  void commit()
+  {
+    closeCursors();
+    if(int rc = mdb_txn_commit(d_txn)) {
+      throw std::runtime_error("committing: " + std::string(mdb_strerror(rc)));
+    }
+    d_parent->decRWTX();
+
+    d_txn=0;
+  }
+
+  void abort()
+  {
+    closeCursors();
+    mdb_txn_abort(d_txn); // XXX check error?
+    d_txn = 0;
+    d_parent->decRWTX();
+  }
+
+  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(d_parent->d_env, d_txn, dbname, flags);
+  }
+
+  MDBRWCursor getCursor(const MDBDbi&);
+
+  void reportCursor(MDBRWCursor* child)
+  {
+    d_cursors.insert(child);
+  }
+  void unreportCursor(MDBRWCursor* child)
+  {
+    d_cursors.erase(child);
+  }
+  
+  void reportCursorMove(MDBRWCursor* from, MDBRWCursor* to)
+  {
+    d_cursors.erase(from);
+    d_cursors.insert(to);
+  }
+  
+  operator MDB_txn*&()
+  {
+    return d_txn;
+  }
+
+
+  
+  std::set<MDBRWCursor*> d_cursors;
+  MDBEnv* d_parent;
+  MDB_txn* d_txn;
+};
+
+/* "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<MDBRWTransaction>
+{
+public:
+  MDBRWCursor(MDBRWTransaction* parent, const MDB_dbi& dbi) : MDBGenCursor<MDBRWTransaction>(parent)
+  {
+    int rc= mdb_cursor_open(d_parent->d_txn, dbi, &d_cursor);
+    if(rc) {
+      throw std::runtime_error("Error creating RW cursor: "+std::string(mdb_strerror(rc)));
+    }
+    d_parent->reportCursor(this);
+  }
+  MDBRWCursor(MDBRWCursor&& rhs) : MDBGenCursor<MDBRWTransaction>(rhs.d_parent)
+  {
+    d_cursor = rhs.d_cursor;
+    rhs.d_cursor=0;
+    d_parent->reportCursorMove(&rhs, this);
+  }
+
+  void close()
+  {
+    if(d_cursor)
+      mdb_cursor_close(d_cursor);
+    d_cursor=0;
+  }
+  
+  ~MDBRWCursor()
+  {
+    if(d_cursor)
+      mdb_cursor_close(d_cursor);
+    d_parent->unreportCursor(this);
+  }
+
+
+  void put(const MDBOutVal& key, const MDBInVal& data)
+  {
+    int rc = mdb_cursor_put(d_cursor,
+                          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(d_cursor,
+                          const_cast<MDB_val*>(&key.d_mdbval),
+                          const_cast<MDB_val*>(&data.d_mdbval), flags);
+  }
+
+  int del(int flags=0)
+  {
+    return mdb_cursor_del(d_cursor, flags);
+  }
+};
+
diff --git a/modules/lmdbbackend/lmdb-typed.cc b/modules/lmdbbackend/lmdb-typed.cc
new file mode 100644
index 0000000000..95206e3aac
--- /dev/null
+++ b/modules/lmdbbackend/lmdb-typed.cc
@@ -0,0 +1,14 @@
+#include "lmdb-typed.hh"
+
+unsigned int MDBGetMaxID(MDBRWTransaction& txn, MDBDbi& dbi)
+{
+  auto cursor = txn.getCursor(dbi);
+  MDBOutVal maxidval, maxcontent;
+  unsigned int maxid{0};
+  if(!cursor.get(maxidval, maxcontent, MDB_LAST)) {
+    maxid = maxidval.get<unsigned int>();
+  }
+  return maxid;
+}
+
+
diff --git a/modules/lmdbbackend/lmdb-typed.hh b/modules/lmdbbackend/lmdb-typed.hh
new file mode 100644
index 0000000000..a64dd1110a
--- /dev/null
+++ b/modules/lmdbbackend/lmdb-typed.hh
@@ -0,0 +1,723 @@
+#pragma once
+#include <iostream>
+#include "lmdb-safe.hh"
+#include <boost/archive/binary_oarchive.hpp>
+#include <boost/archive/binary_iarchive.hpp>
+#include <boost/serialization/vector.hpp>
+#include <boost/serialization/string.hpp>
+#include <boost/serialization/utility.hpp>
+#include <boost/iostreams/stream.hpp>
+#include <boost/iostreams/stream_buffer.hpp>
+#include <boost/iostreams/device/back_inserter.hpp>
+#include <sstream>
+// using std::cout;
+// using std::endl;
+
+
+/* 
+   Open issues:
+
+   Everything should go into a namespace
+   What is an error? What is an exception?
+   could id=0 be magic? ('no such id')
+     yes
+   Is boost the best serializer?
+     good default
+   Perhaps use the separate index concept from multi_index
+   perhaps get eiter to be of same type so for(auto& a : x) works
+     make it more value "like" with unique_ptr
+*/
+
+
+/** Return the highest ID used in a database. Returns 0 for an empty DB.
+    This makes us start everything at ID=1, which might make it possible to 
+    treat id 0 as special
+*/
+unsigned int MDBGetMaxID(MDBRWTransaction& txn, MDBDbi& dbi);
+
+/** This is our serialization interface.
+    You can define your own serToString for your type if you know better
+*/
+template<typename T>
+std::string serToString(const T& t)
+{
+  std::string serial_str;
+  boost::iostreams::back_insert_device<std::string> inserter(serial_str);
+  boost::iostreams::stream<boost::iostreams::back_insert_device<std::string> > s(inserter);
+  boost::archive::binary_oarchive oa(s, boost::archive::no_header | boost::archive::no_codecvt);
+  
+  oa << t;
+  return serial_str;
+}
+
+template<typename T>
+void serFromString(const string_view& str, T& ret)
+{
+  ret = T();
+
+  boost::iostreams::array_source source(&str[0], str.size());
+  boost::iostreams::stream<boost::iostreams::array_source> stream(source);
+  boost::archive::binary_iarchive in_archive(stream, boost::archive::no_header|boost::archive::no_codecvt);
+  in_archive >> ret;
+
+  /*
+  std::istringstream istr{str};
+  boost::archive::binary_iarchive oi(istr,boost::archive::no_header|boost::archive::no_codecvt );
+  oi >> ret;
+  */
+}
+
+
+template <class T, class Enable>
+inline std::string keyConv(const T& t);
+
+template <class T, typename std::enable_if<std::is_arithmetic<T>::value,T>::type* = nullptr>
+inline std::string keyConv(const T& t)
+{
+  return std::string((char*)&t, sizeof(t));
+}
+
+// this is how to override specific types.. it is ugly 
+template<class T, typename std::enable_if<std::is_same<T, std::string>::value,T>::type* = nullptr>
+inline std::string keyConv(const T& t)
+{
+  return t;
+}
+
+
+/** This is a struct that implements index operations, but 
+    only the operations that are broadcast to all indexes.
+    Specifically, to deal with databases with less than the maximum
+    number of interfaces, this only includes calls that should be
+    ignored for empty indexes.
+
+    this only needs methods that must happen for all indexes at once
+    so specifically, not size<t> or get<t>, people ask for those themselves, and
+    should no do that on indexes that don't exist */
+
+template<class Class,typename Type, typename Parent>
+struct LMDBIndexOps
+{
+  explicit LMDBIndexOps(Parent* parent) : d_parent(parent){}
+  void put(MDBRWTransaction& txn, const Class& t, uint32_t id, int flags=0)
+  {
+    txn.put(d_idx, keyConv(d_parent->getMember(t)), id, flags);
+  }
+
+  void del(MDBRWTransaction& txn, const Class& t, uint32_t id)
+  {
+    if(int rc = txn.del(d_idx, keyConv(d_parent->getMember(t)), id)) {
+      throw std::runtime_error("Error deleting from index: " + std::string(mdb_strerror(rc)));
+    }
+  }
+
+  void openDB(std::shared_ptr<MDBEnv>& env, string_view str, int flags)
+  {
+    d_idx = env->openDB(str, flags);
+  }
+  MDBDbi d_idx;
+  Parent* d_parent;
+};
+
+/** This is an index on a field in a struct, it derives from the LMDBIndexOps */
+
+template<class Class,typename Type,Type Class::*PtrToMember>
+struct index_on : LMDBIndexOps<Class, Type, index_on<Class, Type, PtrToMember>>
+{
+  index_on() : LMDBIndexOps<Class, Type, index_on<Class, Type, PtrToMember>>(this)
+  {}
+  static Type getMember(const Class& c)
+  {
+    return c.*PtrToMember;
+  }
+  
+  typedef Type type;
+};
+
+/** This is a calculated index */
+template<class Class, typename Type, class Func>
+struct index_on_function : LMDBIndexOps<Class, Type, index_on_function<Class, Type, Func> >
+{
+  index_on_function() : LMDBIndexOps<Class, Type, index_on_function<Class, Type, Func> >(this)
+  {}
+  static Type getMember(const Class& c)
+  {
+    Func f;
+    return f(c);
+  }
+
+  typedef Type type;           
+};
+
+/** nop index, so we can fill our N indexes, even if you don't use them all */
+struct nullindex_t
+{
+  template<typename Class>
+  void put(MDBRWTransaction& txn, const Class& t, uint32_t id, int flags=0)
+  {}
+  template<typename Class>
+  void del(MDBRWTransaction& txn, const Class& t, uint32_t id)
+  {}
+  
+  void openDB(std::shared_ptr<MDBEnv>& env, string_view str, int flags)
+  {
+    
+  }
+  typedef uint32_t type; // dummy
+};
+
+
+/** The main class. Templatized only on the indexes and typename right now */
+template<typename T, class I1=nullindex_t, class I2=nullindex_t, class I3 = nullindex_t, class I4 = nullindex_t>
+class TypedDBI
+{
+public:
+  TypedDBI(std::shared_ptr<MDBEnv> env, string_view name)
+    : d_env(env), d_name(name)
+  {
+    d_main = d_env->openDB(name, MDB_CREATE | MDB_INTEGERKEY);
+
+    // now you might be tempted to go all MPL on this so we can get rid of the
+    // ugly macro. I'm not very receptive to that idea since it will make things
+    // EVEN uglier.
+#define openMacro(N) std::get<N>(d_tuple).openDB(d_env, std::string(name)+"_"#N, MDB_CREATE | MDB_DUPFIXED | MDB_DUPSORT);
+    openMacro(0);
+    openMacro(1);
+    openMacro(2);
+    openMacro(3);
+#undef openMacro
+  }
+
+  
+  // we get a lot of our smarts from this tuple, it enables get<0> etc
+  typedef std::tuple<I1, I2, I3, I4> tuple_t; 
+  tuple_t d_tuple;
+
+  // We support readonly and rw transactions. Here we put the Readonly operations
+  // which get sourced by both kinds of transactions
+  template<class Parent>
+  struct ReadonlyOperations
+  {
+    ReadonlyOperations(Parent& parent) : d_parent(parent)
+    {}
+
+    //! Number of entries in main database
+    uint32_t size()
+    {
+      MDB_stat stat;
+      mdb_stat(*d_parent.d_txn, d_parent.d_parent->d_main, &stat);
+      return stat.ms_entries;
+    }
+
+    //! Number of entries in the various indexes - should be the same
+    template<int N>
+    uint32_t size()
+    {
+      MDB_stat stat;
+      mdb_stat(*d_parent.d_txn, std::get<N>(d_parent.d_parent->d_tuple).d_idx, &stat);
+      return stat.ms_entries;
+    }
+
+    //! Get item with id, from main table directly
+    bool get(uint32_t id, T& t)
+    {
+      MDBOutVal data;
+      if(d_parent.d_txn->get(d_parent.d_parent->d_main, id, data))
+        return false;
+      
+      serFromString(data.get<std::string>(), t);
+      return true;
+    }
+
+    //! Get item through index N, then via the main database
+    template<int N>
+    uint32_t get(const typename std::tuple_element<N, tuple_t>::type::type& key, T& out)
+    {
+      MDBOutVal id;
+      if(!d_parent.d_txn->get(std::get<N>(d_parent.d_parent->d_tuple).d_idx, keyConv(key), id)) {
+        if(get(id.get<uint32_t>(), out))
+          return id.get<uint32_t>();
+      }
+      return 0;
+    }
+
+    //! Cardinality of index N
+    template<int N>
+    uint32_t cardinality()
+    {
+      auto cursor = d_parent.d_txn->getCursor(std::get<N>(d_parent.d_parent->d_tuple).d_idx);
+      bool first = true;
+      MDBOutVal key, data;
+      uint32_t count = 0;
+      while(!cursor.get(key, data, first ? MDB_FIRST : MDB_NEXT_NODUP)) {
+        ++count;
+        first=false;
+      }
+      return count;
+    }
+
+    //! End iderator type
+    struct eiter_t
+    {};
+
+    // can be on main, or on an index
+    // when on main, return data directly
+    // when on index, indirect
+    // we can be limited to one key, or iterate over entire database
+    // iter requires you to put the cursor in the right place first!
+    struct iter_t
+    {
+      explicit iter_t(Parent* parent, typename Parent::cursor_t&& cursor, bool on_index, bool one_key, bool end=false) :
+        d_parent(parent),
+        d_cursor(std::move(cursor)),
+        d_on_index(on_index), // is this an iterator on main database or on index?
+        d_one_key(one_key),   // should we stop at end of key? (equal range)
+        d_end(end)
+      {
+        if(d_end)
+          return;
+        d_prefix.clear();
+        
+        if(d_cursor.get(d_key, d_id,  MDB_GET_CURRENT)) {
+          d_end = true;
+          return;
+        }
+
+        if(d_on_index) {
+          if(d_parent->d_txn->get(d_parent->d_parent->d_main, d_id, d_data))
+            throw std::runtime_error("Missing id in constructor");
+          serFromString(d_data.get<std::string>(), d_t);
+        }
+        else
+          serFromString(d_id.get<std::string>(), d_t);
+      }
+
+      explicit iter_t(Parent* parent, typename Parent::cursor_t&& cursor, const std::string& prefix) :
+        d_parent(parent),
+        d_cursor(std::move(cursor)),
+        d_on_index(true), // is this an iterator on main database or on index?
+        d_one_key(false),
+        d_prefix(prefix),  
+        d_end(false)
+      {
+        if(d_end)
+          return;
+
+        if(d_cursor.get(d_key, d_id,  MDB_GET_CURRENT)) {
+          d_end = true;
+          return;
+        }
+
+        if(d_on_index) {
+          if(d_parent->d_txn->get(d_parent->d_parent->d_main, d_id, d_data))
+            throw std::runtime_error("Missing id in constructor");
+          serFromString(d_data.get<std::string>(), d_t);
+        }
+        else
+          serFromString(d_id.get<std::string>(), d_t);
+      }
+
+      
+      std::function<bool(const MDBOutVal&)> filter;
+      void del()
+      {
+        d_cursor.del();
+      }
+      
+      bool operator!=(const eiter_t& rhs) const
+      {
+        return !d_end;
+      }
+      
+      bool operator==(const eiter_t& rhs) const
+      {
+        return d_end;
+      }
+      
+      const T& operator*()
+      {
+        return d_t;
+      }
+      
+      const T* operator->()
+      {
+        return &d_t;
+      }
+
+      // implements generic ++ or --
+      iter_t& genoperator(MDB_cursor_op dupop, MDB_cursor_op op)
+      {
+        MDBOutVal data;
+        int rc;
+      next:;
+        rc = d_cursor.get(d_key, d_id, d_one_key ? dupop : op);
+        if(rc == MDB_NOTFOUND) {
+          d_end = true;
+        }
+        else if(rc) {
+          throw std::runtime_error("in genoperator, " + std::string(mdb_strerror(rc)));
+        }
+        else if(!d_prefix.empty() && d_key.get<std::string>().rfind(d_prefix, 0)!=0) {
+          d_end = true;
+        }
+        else {
+          if(d_on_index) {
+            if(d_parent->d_txn->get(d_parent->d_parent->d_main, d_id, data))
+              throw std::runtime_error("Missing id field");
+            if(filter && !filter(data))
+              goto next;
+            
+            serFromString(data.get<std::string>(), d_t);
+          }
+          else {
+            if(filter && !filter(data))
+              goto next;
+                        
+            serFromString(d_id.get<std::string>(), d_t);
+          }
+        }
+        return *this;
+      }
+
+      iter_t& operator++()
+      {
+        return genoperator(MDB_NEXT_DUP, MDB_NEXT);
+      }
+      iter_t& operator--()
+      {
+        return genoperator(MDB_PREV_DUP, MDB_PREV);
+      }
+
+      // get ID this iterator points to
+      uint32_t getID()
+      {
+        if(d_on_index)
+          return d_id.get<uint32_t>();
+        else
+          return d_key.get<uint32_t>();
+      }
+
+      const MDBOutVal& getKey()
+      {
+        return d_key;
+      }
+      
+      
+      // transaction we are part of
+      Parent* d_parent;
+      typename Parent::cursor_t d_cursor;
+
+      // gcc complains if I don't zero-init these, which is worrying XXX
+      MDBOutVal d_key{{0,0}}, d_data{{0,0}}, d_id{{0,0}};
+      bool d_on_index;
+      bool d_one_key;
+      std::string d_prefix;
+      bool d_end{false};
+      T d_t;
+    };
+
+    template<int N>
+    iter_t genbegin(MDB_cursor_op op)
+    {
+      typename Parent::cursor_t cursor = d_parent.d_txn->getCursor(std::get<N>(d_parent.d_parent->d_tuple).d_idx);
+      
+      MDBOutVal out, id;
+      
+      if(cursor.get(out, id,  op)) {
+                                             // on_index, one_key, end
+        return iter_t{&d_parent, std::move(cursor), true, false, true};
+      }
+
+      return iter_t{&d_parent, std::move(cursor), true, false};
+    };
+
+    template<int N>
+    iter_t begin()
+    {
+      return genbegin<N>(MDB_FIRST);
+    }
+
+    template<int N>
+    iter_t rbegin()
+    {
+      return genbegin<N>(MDB_LAST);
+    }
+
+    iter_t begin()
+    {
+      typename Parent::cursor_t cursor = d_parent.d_txn->getCursor(d_parent.d_parent->d_main);
+      
+      MDBOutVal out, id;
+      
+      if(cursor.get(out, id,  MDB_FIRST)) {
+                                              // on_index, one_key, end        
+        return iter_t{&d_parent, std::move(cursor), false, false, true};
+      }
+
+      return iter_t{&d_parent, std::move(cursor), false, false};
+    };
+
+    eiter_t end()
+    {
+      return eiter_t();
+    }
+
+    // basis for find, lower_bound
+    template<int N>
+    iter_t genfind(const typename std::tuple_element<N, tuple_t>::type::type& key, MDB_cursor_op op)
+    {
+      typename Parent::cursor_t cursor = d_parent.d_txn->getCursor(std::get<N>(d_parent.d_parent->d_tuple).d_idx);
+
+      std::string keystr = keyConv(key);
+      MDBInVal in(keystr);
+      MDBOutVal out, id;
+      out.d_mdbval = in.d_mdbval;
+      
+      if(cursor.get(out, id,  op)) {
+                                              // on_index, one_key, end        
+        return iter_t{&d_parent, std::move(cursor), true, false, true};
+      }
+
+      return iter_t{&d_parent, std::move(cursor), true, false};
+    };
+
+    template<int N>
+    iter_t find(const typename std::tuple_element<N, tuple_t>::type::type& key)
+    {
+      return genfind<N>(key, MDB_SET);
+    }
+
+    template<int N>
+    iter_t lower_bound(const typename std::tuple_element<N, tuple_t>::type::type& key)
+    {
+      return genfind<N>(key, MDB_SET_RANGE);
+    }
+
+
+    //! equal range - could possibly be expressed through genfind
+    template<int N>
+    std::pair<iter_t,eiter_t> equal_range(const typename std::tuple_element<N, tuple_t>::type::type& key)
+    {
+      typename Parent::cursor_t cursor = d_parent.d_txn->getCursor(std::get<N>(d_parent.d_parent->d_tuple).d_idx);
+
+      std::string keyString=keyConv(key);
+      MDBInVal in(keyString);
+      MDBOutVal out, id;
+      out.d_mdbval = in.d_mdbval;
+      
+      if(cursor.get(out, id,  MDB_SET)) {
+                                              // on_index, one_key, end        
+        return {iter_t{&d_parent, std::move(cursor), true, true, true}, eiter_t()};
+      }
+
+      return {iter_t{&d_parent, std::move(cursor), true, true}, eiter_t()};
+    };
+
+    //! equal range - could possibly be expressed through genfind
+    template<int N>
+    std::pair<iter_t,eiter_t> prefix_range(const typename std::tuple_element<N, tuple_t>::type::type& key)
+    {
+      typename Parent::cursor_t cursor = d_parent.d_txn->getCursor(std::get<N>(d_parent.d_parent->d_tuple).d_idx);
+
+      std::string keyString=keyConv(key);
+      MDBInVal in(keyString);
+      MDBOutVal out, id;
+      out.d_mdbval = in.d_mdbval;
+      
+      if(cursor.get(out, id,  MDB_SET_RANGE)) {
+                                              // on_index, one_key, end        
+        return {iter_t{&d_parent, std::move(cursor), true, true, true}, eiter_t()};
+      }
+
+      return {iter_t(&d_parent, std::move(cursor), keyString), eiter_t()};
+    };
+
+    
+    Parent& d_parent;
+  };
+  
+  class ROTransaction : public ReadonlyOperations<ROTransaction>
+  {
+  public:
+    explicit ROTransaction(TypedDBI* parent) : ReadonlyOperations<ROTransaction>(*this), d_parent(parent), d_txn(std::make_shared<MDBROTransaction>(d_parent->d_env->getROTransaction())) 
+    {
+    }
+
+    explicit ROTransaction(TypedDBI* parent, std::shared_ptr<MDBROTransaction> txn) : ReadonlyOperations<ROTransaction>(*this), d_parent(parent), d_txn(txn) 
+    {
+    }
+
+    
+    ROTransaction(ROTransaction&& rhs) :
+      ReadonlyOperations<ROTransaction>(*this), d_parent(rhs.d_parent),d_txn(std::move(rhs.d_txn))
+      
+    {
+      rhs.d_parent = 0;
+    }
+
+    std::shared_ptr<MDBROTransaction> getTransactionHandle()
+    {
+      return d_txn;
+    }
+    
+    typedef MDBROCursor cursor_t;
+
+    TypedDBI* d_parent;
+    std::shared_ptr<MDBROTransaction> d_txn;    
+  };    
+
+  
+  class RWTransaction :  public ReadonlyOperations<RWTransaction>
+  {
+  public:
+    explicit RWTransaction(TypedDBI* parent) : ReadonlyOperations<RWTransaction>(*this), d_parent(parent)
+    {
+      d_txn = std::make_shared<MDBRWTransaction>(d_parent->d_env->getRWTransaction());
+    }
+
+    explicit RWTransaction(TypedDBI* parent, std::shared_ptr<MDBRWTransaction> txn) : ReadonlyOperations<RWTransaction>(*this), d_parent(parent), d_txn(txn)
+    {
+    }
+
+    
+    RWTransaction(RWTransaction&& rhs) :
+      ReadonlyOperations<RWTransaction>(*this),
+      d_parent(rhs.d_parent), d_txn(std::move(rhs.d_txn))
+    {
+      rhs.d_parent = 0;
+    }
+
+    // insert something, with possibly a specific id
+    uint32_t put(const T& t, uint32_t id=0)
+    {
+      int flags = 0;
+      if(!id) {
+        id = MDBGetMaxID(*d_txn, d_parent->d_main) + 1;
+        flags = MDB_APPEND;
+      }
+      d_txn->put(d_parent->d_main, id, serToString(t), flags);
+
+#define insertMacro(N) std::get<N>(d_parent->d_tuple).put(*d_txn, t, id);
+      insertMacro(0);
+      insertMacro(1);
+      insertMacro(2);
+      insertMacro(3);
+#undef insertMacro
+
+      return id;
+    }
+
+    // modify an item 'in place', plus update indexes
+    void modify(uint32_t id, std::function<void(T&)> func)
+    {
+      T t;
+      if(!this->get(id, t)) 
+        throw std::runtime_error("Could not modify id "+std::to_string(id));
+      func(t);
+      
+      del(id);  // this is the lazy way. We could test for changed index fields
+      put(t, id);
+    }
+
+    //! delete an item, and from indexes
+    void del(uint32_t id)
+    {
+      T t;
+      if(!this->get(id, t)) 
+        return;
+      
+      d_txn->del(d_parent->d_main, id);
+      clearIndex(id, t);
+    }
+
+    //! clear database & indexes (by hand!)
+    void clear()
+    {
+      auto cursor = d_txn->getCursor(d_parent->d_main);
+      bool first = true;
+      MDBOutVal key, data;
+      while(!cursor.get(key, data, first ? MDB_FIRST : MDB_NEXT)) {
+        first = false;
+        T t;
+        serFromString(data.get<std::string>(), t);
+        clearIndex(key.get<uint32_t>(), t);
+        cursor.del();
+      }
+    }
+
+    //! commit this transaction
+    void commit()
+    {
+      d_txn->commit();
+    }
+
+    //! abort this transaction
+    void abort()
+    {
+      d_txn->abort();
+    }
+
+    typedef MDBRWCursor cursor_t;
+
+    std::shared_ptr<MDBRWTransaction> getTransactionHandle()
+    {
+      return d_txn;
+    }
+
+    
+  private:
+    // clear this ID from all indexes
+    void clearIndex(uint32_t id, const T& t)
+    {
+#define clearMacro(N) std::get<N>(d_parent->d_tuple).del(*d_txn, t, id);
+      clearMacro(0);
+      clearMacro(1);
+      clearMacro(2);
+      clearMacro(3);
+#undef clearMacro      
+    }
+
+  public:
+    TypedDBI* d_parent;
+    std::shared_ptr<MDBRWTransaction> d_txn;
+  };
+
+  //! Get an RW transaction
+  RWTransaction getRWTransaction()
+  {
+    return RWTransaction(this);
+  }
+
+  //! Get an RO transaction
+  ROTransaction getROTransaction()
+  {
+    return ROTransaction(this);
+  }
+
+  //! Get an RW transaction
+  RWTransaction getRWTransaction(std::shared_ptr<MDBRWTransaction> txn)
+  {
+    return RWTransaction(this, txn);
+  }
+
+  //! Get an RO transaction
+  ROTransaction getROTransaction(std::shared_ptr<MDBROTransaction> txn)
+  {
+    return ROTransaction(this, txn);
+  }
+
+  std::shared_ptr<MDBEnv> getEnv()
+  {
+    return d_env;
+  }
+  
+private:
+  std::shared_ptr<MDBEnv> d_env;
+  MDBDbi d_main;
+  std::string d_name;
+};
+
+
+
+
+
diff --git a/modules/lmdbbackend/lmdbbackend.cc b/modules/lmdbbackend/lmdbbackend.cc
new file mode 100644
index 0000000000..9ca3e29e62
--- /dev/null
+++ b/modules/lmdbbackend/lmdbbackend.cc
@@ -0,0 +1,1606 @@
+/*
+ * This file is part of PowerDNS or dnsdist.
+ * Copyright -- PowerDNS.COM B.V. and its contributors
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of version 2 of the GNU General Public License as
+ * published by the Free Software Foundation.
+ *
+ * In addition, for the avoidance of any doubt, permission is granted to
+ * link this program with OpenSSL and to (re)distribute the binaries
+ * produced as the result of such linking.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+ */
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+#include "pdns/utility.hh"
+#include "pdns/dnsbackend.hh"
+#include "pdns/dns.hh"
+#include "pdns/dnspacket.hh"
+#include "pdns/base32.hh"
+#include "pdns/dnssecinfra.hh"
+#include "pdns/pdnsexception.hh"
+#include "pdns/logger.hh"
+#include "pdns/version.hh"
+#include "pdns/arguments.hh"
+#include <boost/archive/binary_oarchive.hpp>
+#include <boost/archive/binary_iarchive.hpp>
+#include <boost/serialization/vector.hpp>
+#include <boost/serialization/string.hpp>
+#include <boost/serialization/utility.hpp>
+// #include <boost/iostreams/stream.hpp>
+// #include <boost/iostreams/stream_buffer.hpp>
+
+#include <boost/iostreams/device/back_inserter.hpp>
+// #include <sstream>
+
+
+#include "lmdbbackend.hh"
+
+LMDBBackend::LMDBBackend(const std::string& suffix)
+{
+  setArgPrefix("lmdb"+suffix);
+  
+  string syncMode = toLower(getArg("sync-mode"));
+
+  if(syncMode == "nosync")
+    d_asyncFlag = MDB_NOSYNC;
+  else if(syncMode == "nometasync")
+    d_asyncFlag = MDB_NOMETASYNC;
+  else if(syncMode == "mapasync")
+    d_asyncFlag = MDB_MAPASYNC;
+  else if(syncMode.empty())
+    d_asyncFlag = 0;
+  else
+    throw std::runtime_error("Unknown sync mode "+syncMode+" requested for LMDB backend");
+
+  d_tdomains = std::make_shared<tdomains_t>(getMDBEnv(getArg("filename").c_str(), MDB_NOSUBDIR | d_asyncFlag, 0600), "domains");
+  d_tmeta = std::make_shared<tmeta_t>(d_tdomains->getEnv(), "metadata");
+  d_tkdb = std::make_shared<tkdb_t>(d_tdomains->getEnv(), "keydata");
+  d_ttsig = std::make_shared<ttsig_t>(d_tdomains->getEnv(), "tsig");
+  
+  auto pdnsdbi = d_tdomains->getEnv()->openDB("pdns", MDB_CREATE);
+  auto txn = d_tdomains->getEnv()->getRWTransaction();
+  MDBOutVal shards;
+  if(!txn.get(pdnsdbi, "shards", shards)) {
+    
+    d_shards = shards.get<uint32_t>();
+    if(d_shards != atoi(getArg("shards").c_str())) {
+      g_log << Logger::Warning<<"Note: configured number of lmdb shards ("<<atoi(getArg("shards").c_str())<<") is different from on-disk ("<<d_shards<<"). Using on-disk shard number"<<endl;
+    }
+  }
+  else {
+    d_shards = atoi(getArg("shards").c_str());
+    txn.put(pdnsdbi, "shards", d_shards);
+    txn.commit();
+  }
+    d_trecords.resize(d_shards);
+  d_dolog = ::arg().mustDo("query-logging");
+}
+
+
+
+namespace boost {
+namespace serialization {
+
+template<class Archive>
+void save(Archive & ar, const DNSName& g, const unsigned int version)
+{
+  if(!g.empty()) {
+    std::string tmp = g.toDNSStringLC(); // g++ 4.8 woes
+    ar & tmp;
+  }
+  else
+    ar & "";
+}
+
+template<class Archive>
+void load(Archive & ar, DNSName& g, const unsigned int version)
+{
+  string tmp;
+  ar & tmp;
+  if(tmp.empty())
+    g = DNSName();
+  else
+    g = DNSName(tmp.c_str(), tmp.size(), 0, false);
+}
+
+template<class Archive>
+void save(Archive & ar, const QType& g, const unsigned int version)
+{
+  uint16_t tmp = g.getCode(); // g++ 4.8 woes
+  ar & tmp;
+}
+
+template<class Archive>
+void load(Archive & ar, QType& g, const unsigned int version)
+{
+  uint16_t tmp;
+  ar & tmp; 
+  g = QType(tmp);
+}
+  
+template<class Archive>
+void serialize(Archive & ar, DomainInfo& g, const unsigned int version)
+{
+  ar & g.zone;
+  ar & g.last_check;
+  ar & g.account;
+  ar & g.masters;
+  ar & g.id;
+  ar & g.notified_serial;
+  ar & g.kind;
+}
+
+template<class Archive>
+void serialize(Archive & ar, LMDBBackend::DomainMeta& g, const unsigned int version)
+{
+  ar & g.domain & g.key & g.value;
+}
+
+template<class Archive>
+void serialize(Archive & ar, LMDBBackend::KeyDataDB& g, const unsigned int version)
+{
+  ar & g.domain & g.content & g.flags & g.active;
+}
+
+template<class Archive>
+void serialize(Archive & ar, TSIGKey& g, const unsigned int version)
+{
+  ar & g.name;
+  ar & g.algorithm; // this is the ordername
+  ar & g.key;
+}
+
+
+  
+} // namespace serialization
+} // namespace boost
+
+BOOST_SERIALIZATION_SPLIT_FREE(DNSName);
+BOOST_SERIALIZATION_SPLIT_FREE(QType);
+BOOST_IS_BITWISE_SERIALIZABLE(ComboAddress);
+
+template<>
+std::string serToString(const DNSResourceRecord& rr)
+{
+  // only does content, ttl, auth
+  std::string ret;
+  uint16_t len = rr.content.length();
+  ret.reserve(2+len+8);
+  
+  ret.assign((const char*)&len, 2);
+  ret += rr.content;
+  ret.append((const char*)&rr.ttl, 4);
+  ret.append(1, (char)rr.auth);
+  ret.append(1, (char)rr.disabled);
+  return ret;
+}
+
+template<>
+void serFromString(const string_view& str, DNSResourceRecord& rr)
+{
+  uint16_t len;
+  memcpy(&len, &str[0], 2);
+  rr.content.assign(&str[2], len);    // len bytes
+  memcpy(&rr.ttl, &str[2] + len, 4);
+  rr.auth = str[str.size()-2];
+  rr.disabled = str[str.size()-1];
+  rr.wildcardname.clear();
+}
+
+
+std::string serializeContent(uint16_t qtype, const DNSName& domain, const std::string& content)
+{
+  auto drc = DNSRecordContent::mastermake(qtype, 1, content);
+  return drc->serialize(domain, false);
+}
+
+std::shared_ptr<DNSRecordContent> unserializeContentZR(uint16_t qtype, const DNSName& qname, const std::string& content)
+{
+  if(qtype == QType::A && content.size() == 4) {
+    return std::make_shared<ARecordContent>(*((uint32_t*)content.c_str()));
+  }
+  return DNSRecordContent::unserialize(qname, qtype, content);
+}
+
+
+/* design. If you ask a question without a zone id, we lookup the best
+   zone id for you, and answer from that. This is different than other backends, but I can't see why it would not work.
+
+   The index we use is "zoneid,canonical relative name". This index is also used
+   for AXFR.
+
+   Note - domain_id, name and type are ONLY present on the index!
+*/
+
+#if BOOST_VERSION <= 105400
+#define StringView string
+#else
+#define StringView string_view
+#endif
+
+void LMDBBackend::deleteDomainRecords(RecordsRWTransaction& txn, uint32_t domain_id, uint16_t qtype)
+{
+  compoundOrdername co;
+  string match = co(domain_id);
+
+  auto cursor = txn.txn.getCursor(txn.db->dbi);
+  MDBOutVal key, val;
+  //  cout<<"Match: "<<makeHexDump(match);
+  if(!cursor.lower_bound(match, key, val) ) {
+    while(key.get<StringView>().rfind(match, 0) == 0) {
+      if(qtype == QType::ANY || co.getQType(key.get<StringView>()) == qtype)
+        cursor.del(MDB_NODUPDATA);
+      if(cursor.next(key, val)) break;
+    } 
+  }
+}
+
+/* Here's the complicated story. Other backends have just one transaction, which is either
+   on or not. 
+   
+   You can't call feedRecord without a transaction started with startTransaction.
+
+   However, other functions can be called after startTransaction() or without startTransaction()
+     (like updateDNSSECOrderNameAndAuth)
+
+
+
+*/
+
+bool LMDBBackend::startTransaction(const DNSName &domain, int domain_id)
+{
+  // cout <<"startTransaction("<<domain<<", "<<domain_id<<")"<<endl;
+  int real_id = domain_id;
+  if(real_id < 0) {
+    auto rotxn = d_tdomains->getROTransaction();
+    DomainInfo di;
+    real_id = rotxn.get<0>(domain, di);
+    // cout<<"real_id = "<<real_id << endl;
+    if(!real_id)
+      return false;
+  }
+  if(d_rwtxn) {
+    throw DBException("Attempt to start a transaction while one was open already");
+  }
+  d_rwtxn = getRecordsRWTransaction(real_id);
+    
+  d_transactiondomain = domain;
+  d_transactiondomainid = real_id;
+  if(domain_id >= 0) {
+    deleteDomainRecords(*d_rwtxn, domain_id);
+  }
+
+  return true;
+}
+
+bool LMDBBackend::commitTransaction()
+{
+  // cout<<"Commit transaction" <<endl;
+  d_rwtxn->txn.commit();
+  d_rwtxn.reset();
+  return true;
+}
+
+bool LMDBBackend::abortTransaction()
+{
+  // cout<<"Abort transaction"<<endl;
+  d_rwtxn->txn.abort();
+  d_rwtxn.reset();
+
+  return true;
+}
+
+// d_rwtxn must be set here
+bool LMDBBackend::feedRecord(const DNSResourceRecord &r, const DNSName &ordername)
+{
+  DNSResourceRecord rr(r);
+  rr.qname.makeUsRelative(d_transactiondomain);
+  rr.content = serializeContent(rr.qtype.getCode(), r.qname, rr.content);
+
+  compoundOrdername co;
+  d_rwtxn->txn.put(d_rwtxn->db->dbi, co(r.domain_id, rr.qname, rr.qtype.getCode()), serToString(rr));
+
+  if(!ordername.empty()) {
+      rr.ttl = 0;
+      rr.auth = 0;
+      rr.content=rr.qname.toDNSStringLC();
+      string ser = serToString(rr);
+      d_rwtxn->txn.put(d_rwtxn->db->dbi, co(r.domain_id, ordername, QType::NSEC3), ser);
+
+      rr.ttl = 1;
+      rr.content = ordername.toDNSString();
+      ser = serToString(rr);
+      d_rwtxn->txn.put(d_rwtxn->db->dbi, co(r.domain_id, rr.qname, QType::NSEC3), ser);
+  }
+  return true;
+}
+
+bool LMDBBackend::feedEnts(int domain_id, map<DNSName,bool>& nonterm)
+{
+  DNSResourceRecord rr;
+  rr.ttl = 0;
+  compoundOrdername co;
+  for(const auto& nt: nonterm) {
+    rr.qname = nt.first.makeRelative(d_transactiondomain);
+    rr.auth = nt.second;
+    std::string ser = serToString(rr);
+
+    d_rwtxn->txn.put(d_rwtxn->db->dbi, co(domain_id, rr.qname, 0), ser);
+  }
+  return true;
+}
+
+bool LMDBBackend::feedEnts3(int domain_id, const DNSName &domain, map<DNSName,bool> &nonterm, const NSEC3PARAMRecordContent& ns3prc, bool narrow)
+{
+  DNSName ordername;
+  DNSResourceRecord rr;
+  compoundOrdername co;
+  for(const auto& nt: nonterm) {
+    rr.qname = nt.first.makeRelative(domain);
+    rr.ttl = 0;
+    rr.auth = nt.second;
+    rr.disabled = true;
+    string ser = serToString(rr);
+
+    d_rwtxn->txn.put(d_rwtxn->db->dbi, co(domain_id, rr.qname, 0), ser);
+
+    if(!narrow && rr.auth) {
+      rr.auth=0;
+      rr.content=rr.qname.toDNSString();
+      ser = serToString(rr);
+
+      ordername=DNSName(toBase32Hex(hashQNameWithSalt(ns3prc, nt.first)));
+      d_rwtxn->txn.put(d_rwtxn->db->dbi, co(domain_id, ordername, QType::NSEC3), ser);
+
+      rr.ttl = 1;
+      rr.content = ordername.toDNSString();
+      ser = serToString(rr);
+      d_rwtxn->txn.put(d_rwtxn->db->dbi, co(domain_id, rr.qname, QType::NSEC3), ser);
+    }
+  }
+  return true;
+}
+
+
+// might be called within a transaction, might also be called alone
+bool LMDBBackend::replaceRRSet(uint32_t domain_id, const DNSName& qname, const QType& qt, const vector<DNSResourceRecord>& rrset)
+{
+  // zonk qname/qtype within domain_id (go through qname, check domain_id && qtype)
+  shared_ptr<RecordsRWTransaction> txn;
+  bool needCommit = false;
+  if(d_rwtxn && d_transactiondomainid==domain_id) {
+    txn = d_rwtxn;
+    //    cout<<"Reusing open transaction"<<endl;
+  }
+  else {
+    //    cout<<"Making a new RW txn for replace rrset"<<endl;
+    txn = getRecordsRWTransaction(domain_id);
+    needCommit = true;
+  }
+
+  DomainInfo di;
+  d_tdomains->getROTransaction().get(domain_id, di); // XX error checking
+  
+  compoundOrdername co;
+  auto cursor = txn->txn.getCursor(txn->db->dbi);
+  MDBOutVal key, val;
+  string match =co(domain_id, qname.makeRelative(di.zone), qt.getCode());
+  if(!cursor.find(match, key, val)) {
+    do {
+      cursor.del(MDB_NODUPDATA);
+    } while(!cursor.next(key, val) && key.get<StringView>().rfind(match, 0) == 0);
+  }
+
+  for(auto rr : rrset) {
+    rr.content = serializeContent(rr.qtype.getCode(), rr.qname, rr.content);
+    rr.qname.makeUsRelative(di.zone);
+    txn->txn.put(txn->db->dbi, match, serToString(rr));
+  }
+
+  if(needCommit)
+    txn->txn.commit();
+
+  return true;
+}
+
+// tempting to templatize these two functions but the pain is not worth it
+std::shared_ptr<LMDBBackend::RecordsRWTransaction> LMDBBackend::getRecordsRWTransaction(uint32_t id)
+{
+  auto& shard =d_trecords[id % d_shards];
+  if(!shard.env) {
+    shard.env = getMDBEnv( (getArg("filename")+"-"+std::to_string(id % d_shards)).c_str(),
+                           MDB_NOSUBDIR | d_asyncFlag, 0600);
+    shard.dbi = shard.env->openDB("records", MDB_CREATE | MDB_DUPSORT);
+  }
+  auto ret = std::make_shared<RecordsRWTransaction>(shard.env->getRWTransaction());
+  ret->db = std::make_shared<RecordsDB>(shard);
+
+  return ret;
+}
+
+std::shared_ptr<LMDBBackend::RecordsROTransaction> LMDBBackend::getRecordsROTransaction(uint32_t id)
+{
+  auto& shard =d_trecords[id % d_shards];
+  if(!shard.env) {
+    shard.env = getMDBEnv( (getArg("filename")+"-"+std::to_string(id % d_shards)).c_str(),
+                           MDB_NOSUBDIR | d_asyncFlag, 0600);
+    shard.dbi = shard.env->openDB("records", MDB_CREATE | MDB_DUPSORT);
+  }
+  
+  auto ret = std::make_shared<RecordsROTransaction>(shard.env->getROTransaction());
+  ret->db = std::make_shared<RecordsDB>(shard);
+  return ret;
+}
+
+
+bool LMDBBackend::deleteDomain(const DNSName &domain)
+{
+  auto doms = d_tdomains->getRWTransaction();
+
+  DomainInfo di;
+  auto id = doms.get<0>(domain, di); 
+  if(!id)
+    return false;
+  
+  shared_ptr<RecordsRWTransaction> txn;
+  bool needCommit = false;
+  if(d_rwtxn && d_transactiondomainid == id) {
+    txn = d_rwtxn;
+    //    cout<<"Reusing open transaction"<<endl;
+  }
+  else {
+    //    cout<<"Making a new RW txn for delete domain"<<endl;
+    txn = getRecordsRWTransaction(id);
+    needCommit = true;
+  }
+
+  
+  doms.del(id);
+  compoundOrdername co;
+  string match=co(id);
+
+  auto cursor = txn->txn.getCursor(txn->db->dbi);
+  MDBOutVal key, val;
+  if(!cursor.find(match, key, val)) {
+    do {
+      cursor.del(MDB_NODUPDATA);
+    } while(!cursor.next(key, val) && key.get<StringView>().rfind(match, 0) == 0);
+  }
+
+  if(needCommit)
+    txn->txn.commit();
+  
+  doms.commit();
+
+  return true;
+}
+
+bool LMDBBackend::list(const DNSName &target, int id, bool include_disabled)
+{
+  d_inlist=true;
+  DomainInfo di;
+  {
+    auto dtxn = d_tdomains->getROTransaction();
+    
+    if((di.id = dtxn.get<0>(target, di))) 
+      ; //      cout<<"Found domain "<<target<<" on domain_id "<<di.id <<", list requested "<<id<<endl;
+    else {
+      // cout<<"Did not find "<<target<<endl;
+      return false;
+    }
+  }
+  
+  d_rotxn = getRecordsROTransaction(di.id);
+  compoundOrdername co;
+  d_matchkey = co(di.id);
+  d_getcursor = std::make_shared<MDBROCursor>(d_rotxn->txn.getCursor(d_rotxn->db->dbi));
+  MDBOutVal key, val;
+  d_inlist = true;
+  
+  if(d_getcursor->lower_bound(d_matchkey, key, val) || key.get<StringView>().rfind(d_matchkey, 0) != 0) {
+    // cout<<"Found nothing for list"<<endl;
+    d_getcursor.reset();
+    return true;
+  }
+  
+  d_lookupqname = target;
+  
+  return true;
+}
+
+void LMDBBackend::lookup(const QType &type, const DNSName &qdomain, DNSPacket *p, int zoneId)
+{
+  if(d_dolog) {
+    g_log << Logger::Warning << "Got lookup for "<<qdomain<<"|"<<type.getName()<<" in zone "<< zoneId<<endl;
+    d_dtime.set();
+  }
+  DNSName hunt(qdomain);
+  if(zoneId < 0) {
+    auto rotxn = d_tdomains->getROTransaction();
+    
+    for(;;) {
+      DomainInfo di;
+      if((zoneId = rotxn.get<0>(hunt, di))) {
+        break;
+      }
+      if(!hunt.chopOff())
+        break;
+    }
+    if(zoneId <= 0) {
+      //      cout << "Did not find zone for "<< qdomain<<endl;
+      d_getcursor.reset();
+      return;
+    }
+  }
+  else {
+    DomainInfo di;
+    if(!d_tdomains->getROTransaction().get(zoneId, di)) {
+      // cout<<"Could not find a zone with id "<<zoneId<<endl;
+      d_getcursor.reset();
+      return;
+    }
+    hunt = di.zone;
+  }
+    
+  DNSName relqname = qdomain.makeRelative(hunt);
+  //  cout<<"get will look for "<<relqname<< " in zone "<<hunt<<" with id "<<zoneId<<endl;
+  d_rotxn = getRecordsROTransaction(zoneId);
+
+  compoundOrdername co;
+  d_getcursor = std::make_shared<MDBROCursor>(d_rotxn->txn.getCursor(d_rotxn->db->dbi));
+  MDBOutVal key, val;
+  if(type.getCode() == QType::ANY) {
+    d_matchkey = co(zoneId,relqname);
+  }
+  else {
+    d_matchkey= co(zoneId,relqname, type.getCode());
+  }
+  d_inlist=false;
+  
+  if(d_getcursor->lower_bound(d_matchkey, key, val) || key.get<StringView>().rfind(d_matchkey, 0) != 0) {
+    d_getcursor.reset();
+    if(d_dolog) {
+      g_log<<Logger::Warning<< "Query "<<((long)(void*)this)<<": "<<d_dtime.udiffNoReset()<<" usec to execute (found nothing)"<<endl;
+    }
+    return;
+  }
+  
+  if(d_dolog) {
+    g_log<<Logger::Warning<< "Query "<<((long)(void*)this)<<": "<<d_dtime.udiffNoReset()<<" usec to execute"<<endl;
+  }
+    
+  d_lookuptype=type;
+  d_lookupqname = qdomain;
+  d_lookupdomain = hunt;
+  d_lookupdomainid = zoneId;
+}
+
+bool LMDBBackend::get(DNSZoneRecord& rr)
+{
+  if(d_inlist)
+    return get_list(rr);
+  else
+    return get_lookup(rr);
+}
+
+bool LMDBBackend::get(DNSResourceRecord& rr)
+{
+  //  cout <<"Old-school get called"<<endl;
+  DNSZoneRecord dzr;
+  if(d_inlist) {
+    if(!get_list(dzr))
+      return false;
+  }
+  else {
+    if(!get_lookup(dzr))
+      return false;
+  }
+  rr.qname = dzr.dr.d_name;
+  rr.ttl = dzr.dr.d_ttl;
+  rr.qtype =dzr.dr.d_type;
+  rr.content = dzr.dr.d_content->getZoneRepresentation();
+  rr.domain_id = dzr.domain_id;
+  //  cout<<"old school called for "<<rr.qname<<", "<<rr.qtype.getName()<<endl;
+  return true;
+}
+
+bool LMDBBackend::getSOA(const DNSName &domain, SOAData &sd)
+{
+  //  cout <<"Native getSOA called"<<endl;
+  lookup(QType(QType::SOA), domain, 0, -1);
+  DNSZoneRecord dzr;
+  bool found=false;
+  while(get(dzr)) {
+    auto src = getRR<SOARecordContent>(dzr.dr);
+    sd.domain_id = dzr.domain_id;
+    sd.ttl = dzr.dr.d_ttl;
+    sd.qname = dzr.dr.d_name;
+    
+    sd.nameserver = src->d_mname;
+    sd.hostmaster = src->d_rname;
+    sd.serial = src->d_st.serial;
+    sd.refresh = src->d_st.refresh;
+    sd.retry = src->d_st.retry;
+    sd.expire = src->d_st.expire;
+    sd.default_ttl = src->d_st.minimum;
+    
+    sd.db = this;
+    found=true;
+  }
+  return found;
+}
+bool LMDBBackend::get_list(DNSZoneRecord& rr)
+{
+  for(;;) {
+    if(!d_getcursor)  {
+      d_rotxn.reset();
+      return false;
+    }
+    
+    MDBOutVal keyv, val;
+
+    d_getcursor->current(keyv, val);
+    DNSResourceRecord drr;
+    serFromString(val.get<string>(), drr);
+  
+    auto key = keyv.get<string_view>();
+    rr.dr.d_name = compoundOrdername::getQName(key) + d_lookupqname;
+    rr.domain_id = compoundOrdername::getDomainID(key);
+    rr.dr.d_type = compoundOrdername::getQType(key).getCode();
+    rr.dr.d_ttl = drr.ttl;
+    rr.auth = drr.auth;
+  
+    if(rr.dr.d_type == QType::NSEC3) {
+      //      cout << "Had a magic NSEC3, skipping it" << endl;
+      if(d_getcursor->next(keyv, val) || keyv.get<StringView>().rfind(d_matchkey, 0) != 0) {
+        d_getcursor.reset();
+      }
+      continue;
+    }
+    rr.dr.d_content = unserializeContentZR(rr.dr.d_type, rr.dr.d_name, drr.content);
+    
+    if(d_getcursor->next(keyv, val) || keyv.get<StringView>().rfind(d_matchkey, 0) != 0) {
+      d_getcursor.reset();
+    }
+    break;
+  }
+  return true;
+}
+
+
+bool LMDBBackend::get_lookup(DNSZoneRecord& rr)
+{
+  for(;;) {
+    if(!d_getcursor) {
+      d_rotxn.reset();
+      return false;
+    }
+    MDBOutVal keyv, val;
+    d_getcursor->current(keyv, val);
+    DNSResourceRecord drr;
+    serFromString(val.get<string>(), drr);
+    
+    auto key = keyv.get<string_view>();
+    
+    rr.dr.d_name = compoundOrdername::getQName(key) + d_lookupdomain;
+    
+    rr.domain_id = compoundOrdername::getDomainID(key);
+    //  cout << "We found "<<rr.qname<< " in zone id "<<rr.domain_id <<endl;
+    rr.dr.d_type = compoundOrdername::getQType(key).getCode();
+    rr.dr.d_ttl = drr.ttl;
+    if(rr.dr.d_type == QType::NSEC3) {
+      //      cout << "Hit a magic NSEC3 skipping" << endl;
+      if(d_getcursor->next(keyv, val) || keyv.get<StringView>().rfind(d_matchkey, 0) != 0) {
+        d_getcursor.reset();
+        d_rotxn.reset();
+      }
+      continue;
+    }
+    
+    rr.dr.d_content = unserializeContentZR(rr.dr.d_type, rr.dr.d_name, drr.content);
+    rr.auth = drr.auth;
+    if(d_getcursor->next(keyv, val) || keyv.get<StringView>().rfind(d_matchkey, 0) != 0) {
+      d_getcursor.reset();
+      d_rotxn.reset();
+    }
+    break;
+  }
+
+  
+  return true;
+}
+
+
+bool LMDBBackend::getDomainInfo(const DNSName &domain, DomainInfo &di, bool getSerial)
+{
+  auto txn = d_tdomains->getROTransaction();
+
+  if(!(di.id=txn.get<0>(domain, di)))
+    return false;
+  di.backend = this;
+  return true;
+}
+
+
+int LMDBBackend::genChangeDomain(const DNSName& domain, std::function<void(DomainInfo&)> func)
+{
+  auto txn = d_tdomains->getRWTransaction();
+
+  DomainInfo di;
+
+  auto id = txn.get<0>(domain, di);
+  func(di);
+  txn.put(di, id);
+  
+  txn.commit();
+  return true;
+}
+
+int LMDBBackend::genChangeDomain(uint32_t id, std::function<void(DomainInfo&)> func)
+{
+  DomainInfo di;
+
+  auto txn = d_tdomains->getRWTransaction();
+
+  if(!txn.get(id , di))
+    return false;
+  
+  func(di);
+  
+  txn.put(di, id);
+
+  txn.commit();
+  return true;
+}
+
+
+bool LMDBBackend::setKind(const DNSName &domain, const DomainInfo::DomainKind kind)
+{
+  return genChangeDomain(domain, [kind](DomainInfo& di) {
+      di.kind = kind;
+    });
+}
+
+bool LMDBBackend::setAccount(const DNSName &domain, const std::string& account)
+{
+  return genChangeDomain(domain, [account](DomainInfo& di) {
+      di.account = account;
+    });
+}
+
+
+void LMDBBackend::setFresh(uint32_t domain_id)
+{
+  genChangeDomain(domain_id, [](DomainInfo& di) {
+      di.last_check = time(0);
+    });
+}
+
+void LMDBBackend::setNotified(uint32_t domain_id, uint32_t serial)
+{
+  genChangeDomain(domain_id, [serial](DomainInfo& di) {
+      di.serial = serial;
+    });
+}
+
+
+bool LMDBBackend::setMaster(const DNSName &domain, const std::string& ips)
+{
+  vector<ComboAddress> masters;
+  vector<string> parts;
+  stringtok(parts, ips, " \t;,");
+  for(const auto& ip : parts) 
+    masters.push_back(ComboAddress(ip)); 
+  
+  return genChangeDomain(domain, [&masters](DomainInfo& di) {
+      di.masters = masters;
+    });
+}
+
+bool LMDBBackend::createDomain(const DNSName &domain)
+{
+  return createDomain(domain, "NATIVE", "", "");
+}
+          
+bool LMDBBackend::createDomain(const DNSName &domain, const string &type, const string &masters, const string &account)
+{
+  DomainInfo di;
+
+  auto txn = d_tdomains->getRWTransaction();
+  if(txn.get<0>(domain, di)) {
+    throw DBException("Domain '"+domain.toLogString()+"' exists already");
+  }
+  
+  di.zone = domain;
+  if(pdns_iequals(type, "master"))
+    di.kind = DomainInfo::Master;
+  else if(pdns_iequals(type, "slave"))
+    di.kind = DomainInfo::Slave;
+  else if(pdns_iequals(type, "native"))
+    di.kind = DomainInfo::Native;
+  else
+    throw DBException("Unable to create domain of unknown type '"+type+"'");
+  di.account = account;
+
+  txn.put(di);
+  txn.commit();
+
+  return true;
+}
+
+void LMDBBackend::getAllDomains(vector<DomainInfo> *domains, bool include_disabled)
+{
+  compoundOrdername co;
+  MDBOutVal val;
+  domains->clear();
+  auto txn = d_tdomains->getROTransaction();
+  for(auto iter = txn.begin(); iter != txn.end(); ++iter) {
+    DomainInfo di=*iter;
+    di.id = iter.getID();
+
+    auto txn = getRecordsROTransaction(iter.getID());
+    if(!txn->txn.get(txn->db->dbi, co(di.id, g_rootdnsname, QType::SOA), val)) {
+      domains->push_back(di);
+    }
+  }
+}
+
+void LMDBBackend::getUnfreshSlaveInfos(vector<DomainInfo>* domains)
+{
+  //  cout<<"Start of getUnfreshSlaveInfos"<<endl;
+  domains->clear();
+  auto txn = d_tdomains->getROTransaction();
+
+  time_t now = time(0);
+  for(auto iter = txn.begin(); iter != txn.end(); ++iter) {
+    if(iter->kind != DomainInfo::Slave)
+      continue;
+
+    auto txn2 = getRecordsROTransaction(iter.getID());
+    compoundOrdername co;
+    MDBOutVal val;
+    uint32_t serial = 0;
+    if(!txn2->txn.get(txn2->db->dbi, co(iter.getID(), g_rootdnsname, QType::SOA), val)) {
+      DNSResourceRecord rr;
+      serFromString(val.get<string_view>(), rr);
+      struct soatimes 
+      {
+        uint32_t serial;
+        uint32_t refresh;
+        uint32_t retry;
+        uint32_t expire;
+        uint32_t minimum;
+      } st;
+
+      memcpy(&st, &rr.content[rr.content.size()-sizeof(soatimes)], sizeof(soatimes));
+
+      if((time_t)(iter->last_check + ntohl(st.refresh)) >= now) { // still fresh
+        continue; // try next domain
+      }
+      //      cout << di.last_check <<" + " <<sdata.refresh<<" > = " << now << "\n";
+      serial = ntohl(st.serial);
+    }
+    else {
+      //      cout << "Could not find SOA for "<<iter->zone<<" with id "<<iter.getID()<<endl;
+      serial=0;  
+    }
+    DomainInfo di=*iter;    
+    di.id = iter.getID();
+    di.serial = serial;
+
+    domains->push_back(di);
+  }
+  //  cout<<"END of getUnfreshSlaveInfos"<<endl;
+}
+
+bool LMDBBackend::getAllDomainMetadata(const DNSName& name, std::map<std::string, std::vector<std::string> >& meta)
+{
+  meta.clear();
+  auto txn = d_tmeta->getROTransaction();
+  auto range = txn.equal_range<0>(name);
+  
+  for(auto& iter = range.first; iter != range.second; ++iter) {
+    meta[iter->key].push_back(iter->value);
+  }
+  return true;
+}
+
+bool LMDBBackend::setDomainMetadata(const DNSName& name, const std::string& kind, const std::vector<std::string>& meta)
+{
+  auto txn = d_tmeta->getRWTransaction();
+
+  auto range = txn.equal_range<0>(name);
+
+  for(auto& iter = range.first; iter != range.second; ++iter) {
+    if(iter-> key == kind)
+      iter.del();
+  }
+
+  for(const auto& m : meta) {
+    DomainMeta dm{name, kind, m};
+    txn.put(dm);
+  }
+  txn.commit();
+  return true;
+
+}
+
+bool LMDBBackend::getDomainKeys(const DNSName& name, std::vector<KeyData>& keys)
+{
+  auto txn = d_tkdb->getROTransaction();
+  auto range = txn.equal_range<0>(name);
+  for(auto& iter = range.first; iter != range.second; ++iter) {
+    KeyData kd{iter->content, iter.getID(), iter->flags, iter->active};
+    keys.push_back(kd);
+  }
+
+  return true;
+}
+
+bool LMDBBackend::removeDomainKey(const DNSName& name, unsigned int id)
+{
+  auto txn = d_tkdb->getRWTransaction();
+  KeyDataDB kdb;
+  if(txn.get(id, kdb)) {
+    if(kdb.domain == name) {
+      txn.del(id);
+      txn.commit();
+      return true;
+    }
+  }
+  // cout << "??? wanted to remove domain key for domain "<<name<<" with id "<<id<<", could not find it"<<endl;
+  return true;
+}
+
+bool LMDBBackend::addDomainKey(const DNSName& name, const KeyData& key, int64_t& id)
+{
+  auto txn = d_tkdb->getRWTransaction();
+  KeyDataDB kdb{name, key.content, key.flags, key.active};
+  id = txn.put(kdb);
+  txn.commit();
+    
+  return true;
+}
+
+bool LMDBBackend::activateDomainKey(const DNSName& name, unsigned int id)
+{
+  auto txn = d_tkdb->getRWTransaction();
+  KeyDataDB kdb;
+  if(txn.get(id, kdb)) {
+    if(kdb.domain == name) {
+      txn.modify(id, [](KeyDataDB& kdb)
+                 {
+                   kdb.active = true;
+                 });
+      txn.commit();
+      return true;
+    }
+  }
+
+  // cout << "??? wanted to activate domain key for domain "<<name<<" with id "<<id<<", could not find it"<<endl;
+  return true;
+}
+
+bool LMDBBackend::deactivateDomainKey(const DNSName& name, unsigned int id)
+{
+  auto txn = d_tkdb->getRWTransaction();
+  KeyDataDB kdb;
+  if(txn.get(id, kdb)) {
+    if(kdb.domain == name) {
+      txn.modify(id, [](KeyDataDB& kdb)
+                 {
+                   kdb.active = false;
+                 });
+      txn.commit();
+      return true;
+    }
+  }
+  // cout << "??? wanted to activate domain key for domain "<<name<<" with id "<<id<<", could not find it"<<endl;
+  return true;
+}
+
+bool LMDBBackend::getBeforeAndAfterNamesAbsolute(uint32_t id, const DNSName& qname, DNSName& unhashed, DNSName& before, DNSName& after) 
+{
+  //  cout << __PRETTY_FUNCTION__<< ": "<<id <<", "<<qname << " " << unhashed<<endl;
+
+  DomainInfo di;
+  if(!d_tdomains->getROTransaction().get(id, di)) {
+    // domain does not exist, tough luck
+    return false;
+  }
+  // cout <<"Zone: "<<di.zone<<endl;
+  
+  compoundOrdername co;
+  auto txn = getRecordsROTransaction(id);
+
+  auto cursor = txn->txn.getCursor(txn->db->dbi);
+  MDBOutVal key, val;
+
+  DNSResourceRecord rr;
+  
+  string matchkey = co(id, qname, QType::NSEC3);
+  if(cursor.lower_bound(matchkey, key, val)) {
+    // this is beyond the end of the database
+    // cout << "Beyond end of database!" << endl;
+    cursor.last(key, val);
+
+    for(;;) {
+      if(co.getDomainID(key.get<StringView>()) != id) {
+        //cout<<"Last record also not part of this zone!"<<endl;
+        // this implies something is wrong in the database, nothing we can do
+        return false;
+      }
+      
+      if(co.getQType(key.get<StringView>()) == QType::NSEC3) {
+        serFromString(val.get<StringView>(), rr);
+        if(!rr.ttl) // the kind of NSEC3 we need
+          break;
+      }
+      if(cursor.prev(key, val)) {
+        // hit beginning of database, again means something is wrong with it
+        return false;
+      }
+    }
+    before = co.getQName(key.get<StringView>());
+    unhashed = DNSName(rr.content.c_str(), rr.content.size(), 0, false) + di.zone;
+
+    // now to find after .. at the beginning of the zone
+    if(cursor.lower_bound(co(id), key, val)) {
+      // cout<<"hit end of zone find when we shouldn't"<<endl;
+      return false;
+    }
+    for(;;) {
+      if(co.getQType(key.get<StringView>()) == QType::NSEC3) {
+        serFromString(val.get<StringView>(), rr);
+        if(!rr.ttl)
+          break;
+      }
+
+      if(cursor.next(key, val) || co.getDomainID(key.get<StringView>()) != id) {
+        // cout<<"hit end of zone or database when we shouldn't"<<endl;
+        return false;
+      }
+    }
+    after = co.getQName(key.get<StringView>());
+    // cout<<"returning: before="<<before<<", after="<<after<<", unhashed: "<<unhashed<<endl;
+    return true;
+  }
+
+  // cout<<"Ended up at "<<co.getQName(key.get<StringView>()) <<endl;
+
+  before = co.getQName(key.get<StringView>());
+  if(before == qname) { 
+    // cout << "Ended up on exact right node" << endl;
+    before = co.getQName(key.get<StringView>());
+    // unhashed should be correct now, maybe check?
+    if(cursor.next(key, val)) {
+      // xxx should find first hash now
+
+      if(cursor.lower_bound(co(id), key, val)) {
+        // cout<<"hit end of zone find when we shouldn't for id "<<id<< __LINE__<<endl;
+        return false;
+      }
+      for(;;) {
+        if(co.getQType(key.get<StringView>()) == QType::NSEC3) {
+          serFromString(val.get<StringView>(), rr);
+          if(!rr.ttl)
+            break;
+        }
+        
+        if(cursor.next(key, val) || co.getDomainID(key.get<StringView>()) != id) {
+          // cout<<"hit end of zone or database when we shouldn't" << __LINE__<<endl;
+          return false;
+        }
+      }
+      after = co.getQName(key.get<StringView>());
+      // cout<<"returning: before="<<before<<", after="<<after<<", unhashed: "<<unhashed<<endl;
+      return true;
+    }
+  }
+  else {
+    // cout <<"Going backwards to find 'before'"<<endl;
+    int count=0;
+    for(;;) {
+      if(co.getQName(key.get<StringView>()).canonCompare(qname) && co.getQType(key.get<StringView>()) == QType::NSEC3) {
+        // cout<<"Potentially stopping traverse at "<< co.getQName(key.get<StringView>()) <<", " << (co.getQName(key.get<StringView>()).canonCompare(qname))<<endl;
+        // cout<<"qname = "<<qname<<endl;
+        // cout<<"here  = "<<co.getQName(key.get<StringView>())<<endl;
+        serFromString(val.get<StringView>(), rr);
+        if(!rr.ttl) 
+          break;
+      }
+      
+      if(cursor.prev(key, val) || co.getDomainID(key.get<StringView>()) != id ) {
+        // cout <<"XXX Hit *beginning* of zone or database"<<endl;
+        // this can happen, must deal with it
+        // should now find the last hash of the zone
+
+        if(cursor.lower_bound(co(id+1), key, val)) {
+          // cout << "Could not find the next higher zone, going to the end of the database then"<<endl;
+          cursor.last(key, val);
+        }
+        else
+          cursor.prev(key, val);
+
+        for(;;) {
+          if(co.getDomainID(key.get<StringView>()) != id) {
+            //cout<<"Last record also not part of this zone!"<<endl;
+            // this implies something is wrong in the database, nothing we can do
+            return false;
+          }
+          
+          if(co.getQType(key.get<StringView>()) == QType::NSEC3) {
+            serFromString(val.get<StringView>(), rr);
+            if(!rr.ttl) // the kind of NSEC3 we need
+              break;
+          }
+          if(cursor.prev(key, val)) {
+            // hit beginning of database, again means something is wrong with it
+            return false;
+          }
+        }
+        before = co.getQName(key.get<StringView>());
+        unhashed = DNSName(rr.content.c_str(), rr.content.size(), 0, false) + di.zone;
+        // cout <<"Should still find 'after'!"<<endl;
+        // for 'after', we need to find the first hash of this zone
+
+        if(cursor.lower_bound(co(id), key, val)) {
+          // cout<<"hit end of zone find when we shouldn't"<<endl;
+          // means database is wrong, nothing we can do
+          return false;
+        }
+        for(;;) {
+          if(co.getQType(key.get<StringView>()) == QType::NSEC3) {
+            serFromString(val.get<StringView>(), rr);
+            if(!rr.ttl)
+              break;
+          }
+          
+          if(cursor.next(key, val)) {
+            // means database is wrong, nothing we can do
+            // cout<<"hit end of zone when we shouldn't 2"<<endl;
+            return false;
+          }
+        }
+        after = co.getQName(key.get<StringView>());
+
+        
+        // cout<<"returning: before="<<before<<", after="<<after<<", unhashed: "<<unhashed<<endl;
+        return true;
+      }
+      ++count;
+    }
+    before = co.getQName(key.get<StringView>());
+    unhashed = DNSName(rr.content.c_str(), rr.content.size(), 0, false) + di.zone;
+    // cout<<"Went backwards, found "<<before<<endl;
+    // return us to starting point
+    while(count--)
+      cursor.next(key, val);
+  }
+  //  cout<<"Now going forward"<<endl;
+  for(int count = 0 ;;++count) {
+    if((count && cursor.next(key, val)) || co.getDomainID(key.get<StringView>()) != id ) {
+      // cout <<"Hit end of database or zone, finding first hash then in zone "<<id<<endl;
+      if(cursor.lower_bound(co(id), key, val)) {
+        // cout<<"hit end of zone find when we shouldn't"<<endl;
+        // means database is wrong, nothing we can do
+        return false;
+      }
+      for(;;) {
+        if(co.getQType(key.get<StringView>()) == QType::NSEC3) {
+          serFromString(val.get<StringView>(), rr);
+          if(!rr.ttl)
+            break;
+        }
+        
+        if(cursor.next(key, val)) {
+          // means database is wrong, nothing we can do
+          // cout<<"hit end of zone when we shouldn't 2"<<endl;
+          return false;
+        }
+        // cout << "Next.. "<<endl;
+      }
+      after = co.getQName(key.get<StringView>());
+      
+      // cout<<"returning: before="<<before<<", after="<<after<<", unhashed: "<<unhashed<<endl;
+      return true;
+    }
+    
+    // cout<<"After "<<co.getQName(key.get<StringView>()) <<endl;
+    if(co.getQType(key.get<StringView>()) == QType::NSEC3) {
+      serFromString(val.get<StringView>(), rr);
+      if(!rr.ttl) {
+        break;
+      }
+    }
+  }
+  after = co.getQName(key.get<StringView>());
+  // cout<<"returning: before="<<before<<", after="<<after<<", unhashed: "<<unhashed<<endl;
+  return true;
+}
+
+bool LMDBBackend::getBeforeAndAfterNames(uint32_t id, const DNSName& zonenameU, const DNSName& qname, DNSName& before, DNSName& after)
+{
+  DNSName zonename = zonenameU.makeLowerCase();
+  //  cout << __PRETTY_FUNCTION__<< ": "<<id <<", "<<zonename << ", '"<<qname<<"'"<<endl;
+
+  auto txn = getRecordsROTransaction(id);
+  compoundOrdername co;
+  DNSName qname2 = qname.makeRelative(zonename);
+  string matchkey=co(id,qname2);
+  auto cursor = txn->txn.getCursor(txn->db->dbi);
+  MDBOutVal key, val;
+  // cout<<"Lower_bound for "<<qname2<<endl;
+  if(cursor.lower_bound(matchkey, key, val)) {
+    // cout << "Hit end of database, bummer"<<endl;
+    cursor.last(key, val);
+    if(co.getDomainID(key.get<string_view>()) == id) {
+      before = co.getQName(key.get<string_view>()) + zonename;
+      after = zonename;
+    }
+    // else
+      // cout << "We were at end of database, but this zone is not there?!"<<endl;
+    return true;
+  }
+  // cout<<"Cursor is at "<<co.getQName(key.get<string_view>()) <<", in zone id "<<co.getDomainID(key.get<string_view>())<< endl;
+
+  if(co.getQType(key.get<string_view>()).getCode() && co.getDomainID(key.get<string_view>()) ==id && co.getQName(key.get<string_view>()) == qname2) { // don't match ENTs
+    // cout << "Had an exact match!"<<endl;
+    before = qname2 + zonename;
+    int rc;
+    for(;;) {
+      rc=cursor.next(key, val);
+      if(rc) break;
+      
+      if(co.getDomainID(key.get<string_view>()) == id && key.get<StringView>().rfind(matchkey, 0)==0)
+        continue;
+      DNSResourceRecord rr;
+      serFromString(val.get<StringView>(), rr);
+      if(co.getQType(key.get<string_view>()).getCode() && (rr.auth || co.getQType(key.get<string_view>()).getCode() == QType::NS))
+        break;
+    }
+    if(rc || co.getDomainID(key.get<string_view>()) != id) {
+      // cout << "We hit the end of the zone or database. 'after' is apex" << endl;
+      after=zonename;
+      return false;
+    }
+    after = co.getQName(key.get<string_view>()) + zonename;
+    return true;
+  }
+
+  
+  if(co.getDomainID(key.get<string_view>()) != id) {
+    // cout << "Ended up in next zone, 'after' is zonename" <<endl;
+    after = zonename;
+    // cout << "Now hunting for previous" << endl;
+    int rc;
+    for(;;) {
+      rc=cursor.prev(key, val);
+      if(rc) {
+        // cout<<"Reversed into zone, but got not found from lmdb" <<endl;
+        return false;
+      }
+      
+      if(co.getDomainID(key.get<string_view>()) != id) {
+        // cout<<"Reversed into zone, but found wrong zone id " << co.getDomainID(key.get<string_view>()) << " != "<<id<<endl;
+        // "this can't happen"
+        return false;
+      }
+      DNSResourceRecord rr;
+      serFromString(val.get<StringView>(), rr);
+      if(co.getQType(key.get<string_view>()).getCode() && (rr.auth || co.getQType(key.get<string_view>()).getCode() == QType::NS))
+        break;
+    }
+
+    before = co.getQName(key.get<string_view>()) + zonename;
+    // cout<<"Found: "<< before<<endl;
+    return true;
+  }
+
+  // cout <<"We ended up after "<<qname<<", on "<<co.getQName(key.get<string_view>())<<endl;
+
+  int skips = 0;
+  for(; ;) {
+    DNSResourceRecord rr;
+    serFromString(val.get<StringView>(), rr);
+    if(co.getQType(key.get<string_view>()).getCode() && (rr.auth || co.getQType(key.get<string_view>()).getCode() == QType::NS)) {
+      after = co.getQName(key.get<string_view>()) + zonename;
+      // cout <<"Found auth ("<<rr.auth<<") or an NS record "<<after<<", type: "<<co.getQType(key.get<string_view>()).getName()<<", ttl = "<<rr.ttl<<endl;
+      // cout << makeHexDump(val.get<string>()) << endl;
+      break;
+    }
+    // cout <<"  oops, " << co.getQName(key.get<string_view>()) << " was not auth "<<rr.auth<< " type=" << rr.qtype.getName()<<" or NS, so need to skip ahead a bit more" << endl;
+    int rc = cursor.next(key, val);
+    if(!rc)
+      ++skips;
+    if(rc || co.getDomainID(key.get<string_view>()) != id ) {
+      // cout << "  oops, hit end of database or zone. This means after is apex" <<endl;
+      after = zonename;
+      break;
+    }
+  }
+  // go back to where we were
+  while(skips--)
+    cursor.prev(key,val);
+  
+  for(;;) {
+    int rc = cursor.prev(key, val);
+    if(rc || co.getDomainID(key.get<string_view>()) != id) {
+      // XX I don't think this case can happen
+      // cout << "We hit the beginning of the zone or database.. now what" << endl;
+      return false;
+    }
+    before = co.getQName(key.get<string_view>()) + zonename;
+    DNSResourceRecord rr;
+    serFromString(val.get<string_view>(), rr);
+    // cout<<"And before to "<<before<<", auth = "<<rr.auth<<endl;
+    if(co.getQType(key.get<string_view>()).getCode() && (rr.auth || co.getQType(key.get<string_view>()) == QType::NS))
+      break;
+    // cout << "Oops, that was wrong, go back one more"<<endl;
+  }
+
+  return true;
+
+}
+
+bool LMDBBackend::updateDNSSECOrderNameAndAuth(uint32_t domain_id, const DNSName& qname, const DNSName& ordername, bool auth, const uint16_t qtype)
+{
+  //  cout << __PRETTY_FUNCTION__<< ": "<< domain_id <<", '"<<qname <<"', '"<<ordername<<"', "<<auth<< ", " << qtype << endl;
+  shared_ptr<RecordsRWTransaction> txn;
+  bool needCommit = false;
+  if(d_rwtxn && d_transactiondomainid==domain_id) {
+    txn = d_rwtxn;
+    //    cout<<"Reusing open transaction"<<endl;
+  }
+  else {
+    //    cout<<"Making a new RW txn for " << __PRETTY_FUNCTION__ <<endl;
+    txn = getRecordsRWTransaction(domain_id);
+    needCommit = true;
+  }
+
+  DomainInfo di;
+  if(!d_tdomains->getROTransaction().get(domain_id, di)) {
+    //    cout<<"Could not find domain_id "<<domain_id <<endl;
+    return false;
+  }
+
+  DNSName rel = qname.makeRelative(di.zone);
+
+  compoundOrdername co;
+  string matchkey = co(domain_id, rel);
+
+  auto cursor = txn->txn.getCursor(txn->db->dbi);
+  MDBOutVal key, val;
+  if(cursor.lower_bound(matchkey, key, val)) {
+    // cout << "Could not find anything"<<endl;
+    return false;
+  }
+
+  bool hasOrderName = !ordername.empty();
+  bool needNSEC3 = hasOrderName;
+
+  for(; key.get<StringView>().rfind(matchkey,0) == 0; ) {
+    DNSResourceRecord rr;
+    rr.qtype = co.getQType(key.get<StringView>());
+
+    if(rr.qtype != QType::NSEC3) {
+      serFromString(val.get<StringView>(), rr);
+      if(!needNSEC3 && qtype != QType::ANY) {
+        needNSEC3 = (rr.disabled && QType(qtype) != rr.qtype);
+      }
+
+      if((qtype == QType::ANY || QType(qtype) == rr.qtype) && (rr.disabled != hasOrderName || rr.auth != auth)) {
+        rr.auth = auth;
+        rr.disabled = hasOrderName;
+        string repl = serToString(rr);
+        cursor.put(key, repl);
+      }
+    }
+
+    if(cursor.next(key, val))
+      break;
+  }
+
+  bool del = false;
+  DNSResourceRecord rr;
+  matchkey = co(domain_id,rel,QType::NSEC3);
+  if(!txn->txn.get(txn->db->dbi, matchkey, val)) {
+    serFromString(val.get<string_view>(), rr);
+
+    if(needNSEC3) {
+      if(hasOrderName && rr.content != ordername.toDNSStringLC()) {
+          del = true;
+      }
+    } else {
+      del = true;
+    }
+    if(del) {
+      txn->txn.del(txn->db->dbi, co(domain_id, DNSName(rr.content.c_str(), rr.content.size(), 0, false), QType::NSEC3));
+      txn->txn.del(txn->db->dbi, matchkey);
+    }
+  } else {
+    del = true;
+  }
+
+  if(hasOrderName && del) {
+    matchkey = co(domain_id,rel,QType::NSEC3);
+
+    rr.ttl=0;
+    rr.auth=0;
+    rr.content=rel.toDNSStringLC();
+
+    string str = serToString(rr);
+    txn->txn.put(txn->db->dbi, co(domain_id,ordername,QType::NSEC3), str);
+    rr.ttl = 1;
+    rr.content = ordername.toDNSStringLC();
+    str = serToString(rr);
+    txn->txn.put(txn->db->dbi, matchkey, str);  // 2
+  }
+
+  if(needCommit)
+    txn->txn.commit();
+  return false;
+}
+
+bool LMDBBackend::updateEmptyNonTerminals(uint32_t domain_id, set<DNSName>& insert, set<DNSName>& erase, bool remove) 
+{
+  // cout << __PRETTY_FUNCTION__<< ": "<< domain_id << ", insert.size() "<<insert.size()<<", "<<erase.size()<<", " <<remove<<endl;
+
+  bool needCommit = false;
+  shared_ptr<RecordsRWTransaction> txn;
+  if(d_rwtxn && d_transactiondomainid == domain_id) {
+    txn = d_rwtxn;
+    //    cout<<"Reusing open transaction"<<endl;
+  }
+  else {
+    //    cout<<"Making a new RW txn for delete domain"<<endl;
+    txn = getRecordsRWTransaction(domain_id);
+    needCommit = true;
+  }
+
+  
+  // if remove is set, all ENTs should be removed & nothing else should be done
+  if(remove) {
+    deleteDomainRecords(*txn, domain_id, 0);
+  }
+  else {
+    DomainInfo di;
+    auto rotxn = d_tdomains->getROTransaction();
+    if(!rotxn.get(domain_id, di)) {
+      // cout <<"No such domain with id "<<domain_id<<endl;
+      return false;
+    }
+    compoundOrdername co;
+    for(const auto& n : insert) {
+      DNSResourceRecord rr;
+      rr.qname = n.makeRelative(di.zone);
+      rr.ttl = 0;
+      rr.auth = true;
+
+      std::string ser = serToString(rr);
+
+      txn->txn.put(txn->db->dbi, co(domain_id, rr.qname, 0), ser);
+
+      DNSResourceRecord rr2;
+      serFromString(ser, rr2);
+      
+      // cout <<" +"<<n<<endl;
+    }
+    for(auto n : erase) {
+      // cout <<" -"<<n<<endl;
+      n.makeUsRelative(di.zone);
+      txn->txn.del(txn->db->dbi, co(domain_id, n, 0));
+    }
+  }
+  if(needCommit)
+    txn->txn.commit();
+  return false;
+}
+
+/* TSIG */
+bool LMDBBackend::getTSIGKey(const DNSName& name, DNSName* algorithm, string* content)
+{
+  auto txn = d_ttsig->getROTransaction();
+
+  TSIGKey tk;
+  if(!txn.get<0>(name, tk))
+    return false;
+  if(algorithm)
+    *algorithm = tk.algorithm;
+  if(content)
+    *content = tk.key;
+  return true;
+
+}
+// this deletes an old key if it has the same algorithm
+bool LMDBBackend::setTSIGKey(const DNSName& name, const DNSName& algorithm, const string& content)
+{
+  auto txn = d_ttsig->getRWTransaction();
+
+  for(auto range = txn.equal_range<0>(name); range.first != range.second; ++range.first) {
+    if(range.first->algorithm == algorithm)
+      range.first.del();
+  }
+
+  TSIGKey tk;
+  tk.name = name;
+  tk.algorithm = algorithm;
+  tk.key=content;
+  
+  txn.put(tk);
+  txn.commit();
+  
+  return true;
+}
+bool LMDBBackend::deleteTSIGKey(const DNSName& name)
+{
+  auto txn = d_ttsig->getRWTransaction();
+  TSIGKey tk;
+
+  for(auto range = txn.equal_range<0>(name); range.first != range.second; ++range.first) {
+    range.first.del();
+  }
+  txn.commit();
+  return true;
+}
+bool LMDBBackend::getTSIGKeys(std::vector< struct TSIGKey > &keys)
+{
+  auto txn = d_ttsig->getROTransaction();
+
+  keys.clear();
+  for(auto iter = txn.begin(); iter != txn.end(); ++iter) {
+    keys.push_back(*iter);
+  }
+  return false;
+}
+
+
+
+
+class LMDBFactory : public BackendFactory
+{
+public:
+  LMDBFactory() : BackendFactory("lmdb") {}
+  void declareArguments(const string &suffix="")
+  {
+    declare(suffix,"filename","Filename for lmdb","./pdns.lmdb");
+    declare(suffix,"sync-mode","Synchronisation mode: nosync, nometasync, mapasync","mapasync");
+    declare(suffix,"shards","Records database will be split into this number of shards","64");
+  }
+  DNSBackend *make(const string &suffix="")
+  {
+    return new LMDBBackend(suffix);
+  }
+};
+
+
+
+
+/* THIRD PART */
+
+class LMDBLoader
+{
+public:
+  LMDBLoader()
+  {
+    BackendMakers().report(new LMDBFactory);
+    g_log << Logger::Info << "[lmdbbackend] This is the lmdb backend version " VERSION
+#ifndef REPRODUCIBLE
+      << " (" __DATE__ " " __TIME__ ")"
+#endif
+      << " reporting" << endl;
+  }  
+};
+
+static LMDBLoader randomLoader;
diff --git a/modules/lmdbbackend/lmdbbackend.hh b/modules/lmdbbackend/lmdbbackend.hh
new file mode 100644
index 0000000000..14cd5499a8
--- /dev/null
+++ b/modules/lmdbbackend/lmdbbackend.hh
@@ -0,0 +1,273 @@
+#pragma once
+#include "pdns/dnsbackend.hh"
+#include "lmdb-typed.hh"
+
+template<class T, typename std::enable_if<std::is_same<T, DNSName>::value,T>::type* = nullptr>
+std::string keyConv(const T& t)
+{
+  /* www.ds9a.nl -> nl0ds9a0www0
+     root -> 0   <- we need this to keep lmdb happy
+     nl -> nl0
+     
+  */
+  if(t.isRoot())
+    return std::string(1, (char)0);
+  std::string in = t.labelReverse().toDNSStringLC(); // www.ds9a.nl is now 2nl4ds9a3www0
+  std::string ret;
+  ret.reserve(in.size());
+  
+  for(auto iter = in.begin(); iter != in.end(); ++iter) {
+    uint8_t len = *iter;
+    if(iter != in.begin())
+      ret.append(1, (char)0);
+    if(!len)
+      break;
+
+    ret.append(&*(iter+1), len);
+    iter+=len;
+  }
+  return ret;
+}
+
+
+class LMDBBackend : public DNSBackend
+{
+public:
+  explicit LMDBBackend(const string &suffix="");
+
+  bool list(const DNSName &target, int id, bool include_disabled) override;
+
+  bool getDomainInfo(const DNSName &domain, DomainInfo &di, bool getSerial=true) override;
+  bool createDomain(const DNSName &domain, const string &type, const string &masters, const string &account);
+
+  bool createDomain(const DNSName &domain) override;
+  
+  bool startTransaction(const DNSName &domain, int domain_id=-1) override;
+  bool commitTransaction() override;
+  bool abortTransaction() override;
+  bool feedRecord(const DNSResourceRecord &r, const DNSName &ordername) override;
+  bool feedEnts(int domain_id, map<DNSName,bool>& nonterm) override;
+  bool feedEnts3(int domain_id, const DNSName &domain, map<DNSName,bool> &nonterm, const NSEC3PARAMRecordContent& ns3prc, bool narrow) override;
+  bool replaceRRSet(uint32_t domain_id, const DNSName& qname, const QType& qt, const vector<DNSResourceRecord>& rrset) override;
+
+  void getAllDomains(vector<DomainInfo> *domains, bool include_disabled=false) override;
+  void lookup(const QType &type, const DNSName &qdomain, DNSPacket *p, int zoneId) override;
+  bool get(DNSResourceRecord &rr) override;
+  bool get(DNSZoneRecord& dzr) override;
+
+  bool getSOA(const DNSName &domain, SOAData &sd) override;
+  void getUnfreshSlaveInfos(vector<DomainInfo>* domains) override;
+  
+  bool setMaster(const DNSName &domain, const string &ip) override;
+  bool setKind(const DNSName &domain, const DomainInfo::DomainKind kind) override;
+  bool getAllDomainMetadata(const DNSName& name, std::map<std::string, std::vector<std::string> >& meta) override;
+  bool getDomainMetadata(const DNSName& name, const std::string& kind, std::vector<std::string>& meta) override
+  {
+    //    std::cout<<"Request for metadata items for zone "<<name<<", kind "<<kind<<endl;
+    meta.clear();
+    std::map<std::string, std::vector<std::string> > metas;
+    if(getAllDomainMetadata(name, metas)) {
+      for(const auto& m : metas) {
+        if(m.first == kind) {
+          meta = m.second;
+          return true;
+        }
+      }
+      return true;
+    }
+    return false;
+  }
+
+  bool setDomainMetadata(const DNSName& name, const std::string& kind, const std::vector<std::string>& meta) override;
+  void setFresh(uint32_t domain_id) override;
+  void setNotified(uint32_t id, uint32_t serial) override;
+  bool setAccount(const DNSName &domain, const std::string& account) override;
+  bool deleteDomain(const DNSName &domain) override;
+
+  bool getDomainKeys(const DNSName& name, std::vector<KeyData>& keys) override;
+  bool removeDomainKey(const DNSName& name, unsigned int id) override;
+  bool addDomainKey(const DNSName& name, const KeyData& key, int64_t& id) override;
+  bool activateDomainKey(const DNSName& name, unsigned int id) override;
+  bool deactivateDomainKey(const DNSName& name, unsigned int id) override;
+
+  // TSIG
+  bool getTSIGKey(const DNSName& name, DNSName* algorithm, string* content) override;
+  bool setTSIGKey(const DNSName& name, const DNSName& algorithm, const string& content) override;
+  bool deleteTSIGKey(const DNSName& name) override;
+  bool getTSIGKeys(std::vector< struct TSIGKey > &keys) override;
+
+  
+  // DNSSEC
+
+  bool getBeforeAndAfterNamesAbsolute(uint32_t id, const DNSName& qname, DNSName& unhashed, DNSName& before, DNSName& after) override;
+
+  virtual bool getBeforeAndAfterNames(uint32_t id, const DNSName& zonename, const DNSName& qname, DNSName& before, DNSName& after) override;
+
+  bool updateDNSSECOrderNameAndAuth(uint32_t domain_id, const DNSName& qname, const DNSName& ordername, bool auth, const uint16_t qtype=QType::ANY) override;
+
+  bool updateEmptyNonTerminals(uint32_t domain_id, set<DNSName>& insert, set<DNSName>& erase, bool remove) override;
+
+  bool doesDNSSEC() override
+  {
+    return true;
+  }
+private:
+
+  struct compoundOrdername
+  {
+    std::string operator()(uint32_t id, const DNSName& t, uint16_t qtype)
+    {
+      std::string ret = operator()(id, t);
+      uint16_t qt = htons(qtype);
+      ret.append((char*)&qt, 2);
+      return ret;
+    }
+    std::string operator()(uint32_t id, const DNSName& t)
+    {
+      std::string ret = operator()(id);
+      ret += keyConv(t);
+      ret.append(1, (char)0); // this means '00' really ends the zone
+      return ret;
+    }
+    std::string operator()(uint32_t id)
+    {
+      std::string ret;
+      id = htonl(id);
+      ret.assign((char*)&id, 4);
+      return ret;
+    }
+
+    std::string operator()(const DNSResourceRecord& rr)
+    {
+      return operator()(rr.domain_id, rr.qname, rr.qtype.getCode());
+    }
+
+    static uint32_t getDomainID(const string_view& key)
+    {
+      uint32_t ret;
+      memcpy(&ret, &key[0], sizeof(ret));
+      return ntohl(ret);
+    }
+
+    static DNSName getQName(const string_view& key)
+    {
+      /* www.ds9a.nl -> nl0ds9a0www0
+         root -> 0   <- we need this to keep lmdb happy
+         nl -> nl0 */
+      DNSName ret;
+      auto iter = key.cbegin() + 4;
+      auto end = key.cend() - 2;
+      while(iter < end) {
+        auto startpos = iter;
+        while(iter != end && *iter)
+          ++iter;
+        if(iter == startpos)
+          break;
+        string part(&*startpos, iter-startpos);
+        ret.prependRawLabel(part);
+        //        cout << "Prepending part: "<<part<<endl;
+        if(iter != end)
+          ++iter;
+      }
+      if(ret.empty())
+        return g_rootdnsname;
+      return ret;
+    }
+
+    static QType getQType(const string_view& key)
+    {
+      uint16_t ret;
+      memcpy(&ret, &key[key.size()-2], sizeof(ret));
+      return QType(ntohs(ret));
+    }
+  };
+
+public:
+  struct DomainMeta
+  {
+    DNSName domain;
+    string key;
+    string value;
+  };
+  struct KeyDataDB
+  {
+    DNSName domain;
+    std::string content;
+    unsigned int flags;
+    bool active;
+  };
+
+private:
+
+  typedef TypedDBI<DomainInfo,
+                   index_on<DomainInfo, DNSName, &DomainInfo::zone>
+          > tdomains_t;
+
+  
+  typedef TypedDBI<DomainMeta,
+            index_on<DomainMeta, DNSName, &DomainMeta::domain>
+          > tmeta_t;
+  
+  typedef TypedDBI<KeyDataDB,
+                   index_on<KeyDataDB, DNSName, &KeyDataDB::domain>                   
+          > tkdb_t;
+
+  typedef TypedDBI<TSIGKey,
+                   index_on<TSIGKey, DNSName, &TSIGKey::name>                   
+          > ttsig_t;
+  
+  int d_shards;
+  int d_asyncFlag;
+
+  struct RecordsDB
+  {
+    shared_ptr<MDBEnv> env;
+    MDBDbi dbi;
+  };
+
+  struct RecordsROTransaction
+  {
+    RecordsROTransaction(MDBROTransaction&& intxn) : txn(std::move(intxn))
+    {}
+    shared_ptr<RecordsDB> db;
+    MDBROTransaction txn;
+  };
+  struct RecordsRWTransaction
+  {
+    RecordsRWTransaction(MDBRWTransaction&& intxn) : txn(std::move(intxn))
+    {}
+    shared_ptr<RecordsDB> db;
+    MDBRWTransaction txn;
+  };
+
+  vector<RecordsDB> d_trecords;;
+
+  std::shared_ptr<MDBROCursor> d_getcursor;
+
+  shared_ptr<tdomains_t> d_tdomains;
+  shared_ptr<tmeta_t> d_tmeta;
+  shared_ptr<tkdb_t> d_tkdb;
+  shared_ptr<ttsig_t> d_ttsig;
+  
+  shared_ptr<RecordsROTransaction> d_rotxn; // for lookup and list
+  shared_ptr<RecordsRWTransaction> d_rwtxn; // for feedrecord within begin/aborttransaction
+  std::shared_ptr<RecordsRWTransaction> getRecordsRWTransaction(uint32_t id);
+  std::shared_ptr<RecordsROTransaction> getRecordsROTransaction(uint32_t id);
+  int genChangeDomain(const DNSName& domain, std::function<void(DomainInfo&)> func);
+  int genChangeDomain(uint32_t id, std::function<void(DomainInfo&)> func);
+  void deleteDomainRecords(RecordsRWTransaction& txn, uint32_t domain_id, uint16_t qtype=QType::ANY);
+  
+  bool get_list(DNSZoneRecord &rr);
+  bool get_lookup(DNSZoneRecord &rr);
+  bool d_inlist{false};
+  QType d_lookuptype;                   // for get after lookup
+  std::string d_matchkey;
+  int32_t d_lookupdomainid;            // for get after lookup
+  DNSName d_lookupqname;
+  DNSName d_lookupdomain;
+  
+  DNSName d_transactiondomain;
+  uint32_t d_transactiondomainid;
+  bool d_dolog;
+  DTime d_dtime; // used only for logging
+};
diff --git a/pdns/pdnsutil.cc b/pdns/pdnsutil.cc
index d5b4d0801c..46cde8ac62 100644
--- a/pdns/pdnsutil.cc
+++ b/pdns/pdnsutil.cc
@@ -786,14 +786,17 @@ int listZone(const DNSName &zone) {
   di.backend->list(zone, di.id);
   DNSResourceRecord rr;
   cout<<"$ORIGIN ."<<endl;
+  cout.sync_with_stdio(false);
+  
   while(di.backend->get(rr)) {
     if(rr.qtype.getCode()) {
       if ( (rr.qtype.getCode() == QType::NS || rr.qtype.getCode() == QType::SRV || rr.qtype.getCode() == QType::MX || rr.qtype.getCode() == QType::CNAME) && !rr.content.empty() && rr.content[rr.content.size()-1] != '.') 
 	rr.content.append(1, '.');
 	
-      cout<<rr.qname<<"\t"<<rr.ttl<<"\tIN\t"<<rr.qtype.getName()<<"\t"<<rr.content<<endl;
+      cout<<rr.qname<<"\t"<<rr.ttl<<"\tIN\t"<<rr.qtype.getName()<<"\t"<<rr.content<<"\n";
     }
   }
+  cout.flush();
   return EXIT_SUCCESS;
 }