#ifndef _GLIM_MDB_HPP_INCLUDED #define _GLIM_MDB_HPP_INCLUDED /** * A C++ wrapper around MDB (http://www.symas.com/mdb/). * @code Copyright 2012 Kozarezov Artem Aleksandrovich Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. * @endcode * @file */ #include #include #include #include #include #include #include #include // htonl, ntohl #include "gstring.hpp" namespace glim { struct MdbEx: public std::runtime_error {MdbEx (std::string message): std::runtime_error (message) {}}; template inline void mdbSerialize (gstring& bytes, const T& data) { gstring_stream stream (bytes); boost::archive::binary_oarchive oa (stream, boost::archive::no_header); oa << data; } template inline void mdbDeserialize (const gstring& bytes, V& data) { gstring_stream stream (const_cast (bytes)); boost::archive::binary_iarchive ia (stream, boost::archive::no_header); ia >> data; } /** uint32_t keys are stored big-endian (network byte order) in order to be compatible with lexicographic ordering. */ template <> inline void mdbSerialize (gstring& bytes, const uint32_t& ui) { uint32_t nui = htonl (ui); bytes.append ((const char*) &nui, sizeof (uint32_t));} /** Deserialize uint32_t from big-endian (network byte order). */ template <> inline void mdbDeserialize (const gstring& bytes, uint32_t& ui) { if (bytes.size() != sizeof (uint32_t)) throw MdbEx ("Not uint32_t, wrong number of bytes"); uint32_t nui = * (uint32_t*) bytes.data(); ui = ntohl (nui);} /** If the data is `gstring` then use the data's buffer directly, no copy. */ template <> inline void mdbSerialize (gstring& bytes, const gstring& data) { bytes = gstring (0, (void*) data.data(), false, data.length());} /** Deserializing into `gstring` copies the bytes into it, reusing its buffer. */ template <> inline void mdbDeserialize (const gstring& bytes, gstring& data) { data.clear() << bytes;} /** * Header-only C++ wrapper around OpenLDAP-MDB.\n * Uses Boost Serialization to pack keys and values (glim::gstring can be used for raw bytes).\n * Allows semi-automatic indexing with triggers.\n * Known issues: http://www.openldap.org/its/index.cgi?findid=7448 */ struct Mdb { std::shared_ptr _env; MDB_dbi _dbi = 0; typedef std::unique_ptr Transaction; /** Holds the current key and value of the Iterator. */ struct IteratorEntry { MDB_val _key = {0, 0}, _val = {0, 0}; /** Zero-copy view of the current key bytes. Should *not* be used after the Iterator is changed or destroyed. */ const gstring keyView() const {return gstring (0, _key.mv_data, false, _key.mv_size, true);} // Zero copy. /** Zero-copy view of the current value bytes. Should *not* be used after the Iterator is changed or destroyed. */ const gstring valueView() const {return gstring (0, _val.mv_data, false, _val.mv_size, true);} // Zero copy. /** Deserialize into `key`. */ template void getKey (T& key) const {mdbDeserialize (keyView(), key);} /** Deserialize the key into a temporary and return it. */ template T getKey() const {T key; getKey (key); return key;} /** Deserialize into `value`. */ template void getValue (T& value) const {mdbDeserialize (valueView(), value);} /** Deserialize the value into a temporary and return it. */ template T getValue() const {T value; getValue (value); return value;} }; /** Holds the Iterator's unique transaction and cursor, allowing the Iterator to be copied. */ struct IteratorImpl: boost::noncopyable { Mdb* _mdb; Transaction _txn; MDB_cursor* _cur; IteratorImpl (Mdb* mdb, Transaction&& txn, MDB_cursor* cur): _mdb (mdb), _txn (std::move (txn)), _cur (cur) {} ~IteratorImpl() { if (_cur) {::mdb_cursor_close (_cur); _cur = nullptr;} if (_mdb && _txn) {_mdb->commitTransaction (_txn); _mdb = nullptr;} } }; /** Wraps MDB cursor and cursor's transaction. */ struct Iterator: public boost::iterator_facade { std::shared_ptr _impl; // Iterator might be copied around, thus we keep the unique things in IteratorImpl. IteratorEntry _entry; bool _stayInKey = false; Iterator (const Iterator&) = default; Iterator (Iterator&&) = default; /** Iterate from the beginning or the end of the database. * @param position can be MDB_FIRST or MDB_LAST */ Iterator (Mdb* mdb, int position = 0): _stayInKey (false) { Transaction txn (mdb->beginTransaction()); MDB_cursor* cur = nullptr; int rc = ::mdb_cursor_open (txn.get(), mdb->_dbi, &cur); if (rc) throw MdbEx ("mdb_cursor_open"); _impl = std::make_shared (mdb, std::move (txn), cur); if (position == ::MDB_FIRST || position == ::MDB_LAST) { rc = ::mdb_cursor_get (cur, &_entry._key, &_entry._val, (MDB_cursor_op) position); if (rc) throw MdbEx ("mdb_cursor_get"); } } /** Iterate over `key` values. * @param stayInKey if `false` then iterator can go farther than the `key`. */ Iterator (Mdb* mdb, const gstring& key, bool stayInKey = true): _stayInKey (stayInKey) { Transaction txn (mdb->beginTransaction()); MDB_cursor* cur = nullptr; int rc = ::mdb_cursor_open (txn.get(), mdb->_dbi, &cur); if (rc) throw MdbEx ("mdb_cursor_open"); _impl = std::make_shared (mdb, std::move (txn), cur); _entry._key = {key.size(), (void*) key.data()}; rc = ::mdb_cursor_get (cur, &_entry._key, &_entry._val, ::MDB_SET_KEY); if (rc == MDB_NOTFOUND) {_entry._key = {0, 0}; _entry._val = {0, 0};} else if (rc) throw MdbEx ("mdb_cursor_get"); } struct EndIteratorFlag {}; /** The "end" iterator does not open an MDB transaction (this is essential for having a pair of iterators without a deadlock). */ Iterator (EndIteratorFlag): _stayInKey (false) {} /** True if the iterator isn't pointing anywhere. */ bool end() const {return _entry._key.mv_size == 0;} bool equal (const Iterator& other) const { IteratorImpl* impl = _impl.get(); if (mdb_cmp (impl->_txn.get(), impl->_mdb->_dbi, &_entry._key, &other._entry._key)) return false; if (mdb_dcmp (impl->_txn.get(), impl->_mdb->_dbi, &_entry._val, &other._entry._val)) return false; return true; } IteratorEntry& dereference() const { // NB: Boost iterator_facade expects the `dereference` to be a `const` method. // I guess Iterator is not modified, so the `dereference` is `const`, even though the Entry can be modified. return const_cast (_entry);} void increment() { int rc = ::mdb_cursor_get (_impl->_cur, &_entry._key, &_entry._val, _stayInKey ? ::MDB_NEXT_DUP : ::MDB_NEXT); if (rc) {_entry._key = {0,0}; _entry._val = {0,0};} } void decrement() { int rc = ::mdb_cursor_get (_impl->_cur, &_entry._key, &_entry._val, _stayInKey ? ::MDB_PREV_DUP : ::MDB_PREV); if (rc) {_entry._key = {0,0}; _entry._val = {0,0};} } }; Iterator begin() {return Iterator (this, ::MDB_FIRST);} const Iterator end() {return Iterator (Iterator::EndIteratorFlag());} /** Position the cursor at the first `key` record.\n * The iterator increment will use `MDB_NEXT_DUP`, staying withing the `key`.\n * See also the `all` method. */ template Iterator values (const K& key) { char kbuf[64]; // Allow up to 64 bytes to be serialized without heap allocations. gstring kbytes (sizeof (kbuf), kbuf, false, 0); mdbSerialize (kbytes, key); return Iterator (this, kbytes); } /** Range over the `key` values.\n * See also the `all` method. */ template boost::iterator_range valuesRange (const K& key) {return boost::iterator_range (values (key), end());} struct Trigger { virtual gstring getTriggerName() const {return C2GSTRING ("defaultTriggerName");}; virtual void add (Mdb& mdb, void* key, gstring& kbytes, void* value, gstring& vbytes, Transaction& txn) = 0; virtual void erase (Mdb& mdb, void* key, gstring& kbytes, Transaction& txn) = 0; virtual void eraseKV (Mdb& mdb, void* key, gstring& kbytes, void* value, gstring& vbytes, Transaction& txn) = 0; }; std::map> _triggers; void setTrigger (std::shared_ptr trigger) { _triggers[trigger->getTriggerName()] = trigger; } /** `flags` can be `MDB_RDONLY` */ Transaction beginTransaction (unsigned flags = 0) { MDB_txn* txn = 0; int rc = ::mdb_txn_begin (_env.get(), nullptr, flags, &txn); if (rc) throw MdbEx (std::string ("mdb_txn_begin: ") + ::strerror (rc)); return Transaction (txn, ::mdb_txn_abort); } void commitTransaction (Transaction& txn) { int rc = ::mdb_txn_commit (txn.get()); txn.release(); // Must prevent `mdb_txn_abort` from happening (even if rc != 0). if (rc) throw MdbEx (std::string ("mdb_txn_commit: ") + ::strerror (rc)); } virtual unsigned envFlags (uint8_t sync) { unsigned flags = MDB_NOSUBDIR; if (sync < 1) flags |= MDB_NOSYNC; else if (sync < 2) flags |= MDB_NOMETASYNC; return flags; } /** Used before `mdb_env_open`. By default sets the number of database to 32. */ virtual void envConf (MDB_env* env) { int rc = ::mdb_env_set_maxdbs (env, 32); if (rc) throw MdbEx (std::string ("envConf: ") + ::strerror (rc)); } virtual void dbFlags (unsigned& flags) {} protected: void open (const char* dbName, bool dup) { auto txn = beginTransaction(); unsigned flags = MDB_CREATE; if (dup) flags |= MDB_DUPSORT; dbFlags (flags); int rc = ::mdb_open (txn.get(), dbName, flags, &_dbi); if (rc) throw MdbEx (std::string ("mdb_open (") + dbName + "): " + ::strerror (rc)); commitTransaction (txn); } public: /** Opens MDB environment and MDB database. */ Mdb (const char* path, size_t maxSizeMb = 1024, const char* dbName = "main", uint8_t sync = 0, bool dup = true, mode_t mode = 0660) { MDB_env* env = 0; int rc = ::mdb_env_create (&env); if (rc) throw MdbEx (std::string ("mdb_env_create: ") + ::strerror (rc)); _env.reset (env, ::mdb_env_close); rc = ::mdb_env_set_mapsize (env, maxSizeMb * 1024 * 1024); if (rc) throw MdbEx (std::string ("mdb_env_set_mapsize: ") + ::strerror (rc)); envConf (env); rc = ::mdb_env_open (env, path, envFlags (sync), mode); _dbi = 0; open (dbName, dup); } /** Opens MDB database in the provided environment. */ Mdb (std::shared_ptr env, const char* dbName, bool dup = true): _env (env), _dbi (0) { open (dbName, dup); } template void add (const K& key, const V& value, Transaction& txn) { char kbuf[64]; // Allow up to 64 bytes to be serialized without heap allocations. gstring kbytes (sizeof (kbuf), kbuf, false, 0); mdbSerialize (kbytes, key); MDB_val mkey = {kbytes.size(), (void*) kbytes.data()}; char vbuf[64]; // Allow up to 64 bytes to be serialized without heap allocations. gstring vbytes (sizeof (vbuf), vbuf, false, 0); mdbSerialize (vbytes, value); MDB_val mvalue = {vbytes.size(), (void*) vbytes.data()}; for (auto& trigger: _triggers) trigger.second->add (*this, (void*) &key, kbytes, (void*) &value, vbytes, txn); int rc = ::mdb_put (txn.get(), _dbi, &mkey, &mvalue, 0); if (rc) throw MdbEx (std::string ("mdb_put: ") + ::strerror (rc)); } template void add (const K& key, const V& value) { Transaction txn (beginTransaction()); add (key, value, txn); commitTransaction (txn); } template bool first (const K& key, V& value, Transaction& txn) { char kbuf[64]; // Allow up to 64 bytes to be serialized without heap allocations. gstring kbytes (sizeof (kbuf), kbuf, false, 0); mdbSerialize (kbytes, key); MDB_val mkey = {kbytes.size(), (void*) kbytes.data()}; MDB_val mvalue; int rc = ::mdb_get (txn.get(), _dbi, &mkey, &mvalue); if (rc == MDB_NOTFOUND) return false; if (rc) throw MdbEx (std::string ("mdb_get: ") + ::strerror (rc)); gstring vstr (0, mvalue.mv_data, false, mvalue.mv_size); mdbDeserialize (vstr, value); return true; } template bool first (const K& key, V& value) { Transaction txn (beginTransaction (MDB_RDONLY)); bool rb = first (key, value, txn); commitTransaction (txn); return rb; } /** Iterate over `key` values until `visitor` returns `false`. Return the number of values visited. */ template int32_t all (const K& key, std::function visitor, Transaction& txn) { char kbuf[64]; // Allow up to 64 bytes to be serialized without heap allocations. gstring kbytes (sizeof (kbuf), kbuf, false, 0); mdbSerialize (kbytes, key); MDB_val mkey = {kbytes.size(), (void*) kbytes.data()}; MDB_cursor* cur = 0; int rc = ::mdb_cursor_open (txn.get(), _dbi, &cur); if (rc) throw MdbEx (std::string ("mdb_cursor_open: ") + ::strerror (rc)); std::unique_ptr curHolder (cur, ::mdb_cursor_close); MDB_val mval = {0, 0}; rc = ::mdb_cursor_get (cur, &mkey, &mval, ::MDB_SET_KEY); if (rc == MDB_NOTFOUND) return 0; if (rc) throw MdbEx (std::string ("mdb_cursor_get: ") + ::strerror (rc)); V value; gstring vstr (0, mval.mv_data, false, mval.mv_size); mdbDeserialize (vstr, value); bool goOn = visitor (value); int32_t count = 1; while (goOn) { rc = ::mdb_cursor_get (cur, &mkey, &mval, ::MDB_NEXT_DUP); if (rc == MDB_NOTFOUND) return count; if (rc) throw MdbEx (std::string ("mdb_cursor_get: ") + ::strerror (rc)); gstring vstr (0, mval.mv_data, false, mval.mv_size); mdbDeserialize (vstr, value); goOn = visitor (value); ++count; } return count; } /** Iterate over `key` values until `visitor` returns `false`. Return the number of values visited. */ template int32_t all (const K& key, std::function visitor) { Transaction txn (beginTransaction (MDB_RDONLY)); int32_t count = all (key, visitor, txn); commitTransaction (txn); return count; } template bool eraseKV (const K& key, const V& value, Transaction& txn) { char kbuf[64]; // Allow up to 64 bytes to be serialized without heap allocations. gstring kbytes (sizeof (kbuf), kbuf, false, 0); mdbSerialize (kbytes, key); if (kbytes.empty()) throw MdbEx ("eraseKV: key is empty"); MDB_val mkey = {kbytes.size(), (void*) kbytes.data()}; char vbuf[64]; // Allow up to 64 bytes to be serialized without heap allocations. gstring vbytes (sizeof (vbuf), vbuf, false, 0); mdbSerialize (vbytes, value); MDB_val mvalue = {vbytes.size(), (void*) vbytes.data()}; for (auto& trigger: _triggers) trigger.second->eraseKV (*this, (void*) &key, kbytes, (void*) &value, vbytes, txn); int rc = ::mdb_del (txn.get(), _dbi, &mkey, &mvalue); if (rc == MDB_NOTFOUND) return false; if (rc) throw MdbEx (std::string ("mdb_del: ") + ::strerror (rc)); return true; } template bool eraseKV (const K& key, const V& value) { Transaction txn (beginTransaction()); bool rb = eraseKV (key, value, txn); commitTransaction (txn); return rb; } /** Erase all values of the `key`. */ template bool erase (const K& key, Transaction& txn) { char kbuf[64]; // Allow up to 64 bytes to be serialized without heap allocations. gstring kbytes (sizeof (kbuf), kbuf, false, 0); mdbSerialize (kbytes, key); if (kbytes.empty()) throw MdbEx ("erase: key is empty"); MDB_val mkey = {kbytes.size(), (void*) kbytes.data()}; for (auto& trigger: _triggers) trigger.second->erase (*this, (void*) &key, kbytes, txn); int rc = ::mdb_del (txn.get(), _dbi, &mkey, nullptr); if (rc == MDB_NOTFOUND) return false; if (rc) throw MdbEx (std::string ("mdb_del: ") + ::strerror (rc)); return true; } /** Erase all values of the `key`. */ template bool erase (const K& key) { Transaction txn (beginTransaction()); bool rb = erase (key, txn); commitTransaction (txn); return rb; } static void test (Mdb& mdb) { mdb.add (std::string ("foo"), std::string ("bar")); // NB: "MDB_DUPSORT doesn't allow duplicate duplicates" (Howard Chu) mdb.add (std::string ("foo"), std::string ("bar")); mdb.add ((uint32_t) 123, 1); mdb.add ((uint32_t) 123, 2); mdb.add (C2GSTRING ("foo"), 3); mdb.add (C2GSTRING ("foo"), 4); mdb.add (C2GSTRING ("gsk"), C2GSTRING ("gsv")); string ts; int ti; gstring tgs; auto fail = [](string msg) {throw std::runtime_error ("assertion failed: " + msg);}; if (!mdb.first (std::string ("foo"), ts) || ts != "bar") fail ("!foo=bar"); if (!mdb.first ((uint32_t) 123, ti) || ti < 1 || ti > 2) fail ("!123"); if (!mdb.first (C2GSTRING ("foo"), ti) || ti < 3 || ti > 4) fail ("!foo=3,4"); if (!mdb.first (C2GSTRING ("gsk"), tgs) || tgs != "gsv") fail ("!gsk=gsv"); // Test range-based for. int count = 0; bool haveGskGsv = false; for (auto&& entry: mdb) { if (!entry._key.mv_size || !entry._val.mv_size) fail ("!entry"); if (entry.keyView() == "gsk") { if (entry.getKey() != "gsk") fail ("getKey(gsk)!=gsk"); if (entry.getValue() != "gsv") fail ("getValue(gsk)!=gsv"); haveGskGsv = true; } ++count;} if (count != 6) fail ("count!=6"); // foo=bar, 123=1, 123=2, foo=3, foo=4, gsk=gsv if (!haveGskGsv) fail ("!haveGskGsv"); // Test `values`. count = 0; int sum = 0; for (auto&& entry: mdb.valuesRange ((uint32_t) 123)) { if (entry.getKey() != (uint32_t) 123) fail("values(123).key!=123"); ++count; sum += entry.getValue(); } if (count != 2) fail ("count(123)!=2"); if (sum != 3) fail ("sum(123)!=3"); if (!mdb.eraseKV ((uint32_t) 123, 1)) fail ("!eraseKV(123,1)"); if (!mdb.first ((uint32_t) 123, ti) || ti != 2) fail ("!123=2"); if (!mdb.eraseKV ((uint32_t) 123, 2)) fail ("!eraseKV(123,2)"); if (mdb.first ((uint32_t) 123, ti)) fail ("123"); if (!mdb.erase (C2GSTRING ("foo"))) fail ("!erase(g(foo))"); if (mdb.first (C2GSTRING ("foo"), ti)) fail ("foo"); if (!mdb.erase (std::string ("foo"))) fail ("!erase(str(foo))"); { // We've erased "123" and "foo", the only key left is "gsk" (gsk=gsv), let's test the iterator boundaries on this small dataset. auto&& it = mdb.begin(); if (it->getKey() != "gsk") fail ("first key !gsk " + it->keyView().str()); if (!(++it).end()) fail ("++it != end"); if ((--it)->getKey() != "gsk") fail ("can't go back to gsk"); if (!(--it).end()) fail ("--it != end"); if ((++it)->getKey() != "gsk") fail ("can't go forward to gsk"); } struct SimpleIndexTrigger: public Trigger { const char* _name; Mdb _indexDb; SimpleIndexTrigger (Mdb& mdb, const char* name = "index"): _name (name), _indexDb (mdb._env, name) {} gstring getTriggerName() {return gstring (0, (void*) _name, false, strlen (_name), true);} void add (Mdb& mdb, void* key, gstring& kbytes, void* value, gstring& vbytes, Transaction& txn) { MDB_val mkey = {vbytes.size(), (void*) vbytes.data()}; MDB_val mvalue = {kbytes.size(), (void*) kbytes.data()}; int rc = ::mdb_put (txn.get(), _indexDb._dbi, &mkey, &mvalue, 0); if (rc) throw MdbEx (std::string ("index, mdb_put: ") + ::strerror (rc)); } void erase (Mdb& mdb, void* ekey, gstring& kbytes, Transaction& txn) { // Get all the values and remove them from the index. MDB_cursor* cur = 0; int rc = ::mdb_cursor_open (txn.get(), mdb._dbi, &cur); if (rc) throw MdbEx (std::string ("index, erase, mdb_cursor_open: ") + ::strerror (rc)); std::unique_ptr curHolder (cur, ::mdb_cursor_close); MDB_val mkey = {kbytes.size(), (void*) kbytes.data()}, val = {0, 0}; rc = ::mdb_cursor_get (cur, &mkey, &val, ::MDB_SET_KEY); if (rc == MDB_NOTFOUND) return; if (rc) throw MdbEx (std::string ("index, erase, mdb_cursor_get: ") + ::strerror (rc)); rc = ::mdb_del (txn.get(), _indexDb._dbi, &val, &mkey); if (rc && rc != MDB_NOTFOUND) throw MdbEx (std::string ("index, erase, mdb_del: ") + ::strerror (rc)); for (;;) { rc = ::mdb_cursor_get (cur, &mkey, &val, ::MDB_NEXT_DUP); if (rc == MDB_NOTFOUND) return; if (rc) throw MdbEx (std::string ("index, erase, mdb_cursor_get: ") + ::strerror (rc)); rc = ::mdb_del (txn.get(), _indexDb._dbi, &val, &mkey); if (rc && rc != MDB_NOTFOUND) throw MdbEx (std::string ("index, erase, mdb_del: ") + ::strerror (rc)); } } void eraseKV (Mdb& mdb, void* key, gstring& kbytes, void* value, gstring& vbytes, Transaction& txn) { MDB_val mkey = {vbytes.size(), (void*) vbytes.data()}; MDB_val mvalue = {kbytes.size(), (void*) kbytes.data()}; int rc = ::mdb_del (txn.get(), _indexDb._dbi, &mkey, &mvalue); if (rc && rc != MDB_NOTFOUND) throw MdbEx (std::string ("index, mdb_del: ") + ::strerror (rc)); } }; auto indexTrigger = std::make_shared (mdb); mdb.setTrigger (indexTrigger); auto& indexDb = indexTrigger->_indexDb; mdb.erase (C2GSTRING ("gsk")); // NB: "gsk" wasn't indexed here. `IndexTrigger.erase` should handle this gracefully. // Add indexed. mdb.add (C2GSTRING ("ik"), C2GSTRING ("iv1")); mdb.add (C2GSTRING ("ik"), string ("iv2")); mdb.add (C2GSTRING ("ik"), 3); // Check the index. gstring ik; if (!indexDb.first (C2GSTRING ("iv1"), ik) || ik != "ik") fail ("!iv1=ik"); if (!indexDb.first (string ("iv2"), ik) || ik != "ik") fail ("!iv2=ik"); if (!indexDb.first (3, ik) || ik != "ik") fail ("!iv3=ik"); // Remove indexed. mdb.eraseKV (C2GSTRING ("ik"), string ("iv2")); // Check the index. if (!indexDb.first (C2GSTRING ("iv1"), ik) || ik != "ik") fail ("!iv1=ik"); if (indexDb.first (string ("iv2"), ik)) fail ("iv2=ik"); if (!indexDb.first (3, ik) || ik != "ik") fail ("!iv3=ik"); // Remove indexed. mdb.erase (C2GSTRING ("ik")); // Check the index. if (indexDb.first (C2GSTRING ("iv1"), ik)) fail ("iv1"); if (indexDb.first (3, ik)) fail ("iv3"); // Check the data. if (mdb.first (C2GSTRING ("ik"), ik)) fail ("ik"); } virtual ~Mdb() { _triggers.clear(); // Destroy triggers before closing the database. if (_dbi) {::mdb_close (_env.get(), _dbi); _dbi = 0;} } }; } // namespace glim #endif // _GLIM_MDB_HPP_INCLUDED