diff options
Diffstat (limited to 'plugins/Dbx_kyoto/src/kyotocabinet/kcdbext.h')
| -rw-r--r-- | plugins/Dbx_kyoto/src/kyotocabinet/kcdbext.h | 1688 |
1 files changed, 0 insertions, 1688 deletions
diff --git a/plugins/Dbx_kyoto/src/kyotocabinet/kcdbext.h b/plugins/Dbx_kyoto/src/kyotocabinet/kcdbext.h deleted file mode 100644 index 9a6beb1360..0000000000 --- a/plugins/Dbx_kyoto/src/kyotocabinet/kcdbext.h +++ /dev/null @@ -1,1688 +0,0 @@ -/************************************************************************************************* - * Database extension - * Copyright (C) 2009-2012 FAL Labs - * This file is part of Kyoto Cabinet. - * This program is free software: you can redistribute it and/or modify it under the terms of - * the GNU General Public License as published by the Free Software Foundation, either version - * 3 of the License, or any later version. - * 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, see <http://www.gnu.org/licenses/>. - *************************************************************************************************/ - - -#ifndef _KCDBEXT_H // duplication check -#define _KCDBEXT_H - -#include <kccommon.h> -#include <kcutil.h> -#include <kcthread.h> -#include <kcfile.h> -#include <kccompress.h> -#include <kccompare.h> -#include <kcmap.h> -#include <kcregex.h> -#include <kcdb.h> -#include <kcplantdb.h> -#include <kcprotodb.h> -#include <kcstashdb.h> -#include <kccachedb.h> -#include <kchashdb.h> -#include <kcdirdb.h> -#include <kcpolydb.h> - -namespace kyotocabinet { // common namespace - - -/** - * MapReduce framework. - * @note Although this framework is not distributed or concurrent, it is useful for aggregate - * calculation with less CPU loading and less memory usage. - */ -class MapReduce { - public: - class ValueIterator; - private: - class FlushThread; - class ReduceTaskQueue; - class MapVisitor; - struct MergeLine; - /** An alias of vector of loaded values. */ - typedef std::vector<std::string> Values; - /** The default number of temporary databases. */ - static const size_t DEFDBNUM = 8; - /** The maxinum number of temporary databases. */ - static const size_t MAXDBNUM = 256; - /** The default cache limit. */ - static const int64_t DEFCLIM = 512LL << 20; - /** The default cache bucket numer. */ - static const int64_t DEFCBNUM = 1048583LL; - /** The bucket number of temprary databases. */ - static const int64_t DBBNUM = 512LL << 10; - /** The page size of temprary databases. */ - static const int32_t DBPSIZ = 32768; - /** The mapped size of temprary databases. */ - static const int64_t DBMSIZ = 516LL * 4096; - /** The page cache capacity of temprary databases. */ - static const int64_t DBPCCAP = 16LL << 20; - /** The default number of threads in parallel mode. */ - static const size_t DEFTHNUM = 8; - /** The number of slots of the record lock. */ - static const int32_t RLOCKSLOT = 256; - public: - /** - * Value iterator for the reducer. - */ - class ValueIterator { - friend class MapReduce; - public: - /** - * Get the next value. - * @param sp the pointer to the variable into which the size of the region of the return - * value is assigned. - * @return the pointer to the next value region, or NULL if no value remains. - */ - const char* next(size_t* sp) { - _assert_(sp); - if (!vptr_) { - if (vit_ == vend_) return NULL; - vptr_ = vit_->data(); - vsiz_ = vit_->size(); - vit_++; - } - uint64_t vsiz; - size_t step = readvarnum(vptr_, vsiz_, &vsiz); - vptr_ += step; - vsiz_ -= step; - const char* vbuf = vptr_; - *sp = vsiz; - vptr_ += vsiz; - vsiz_ -= vsiz; - if (vsiz_ < 1) vptr_ = NULL; - return vbuf; - } - private: - /** - * Default constructor. - */ - explicit ValueIterator(Values::const_iterator vit, Values::const_iterator vend) : - vit_(vit), vend_(vend), vptr_(NULL), vsiz_(0) { - _assert_(true); - } - /** - * Destructor. - */ - ~ValueIterator() { - _assert_(true); - } - /** Dummy constructor to forbid the use. */ - ValueIterator(const ValueIterator&); - /** Dummy Operator to forbid the use. */ - ValueIterator& operator =(const ValueIterator&); - /** The current iterator of loaded values. */ - Values::const_iterator vit_; - /** The ending iterator of loaded values. */ - Values::const_iterator vend_; - /** The pointer of the current value. */ - const char* vptr_; - /** The size of the current value. */ - size_t vsiz_; - }; - /** - * Execution options. - */ - enum Option { - XNOLOCK = 1 << 0, ///< avoid locking against update operations - XPARAMAP = 1 << 1, ///< run mappers in parallel - XPARARED = 1 << 2, ///< run reducers in parallel - XPARAFLS = 1 << 3, ///< run cache flushers in parallel - XNOCOMP = 1 << 8 ///< avoid compression of temporary databases - }; - /** - * Default constructor. - */ - explicit MapReduce() : - db_(NULL), rcomp_(NULL), tmpdbs_(NULL), dbnum_(DEFDBNUM), dbclock_(0), - mapthnum_(DEFTHNUM), redthnum_(DEFTHNUM), flsthnum_(DEFTHNUM), - cache_(NULL), csiz_(0), clim_(DEFCLIM), cbnum_(DEFCBNUM), flsths_(NULL), - redtasks_(NULL), redaborted_(false), rlocks_(NULL) { - _assert_(true); - } - /** - * Destructor. - */ - virtual ~MapReduce() { - _assert_(true); - } - /** - * Map a record data. - * @param kbuf the pointer to the key region. - * @param ksiz the size of the key region. - * @param vbuf the pointer to the value region. - * @param vsiz the size of the value region. - * @return true on success, or false on failure. - * @note This function can call the MapReduce::emit method to emit a record. To avoid - * deadlock, any explicit database operation must not be performed in this function. - */ - virtual bool map(const char* kbuf, size_t ksiz, const char* vbuf, size_t vsiz) = 0; - /** - * Reduce a record data. - * @param kbuf the pointer to the key region. - * @param ksiz the size of the key region. - * @param iter the iterator to get the values. - * @return true on success, or false on failure. - * @note To avoid deadlock, any explicit database operation must not be performed in this - * function. - */ - virtual bool reduce(const char* kbuf, size_t ksiz, ValueIterator* iter) = 0; - /** - * Preprocess the map operations. - * @return true on success, or false on failure. - * @note This function can call the MapReduce::emit method to emit a record. To avoid - * deadlock, any explicit database operation must not be performed in this function. - */ - virtual bool preprocess() { - _assert_(true); - return true; - } - /** - * Mediate between the map and the reduce phases. - * @return true on success, or false on failure. - * @note This function can call the MapReduce::emit method to emit a record. To avoid - * deadlock, any explicit database operation must not be performed in this function. - */ - virtual bool midprocess() { - _assert_(true); - return true; - } - /** - * Postprocess the reduce operations. - * @return true on success, or false on failure. - * @note To avoid deadlock, any explicit database operation must not be performed in this - * function. - */ - virtual bool postprocess() { - _assert_(true); - return true; - } - /** - * Process a log message. - * @param name the name of the event. - * @param message a supplement message. - * @return true on success, or false on failure. - */ - virtual bool log(const char* name, const char* message) { - _assert_(name && message); - return true; - } - /** - * Execute the MapReduce process about a database. - * @param db the source database. - * @param tmppath the path of a directory for the temporary data storage. If it is an empty - * string, temporary data are handled on memory. - * @param opts the optional features by bitwise-or: MapReduce::XNOLOCK to avoid locking - * against update operations by other threads, MapReduce::XPARAMAP to run the mapper in - * parallel, MapReduce::XPARARED to run the reducer in parallel, MapReduce::XNOCOMP to avoid - * compression of temporary databases. - * @return true on success, or false on failure. - */ - bool execute(BasicDB* db, const std::string& tmppath = "", uint32_t opts = 0) { - int64_t count = db->count(); - if (count < 0) { - if (db->error() != BasicDB::Error::NOIMPL) return false; - count = 0; - } - bool err = false; - double stime, etime; - db_ = db; - rcomp_ = LEXICALCOMP; - BasicDB* idb = db; - if (typeid(*db) == typeid(PolyDB)) { - PolyDB* pdb = (PolyDB*)idb; - idb = pdb->reveal_inner_db(); - } - const std::type_info& info = typeid(*idb); - if (info == typeid(GrassDB)) { - GrassDB* gdb = (GrassDB*)idb; - rcomp_ = gdb->rcomp(); - } else if (info == typeid(TreeDB)) { - TreeDB* tdb = (TreeDB*)idb; - rcomp_ = tdb->rcomp(); - } else if (info == typeid(ForestDB)) { - ForestDB* fdb = (ForestDB*)idb; - rcomp_ = fdb->rcomp(); - } - tmpdbs_ = new BasicDB*[dbnum_]; - if (tmppath.empty()) { - if (!logf("prepare", "started to open temporary databases on memory")) err = true; - stime = time(); - for (size_t i = 0; i < dbnum_; i++) { - GrassDB* gdb = new GrassDB; - int32_t myopts = 0; - if (!(opts & XNOCOMP)) myopts |= GrassDB::TCOMPRESS; - gdb->tune_options(myopts); - gdb->tune_buckets(DBBNUM / 2); - gdb->tune_page(DBPSIZ); - gdb->tune_page_cache(DBPCCAP); - gdb->tune_comparator(rcomp_); - gdb->open("%", GrassDB::OWRITER | GrassDB::OCREATE | GrassDB::OTRUNCATE); - tmpdbs_[i] = gdb; - } - etime = time(); - if (!logf("prepare", "opening temporary databases finished: time=%.6f", etime - stime)) - err = true; - if (err) { - delete[] tmpdbs_; - return false; - } - } else { - File::Status sbuf; - if (!File::status(tmppath, &sbuf) || !sbuf.isdir) { - db->set_error(_KCCODELINE_, BasicDB::Error::NOREPOS, "no such directory"); - delete[] tmpdbs_; - return false; - } - if (!logf("prepare", "started to open temporary databases under %s", tmppath.c_str())) - err = true; - stime = time(); - uint32_t pid = getpid() & UINT16MAX; - uint32_t tid = Thread::hash() & UINT16MAX; - uint32_t ts = time() * 1000; - for (size_t i = 0; i < dbnum_; i++) { - std::string childpath = - strprintf("%s%cmr-%04x-%04x-%08x-%03d%ckct", - tmppath.c_str(), File::PATHCHR, pid, tid, ts, (int)(i + 1), File::EXTCHR); - TreeDB* tdb = new TreeDB; - int32_t myopts = TreeDB::TSMALL | TreeDB::TLINEAR; - if (!(opts & XNOCOMP)) myopts |= TreeDB::TCOMPRESS; - tdb->tune_options(myopts); - tdb->tune_buckets(DBBNUM); - tdb->tune_page(DBPSIZ); - tdb->tune_map(DBMSIZ); - tdb->tune_page_cache(DBPCCAP); - tdb->tune_comparator(rcomp_); - if (!tdb->open(childpath, TreeDB::OWRITER | TreeDB::OCREATE | TreeDB::OTRUNCATE)) { - const BasicDB::Error& e = tdb->error(); - db->set_error(_KCCODELINE_, e.code(), e.message()); - err = true; - } - tmpdbs_[i] = tdb; - } - etime = time(); - if (!logf("prepare", "opening temporary databases finished: time=%.6f", etime - stime)) - err = true; - if (err) { - for (size_t i = 0; i < dbnum_; i++) { - delete tmpdbs_[i]; - } - delete[] tmpdbs_; - return false; - } - } - if (opts & XPARARED) redtasks_ = new ReduceTaskQueue; - if (opts & XPARAFLS) flsths_ = new std::deque<FlushThread*>; - if (opts & XNOLOCK) { - MapChecker mapchecker; - MapVisitor mapvisitor(this, &mapchecker, count); - mapvisitor.visit_before(); - if (!err) { - BasicDB::Cursor* cur = db->cursor(); - if (!cur->jump() && cur->error() != BasicDB::Error::NOREC) err = true; - while (!err) { - if (!cur->accept(&mapvisitor, false, true)) { - if (cur->error() != BasicDB::Error::NOREC) err = true; - break; - } - } - delete cur; - } - if (mapvisitor.error()) { - db_->set_error(_KCCODELINE_, BasicDB::Error::LOGIC, "mapper failed"); - err = true; - } - mapvisitor.visit_after(); - } else if (opts & XPARAMAP) { - MapChecker mapchecker; - MapVisitor mapvisitor(this, &mapchecker, count); - rlocks_ = new SlottedMutex(RLOCKSLOT); - if (!err && !db->scan_parallel(&mapvisitor, mapthnum_, &mapchecker)) { - db_->set_error(_KCCODELINE_, BasicDB::Error::LOGIC, "mapper failed"); - err = true; - } - delete rlocks_; - rlocks_ = NULL; - if (mapvisitor.error()) err = true; - } else { - MapChecker mapchecker; - MapVisitor mapvisitor(this, &mapchecker, count); - if (!err && !db->iterate(&mapvisitor, false, &mapchecker)) err = true; - if (mapvisitor.error()) { - db_->set_error(_KCCODELINE_, BasicDB::Error::LOGIC, "mapper failed"); - err = true; - } - } - if (flsths_) { - delete flsths_; - flsths_ = NULL; - } - if (redtasks_) { - delete redtasks_; - redtasks_ = NULL; - } - if (!logf("clean", "closing the temporary databases")) err = true; - stime = time(); - for (size_t i = 0; i < dbnum_; i++) { - const std::string& path = tmpdbs_[i]->path(); - if (!tmpdbs_[i]->clear()) { - const BasicDB::Error& e = tmpdbs_[i]->error(); - db->set_error(_KCCODELINE_, e.code(), e.message()); - err = true; - } - if (!tmpdbs_[i]->close()) { - const BasicDB::Error& e = tmpdbs_[i]->error(); - db->set_error(_KCCODELINE_, e.code(), e.message()); - err = true; - } - if (!tmppath.empty()) File::remove(path); - delete tmpdbs_[i]; - } - etime = time(); - if (!logf("clean", "closing the temporary databases finished: time=%.6f", - etime - stime)) err = true; - delete[] tmpdbs_; - return !err; - } - /** - * Set the storage configurations. - * @param dbnum the number of temporary databases. - * @param clim the limit size of the internal cache. - * @param cbnum the bucket number of the internal cache. - */ - void tune_storage(int32_t dbnum, int64_t clim, int64_t cbnum) { - _assert_(true); - dbnum_ = dbnum > 0 ? dbnum : DEFDBNUM; - if (dbnum_ > MAXDBNUM) dbnum_ = MAXDBNUM; - clim_ = clim > 0 ? clim : DEFCLIM; - cbnum_ = cbnum > 0 ? cbnum : DEFCBNUM; - if (cbnum_ > INT16MAX) cbnum_ = nearbyprime(cbnum_); - } - /** - * Set the thread configurations. - * @param mapthnum the number of threads for the mapper. - * @param redthnum the number of threads for the reducer. - * @param flsthnum the number of threads for the internal flusher. - */ - void tune_thread(int32_t mapthnum, int32_t redthnum, int32_t flsthnum) { - _assert_(true); - mapthnum_ = mapthnum > 0 ? mapthnum : DEFTHNUM; - redthnum_ = redthnum > 0 ? redthnum : DEFTHNUM; - flsthnum_ = flsthnum > 0 ? flsthnum : DEFTHNUM; - } - protected: - /** - * Emit a record from the mapper. - * @param kbuf the pointer to the key region. - * @param ksiz the size of the key region. - * @param vbuf the pointer to the value region. - * @param vsiz the size of the value region. - * @return true on success, or false on failure. - */ - bool emit(const char* kbuf, size_t ksiz, const char* vbuf, size_t vsiz) { - _assert_(kbuf && ksiz <= MEMMAXSIZ && vbuf && vsiz <= MEMMAXSIZ); - bool err = false; - size_t rsiz = sizevarnum(vsiz) + vsiz; - char stack[NUMBUFSIZ*4]; - char* rbuf = rsiz > sizeof(stack) ? new char[rsiz] : stack; - char* wp = rbuf; - wp += writevarnum(rbuf, vsiz); - std::memcpy(wp, vbuf, vsiz); - if (rlocks_) { - size_t bidx = TinyHashMap::hash_record(kbuf, ksiz) % cbnum_; - size_t lidx = bidx % RLOCKSLOT; - rlocks_->lock(lidx); - cache_->append(kbuf, ksiz, rbuf, rsiz); - rlocks_->unlock(lidx); - } else { - cache_->append(kbuf, ksiz, rbuf, rsiz); - } - if (rbuf != stack) delete[] rbuf; - csiz_ += sizevarnum(ksiz) + ksiz + rsiz; - return !err; - } - private: - /** - * Cache flusher. - */ - class FlushThread : public Thread { - public: - /** constructor */ - explicit FlushThread(MapReduce* mr, BasicDB* tmpdb, - TinyHashMap* cache, size_t csiz, bool cown) : - mr_(mr), tmpdb_(tmpdb), cache_(cache), csiz_(csiz), cown_(cown), err_(false) {} - /** perform the concrete process */ - void run() { - if (!mr_->logf("map", "started to flushing the cache: count=%lld size=%lld", - (long long)cache_->count(), (long long)csiz_)) err_ = true; - double stime = time(); - BasicDB* tmpdb = tmpdb_; - TinyHashMap* cache = cache_; - bool cown = cown_; - TinyHashMap::Sorter sorter(cache); - const char* kbuf, *vbuf; - size_t ksiz, vsiz; - while ((kbuf = sorter.get(&ksiz, &vbuf, &vsiz)) != NULL) { - if (!tmpdb->append(kbuf, ksiz, vbuf, vsiz)) { - const BasicDB::Error& e = tmpdb->error(); - mr_->db_->set_error(_KCCODELINE_, e.code(), e.message()); - err_ = true; - } - sorter.step(); - if (cown) cache->remove(kbuf, ksiz); - } - double etime = time(); - if (!mr_->logf("map", "flushing the cache finished: time=%.6f", etime - stime)) - err_ = true; - if (cown) delete cache; - } - /** check the error flag. */ - bool error() { - return err_; - } - private: - MapReduce* mr_; ///< driver - BasicDB* tmpdb_; ///< temprary database - TinyHashMap* cache_; ///< cache for emitter - size_t csiz_; ///< current cache size - bool cown_; ///< cache ownership flag - bool err_; ///< error flag - }; - /** - * Task queue for parallel reducer. - */ - class ReduceTaskQueue : public TaskQueue { - public: - /** - * Task for parallel reducer. - */ - class ReduceTask : public Task { - friend class ReduceTaskQueue; - public: - /** constructor */ - explicit ReduceTask(MapReduce* mr, const char* kbuf, size_t ksiz, const Values& values) : - mr_(mr), key_(kbuf, ksiz), values_(values) {} - private: - MapReduce* mr_; ///< driver - std::string key_; ///< key - Values values_; ///< values - }; - /** constructor */ - explicit ReduceTaskQueue() {} - private: - /** process a task */ - void do_task(Task* task) { - ReduceTask* rtask = (ReduceTask*)task; - ValueIterator iter(rtask->values_.begin(), rtask->values_.end()); - if (!rtask->mr_->reduce(rtask->key_.data(), rtask->key_.size(), &iter)) - rtask->mr_->redaborted_ = true; - delete rtask; - } - }; - /** - * Checker for the map process. - */ - class MapChecker : public BasicDB::ProgressChecker { - public: - /** constructor */ - explicit MapChecker() : stop_(false) {} - /** stop the process */ - void stop() { - stop_ = true; - } - /** check whether stopped */ - bool stopped() { - return stop_; - } - private: - /** check whether stopped */ - bool check(const char* name, const char* message, int64_t curcnt, int64_t allcnt) { - return !stop_; - } - bool stop_; ///< flag for stop - }; - /** - * Visitor for the map process. - */ - class MapVisitor : public BasicDB::Visitor { - public: - /** constructor */ - explicit MapVisitor(MapReduce* mr, MapChecker* checker, int64_t scale) : - mr_(mr), checker_(checker), scale_(scale), stime_(0), err_(false) {} - /** get the error flag */ - bool error() { - return err_; - } - /** preprocess the mappter */ - void visit_before() { - mr_->dbclock_ = 0; - mr_->cache_ = new TinyHashMap(mr_->cbnum_); - mr_->csiz_ = 0; - if (!mr_->preprocess()) err_ = true; - if (mr_->csiz_ > 0 && !mr_->flush_cache()) err_ = true; - if (!mr_->logf("map", "started the map process: scale=%lld", (long long)scale_)) - err_ = true; - stime_ = time(); - } - /** postprocess the mappter and call the reducer */ - void visit_after() { - if (mr_->csiz_ > 0 && !mr_->flush_cache()) err_ = true; - double etime = time(); - if (!mr_->logf("map", "the map process finished: time=%.6f", etime - stime_)) - err_ = true; - if (!mr_->midprocess()) err_ = true; - if (mr_->csiz_ > 0 && !mr_->flush_cache()) err_ = true; - delete mr_->cache_; - if (mr_->flsths_ && !mr_->flsths_->empty()) { - std::deque<FlushThread*>::iterator flthit = mr_->flsths_->begin(); - std::deque<FlushThread*>::iterator flthitend = mr_->flsths_->end(); - while (flthit != flthitend) { - FlushThread* flth = *flthit; - flth->join(); - if (flth->error()) err_ = true; - delete flth; - ++flthit; - } - } - if (!err_ && !mr_->execute_reduce()) err_ = true; - if (!mr_->postprocess()) err_ = true; - } - private: - /** visit a record */ - const char* visit_full(const char* kbuf, size_t ksiz, - const char* vbuf, size_t vsiz, size_t* sp) { - if (!mr_->map(kbuf, ksiz, vbuf, vsiz)) { - checker_->stop(); - err_ = true; - } - if (mr_->rlocks_) { - if (mr_->csiz_ >= mr_->clim_) { - mr_->rlocks_->lock_all(); - if (mr_->csiz_ >= mr_->clim_ && !mr_->flush_cache()) { - checker_->stop(); - err_ = true; - } - mr_->rlocks_->unlock_all(); - } - } else { - if (mr_->csiz_ >= mr_->clim_ && !mr_->flush_cache()) { - checker_->stop(); - err_ = true; - } - } - return NOP; - } - MapReduce* mr_; ///< driver - MapChecker* checker_; ///< checker - int64_t scale_; ///< number of records - double stime_; ///< start time - bool err_; ///< error flag - }; - /** - * Front line of a merging list. - */ - struct MergeLine { - BasicDB::Cursor* cur; ///< cursor - Comparator* rcomp; ///< record comparator - char* kbuf; ///< pointer to the key - size_t ksiz; ///< size of the key - const char* vbuf; ///< pointer to the value - size_t vsiz; ///< size of the value - /** comparing operator */ - bool operator <(const MergeLine& right) const { - return rcomp->compare(kbuf, ksiz, right.kbuf, right.ksiz) > 0; - } - }; - /** - * Process a log message. - * @param name the name of the event. - * @param format the printf-like format string. - * @param ... used according to the format string. - * @return true on success, or false on failure. - */ - bool logf(const char* name, const char* format, ...) { - _assert_(name && format); - va_list ap; - va_start(ap, format); - std::string message; - vstrprintf(&message, format, ap); - va_end(ap); - return log(name, message.c_str()); - } - /** - * Flush all cache records. - * @return true on success, or false on failure. - */ - bool flush_cache() { - _assert_(true); - bool err = false; - BasicDB* tmpdb = tmpdbs_[dbclock_]; - dbclock_ = (dbclock_ + 1) % dbnum_; - if (flsths_) { - size_t num = flsths_->size(); - if (num >= flsthnum_ || num >= dbnum_) { - FlushThread* flth = flsths_->front(); - flsths_->pop_front(); - flth->join(); - if (flth->error()) err = true; - delete flth; - } - FlushThread* flth = new FlushThread(this, tmpdb, cache_, csiz_, true); - cache_ = new TinyHashMap(cbnum_); - csiz_ = 0; - flth->start(); - flsths_->push_back(flth); - } else { - FlushThread flth(this, tmpdb, cache_, csiz_, false); - flth.run(); - if (flth.error()) err = true; - cache_->clear(); - csiz_ = 0; - } - return !err; - } - /** - * Execute the reduce part. - * @return true on success, or false on failure. - */ - bool execute_reduce() { - bool err = false; - int64_t scale = 0; - for (size_t i = 0; i < dbnum_; i++) { - scale += tmpdbs_[i]->count(); - } - if (!logf("reduce", "started the reduce process: scale=%lld", (long long)scale)) err = true; - double stime = time(); - if (redtasks_) redtasks_->start(redthnum_); - std::priority_queue<MergeLine> lines; - for (size_t i = 0; i < dbnum_; i++) { - MergeLine line; - line.cur = tmpdbs_[i]->cursor(); - line.rcomp = rcomp_; - line.cur->jump(); - line.kbuf = line.cur->get(&line.ksiz, &line.vbuf, &line.vsiz, true); - if (line.kbuf) { - lines.push(line); - } else { - delete line.cur; - } - } - char* lkbuf = NULL; - size_t lksiz = 0; - Values values; - while (!err && !lines.empty()) { - MergeLine line = lines.top(); - lines.pop(); - if (lkbuf && (lksiz != line.ksiz || std::memcmp(lkbuf, line.kbuf, lksiz))) { - if (!call_reducer(lkbuf, lksiz, values)) { - db_->set_error(_KCCODELINE_, BasicDB::Error::LOGIC, "reducer failed"); - err = true; - } - values.clear(); - } - delete[] lkbuf; - lkbuf = line.kbuf; - lksiz = line.ksiz; - values.push_back(std::string(line.vbuf, line.vsiz)); - line.kbuf = line.cur->get(&line.ksiz, &line.vbuf, &line.vsiz, true); - if (line.kbuf) { - lines.push(line); - } else { - delete line.cur; - } - } - if (lkbuf) { - if (!err && !call_reducer(lkbuf, lksiz, values)) { - db_->set_error(_KCCODELINE_, BasicDB::Error::LOGIC, "reducer failed"); - err = true; - } - delete[] lkbuf; - } - while (!lines.empty()) { - MergeLine line = lines.top(); - lines.pop(); - delete[] line.kbuf; - delete line.cur; - } - if (redtasks_) redtasks_->finish(); - double etime = time(); - if (!logf("reduce", "the reduce process finished: time=%.6f", etime - stime)) err = true; - return !err; - } - /** - * Call the reducer. - * @param kbuf the pointer to the key region. - * @param ksiz the size of the key region. - * @param values a vector of the values. - * @return true on success, or false on failure. - */ - bool call_reducer(const char* kbuf, size_t ksiz, const Values& values) { - _assert_(kbuf && ksiz <= MEMMAXSIZ); - if (redtasks_) { - if (redaborted_) return false; - ReduceTaskQueue::ReduceTask* task = - new ReduceTaskQueue::ReduceTask(this, kbuf, ksiz, values); - redtasks_->add_task(task); - return true; - } - bool err = false; - ValueIterator iter(values.begin(), values.end()); - if (!reduce(kbuf, ksiz, &iter)) err = true; - return !err; - } - /** Dummy constructor to forbid the use. */ - MapReduce(const MapReduce&); - /** Dummy Operator to forbid the use. */ - MapReduce& operator =(const MapReduce&); - /** The internal database. */ - BasicDB* db_; - /** The record comparator. */ - Comparator* rcomp_; - /** The temporary databases. */ - BasicDB** tmpdbs_; - /** The number of temporary databases. */ - size_t dbnum_; - /** The logical clock for temporary databases. */ - int64_t dbclock_; - /** The number of the mapper threads. */ - size_t mapthnum_; - /** The number of the reducer threads. */ - size_t redthnum_; - /** The number of the flusher threads. */ - size_t flsthnum_; - /** The cache for emitter. */ - TinyHashMap* cache_; - /** The current size of the cache for emitter. */ - AtomicInt64 csiz_; - /** The limit size of the cache for emitter. */ - int64_t clim_; - /** The bucket number of the cache for emitter. */ - int64_t cbnum_; - /** The flush threads. */ - std::deque<FlushThread*>* flsths_; - /** The task queue for parallel reducer. */ - TaskQueue* redtasks_; - /** The flag whether aborted. */ - bool redaborted_; - /** The whole lock. */ - SlottedMutex* rlocks_; -}; - - -/** - * Index database. - * @note This class is designed to implement an indexing storage with an efficient appending - * operation for the existing record values. This class is a wrapper of the polymorphic - * database, featuring buffering mechanism to alleviate IO overhead in the database layer. This - * class can be inherited but overwriting methods is forbidden. Before every database operation, - * it is necessary to call the IndexDB::open method in order to open a database file and connect - * the database object to it. To avoid data missing or corruption, it is important to close - * every database file by the IndexDB::close method when the database is no longer in use. It - * is forbidden for multible database objects in a process to open the same database at the same - * time. It is forbidden to share a database object with child processes. - */ -class IndexDB { - private: - /** The default number of temporary databases. */ - static const size_t DEFDBNUM = 8; - /** The maxinum number of temporary databases. */ - static const size_t MAXDBNUM = 256; - /** The default cache limit size. */ - static const int64_t DEFCLIM = 256LL << 20; - /** The default cache bucket number. */ - static const int64_t DEFCBNUM = 1048583LL; - /** The bucket number of temprary databases. */ - static const int64_t DBBNUM = 512LL << 10; - /** The page size of temprary databases. */ - static const int32_t DBPSIZ = 32768; - /** The mapped size of temprary databases. */ - static const int64_t DBMSIZ = 516LL * 4096; - /** The page cache capacity of temprary databases. */ - static const int64_t DBPCCAP = 16LL << 20; - public: - /** - * Default constructor. - */ - explicit IndexDB() : - db_(), omode_(0), - rcomp_(NULL), tmppath_(""), tmpdbs_(NULL), dbnum_(DEFDBNUM), dbclock_(0), - cache_(NULL), csiz_(0), clim_(0) { - _assert_(true); - } - /** - * Destructor. - * @note If the database is not closed, it is closed implicitly. - */ - virtual ~IndexDB() { - _assert_(true); - if (omode_ != 0) close(); - } - /** - * Get the last happened error. - * @return the last happened error. - */ - BasicDB::Error error() const { - _assert_(true); - return db_.error(); - } - /** - * Set the error information. - * @param file the file name of the program source code. - * @param line the line number of the program source code. - * @param func the function name of the program source code. - * @param code an error code. - * @param message a supplement message. - */ - void set_error(const char* file, int32_t line, const char* func, - BasicDB::Error::Code code, const char* message) { - _assert_(file && line > 0 && func && message); - db_.set_error(file, line, func, code, message); - } - /** - * Set the error information without source code information. - * @param code an error code. - * @param message a supplement message. - */ - void set_error(BasicDB::Error::Code code, const char* message) { - _assert_(message); - db_.set_error(_KCCODELINE_, code, message); - } - /** - * Open a database file. - * @param path the path of a database file. The same as with PolyDB. In addition, the - * following tuning parameters are supported. "idxclim" specifies the limit size of the - * internal cache. "idxcbnum" the bucket number of the internal cache. "idxdbnum" specifies - * the number of internal databases. "idxtmppath' specifies the path of the temporary - * directory. - * @param mode the connection mode. The same as with PolyDB. - * @return true on success, or false on failure. - * @note Every opened database must be closed by the IndexDB::close method when it is no longer - * in use. It is not allowed for two or more database objects in the same process to keep - * their connections to the same database file at the same time. - */ - bool open(const std::string& path = ":", - uint32_t mode = BasicDB::OWRITER | BasicDB::OCREATE) { - _assert_(true); - // ScopedRWLock lock(&mlock_, true); - if (omode_ != 0) { - set_error(_KCCODELINE_, BasicDB::Error::INVALID, "already opened"); - return false; - } - std::vector<std::string> elems; - strsplit(path, '#', &elems); - int64_t clim = 0; - int64_t cbnum = 0; - size_t dbnum = 0; - std::string tmppath = ""; - std::vector<std::string>::iterator it = elems.begin(); - std::vector<std::string>::iterator itend = elems.end(); - if (it != itend) ++it; - while (it != itend) { - std::vector<std::string> fields; - if (strsplit(*it, '=', &fields) > 1) { - const char* key = fields[0].c_str(); - const char* value = fields[1].c_str(); - if (!std::strcmp(key, "idxclim") || !std::strcmp(key, "idxcachelimit")) { - clim = atoix(value); - } else if (!std::strcmp(key, "idxcbnum") || !std::strcmp(key, "idxcachebuckets")) { - cbnum = atoix(value); - } else if (!std::strcmp(key, "idxdbnum")) { - dbnum = atoix(value); - } else if (!std::strcmp(key, "idxtmppath")) { - tmppath = value; - } - } - ++it; - } - if (!db_.open(path, mode)) return false; - tmppath_ = tmppath; - rcomp_ = LEXICALCOMP; - BasicDB* idb = &db_; - if (typeid(db_) == typeid(PolyDB)) { - PolyDB* pdb = (PolyDB*)idb; - idb = pdb->reveal_inner_db(); - } - const std::type_info& info = typeid(*idb); - if (info == typeid(GrassDB)) { - GrassDB* gdb = (GrassDB*)idb; - rcomp_ = gdb->rcomp(); - } else if (info == typeid(TreeDB)) { - TreeDB* tdb = (TreeDB*)idb; - rcomp_ = tdb->rcomp(); - } else if (info == typeid(ForestDB)) { - ForestDB* fdb = (ForestDB*)idb; - rcomp_ = fdb->rcomp(); - } - dbnum_ = dbnum < MAXDBNUM ? dbnum : MAXDBNUM; - dbclock_ = 0; - if ((mode & BasicDB::OWRITER) && dbnum > 0) { - tmpdbs_ = new BasicDB*[dbnum_]; - if (tmppath_.empty()) { - report(_KCCODELINE_, "started to open temporary databases on memory"); - double stime = time(); - for (size_t i = 0; i < dbnum_; i++) { - GrassDB* gdb = new GrassDB; - gdb->tune_options(GrassDB::TCOMPRESS); - gdb->tune_buckets(DBBNUM / 2); - gdb->tune_page(DBPSIZ); - gdb->tune_page_cache(DBPCCAP); - gdb->tune_comparator(rcomp_); - gdb->open("%", GrassDB::OWRITER | GrassDB::OCREATE | GrassDB::OTRUNCATE); - tmpdbs_[i] = gdb; - } - double etime = time(); - report(_KCCODELINE_, "opening temporary databases finished: time=%.6f", etime - stime); - } else { - File::Status sbuf; - if (!File::status(tmppath_, &sbuf) || !sbuf.isdir) { - set_error(_KCCODELINE_, BasicDB::Error::NOREPOS, "no such directory"); - delete[] tmpdbs_; - db_.close(); - return false; - } - report(_KCCODELINE_, "started to open temporary databases under %s", tmppath.c_str()); - double stime = time(); - uint32_t pid = getpid() & UINT16MAX; - uint32_t tid = Thread::hash() & UINT16MAX; - uint32_t ts = time() * 1000; - bool err = false; - for (size_t i = 0; i < dbnum_; i++) { - std::string childpath = - strprintf("%s%cidx-%04x-%04x-%08x-%03d%ckct", - tmppath_.c_str(), File::PATHCHR, pid, tid, ts, - (int)(i + 1), File::EXTCHR); - TreeDB* tdb = new TreeDB; - tdb->tune_options(TreeDB::TSMALL | TreeDB::TLINEAR); - tdb->tune_buckets(DBBNUM); - tdb->tune_page(DBPSIZ); - tdb->tune_map(DBMSIZ); - tdb->tune_page_cache(DBPCCAP); - tdb->tune_comparator(rcomp_); - if (!tdb->open(childpath, TreeDB::OWRITER | TreeDB::OCREATE | TreeDB::OTRUNCATE)) { - const BasicDB::Error& e = tdb->error(); - set_error(_KCCODELINE_, e.code(), e.message()); - err = true; - } - tmpdbs_[i] = tdb; - } - double etime = time(); - report(_KCCODELINE_, "opening temporary databases finished: time=%.6f", etime - stime); - if (err) { - for (size_t i = 0; i < dbnum_; i++) { - delete tmpdbs_[i]; - } - delete[] tmpdbs_; - db_.close(); - return false; - } - } - } else { - tmpdbs_ = NULL; - } - if (mode & BasicDB::OWRITER) { - cache_ = new TinyHashMap(cbnum > 0 ? cbnum : DEFCBNUM); - } else { - cache_ = NULL; - } - clim_ = clim > 0 ? clim : DEFCLIM; - csiz_ = 0; - omode_ = mode; - return true; - } - /** - * Close the database file. - * @return true on success, or false on failure. - */ - bool close() { - _assert_(true); - // ScopedRWLock lock(&mlock_, true); - if (omode_ == 0) { - set_error(_KCCODELINE_, BasicDB::Error::INVALID, "not opened"); - return false; - } - bool err = false; - if (cache_) { - if (!flush_cache()) err = true; - delete cache_; - if (tmpdbs_) { - if (!merge_tmpdbs()) err = true; - report(_KCCODELINE_, "closing the temporary databases"); - double stime = time(); - for (size_t i = 0; i < dbnum_; i++) { - BasicDB* tmpdb = tmpdbs_[i]; - const std::string& path = tmpdb->path(); - if (!tmpdb->close()) { - const BasicDB::Error& e = tmpdb->error(); - set_error(_KCCODELINE_, e.code(), e.message()); - err = true; - } - if (!tmppath_.empty()) File::remove(path); - delete tmpdb; - } - double etime = time(); - report(_KCCODELINE_, "closing the temporary databases finished: %.6f", etime - stime); - delete[] tmpdbs_; - } - } - if (!db_.close()) err = true; - omode_ = 0; - return !err; - } - /** - * Set the value of a record. - * @param kbuf the pointer to the key region. - * @param ksiz the size of the key region. - * @param vbuf the pointer to the value region. - * @param vsiz the size of the value region. - * @return true on success, or false on failure. - * @note If no record corresponds to the key, a new record is created. If the corresponding - * record exists, the value is overwritten. - */ - bool set(const char* kbuf, size_t ksiz, const char* vbuf, size_t vsiz) { - _assert_(kbuf && ksiz <= MEMMAXSIZ && vbuf && vsiz <= MEMMAXSIZ); - // ScopedRWLock lock(&mlock_, true); - if (omode_ == 0) { - set_error(_KCCODELINE_, BasicDB::Error::INVALID, "not opened"); - return false; - } - if (!cache_) { - set_error(_KCCODELINE_, BasicDB::Error::INVALID, "permission denied"); - return false; - } - bool err = false; - if (!clean_dbs(kbuf, ksiz)) err = true; - cache_->set(kbuf, ksiz, vbuf, vsiz); - csiz_ += ksiz + vsiz; - if (csiz_ > clim_ && !flush_cache()) err = false; - return !err; - } - /** - * Set the value of a record. - * @note Equal to the original DB::set method except that the parameters are std::string. - */ - bool set(const std::string& key, const std::string& value) { - _assert_(true); - return set(key.c_str(), key.size(), value.c_str(), value.size()); - } - /** - * Add a record. - * @param kbuf the pointer to the key region. - * @param ksiz the size of the key region. - * @param vbuf the pointer to the value region. - * @param vsiz the size of the value region. - * @return true on success, or false on failure. - * @note If no record corresponds to the key, a new record is created. If the corresponding - * record exists, the record is not modified and false is returned. - */ - bool add(const char* kbuf, size_t ksiz, const char* vbuf, size_t vsiz) { - _assert_(kbuf && ksiz <= MEMMAXSIZ && vbuf && vsiz <= MEMMAXSIZ); - // ScopedRWLock lock(&mlock_, true); - if (omode_ == 0) { - set_error(_KCCODELINE_, BasicDB::Error::INVALID, "not opened"); - return false; - } - if (!cache_) { - set_error(_KCCODELINE_, BasicDB::Error::INVALID, "permission denied"); - return false; - } - if (check_impl(kbuf, ksiz)) { - set_error(_KCCODELINE_, BasicDB::Error::DUPREC, "record duplication"); - return false; - } - bool err = false; - cache_->set(kbuf, ksiz, vbuf, vsiz); - csiz_ += ksiz + vsiz; - if (csiz_ > clim_ && !flush_cache()) err = false; - return !err; - } - /** - * Set the value of a record. - * @note Equal to the original DB::add method except that the parameters are std::string. - */ - bool add(const std::string& key, const std::string& value) { - _assert_(true); - return add(key.c_str(), key.size(), value.c_str(), value.size()); - } - /** - * Replace the value of a record. - * @param kbuf the pointer to the key region. - * @param ksiz the size of the key region. - * @param vbuf the pointer to the value region. - * @param vsiz the size of the value region. - * @return true on success, or false on failure. - * @note If no record corresponds to the key, no new record is created and false is returned. - * If the corresponding record exists, the value is modified. - */ - bool replace(const char* kbuf, size_t ksiz, const char* vbuf, size_t vsiz) { - _assert_(kbuf && ksiz <= MEMMAXSIZ && vbuf && vsiz <= MEMMAXSIZ); - // ScopedRWLock lock(&mlock_, true); - if (omode_ == 0) { - set_error(_KCCODELINE_, BasicDB::Error::INVALID, "not opened"); - return false; - } - if (!cache_) { - set_error(_KCCODELINE_, BasicDB::Error::INVALID, "permission denied"); - return false; - } - if (!check_impl(kbuf, ksiz)) { - set_error(_KCCODELINE_, BasicDB::Error::NOREC, "no record"); - return false; - } - bool err = false; - if (!clean_dbs(kbuf, ksiz)) err = true; - cache_->set(kbuf, ksiz, vbuf, vsiz); - csiz_ += ksiz + vsiz; - if (csiz_ > clim_ && !flush_cache()) err = false; - return !err; - } - /** - * Replace the value of a record. - * @note Equal to the original DB::replace method except that the parameters are std::string. - */ - bool replace(const std::string& key, const std::string& value) { - _assert_(true); - return replace(key.c_str(), key.size(), value.c_str(), value.size()); - } - /** - * Append the value of a record. - * @param kbuf the pointer to the key region. - * @param ksiz the size of the key region. - * @param vbuf the pointer to the value region. - * @param vsiz the size of the value region. - * @return true on success, or false on failure. - * @note If no record corresponds to the key, a new record is created. If the corresponding - * record exists, the given value is appended at the end of the existing value. - */ - bool append(const char* kbuf, size_t ksiz, const char* vbuf, size_t vsiz) { - _assert_(kbuf && ksiz <= MEMMAXSIZ && vbuf && vsiz <= MEMMAXSIZ); - // ScopedRWLock lock(&mlock_, true); - if (omode_ == 0) { - set_error(_KCCODELINE_, BasicDB::Error::INVALID, "not opened"); - return false; - } - if (!cache_) { - set_error(_KCCODELINE_, BasicDB::Error::INVALID, "permission denied"); - return false; - } - bool err = false; - cache_->append(kbuf, ksiz, vbuf, vsiz); - csiz_ += ksiz + vsiz; - if (csiz_ > clim_ && !flush_cache()) err = false; - return !err; - } - /** - * Set the value of a record. - * @note Equal to the original DB::append method except that the parameters are std::string. - */ - bool append(const std::string& key, const std::string& value) { - _assert_(true); - return append(key.c_str(), key.size(), value.c_str(), value.size()); - } - /** - * Remove a record. - * @param kbuf the pointer to the key region. - * @param ksiz the size of the key region. - * @return true on success, or false on failure. - * @note If no record corresponds to the key, false is returned. - */ - bool remove(const char* kbuf, size_t ksiz) { - _assert_(kbuf && ksiz <= MEMMAXSIZ); - // ScopedRWLock lock(&mlock_, true); - if (omode_ == 0) { - set_error(_KCCODELINE_, BasicDB::Error::INVALID, "not opened"); - return false; - } - if (!cache_) { - set_error(_KCCODELINE_, BasicDB::Error::INVALID, "permission denied"); - return false; - } - bool err = false; - if (!clean_dbs(kbuf, ksiz)) err = true; - cache_->remove(kbuf, ksiz); - return !err; - } - /** - * Remove a record. - * @note Equal to the original DB::remove method except that the parameter is std::string. - */ - bool remove(const std::string& key) { - _assert_(true); - return remove(key.c_str(), key.size()); - } - /** - * Retrieve the value of a record. - * @param kbuf the pointer to the key region. - * @param ksiz the size of the key region. - * @param sp the pointer to the variable into which the size of the region of the return - * value is assigned. - * @return the pointer to the value region of the corresponding record, or NULL on failure. - * @note If no record corresponds to the key, NULL is returned. Because an additional zero - * code is appended at the end of the region of the return value, the return value can be - * treated as a C-style string. Because the region of the return value is allocated with the - * the new[] operator, it should be released with the delete[] operator when it is no longer - * in use. - */ - char* get(const char* kbuf, size_t ksiz, size_t* sp) { - _assert_(kbuf && ksiz <= MEMMAXSIZ && sp); - // ScopedRWLock lock(&mlock_, false); - if (omode_ == 0) { - set_error(_KCCODELINE_, BasicDB::Error::INVALID, "not opened"); - *sp = 0; - return false; - } - if (!cache_) return db_.get(kbuf, ksiz, sp); - size_t dvsiz = 0; - char* dvbuf = db_.get(kbuf, ksiz, &dvsiz); - size_t cvsiz = 0; - const char* cvbuf = cache_->get(kbuf, ksiz, &cvsiz); - struct Record { - char* buf; - size_t size; - }; - Record* recs = NULL; - bool hit = false; - size_t rsiz = 0; - if (tmpdbs_) { - recs = new Record[dbnum_]; - for (size_t i = 0; i < dbnum_; i++) { - BasicDB* tmpdb = tmpdbs_[i]; - Record* rp = recs + i; - rp->buf = tmpdb->get(kbuf, ksiz, &rp->size); - if (rp->buf) { - rsiz += rp->size; - hit = true; - } - } - } - if (!hit) { - delete[] recs; - if (!dvbuf && !cvbuf) { - *sp = 0; - return NULL; - } - if (!dvbuf) { - dvbuf = new char[cvsiz+1]; - std::memcpy(dvbuf, cvbuf, cvsiz); - *sp = cvsiz; - return dvbuf; - } - if (!cvbuf) { - *sp = dvsiz; - return dvbuf; - } - char* rbuf = new char[dvsiz+cvsiz+1]; - std::memcpy(rbuf, dvbuf, dvsiz); - std::memcpy(rbuf + dvsiz, cvbuf, cvsiz); - delete[] dvbuf; - *sp = dvsiz + cvsiz; - return rbuf; - } - if (dvbuf) rsiz += dvsiz; - if (cvbuf) rsiz += cvsiz; - char* rbuf = new char[rsiz+1]; - char* wp = rbuf; - if (dvbuf) { - std::memcpy(wp, dvbuf, dvsiz); - wp += dvsiz; - delete[] dvbuf; - } - if (cvbuf) { - std::memcpy(wp, cvbuf, cvsiz); - wp += cvsiz; - } - if (recs) { - for (size_t i = 0; i < dbnum_; i++) { - Record* rp = recs + i; - if (rp->buf) { - std::memcpy(wp, rp->buf, rp->size); - wp += rp->size; - delete[] rp->buf; - } - } - delete[] recs; - } - *sp = rsiz; - return rbuf; - } - /** - * Retrieve the value of a record. - * @note Equal to the original DB::get method except that the first parameters is the key - * string and the second parameter is a string to contain the result and the return value is - * bool for success. - */ - bool get(const std::string& key, std::string* value) { - _assert_(value); - size_t vsiz; - char* vbuf = get(key.c_str(), key.size(), &vsiz); - if (!vbuf) return false; - value->clear(); - value->append(vbuf, vsiz); - delete[] vbuf; - return true; - } - /** - * Synchronize updated contents with the file and the device. - * @param hard true for physical synchronization with the device, or false for logical - * synchronization with the file system. - * @param proc a postprocessor object. If it is NULL, no postprocessing is performed. - * @return true on success, or false on failure. - * @note The operation of the postprocessor is performed atomically and other threads accessing - * the same record are blocked. To avoid deadlock, any explicit database operation must not - * be performed in this function. - */ - bool synchronize(bool hard = false, BasicDB::FileProcessor* proc = NULL) { - _assert_(true); - // ScopedRWLock lock(&mlock_, true); - if (omode_ == 0) { - set_error(_KCCODELINE_, BasicDB::Error::INVALID, "not opened"); - return false; - } - if (!cache_) { - set_error(_KCCODELINE_, BasicDB::Error::INVALID, "permission denied"); - return false; - } - bool err = false; - if (!flush_cache()) err = true; - if (tmpdbs_ && !merge_tmpdbs()) err = true; - if (!db_.synchronize(hard, proc)) err = true; - return !err; - } - /** - * Remove all records. - * @return true on success, or false on failure. - */ - bool clear() { - _assert_(true); - // ScopedRWLock lock(&mlock_, true); - if (omode_ == 0) { - set_error(_KCCODELINE_, BasicDB::Error::INVALID, "not opened"); - return false; - } - if (!cache_) { - set_error(_KCCODELINE_, BasicDB::Error::INVALID, "permission denied"); - return false; - } - cache_->clear(); - csiz_ = 0; - return db_.clear(); - } - /** - * Get the number of records. - * @return the number of records, or -1 on failure. - */ - int64_t count() { - _assert_(true); - // ScopedRWLock lock(&mlock_, false); - return count_impl(); - } - /** - * Get the size of the database file. - * @return the size of the database file in bytes, or -1 on failure. - */ - int64_t size() { - _assert_(true); - // ScopedRWLock lock(&mlock_, false); - return size_impl(); - } - /** - * Get the path of the database file. - * @return the path of the database file, or an empty string on failure. - */ - std::string path() { - _assert_(true); - return db_.path(); - } - /** - * Get the miscellaneous status information. - * @param strmap a string map to contain the result. - * @return true on success, or false on failure. - */ - bool status(std::map<std::string, std::string>* strmap) { - _assert_(strmap); - return db_.status(strmap); - } - /** - * Reveal the inner database object. - * @return the inner database object, or NULL on failure. - */ - PolyDB* reveal_inner_db() { - _assert_(true); - return &db_; - } - /** - * Create a cursor object. - * @return the return value is the created cursor object. - * @note Because the object of the return value is allocated by the constructor, it should be - * released with the delete operator when it is no longer in use. - */ - BasicDB::Cursor* cursor() { - _assert_(true); - return db_.cursor(); - } - /** - * Write a log message. - * @param file the file name of the program source code. - * @param line the line number of the program source code. - * @param func the function name of the program source code. - * @param kind the kind of the event. Logger::DEBUG for debugging, Logger::INFO for normal - * information, Logger::WARN for warning, and Logger::ERROR for fatal error. - * @param message the supplement message. - */ - void log(const char* file, int32_t line, const char* func, BasicDB::Logger::Kind kind, - const char* message) { - _assert_(file && line > 0 && func && message); - db_.log(file, line, func, kind, message); - } - /** - * Set the internal logger. - * @param logger the logger object. - * @param kinds kinds of logged messages by bitwise-or: Logger::DEBUG for debugging, - * Logger::INFO for normal information, Logger::WARN for warning, and Logger::ERROR for fatal - * error. - * @return true on success, or false on failure. - */ - bool tune_logger(BasicDB::Logger* logger, - uint32_t kinds = BasicDB::Logger::WARN | BasicDB::Logger::ERROR) { - _assert_(logger); - return db_.tune_logger(logger, kinds); - } - /** - * Set the internal meta operation trigger. - * @param trigger the trigger object. - * @return true on success, or false on failure. - */ - bool tune_meta_trigger(BasicDB::MetaTrigger* trigger) { - _assert_(trigger); - return db_.tune_meta_trigger(trigger); - } - protected: - /** - * Report a message for debugging. - * @param file the file name of the program source code. - * @param line the line number of the program source code. - * @param func the function name of the program source code. - * @param format the printf-like format string. - * @param ... used according to the format string. - */ - void report(const char* file, int32_t line, const char* func, const char* format, ...) { - _assert_(file && line > 0 && func && format); - std::string message; - va_list ap; - va_start(ap, format); - vstrprintf(&message, format, ap); - va_end(ap); - db_.log(file, line, func, BasicDB::Logger::INFO, message.c_str()); - } - private: - /** - * Flush all cache records. - * @return true on success, or false on failure. - */ - bool flush_cache() { - _assert_(true); - bool err = false; - double stime = time(); - report(_KCCODELINE_, "flushing the cache"); - if (tmpdbs_) { - BasicDB* tmpdb = tmpdbs_[dbclock_]; - TinyHashMap::Sorter sorter(cache_); - const char* kbuf, *vbuf; - size_t ksiz, vsiz; - while ((kbuf = sorter.get(&ksiz, &vbuf, &vsiz)) != NULL) { - if (!tmpdb->append(kbuf, ksiz, vbuf, vsiz)) { - const BasicDB::Error& e = tmpdb->error(); - db_.set_error(_KCCODELINE_, e.code(), e.message()); - err = true; - } - sorter.step(); - } - dbclock_ = (dbclock_ + 1) % dbnum_; - } else { - TinyHashMap::Sorter sorter(cache_); - const char* kbuf, *vbuf; - size_t ksiz, vsiz; - while ((kbuf = sorter.get(&ksiz, &vbuf, &vsiz)) != NULL) { - if (!db_.append(kbuf, ksiz, vbuf, vsiz)) err = true; - sorter.step(); - } - } - cache_->clear(); - csiz_ = 0; - double etime = time(); - report(_KCCODELINE_, "flushing the cache finished: time=%.6f", etime - stime); - return !err; - } - /** - * Merge temporary databases. - * @return true on success, or false on failure. - */ - bool merge_tmpdbs() { - _assert_(true); - bool err = false; - report(_KCCODELINE_, "merging the temporary databases"); - double stime = time(); - if (!db_.merge(tmpdbs_, dbnum_, PolyDB::MAPPEND)) err = true; - dbclock_ = 0; - for (size_t i = 0; i < dbnum_; i++) { - BasicDB* tmpdb = tmpdbs_[i]; - if (!tmpdb->clear()) { - const BasicDB::Error& e = tmpdb->error(); - set_error(_KCCODELINE_, e.code(), e.message()); - err = true; - } - } - double etime = time(); - report(_KCCODELINE_, "merging the temporary databases finished: %.6f", etime - stime); - return !err; - } - /** - * Remove a record from databases. - * @param kbuf the pointer to the key region. - * @param ksiz the size of the key region. - * @return true on success, or false on failure. - */ - bool clean_dbs(const char* kbuf, size_t ksiz) { - _assert_(kbuf && ksiz <= MEMMAXSIZ); - if (db_.remove(kbuf, ksiz)) return true; - bool err = false; - if (db_.error() != BasicDB::Error::NOREC) err = true; - if (tmpdbs_) { - for (size_t i = 0; i < dbnum_; i++) { - BasicDB* tmpdb = tmpdbs_[i]; - if (!tmpdb->remove(kbuf, ksiz)) { - const BasicDB::Error& e = tmpdb->error(); - if (e != BasicDB::Error::NOREC) { - set_error(_KCCODELINE_, e.code(), e.message()); - err = true; - } - } - } - } - return !err; - } - /** - * Check whether a record exists. - * @param kbuf the pointer to the key region. - * @param ksiz the size of the key region. - * @return true if the record exists, or false if not. - */ - bool check_impl(const char* kbuf, size_t ksiz) { - _assert_(kbuf && ksiz <= MEMMAXSIZ); - char vbuf; - if (db_.get(kbuf, ksiz, &vbuf, 1) >= 0) return true; - if (cache_) { - size_t vsiz; - if (cache_->get(kbuf, ksiz, &vsiz)) return true; - if (tmpdbs_) { - for (size_t i = 0; i < dbnum_; i++) { - BasicDB* tmpdb = tmpdbs_[i]; - if (tmpdb->get(kbuf, ksiz, &vbuf, 1)) return true; - } - } - } - return false; - } - /** - * Get the number of records. - * @return the number of records, or -1 on failure. - */ - int64_t count_impl() { - _assert_(true); - int64_t dbcnt = db_.count(); - if (dbcnt < 0) return -1; - if (!cache_) return dbcnt; - int64_t ccnt = cache_->count(); - return dbcnt > ccnt ? dbcnt : ccnt; - } - /** - * Get the size of the database file. - * @return the size of the database file in bytes. - */ - int64_t size_impl() { - _assert_(true); - int64_t dbsiz = db_.size(); - if (dbsiz < 0) return -1; - return dbsiz + csiz_; - } - /** Dummy constructor to forbid the use. */ - IndexDB(const IndexDB&); - /** Dummy Operator to forbid the use. */ - IndexDB& operator =(const IndexDB&); - /** The internal database. */ - PolyDB db_; - /** The open mode. */ - uint32_t omode_; - /** The record comparator. */ - Comparator* rcomp_; - /** The base path of temporary databases. */ - std::string tmppath_; - /** The temporary databases. */ - BasicDB** tmpdbs_; - /** The number of temporary databases. */ - size_t dbnum_; - /** The logical clock for temporary databases. */ - int64_t dbclock_; - /** The internal cache. */ - TinyHashMap* cache_; - /** The current size of the internal cache. */ - int64_t csiz_; - /** The limit size of the internal cache. */ - int64_t clim_; -}; - - -} // common namespace - -#endif // duplication check - -// END OF FILE |