ext/lmdb-safe/lmdb-safe.cc

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