path: root/plugins/Dbx_kyoto/src/kyotocabinet/kcdbext.h

/*************************************************************************************************
 * 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() :
      mlock_(), 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 method lock. */
  RWLock mlock_;
  /** 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