[thirdparty/pdns.git] / ext / lmdb-safe / lmdb-safe.cc

#include "lmdb-safe.hh"
#include <fcntl.h>
#include <mutex>
#include <memory>
#include <sys/stat.h>
#include <string.h>
#include <map>

using std::string;
using std::runtime_error;
using std::tuple;
using std::weak_ptr;

static string MDBError(int rc)
{
  return mdb_strerror(rc);
}

MDBDbi::MDBDbi(MDB_env* env, MDB_txn* txn, const string_view dbname, int flags)
{
  // A transaction that uses this function must finish (either commit or abort) before any other transaction in the process may use this function.
  
  int rc = mdb_dbi_open(txn, dbname.empty() ? 0 : &dbname[0], flags, &d_dbi);
  if(rc)
    throw std::runtime_error("Unable to open named database: " + MDBError(rc));
  
  // Database names are keys in the unnamed database, and may be read but not written.
}

MDBEnv::MDBEnv(const char* fname, int flags, int mode)
{
  mdb_env_create(&d_env);
  uint64_t mapsizeMB = (sizeof(long)==4) ? 100 : 16000;
  // on 32 bit platforms, there is just no room for more
  if(mdb_env_set_mapsize(d_env, mapsizeMB * 1048576))
    throw std::runtime_error("setting map size");
    /*
Various other options may also need to be set before opening the handle, e.g. mdb_env_set_mapsize(), mdb_env_set_maxreaders(), mdb_env_set_maxdbs(),
    */

  mdb_env_set_maxdbs(d_env, 128);

  // we need MDB_NOTLS since we rely on its semantics
  if(int rc=mdb_env_open(d_env, fname, flags | MDB_NOTLS, mode)) {
    // If this function fails, mdb_env_close() must be called to discard the MDB_env handle.
    mdb_env_close(d_env);
    throw std::runtime_error("Unable to open database file "+std::string(fname)+": " + MDBError(rc));
  }
}

void MDBEnv::incROTX()
{
  std::lock_guard<std::mutex> l(d_countmutex);
  ++d_ROtransactionsOut[std::this_thread::get_id()];
}

void MDBEnv::decROTX()
{
  std::lock_guard<std::mutex> l(d_countmutex);
  --d_ROtransactionsOut[std::this_thread::get_id()];
}

void MDBEnv::incRWTX()
{
  std::lock_guard<std::mutex> l(d_countmutex);
  ++d_RWtransactionsOut[std::this_thread::get_id()];
}

void MDBEnv::decRWTX()
{
  std::lock_guard<std::mutex> l(d_countmutex);
  --d_RWtransactionsOut[std::this_thread::get_id()];
}

int MDBEnv::getRWTX()
{
  std::lock_guard<std::mutex> l(d_countmutex);
  return d_RWtransactionsOut[std::this_thread::get_id()];
}
int MDBEnv::getROTX()
{
  std::lock_guard<std::mutex> l(d_countmutex);
  return d_ROtransactionsOut[std::this_thread::get_id()];
}


std::shared_ptr<MDBEnv> getMDBEnv(const char* fname, int flags, int mode)
{
  struct Value
  {
    weak_ptr<MDBEnv> wp;
    int flags;
  };
  
  static std::map<tuple<dev_t, ino_t>, Value> s_envs;
  static std::mutex mut;
  
  struct stat statbuf;
  if(stat(fname, &statbuf)) {
    if(errno != ENOENT)
      throw std::runtime_error("Unable to stat prospective mdb database: "+string(strerror(errno)));
    else {
      std::lock_guard<std::mutex> l(mut);
      auto fresh = std::make_shared<MDBEnv>(fname, flags, mode);
      if(stat(fname, &statbuf))
        throw std::runtime_error("Unable to stat prospective mdb database: "+string(strerror(errno)));
      auto key = std::tie(statbuf.st_dev, statbuf.st_ino);
      s_envs[key] = {fresh, flags};
      return fresh;
    }
  }

  std::lock_guard<std::mutex> l(mut);
  auto key = std::tie(statbuf.st_dev, statbuf.st_ino);
  auto iter = s_envs.find(key);
  if(iter != s_envs.end()) {
    auto sp = iter->second.wp.lock();
    if(sp) {
      if(iter->second.flags != flags)
        throw std::runtime_error("Can't open mdb with differing flags");

      return sp;
    }
    else {
      s_envs.erase(iter); // useful if make_shared fails
    }
  }

  auto fresh = std::make_shared<MDBEnv>(fname, flags, mode);
  s_envs[key] = {fresh, flags};
  
  return fresh;
}


MDBDbi MDBEnv::openDB(const string_view dbname, int flags)
{
  unsigned int envflags;
  mdb_env_get_flags(d_env, &envflags);
  /*
    This function must not be called from multiple concurrent transactions in the same process. A transaction that uses this function must finish (either commit or abort) before any other transaction in the process may use this function.
  */
  std::lock_guard<std::mutex> l(d_openmut);
  
  if(!(envflags & MDB_RDONLY)) {
    auto rwt = getRWTransaction();
    MDBDbi ret = rwt->openDB(dbname, flags);
    rwt->commit();
    return ret;
  }

  MDBDbi ret;
  {
    auto rwt = getROTransaction(); 
    ret = rwt->openDB(dbname, flags);
  }
  return ret;
}

MDBRWTransactionImpl::MDBRWTransactionImpl(MDBEnv *parent, MDB_txn *txn):
  MDBROTransactionImpl(parent, txn)

{

}

MDB_txn *MDBRWTransactionImpl::openRWTransaction(MDBEnv *env, MDB_txn *parent, int flags)
{
  MDB_txn *result;
  if(env->getROTX() || env->getRWTX())
    throw std::runtime_error("Duplicate RW transaction");

  for(int tries =0 ; tries < 3; ++tries) { // it might happen twice, who knows
    if(int rc=mdb_txn_begin(env->d_env, parent, flags, &result)) {
      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(env->d_env, 0);
        continue;
      }
      throw std::runtime_error("Unable to start RW transaction: "+std::string(mdb_strerror(rc)));
    }
    break;
  }
  env->incRWTX();
  return result;
}

MDBRWTransactionImpl::MDBRWTransactionImpl(MDBEnv* parent, int flags):
  MDBRWTransactionImpl(parent, openRWTransaction(parent, nullptr, flags))
{
}

MDBRWTransactionImpl::~MDBRWTransactionImpl()
{
  abort();
}

void MDBRWTransactionImpl::commit()
{
  closeRORWCursors();
  if (!d_txn) {
    return;
  }

  if(int rc = mdb_txn_commit(d_txn)) {
    throw std::runtime_error("committing: " + std::string(mdb_strerror(rc)));
  }
  environment().decRWTX();
  d_txn = nullptr;
}

void MDBRWTransactionImpl::abort()
{
  closeRORWCursors();
  if (!d_txn) {
    return;
  }

  mdb_txn_abort(d_txn);
  // prevent the RO destructor from cleaning up the transaction itself
  environment().decRWTX();
  d_txn = nullptr;
}

MDBROTransactionImpl::MDBROTransactionImpl(MDBEnv *parent, MDB_txn *txn):
  d_parent(parent),
  d_cursors(),
  d_txn(txn)
{

}

MDB_txn *MDBROTransactionImpl::openROTransaction(MDBEnv *env, MDB_txn *parent, int flags)
{
  if(env->getRWTX())
    throw std::runtime_error("Duplicate RO transaction");
  
  /*
    A transaction and its cursors must only be used by a single thread, and a thread may only have a single transaction at a time. If MDB_NOTLS is in use, this does not apply to read-only transactions. */
  MDB_txn *result = nullptr;
  for(int tries =0 ; tries < 3; ++tries) { // it might happen twice, who knows
    if(int rc=mdb_txn_begin(env->d_env, parent, MDB_RDONLY | flags, &result)) {
      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(env->d_env, 0);
        continue;
      }

      throw std::runtime_error("Unable to start RO transaction: "+string(mdb_strerror(rc)));
    }
    break;
  }
  env->incROTX();

  return result;
}

void MDBROTransactionImpl::closeROCursors()
{
  // we need to move the vector away to ensure that the cursors don’t mess with our iteration.
  std::vector<MDBROCursor*> buf;
  std::swap(d_cursors, buf);
  for (auto &cursor: buf) {
    cursor->close();
  }
}

MDBROTransactionImpl::MDBROTransactionImpl(MDBEnv *parent, int flags):
    MDBROTransactionImpl(parent, openROTransaction(parent, nullptr, flags))
{

}

MDBROTransactionImpl::~MDBROTransactionImpl()
{
  // this is safe because C++ will not call overrides of virtual methods in destructors.
  commit();
}

void MDBROTransactionImpl::abort()
{
  closeROCursors();
  // if d_txn is non-nullptr here, either the transaction object was invalidated earlier (e.g. by moving from it), or it is an RW transaction which has already cleaned up the d_txn pointer (with an abort).
  if (d_txn) {
    d_parent->decROTX();
    mdb_txn_abort(d_txn); // this appears to work better than abort for r/o database opening
    d_txn = nullptr;
  }
}

void MDBROTransactionImpl::commit()
{
  closeROCursors();
  // if d_txn is non-nullptr here, either the transaction object was invalidated earlier (e.g. by moving from it), or it is an RW transaction which has already cleaned up the d_txn pointer (with an abort).
  if (d_txn) {
    d_parent->decROTX();
    mdb_txn_commit(d_txn); // this appears to work better than abort for r/o database opening
    d_txn = nullptr;
  }
}


void MDBRWTransactionImpl::clear(MDB_dbi dbi)
{
  if(int rc = mdb_drop(d_txn, dbi, 0)) {
    throw runtime_error("Error clearing database: " + MDBError(rc));
  }
}

MDBRWCursor MDBRWTransactionImpl::getRWCursor(const MDBDbi& dbi)
{
  MDB_cursor *cursor;
  int rc= mdb_cursor_open(d_txn, dbi, &cursor);
  if(rc) {
    throw std::runtime_error("Error creating RO cursor: "+std::string(mdb_strerror(rc)));
  }
  return MDBRWCursor(d_rw_cursors, cursor);
}

MDBRWCursor MDBRWTransactionImpl::getCursor(const MDBDbi &dbi)
{
  return getRWCursor(dbi);
}

MDBRWTransaction MDBRWTransactionImpl::getRWTransaction()
{
  MDB_txn *txn;
  if (int rc = mdb_txn_begin(environment(), *this, 0, &txn)) {
    throw std::runtime_error(std::string("failed to start child transaction: ")+mdb_strerror(rc));
  }
  // we need to increase the counter here because commit/abort on the child transaction will decrease it
  environment().incRWTX();
  return MDBRWTransaction(new MDBRWTransactionImpl(&environment(), txn));
}

MDBROTransaction MDBRWTransactionImpl::getROTransaction()
{
  return getRWTransaction();
}

MDBROTransaction MDBEnv::getROTransaction()
{
  return MDBROTransaction(new MDBROTransactionImpl(this));
}
MDBRWTransaction MDBEnv::getRWTransaction()
{
  return MDBRWTransaction(new MDBRWTransactionImpl(this));
}


void MDBRWTransactionImpl::closeRWCursors()
{
  decltype(d_rw_cursors) buf;
  std::swap(d_rw_cursors, buf);
  for (auto &cursor: buf) {
    cursor->close();
  }
}

MDBROCursor MDBROTransactionImpl::getCursor(const MDBDbi& dbi)
{
  return getROCursor(dbi);
}

MDBROCursor MDBROTransactionImpl::getROCursor(const MDBDbi &dbi)
{
  MDB_cursor *cursor;
  int rc= mdb_cursor_open(d_txn, dbi, &cursor);
  if(rc) {
    throw std::runtime_error("Error creating RO cursor: "+std::string(mdb_strerror(rc)));
  }
  return MDBROCursor(d_cursors, cursor);
}
Commit	Line	Data
6a40fc00	1	#include "lmdb-safe.hh"
	2	#include <fcntl.h>
	3	#include <mutex>
	4	#include <memory>
	5	#include <sys/stat.h>
	6	#include <string.h>
	7	#include <map>
	8
6910a23b PD	9	using std::string;
	10	using std::runtime_error;
	11	using std::tuple;
	12	using std::weak_ptr;
6a40fc00	13
	14	static string MDBError(int rc)
	15	{
	16	return mdb_strerror(rc);
	17	}
	18
	19	MDBDbi::MDBDbi(MDB_env* env, MDB_txn* txn, const string_view dbname, int flags)
	20	{
	21	// A transaction that uses this function must finish (either commit or abort) before any other transaction in the process may use this function.
	22
	23	int rc = mdb_dbi_open(txn, dbname.empty() ? 0 : &dbname[0], flags, &d_dbi);
	24	if(rc)
	25	throw std::runtime_error("Unable to open named database: " + MDBError(rc));
	26
	27	// Database names are keys in the unnamed database, and may be read but not written.
	28	}
	29
	30	MDBEnv::MDBEnv(const char* fname, int flags, int mode)
	31	{
fdfc4167 PD	32	mdb_env_create(&d_env);
	33	uint64_t mapsizeMB = (sizeof(long)==4) ? 100 : 16000;
	34	// on 32 bit platforms, there is just no room for more
	35	if(mdb_env_set_mapsize(d_env, mapsizeMB * 1048576))
6a40fc00	36	throw std::runtime_error("setting map size");
	37	/*
	38	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(),
	39	*/
	40
	41	mdb_env_set_maxdbs(d_env, 128);
	42
	43	// we need MDB_NOTLS since we rely on its semantics
	44	if(int rc=mdb_env_open(d_env, fname, flags \| MDB_NOTLS, mode)) {
	45	// If this function fails, mdb_env_close() must be called to discard the MDB_env handle.
	46	mdb_env_close(d_env);
	47	throw std::runtime_error("Unable to open database file "+std::string(fname)+": " + MDBError(rc));
	48	}
	49	}
	50
	51	void MDBEnv::incROTX()
	52	{
	53	std::lock_guard<std::mutex> l(d_countmutex);
	54	++d_ROtransactionsOut[std::this_thread::get_id()];
	55	}
	56
	57	void MDBEnv::decROTX()
	58	{
	59	std::lock_guard<std::mutex> l(d_countmutex);
	60	--d_ROtransactionsOut[std::this_thread::get_id()];
	61	}
	62
	63	void MDBEnv::incRWTX()
	64	{
	65	std::lock_guard<std::mutex> l(d_countmutex);
	66	++d_RWtransactionsOut[std::this_thread::get_id()];
	67	}
	68
	69	void MDBEnv::decRWTX()
	70	{
	71	std::lock_guard<std::mutex> l(d_countmutex);
	72	--d_RWtransactionsOut[std::this_thread::get_id()];
	73	}
	74
	75	int MDBEnv::getRWTX()
	76	{
	77	std::lock_guard<std::mutex> l(d_countmutex);
	78	return d_RWtransactionsOut[std::this_thread::get_id()];
	79	}
	80	int MDBEnv::getROTX()
	81	{
	82	std::lock_guard<std::mutex> l(d_countmutex);
	83	return d_ROtransactionsOut[std::this_thread::get_id()];
	84	}
	85
	86
	87	std::shared_ptr<MDBEnv> getMDBEnv(const char* fname, int flags, int mode)
	88	{
	89	struct Value
	90	{
	91	weak_ptr<MDBEnv> wp;
	92	int flags;
	93	};
	94
	95	static std::map<tuple<dev_t, ino_t>, Value> s_envs;
	96	static std::mutex mut;
	97
	98	struct stat statbuf;
	99	if(stat(fname, &statbuf)) {
100	if(errno != ENOENT)
101	throw std::runtime_error("Unable to stat prospective mdb database: "+string(strerror(errno)));
102	else {
103	std::lock_guard<std::mutex> l(mut);
104	auto fresh = std::make_shared<MDBEnv>(fname, flags, mode);
105	if(stat(fname, &statbuf))
106	throw std::runtime_error("Unable to stat prospective mdb database: "+string(strerror(errno)));
107	auto key = std::tie(statbuf.st_dev, statbuf.st_ino);
108	s_envs[key] = {fresh, flags};
109	return fresh;
110	}
111	}
112
113	std::lock_guard<std::mutex> l(mut);
114	auto key = std::tie(statbuf.st_dev, statbuf.st_ino);
115	auto iter = s_envs.find(key);
116	if(iter != s_envs.end()) {
117	auto sp = iter->second.wp.lock();
118	if(sp) {
119	if(iter->second.flags != flags)
120	throw std::runtime_error("Can't open mdb with differing flags");
121
122	return sp;
123	}
124	else {
125	s_envs.erase(iter); // useful if make_shared fails
126	}
127	}
128
129	auto fresh = std::make_shared<MDBEnv>(fname, flags, mode);
130	s_envs[key] = {fresh, flags};
131
132	return fresh;
133	}
134
135
136	MDBDbi MDBEnv::openDB(const string_view dbname, int flags)
137	{
138	unsigned int envflags;
139	mdb_env_get_flags(d_env, &envflags);
140	/*
141	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.
142	*/
143	std::lock_guard<std::mutex> l(d_openmut);
144
145	if(!(envflags & MDB_RDONLY)) {
146	auto rwt = getRWTransaction();
9c1e5491 PL	147	MDBDbi ret = rwt->openDB(dbname, flags);
9c1e5491 PL	148	rwt->commit();
6a40fc00	149	return ret;
	150	}
	151
	152	MDBDbi ret;
	153	{
	154	auto rwt = getROTransaction();
9c1e5491	155	ret = rwt->openDB(dbname, flags);
6a40fc00	156	}
	157	return ret;
	158	}
	159
9c1e5491 PL	160	MDBRWTransactionImpl::MDBRWTransactionImpl(MDBEnv parent, MDB_txn txn):
	161	MDBROTransactionImpl(parent, txn)
	162
	163	{
	164
	165	}
	166
	167	MDB_txn MDBRWTransactionImpl::openRWTransaction(MDBEnv env, MDB_txn *parent, int flags)
6a40fc00	168	{
9c1e5491 PL	169	MDB_txn *result;
9c1e5491 PL	170	if(env->getROTX() \|\| env->getRWTX())
6a40fc00	171	throw std::runtime_error("Duplicate RW transaction");
	172
	173	for(int tries =0 ; tries < 3; ++tries) { // it might happen twice, who knows
9c1e5491	174	if(int rc=mdb_txn_begin(env->d_env, parent, flags, &result)) {
6a40fc00	175	if(rc == MDB_MAP_RESIZED && tries < 2) {
	176	// "If the mapsize is increased by another process (..) mdb_txn_begin() will return MDB_MAP_RESIZED.
	177	// call mdb_env_set_mapsize with a size of zero to adopt the new size."
9c1e5491	178	mdb_env_set_mapsize(env->d_env, 0);
6a40fc00	179	continue;
	180	}
	181	throw std::runtime_error("Unable to start RW transaction: "+std::string(mdb_strerror(rc)));
	182	}
	183	break;
	184	}
9c1e5491 PL	185	env->incRWTX();
	186	return result;
	187	}
	188
	189	MDBRWTransactionImpl::MDBRWTransactionImpl(MDBEnv* parent, int flags):
	190	MDBRWTransactionImpl(parent, openRWTransaction(parent, nullptr, flags))
	191	{
	192	}
	193
	194	MDBRWTransactionImpl::~MDBRWTransactionImpl()
	195	{
	196	abort();
	197	}
	198
	199	void MDBRWTransactionImpl::commit()
	200	{
	201	closeRORWCursors();
	202	if (!d_txn) {
	203	return;
	204	}
	205
	206	if(int rc = mdb_txn_commit(d_txn)) {
	207	throw std::runtime_error("committing: " + std::string(mdb_strerror(rc)));
	208	}
	209	environment().decRWTX();
	210	d_txn = nullptr;
	211	}
	212
	213	void MDBRWTransactionImpl::abort()
	214	{
	215	closeRORWCursors();
	216	if (!d_txn) {
	217	return;
	218	}
	219
	220	mdb_txn_abort(d_txn);
	221	// prevent the RO destructor from cleaning up the transaction itself
	222	environment().decRWTX();
	223	d_txn = nullptr;
	224	}
	225
	226	MDBROTransactionImpl::MDBROTransactionImpl(MDBEnv parent, MDB_txn txn):
	227	d_parent(parent),
	228	d_cursors(),
	229	d_txn(txn)
	230	{
	231
6a40fc00	232	}
6a40fc00	233
9c1e5491	234	MDB_txn MDBROTransactionImpl::openROTransaction(MDBEnv env, MDB_txn *parent, int flags)
6a40fc00	235	{
9c1e5491	236	if(env->getRWTX())
6a40fc00	237	throw std::runtime_error("Duplicate RO transaction");
	238
	239	/*
	240	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. */
9c1e5491	241	MDB_txn *result = nullptr;
6a40fc00	242	for(int tries =0 ; tries < 3; ++tries) { // it might happen twice, who knows
9c1e5491	243	if(int rc=mdb_txn_begin(env->d_env, parent, MDB_RDONLY \| flags, &result)) {
6a40fc00	244	if(rc == MDB_MAP_RESIZED && tries < 2) {
	245	// "If the mapsize is increased by another process (..) mdb_txn_begin() will return MDB_MAP_RESIZED.
	246	// call mdb_env_set_mapsize with a size of zero to adopt the new size."
9c1e5491	247	mdb_env_set_mapsize(env->d_env, 0);
6a40fc00	248	continue;
	249	}
	250
	251	throw std::runtime_error("Unable to start RO transaction: "+string(mdb_strerror(rc)));
	252	}
	253	break;
	254	}
9c1e5491 PL	255	env->incROTX();
	256
	257	return result;
6a40fc00	258	}
6a40fc00	259
9c1e5491 PL	260	void MDBROTransactionImpl::closeROCursors()
	261	{
	262	// we need to move the vector away to ensure that the cursors don’t mess with our iteration.
	263	std::vector<MDBROCursor*> buf;
	264	std::swap(d_cursors, buf);
	265	for (auto &cursor: buf) {
	266	cursor->close();
	267	}
	268	}
6a40fc00	269
9c1e5491 PL	270	MDBROTransactionImpl::MDBROTransactionImpl(MDBEnv *parent, int flags):
	271	MDBROTransactionImpl(parent, openROTransaction(parent, nullptr, flags))
	272	{
6a40fc00	273
9c1e5491 PL	274	}
	275
	276	MDBROTransactionImpl::~MDBROTransactionImpl()
	277	{
	278	// this is safe because C++ will not call overrides of virtual methods in destructors.
	279	commit();
	280	}
	281
	282	void MDBROTransactionImpl::abort()
	283	{
	284	closeROCursors();
	285	// if d_txn is non-nullptr here, either the transaction object was invalidated earlier (e.g. by moving from it), or it is an RW transaction which has already cleaned up the d_txn pointer (with an abort).
	286	if (d_txn) {
	287	d_parent->decROTX();
	288	mdb_txn_abort(d_txn); // this appears to work better than abort for r/o database opening
	289	d_txn = nullptr;
	290	}
	291	}
	292
	293	void MDBROTransactionImpl::commit()
	294	{
	295	closeROCursors();
	296	// if d_txn is non-nullptr here, either the transaction object was invalidated earlier (e.g. by moving from it), or it is an RW transaction which has already cleaned up the d_txn pointer (with an abort).
	297	if (d_txn) {
	298	d_parent->decROTX();
	299	mdb_txn_commit(d_txn); // this appears to work better than abort for r/o database opening
	300	d_txn = nullptr;
	301	}
	302	}
	303
	304
	305
	306	void MDBRWTransactionImpl::clear(MDB_dbi dbi)
6a40fc00	307	{
	308	if(int rc = mdb_drop(d_txn, dbi, 0)) {
	309	throw runtime_error("Error clearing database: " + MDBError(rc));
	310	}
	311	}
	312
9c1e5491	313	MDBRWCursor MDBRWTransactionImpl::getRWCursor(const MDBDbi& dbi)
6a40fc00	314	{
9c1e5491 PL	315	MDB_cursor *cursor;
	316	int rc= mdb_cursor_open(d_txn, dbi, &cursor);
	317	if(rc) {
	318	throw std::runtime_error("Error creating RO cursor: "+std::string(mdb_strerror(rc)));
	319	}
	320	return MDBRWCursor(d_rw_cursors, cursor);
	321	}
	322
	323	MDBRWCursor MDBRWTransactionImpl::getCursor(const MDBDbi &dbi)
	324	{
	325	return getRWCursor(dbi);
	326	}
	327
	328	MDBRWTransaction MDBRWTransactionImpl::getRWTransaction()
	329	{
	330	MDB_txn *txn;
	331	if (int rc = mdb_txn_begin(environment(), *this, 0, &txn)) {
	332	throw std::runtime_error(std::string("failed to start child transaction: ")+mdb_strerror(rc));
	333	}
	334	// we need to increase the counter here because commit/abort on the child transaction will decrease it
	335	environment().incRWTX();
	336	return MDBRWTransaction(new MDBRWTransactionImpl(&environment(), txn));
	337	}
	338
	339	MDBROTransaction MDBRWTransactionImpl::getROTransaction()
	340	{
3c04af8c	341	return getRWTransaction();
6a40fc00	342	}
	343
	344	MDBROTransaction MDBEnv::getROTransaction()
	345	{
9c1e5491	346	return MDBROTransaction(new MDBROTransactionImpl(this));
6a40fc00	347	}
	348	MDBRWTransaction MDBEnv::getRWTransaction()
	349	{
9c1e5491	350	return MDBRWTransaction(new MDBRWTransactionImpl(this));
6a40fc00	351	}
	352
	353
9c1e5491	354	void MDBRWTransactionImpl::closeRWCursors()
6a40fc00	355	{
9c1e5491 PL	356	decltype(d_rw_cursors) buf;
	357	std::swap(d_rw_cursors, buf);
	358	for (auto &cursor: buf) {
	359	cursor->close();
	360	}
6a40fc00	361	}
6a40fc00	362
9c1e5491	363	MDBROCursor MDBROTransactionImpl::getCursor(const MDBDbi& dbi)
6a40fc00	364	{
9c1e5491	365	return getROCursor(dbi);
6a40fc00	366	}
6a40fc00	367
9c1e5491 PL	368	MDBROCursor MDBROTransactionImpl::getROCursor(const MDBDbi &dbi)
	369	{
	370	MDB_cursor *cursor;
	371	int rc= mdb_cursor_open(d_txn, dbi, &cursor);
	372	if(rc) {
	373	throw std::runtime_error("Error creating RO cursor: "+std::string(mdb_strerror(rc)));
	374	}
	375	return MDBROCursor(d_cursors, cursor);
	376	}