kctextdb.h

/*************************************************************************************************
 * Plain text database
 *                                                               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 _KCTEXTDB_H                      // duplication check
#define _KCTEXTDB_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>

namespace kyotocabinet {                 // common namespace


/**
 * Plain text database.
 * @note Although this class is designed to use a text file as a database file, not all methods
 * are implemented.  Each line in the text is treated as a record.  When storing a record, the
 * key is ignored and the value only is appended at the end of the file.  Records can be
 * retrieved only by the iterator and the cursor mechanisms.  No record can be retrieved by
 * specifying the key.  When accessing a record by the iterator, the key is given as the offset
 * from the beginning of the file for descriptive purposes.  Any existing record cannot be
 * modified and deleted.
 */
class TextDB : public BasicDB {
 public:
  class Cursor;
 private:
  class ScopedVisitor;
  /** An alias of list of cursors. */
  typedef std::list<Cursor*> CursorList;
  /** An alias of a past record. */
  typedef std::pair<int64_t, std::string> Record;
  /** The size of the IO buffer. */
  static const size_t IOBUFSIZ = 1024;
 public:
  /**
   * Cursor to indicate a record.
   */
  class Cursor : public BasicDB::Cursor {
    friend class TextDB;
   public:
    /**
     * Constructor.
     * @param db the container database object.
     */
    explicit Cursor(TextDB* db) : db_(db), off_(INT64MAX), end_(0), queue_(), line_() {
      _assert_(db);
      ScopedRWLock lock(&db_->mlock_, true);
      db_->curs_.push_back(this);
    }
    /**
     * Destructor.
     */
    virtual ~Cursor() {
      _assert_(true);
      if (!db_) return;
      ScopedRWLock lock(&db_->mlock_, true);
      db_->curs_.remove(this);
    }
    /**
     * Accept a visitor to the current record.
     * @param visitor a visitor object.
     * @param writable true for writable operation, or false for read-only operation.
     * @param step true to move the cursor to the next record, or false for no move.
     * @return true on success, or false on failure.
     * @note The key is generated from the offset of each record.  To avoid deadlock, any
     * explicit database operation must not be performed in this function.
     */
    bool accept(Visitor* visitor, bool writable = true, bool step = false) {
      _assert_(visitor);
      ScopedRWLock lock(&db_->mlock_, false);
      if (db_->omode_ == 0) {
        db_->set_error(_KCCODELINE_, Error::INVALID, "not opened");
        return false;
      }
      if (writable && !db_->writer_) {
        db_->set_error(_KCCODELINE_, Error::NOPERM, "permission denied");
        return false;
      }
      bool err = false;
      if (!accept_impl(visitor, step)) err = true;
      return !err;
    }
    /**
     * Jump the cursor to the first record for forward scan.
     * @return true on success, or false on failure.
     */
    bool jump() {
      _assert_(true);
      ScopedRWLock lock(&db_->mlock_, false);
      if (db_->omode_ == 0) {
        db_->set_error(_KCCODELINE_, Error::INVALID, "not opened");
        return false;
      }
      off_ = 0;
      end_ = db_->file_.size();
      queue_.clear();
      line_.clear();
      if (off_ >= end_) {
        db_->set_error(_KCCODELINE_, Error::NOREC, "no record");
        return false;
      }
      return true;
    }
    /**
     * Jump the cursor to a record for forward scan.
     * @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 jump(const char* kbuf, size_t ksiz) {
      _assert_(kbuf && ksiz <= MEMMAXSIZ);
      ScopedRWLock lock(&db_->mlock_, true);
      if (db_->omode_ == 0) {
        db_->set_error(_KCCODELINE_, Error::INVALID, "not opened");
        return false;
      }
      off_ = atoin(kbuf, ksiz);
      end_ = db_->file_.size();
      queue_.clear();
      line_.clear();
      if (off_ >= end_) {
        db_->set_error(_KCCODELINE_, Error::NOREC, "no record");
        return false;
      }
      return true;
    }
    /**
     * Jump the cursor to a record for forward scan.
     * @note Equal to the original Cursor::jump method except that the parameter is std::string.
     */
    bool jump(const std::string& key) {
      _assert_(true);
      return jump(key.c_str(), key.size());
    }
    /**
     * Jump the cursor to the last record for backward scan.
     * @note This is a dummy implementation for compatibility.
     */
    bool jump_back() {
      _assert_(true);
      ScopedRWLock lock(&db_->mlock_, true);
      if (db_->omode_ == 0) {
        db_->set_error(_KCCODELINE_, Error::INVALID, "not opened");
        return false;
      }
      db_->set_error(_KCCODELINE_, Error::NOIMPL, "not implemented");
      return false;
    }
    /**
     * Jump the cursor to a record for backward scan.
     * @note This is a dummy implementation for compatibility.
     */
    bool jump_back(const char* kbuf, size_t ksiz) {
      _assert_(kbuf && ksiz <= MEMMAXSIZ);
      ScopedRWLock lock(&db_->mlock_, true);
      if (db_->omode_ == 0) {
        db_->set_error(_KCCODELINE_, Error::INVALID, "not opened");
        return false;
      }
      db_->set_error(_KCCODELINE_, Error::NOIMPL, "not implemented");
      return false;
    }
    /**
     * Jump the cursor to a record for backward scan.
     * @note This is a dummy implementation for compatibility.
     */
    bool jump_back(const std::string& key) {
      _assert_(true);
      return jump_back(key.c_str(), key.size());
    }
    /**
     * Step the cursor to the next record.
     * @return true on success, or false on failure.
     */
    bool step() {
      _assert_(true);
      ScopedRWLock lock(&db_->mlock_, false);
      if (db_->omode_ == 0) {
        db_->set_error(_KCCODELINE_, Error::INVALID, "not opened");
        return false;
      }
      if (queue_.empty() && !read_next()) return false;
      if (queue_.empty()) {
        db_->set_error(_KCCODELINE_, Error::NOREC, "no record");
        return false;
      }
      queue_.pop_front();
      return true;
    }
    /**
     * Step the cursor to the previous record.
     * @note This is a dummy implementation for compatibility.
     */
    bool step_back() {
      _assert_(true);
      ScopedRWLock lock(&db_->mlock_, true);
      if (db_->omode_ == 0) {
        db_->set_error(_KCCODELINE_, Error::INVALID, "not opened");
        return false;
      }
      db_->set_error(_KCCODELINE_, Error::NOIMPL, "not implemented");
      return false;
    }
    /**
     * Get the database object.
     * @return the database object.
     */
    TextDB* db() {
      _assert_(true);
      return db_;
    }
   private:
    /**
     * Accept a visitor to the current record.
     * @param visitor a visitor object.
     * @param step true to move the cursor to the next record, or false for no move.
     * @return true on success, or false on failure.
     */
    bool accept_impl(Visitor* visitor, bool step) {
      _assert_(visitor);
      if (queue_.empty() && !read_next()) return false;
      if (queue_.empty()) {
        db_->set_error(_KCCODELINE_, Error::NOREC, "no record");
        return false;
      }
      bool err = false;
      const Record& rec = queue_.front();
      char kbuf[NUMBUFSIZ];
      size_t ksiz = db_->write_key(kbuf, rec.first);
      size_t vsiz;
      const char* vbuf = visitor->visit_full(kbuf, ksiz,
                                             rec.second.data(), rec.second.size(), &vsiz);
      if (vbuf != Visitor::NOP && vbuf != Visitor::REMOVE) {
        char stack[IOBUFSIZ];
        size_t rsiz = vsiz + 1;
        char* rbuf = rsiz > sizeof(stack) ? new char[rsiz] : stack;
        std::memcpy(rbuf, vbuf, vsiz);
        rbuf[vsiz] = '\n';
        if (!db_->file_.append(rbuf, rsiz)) {
          db_->set_error(_KCCODELINE_, Error::SYSTEM, db_->file_.error());
          err = true;
        }
        if (rbuf != stack) delete[] rbuf;
        if (db_->autosync_ && !db_->file_.synchronize(true)) {
          db_->set_error(_KCCODELINE_, Error::SYSTEM, db_->file_.error());
          err = true;
        }
      }
      if (step) queue_.pop_front();
      return !err;
    }
    /**
     * Read the next record.
     * @return true on success, or false on failure.
     */
    bool read_next() {
      _assert_(true);
      while (off_ < end_) {
        char stack[IOBUFSIZ];
        int64_t rsiz = end_ - off_;
        if (rsiz > (int64_t)sizeof(stack)) rsiz = sizeof(stack);
        if (!db_->file_.read_fast(off_, stack, rsiz)) {
          db_->set_error(_KCCODELINE_, Error::SYSTEM, db_->file_.error());
          return false;
        }
        const char* rp = stack;
        const char* pv = rp;
        const char* ep = rp + rsiz;
        while (rp < ep) {
          if (*rp == '\n') {
            line_.append(pv, rp - pv);
            Record rec;
            rec.first = off_ + pv - stack;
            rec.second = line_;
            queue_.push_back(rec);
            line_.clear();
            rp++;
            pv = rp;
          } else {
            rp++;
          }
        }
        line_.append(pv, rp - pv);
        off_ += rsiz;
        if (!queue_.empty()) break;
      }
      return true;
    }
    /** Dummy constructor to forbid the use. */
    Cursor(const Cursor&);
    /** Dummy Operator to forbid the use. */
    Cursor& operator =(const Cursor&);
    /** The inner database. */
    TextDB* db_;
    /** The current offset. */
    int64_t off_;
    /** The end offset. */
    int64_t end_;
    /** The queue of read lines. */
    std::deque<Record> queue_;
    /** The current line. */
    std::string line_;
  };
  /**
   * Default constructor.
   */
  explicit TextDB() :
      error_(), logger_(NULL), logkinds_(0), mtrigger_(NULL),
      omode_(0), writer_(false), autotran_(false), autosync_(false),
      file_(), curs_(), path_("") {
    _assert_(true);
  }
  /**
   * Destructor.
   * @note If the database is not closed, it is closed implicitly.
   */
  virtual ~TextDB() {
    _assert_(true);
    if (omode_ != 0) close();
    if (!curs_.empty()) {
      CursorList::const_iterator cit = curs_.begin();
      CursorList::const_iterator citend = curs_.end();
      while (cit != citend) {
        Cursor* cur = *cit;
        cur->db_ = NULL;
        ++cit;
      }
    }
  }
  /**
   * Accept a visitor to a record.
   * @param kbuf the pointer to the key region.
   * @param ksiz the size of the key region.
   * @param visitor a visitor object.
   * @param writable true for writable operation, or false for read-only operation.
   * @return true on success, or false on failure.
   * @note No record can be retrieved by specifying the key and the Visitor::visit_empty method
   * is always called.  To avoid deadlock, any explicit database operation must not be performed
   * in this function.
   */
  bool accept(const char* kbuf, size_t ksiz, Visitor* visitor, bool writable = true) {
    _assert_(kbuf && ksiz <= MEMMAXSIZ && visitor);
    ScopedRWLock lock(&mlock_, false);
    if (omode_ == 0) {
      set_error(_KCCODELINE_, Error::INVALID, "not opened");
      return false;
    }
    if (writable && !writer_) {
      set_error(_KCCODELINE_, Error::NOPERM, "permission denied");
      return false;
    }
    bool err = false;
    if (!accept_impl(kbuf, ksiz, visitor)) err = true;
    return !err;
  }
  /**
   * Accept a visitor to multiple records at once.
   * @param keys specifies a string vector of the keys.
   * @param visitor a visitor object.
   * @param writable true for writable operation, or false for read-only operation.
   * @return true on success, or false on failure.
   * @note No record can be retrieved by specifying the key and the Visitor::visit_empty method
   * is always called.  To avoid deadlock, any explicit database operation must not be performed
   * in this function.
   */
  bool accept_bulk(const std::vector<std::string>& keys, Visitor* visitor,
                   bool writable = true) {
    _assert_(visitor);
    ScopedRWLock lock(&mlock_, true);
    if (omode_ == 0) {
      set_error(_KCCODELINE_, Error::INVALID, "not opened");
      return false;
    }
    if (writable && !writer_) {
      set_error(_KCCODELINE_, Error::NOPERM, "permission denied");
      return false;
    }
    ScopedVisitor svis(visitor);
    bool err = false;
    std::vector<std::string>::const_iterator kit = keys.begin();
    std::vector<std::string>::const_iterator kitend = keys.end();
    while (kit != kitend) {
      if (!accept_impl(kit->data(), kit->size(), visitor)) err = true;
      ++kit;
    }
    return !err;
  }
  /**
   * Iterate to accept a visitor for each record.
   * @param visitor a visitor object.
   * @param writable true for writable operation, or false for read-only operation.
   * @param checker a progress checker object.  If it is NULL, no checking is performed.
   * @return true on success, or false on failure.
   * @note The key is generated from the offset of each record.  To avoid deadlock, any explicit
   * database operation must not be performed in this function.
   */
  bool iterate(Visitor *visitor, bool writable = true, ProgressChecker* checker = NULL) {
    _assert_(visitor);
    ScopedRWLock lock(&mlock_, true);
    if (omode_ == 0) {
      set_error(_KCCODELINE_, Error::INVALID, "not opened");
      return false;
    }
    if (writable && !writer_) {
      set_error(_KCCODELINE_, Error::NOPERM, "permission denied");
      return false;
    }
    ScopedVisitor svis(visitor);
    bool err = false;
    if (!iterate_impl(visitor, checker)) err = true;
    trigger_meta(MetaTrigger::ITERATE, "iterate");
    return !err;
  }
  /**
   * Scan each record in parallel.
   * @param visitor a visitor object.
   * @param thnum the number of worker threads.
   * @param checker a progress checker object.  If it is NULL, no checking is performed.
   * @return true on success, or false on failure.
   * @note This function is for reading records and not for updating ones.  The return value of
   * the visitor is just ignored.  The key is generated from the offset of each record.  To
   * avoid deadlock, any explicit database operation must not be performed in this function.
   */
  bool scan_parallel(Visitor *visitor, size_t thnum, ProgressChecker* checker = NULL) {
    _assert_(visitor && thnum <= MEMMAXSIZ);
    ScopedRWLock lock(&mlock_, false);
    if (omode_ == 0) {
      set_error(_KCCODELINE_, Error::INVALID, "not opened");
      return false;
    }
    if (thnum < 1) thnum = 1;
    if (thnum > (size_t)INT8MAX) thnum = INT8MAX;
    ScopedVisitor svis(visitor);
    bool err = false;
    if (!scan_parallel_impl(visitor, thnum, checker)) err = true;
    trigger_meta(MetaTrigger::ITERATE, "scan_parallel");
    return !err;
  }
  /**
   * Get the last happened error.
   * @return the last happened error.
   */
  Error error() const {
    _assert_(true);
    return 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,
                 Error::Code code, const char* message) {
    _assert_(file && line > 0 && func && message);
    error_->set(code, message);
    if (logger_) {
      Logger::Kind kind = code == Error::BROKEN || code == Error::SYSTEM ?
          Logger::ERROR : Logger::INFO;
      if (kind & logkinds_)
        report(file, line, func, kind, "%d: %s: %s", code, Error::codename(code), message);
    }
  }
  /**
   * Open a database file.
   * @param path the path of a database file.
   * @param mode the connection mode.  TextDB::OWRITER as a writer, TextDB::OREADER as a
   * reader.  The following may be added to the writer mode by bitwise-or: TextDB::OCREATE,
   * which means it creates a new database if the file does not exist, TextDB::OTRUNCATE, which
   * means it creates a new database regardless if the file exists, TextDB::OAUTOTRAN, which
   * means each updating operation is performed in implicit transaction, TextDB::OAUTOSYNC,
   * which means each updating operation is followed by implicit synchronization with the file
   * system.  The following may be added to both of the reader mode and the writer mode by
   * bitwise-or: TextDB::ONOLOCK, which means it opens the database file without file locking,
   * TextDB::OTRYLOCK, which means locking is performed without blocking, TextDB::ONOREPAIR,
   * which means the database file is not repaired implicitly even if file destruction is
   * detected.
   * @return true on success, or false on failure.
   * @note Every opened database must be closed by the TextDB::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 = OWRITER | OCREATE) {
    _assert_(true);
    ScopedRWLock lock(&mlock_, true);
    if (omode_ != 0) {
      set_error(_KCCODELINE_, Error::INVALID, "already opened");
      return false;
    }
    report(_KCCODELINE_, Logger::DEBUG, "opening the database (path=%s)", path.c_str());
    writer_ = false;
    autotran_ = false;
    autosync_ = false;
    uint32_t fmode = File::OREADER;
    if (mode & OWRITER) {
      writer_ = true;
      fmode = File::OWRITER;
      if (mode & OCREATE) fmode |= File::OCREATE;
      if (mode & OTRUNCATE) fmode |= File::OTRUNCATE;
      if (mode & OAUTOTRAN) autotran_ = true;
      if (mode & OAUTOSYNC) autosync_ = true;
    }
    if (mode & ONOLOCK) fmode |= File::ONOLOCK;
    if (mode & OTRYLOCK) fmode |= File::OTRYLOCK;
    if (!file_.open(path, fmode, 0)) {
      const char* emsg = file_.error();
      Error::Code code = Error::SYSTEM;
      if (std::strstr(emsg, "(permission denied)") || std::strstr(emsg, "(directory)")) {
        code = Error::NOPERM;
      } else if (std::strstr(emsg, "(file not found)") || std::strstr(emsg, "(invalid path)")) {
        code = Error::NOREPOS;
      }
      set_error(_KCCODELINE_, code, emsg);
      return false;
    }
    if (autosync_ && !File::synchronize_whole()) {
      set_error(_KCCODELINE_, Error::SYSTEM, "synchronizing the file system failed");
      file_.close();
      return false;
    }
    path_.append(path);
    omode_ = mode;
    trigger_meta(MetaTrigger::OPEN, "open");
    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_, Error::INVALID, "not opened");
      return false;
    }
    report(_KCCODELINE_, Logger::DEBUG, "closing the database (path=%s)", path_.c_str());
    bool err = false;
    disable_cursors();
    if (!file_.close()) {
      set_error(_KCCODELINE_, Error::SYSTEM, file_.error());
      err = true;
    }
    omode_ = 0;
    path_.clear();
    trigger_meta(MetaTrigger::CLOSE, "close");
    return !err;
  }
  /**
   * 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.
   * @param checker a progress checker object.  If it is NULL, no checking 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, FileProcessor* proc = NULL,
                   ProgressChecker* checker = NULL) {
    _assert_(true);
    ScopedRWLock lock(&mlock_, false);
    if (omode_ == 0) {
      set_error(_KCCODELINE_, Error::INVALID, "not opened");
      return false;
    }
    bool err = false;
    if (!synchronize_impl(hard, proc, checker)) err = true;
    trigger_meta(MetaTrigger::SYNCHRONIZE, "synchronize");
    return !err;
  }
  /**
   * Occupy database by locking and do something meanwhile.
   * @param writable true to use writer lock, or false to use reader lock.
   * @param proc a processor object.  If it is NULL, no processing is performed.
   * @return true on success, or false on failure.
   * @note The operation of the processor 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 occupy(bool writable = true, FileProcessor* proc = NULL) {
    _assert_(true);
    ScopedRWLock lock(&mlock_, writable);
    bool err = false;
    if (proc && !proc->process(path_, -1, file_.size())) {
      set_error(_KCCODELINE_, Error::LOGIC, "processing failed");
      err = true;
    }
    trigger_meta(MetaTrigger::OCCUPY, "occupy");
    return !err;
  }
  /**
   * Begin transaction.
   * @param hard true for physical synchronization with the device, or false for logical
   * synchronization with the file system.
   * @return true on success, or false on failure.
   */
  bool begin_transaction(bool hard = false) {
    _assert_(true);
    ScopedRWLock lock(&mlock_, true);
    if (omode_ == 0) {
      set_error(_KCCODELINE_, Error::INVALID, "not opened");
      return false;
    }
    set_error(_KCCODELINE_, Error::NOIMPL, "not implemented");
    return false;
  }
  /**
   * Try to begin transaction.
   * @param hard true for physical synchronization with the device, or false for logical
   * synchronization with the file system.
   * @return true on success, or false on failure.
   */
  bool begin_transaction_try(bool hard = false) {
    _assert_(true);
    ScopedRWLock lock(&mlock_, true);
    if (omode_ == 0) {
      set_error(_KCCODELINE_, Error::INVALID, "not opened");
      return false;
    }
    set_error(_KCCODELINE_, Error::NOIMPL, "not implemented");
    return false;
  }
  /**
   * End transaction.
   * @param commit true to commit the transaction, or false to abort the transaction.
   * @return true on success, or false on failure.
   */
  bool end_transaction(bool commit = true) {
    _assert_(true);
    ScopedRWLock lock(&mlock_, true);
    if (omode_ == 0) {
      set_error(_KCCODELINE_, Error::INVALID, "not opened");
      return false;
    }
    set_error(_KCCODELINE_, Error::NOIMPL, "not implemented");
    return false;
  }
  /**
   * 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_, Error::INVALID, "not opened");
      return false;
    }
    if (!writer_) {
      set_error(_KCCODELINE_, Error::NOPERM, "permission denied");
      return false;
    }
    disable_cursors();
    if (!file_.truncate(0)) {
      set_error(_KCCODELINE_, Error::SYSTEM, file_.error());
      return false;
    }
    if (autosync_ && !file_.synchronize(true)) {
      set_error(_KCCODELINE_, Error::SYSTEM, file_.error());
      return false;
    }
    trigger_meta(MetaTrigger::CLEAR, "clear");
    return true;
  }
  /**
   * Get the number of records.
   * @return the number of records, or -1 on failure.
   */
  int64_t count() {
    _assert_(true);
    ScopedRWLock lock(&mlock_, false);
    if (omode_ == 0) {
      set_error(_KCCODELINE_, Error::INVALID, "not opened");
      return -1;
    }
    set_error(_KCCODELINE_, Error::NOIMPL, "not implemented");
    return -1;
  }
  /**
   * 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);
    if (omode_ == 0) {
      set_error(_KCCODELINE_, Error::INVALID, "not opened");
      return -1;
    }
    return file_.size();
  }
  /**
   * 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 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);
    ScopedRWLock lock(&mlock_, true);
    if (omode_ == 0) {
      set_error(_KCCODELINE_, Error::INVALID, "not opened");
      return false;
    }
    (*strmap)["type"] = strprintf("%u", (unsigned)TYPETEXT);
    (*strmap)["realtype"] = strprintf("%u", (unsigned)TYPETEXT);
    (*strmap)["path"] = path_;
    (*strmap)["size"] = strprintf("%lld", (long long)file_.size());
    return true;
  }
  /**
   * 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.
   */
  Cursor* cursor() {
    _assert_(true);
    return new Cursor(this);
  }
  /**
   * 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, Logger::Kind kind,
           const char* message) {
    _assert_(file && line > 0 && func && message);
    ScopedRWLock lock(&mlock_, false);
    if (!logger_) return;
    logger_->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(Logger* logger, uint32_t kinds = Logger::WARN | Logger::ERROR) {
    _assert_(logger);
    ScopedRWLock lock(&mlock_, true);
    if (omode_ != 0) {
      set_error(_KCCODELINE_, Error::INVALID, "already opened");
      return false;
    }
    logger_ = logger;
    logkinds_ = kinds;
    return true;
  }
  /**
   * Set the internal meta operation trigger.
   * @param trigger the trigger object.
   * @return true on success, or false on failure.
   */
  bool tune_meta_trigger(MetaTrigger* trigger) {
    _assert_(trigger);
    ScopedRWLock lock(&mlock_, true);
    if (omode_ != 0) {
      set_error(_KCCODELINE_, Error::INVALID, "already opened");
      return false;
    }
    mtrigger_ = trigger;
    return true;
  }
 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 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 format the printf-like format string.
   * @param ... used according to the format string.
   */
  void report(const char* file, int32_t line, const char* func, Logger::Kind kind,
              const char* format, ...) {
    _assert_(file && line > 0 && func && format);
    if (!logger_ || !(kind & logkinds_)) return;
    std::string message;
    strprintf(&message, "%s: ", path_.empty() ? "-" : path_.c_str());
    va_list ap;
    va_start(ap, format);
    vstrprintf(&message, format, ap);
    va_end(ap);
    logger_->log(file, line, func, kind, message.c_str());
  }
  /**
   * Report a message for debugging with variable number of arguments.
   * @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 format the printf-like format string.
   * @param ap used according to the format string.
   */
  void report_valist(const char* file, int32_t line, const char* func, Logger::Kind kind,
                     const char* format, va_list ap) {
    _assert_(file && line > 0 && func && format);
    if (!logger_ || !(kind & logkinds_)) return;
    std::string message;
    strprintf(&message, "%s: ", path_.empty() ? "-" : path_.c_str());
    vstrprintf(&message, format, ap);
    logger_->log(file, line, func, kind, message.c_str());
  }
  /**
   * Report the content of a binary buffer for debugging.
   * @param file the file name of the epicenter.
   * @param line the line number of the epicenter.
   * @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 name the name of the information.
   * @param buf the binary buffer.
   * @param size the size of the binary buffer
   */
  void report_binary(const char* file, int32_t line, const char* func, Logger::Kind kind,
                     const char* name, const char* buf, size_t size) {
    _assert_(file && line > 0 && func && name && buf && size <= MEMMAXSIZ);
    if (!logger_) return;
    char* hex = hexencode(buf, size);
    report(file, line, func, kind, "%s=%s", name, hex);
    delete[] hex;
  }
  /**
   * Trigger a meta database operation.
   * @param kind the kind of the event.  MetaTrigger::OPEN for opening, MetaTrigger::CLOSE for
   * closing, MetaTrigger::CLEAR for clearing, MetaTrigger::ITERATE for iteration,
   * MetaTrigger::SYNCHRONIZE for synchronization, MetaTrigger::BEGINTRAN for beginning
   * transaction, MetaTrigger::COMMITTRAN for committing transaction, MetaTrigger::ABORTTRAN
   * for aborting transaction, and MetaTrigger::MISC for miscellaneous operations.
   * @param message the supplement message.
   */
  void trigger_meta(MetaTrigger::Kind kind, const char* message) {
    _assert_(message);
    if (mtrigger_) mtrigger_->trigger(kind, message);
  }
 private:
  /**
   * Scoped visitor.
   */
  class ScopedVisitor {
   public:
    /** constructor */
    explicit ScopedVisitor(Visitor* visitor) : visitor_(visitor) {
      _assert_(visitor);
      visitor_->visit_before();
    }
    /** destructor */
    ~ScopedVisitor() {
      _assert_(true);
      visitor_->visit_after();
    }
   private:
    Visitor* visitor_;                   ///< visitor
  };
  /**
   * Accept a visitor to a record.
   * @param kbuf the pointer to the key region.
   * @param ksiz the size of the key region.
   * @param visitor a visitor object.
   * @return true on success, or false on failure.
   */
  bool accept_impl(const char* kbuf, size_t ksiz, Visitor* visitor) {
    _assert_(kbuf && ksiz <= MEMMAXSIZ && visitor);
    bool err = false;
    size_t vsiz;
    const char* vbuf = visitor->visit_empty(kbuf, ksiz, &vsiz);
    if (vbuf != Visitor::NOP && vbuf != Visitor::REMOVE) {
      size_t rsiz = vsiz + 1;
      char stack[IOBUFSIZ];
      char* rbuf = rsiz > sizeof(stack) ? new char[rsiz] : stack;
      std::memcpy(rbuf, vbuf, vsiz);
      rbuf[vsiz] = '\n';
      if (!file_.append(rbuf, rsiz)) {
        set_error(_KCCODELINE_, Error::SYSTEM, file_.error());
        err = true;
      }
      if (rbuf != stack) delete[] rbuf;
      if (autosync_ && !file_.synchronize(true)) {
        set_error(_KCCODELINE_, Error::SYSTEM, file_.error());
        err = true;
      }
    }
    return !err;
  }
  /**
   * Iterate to accept a visitor for each record.
   * @param visitor a visitor object.
   * @param checker a progress checker object.
   * @return true on success, or false on failure.
   */
  bool iterate_impl(Visitor* visitor, ProgressChecker* checker) {
    _assert_(visitor);
    if (checker && !checker->check("iterate", "beginning", 0, -1)) {
      set_error(_KCCODELINE_, Error::LOGIC, "checker failed");
      return false;
    }
    int64_t off = 0;
    int64_t end = file_.size();
    int64_t curcnt = 0;
    std::string line;
    char stack[IOBUFSIZ*4];
    while (off < end) {
      int64_t rsiz = end - off;
      if (rsiz > (int64_t)sizeof(stack)) rsiz = sizeof(stack);
      if (!file_.read_fast(off, stack, rsiz)) {
        set_error(_KCCODELINE_, Error::SYSTEM, file_.error());
        return false;
      }
      const char* rp = stack;
      const char* pv = rp;
      const char* ep = rp + rsiz;
      while (rp < ep) {
        if (*rp == '\n') {
          char kbuf[NUMBUFSIZ];
          size_t ksiz = write_key(kbuf, off + pv - stack);
          const char* vbuf;
          size_t vsiz;
          if (line.empty()) {
            vbuf = visitor->visit_full(kbuf, ksiz, pv, rp - pv, &vsiz);
          } else {
            line.append(pv, rp - pv);
            vbuf = visitor->visit_full(kbuf, ksiz, line.data(), line.size(), &vsiz);
            line.clear();
          }
          if (vbuf != Visitor::NOP && vbuf != Visitor::REMOVE) {
            char tstack[IOBUFSIZ];
            size_t trsiz = vsiz + 1;
            char* trbuf = trsiz > sizeof(tstack) ? new char[trsiz] : tstack;
            std::memcpy(trbuf, vbuf, vsiz);
            trbuf[vsiz] = '\n';
            if (!file_.append(trbuf, trsiz)) {
              set_error(_KCCODELINE_, Error::SYSTEM, file_.error());
              if (trbuf != stack) delete[] trbuf;
              return false;
            }
            if (trbuf != tstack) delete[] trbuf;
          }
          curcnt++;
          if (checker && !checker->check("iterate", "processing", curcnt, -1)) {
            set_error(_KCCODELINE_, Error::LOGIC, "checker failed");
            return false;
          }
          rp++;
          pv = rp;
        } else {
          rp++;
        }
      }
      line.append(pv, rp - pv);
      off += rsiz;
    }
    if (checker && !checker->check("iterate", "ending", -1, -1)) {
      set_error(_KCCODELINE_, Error::LOGIC, "checker failed");
      return false;
    }
    return true;
  }
  /**
   * Scan each record in parallel.
   * @param visitor a visitor object.
   * @param thnum the number of worker threads.
   * @param checker a progress checker object.
   * @return true on success, or false on failure.
   */
  bool scan_parallel_impl(Visitor *visitor, size_t thnum, ProgressChecker* checker) {
    _assert_(visitor && thnum <= MEMMAXSIZ);
    if (checker && !checker->check("scan_parallel", "beginning", -1, -1)) {
      set_error(_KCCODELINE_, Error::LOGIC, "checker failed");
      return false;
    }
    int64_t off = 0;
    int64_t end = file_.size();
    int64_t gap = (end - off) / thnum;
    std::vector<int64_t> offs;
    while (off < end) {
      offs.push_back(off);
      int64_t edge = off + gap;
      while (true) {
        if (edge >= end) {
          off = end;
          break;
        }
        char rbuf[IOBUFSIZ];
        int64_t rsiz = end - edge;
        if (rsiz > (int64_t)sizeof(rbuf)) rsiz = sizeof(rbuf);
        if (!file_.read_fast(edge, rbuf, rsiz)) {
          set_error(_KCCODELINE_, Error::SYSTEM, file_.error());
          return false;
        }
        int64_t noff = -1;
        const char* rp = rbuf;
        const char* ep = rp + rsiz;
        while (rp < ep) {
          if (*rp == '\n') {
            noff = edge + (rp - rbuf);
            break;
          }
          rp++;
        }
        if (noff >= 0) {
          off = noff + 1;
          break;
        }
        edge += rsiz;
      }
    }
    bool err = false;
    size_t onum = offs.size();
    if (onum > 0) {
      class ThreadImpl : public Thread {
       public:
        explicit ThreadImpl() :
            db_(NULL), visitor_(NULL), checker_(NULL), begoff_(0), endoff_(0), error_() {}
        void init(TextDB* db, Visitor* visitor, ProgressChecker* checker,
                  int64_t begoff, int64_t endoff) {
          db_ = db;
          visitor_ = visitor;
          checker_ = checker;
          begoff_ = begoff;
          endoff_ = endoff;
        }
        const Error& error() {
          return error_;
        }
       private:
        void run() {
          TextDB* db = db_;
          File* file = &db->file_;
          Visitor* visitor = visitor_;
          ProgressChecker* checker = checker_;
          int64_t off = begoff_;
          int64_t end = endoff_;
          std::string line;
          char stack[IOBUFSIZ*4];
          while (off < end) {
            int64_t rsiz = end - off;
            if (rsiz > (int64_t)sizeof(stack)) rsiz = sizeof(stack);
            if (!file->read_fast(off, stack, rsiz)) {
              db->set_error(_KCCODELINE_, Error::SYSTEM, file->error());
              return;
            }
            const char* rp = stack;
            const char* pv = rp;
            const char* ep = rp + rsiz;
            while (rp < ep) {
              if (*rp == '\n') {
                char kbuf[NUMBUFSIZ];
                size_t ksiz = db->write_key(kbuf, off + pv - stack);
                if (line.empty()) {
                  size_t vsiz;
                  visitor->visit_full(kbuf, ksiz, pv, rp - pv, &vsiz);
                } else {
                  line.append(pv, rp - pv);
                  size_t vsiz;
                  visitor->visit_full(kbuf, ksiz, line.data(), line.size(), &vsiz);
                  line.clear();
                }
                if (checker && !checker->check("iterate", "processing", -1, -1)) {
                  db->set_error(_KCCODELINE_, Error::LOGIC, "checker failed");
                  return;
                }
                rp++;
                pv = rp;
              } else {
                rp++;
              }
            }
            line.append(pv, rp - pv);
            off += rsiz;
          }
        }
        TextDB* db_;
        Visitor* visitor_;
        ProgressChecker* checker_;
        int64_t begoff_;
        int64_t endoff_;
        Error error_;
      };
      ThreadImpl* threads = new ThreadImpl[onum];
      for (size_t i = 0; i < onum; i++) {
        int64_t begoff = offs[i];
        int64_t endoff = i < onum - 1 ? offs[i+1] : end;
        ThreadImpl* thread = threads + i;
        thread->init(this, visitor, checker, begoff, endoff);
        thread->start();
      }
      for (size_t i = 0; i < onum; i++) {
        ThreadImpl* thread = threads + i;
        thread->join();
        if (thread->error() != Error::SUCCESS) {
          *error_ = thread->error();
          err = true;
        }
      }
      delete[] threads;
    }
    if (checker && !checker->check("scan_parallel", "ending", -1, -1)) {
      set_error(_KCCODELINE_, Error::LOGIC, "checker failed");
      err = true;
    }
    return !err;
  }
  /**
   * 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.
   * @param checker a progress checker object.
   * @return true on success, or false on failure.
   */
  bool synchronize_impl(bool hard, FileProcessor* proc, ProgressChecker* checker) {
    _assert_(true);
    bool err = false;
    if (writer_) {
      if (checker && !checker->check("synchronize", "synchronizing the file", -1, -1)) {
        set_error(_KCCODELINE_, Error::LOGIC, "checker failed");
        return false;
      }
      if (!file_.synchronize(hard)) {
        set_error(_KCCODELINE_, Error::SYSTEM, file_.error());
        err = true;
      }
    }
    if (proc) {
      if (checker && !checker->check("synchronize", "running the post processor", -1, -1)) {
        set_error(_KCCODELINE_, Error::LOGIC, "checker failed");
        return false;
      }
      if (!proc->process(path_, -1, file_.size())) {
        set_error(_KCCODELINE_, Error::LOGIC, "postprocessing failed");
        err = true;
      }
    }
    return !err;
  }
  /**
   * Disable all cursors.
   */
  void disable_cursors() {
    _assert_(true);
    if (curs_.empty()) return;
    CursorList::const_iterator cit = curs_.begin();
    CursorList::const_iterator citend = curs_.end();
    while (cit != citend) {
      Cursor* cur = *cit;
      cur->off_ = INT64MAX;
      ++cit;
    }
  }
  /**
   * Write the key pattern into a buffer.
   * @param kbuf the destination buffer.
   * @param off the offset of the record.
   * @return the size of the key pattern.
   */
  size_t write_key(char* kbuf, int64_t off) {
    _assert_(kbuf && off >= 0);
    for (size_t i = 0; i < sizeof(off); i++) {
      uint8_t c = off >> ((sizeof(off) - 1 - i) * 8);
      uint8_t h = c >> 4;
      if (h < 10) {
        *(kbuf++) = '0' + h;
      } else {
        *(kbuf++) = 'A' - 10 + h;
      }
      uint8_t l = c & 0xf;
      if (l < 10) {
        *(kbuf++) = '0' + l;
      } else {
        *(kbuf++) = 'A' - 10 + l;
      }
    }
    return sizeof(off) * 2;
  }
  /** Dummy constructor to forbid the use. */
  TextDB(const TextDB&);
  /** Dummy Operator to forbid the use. */
  TextDB& operator =(const TextDB&);
  /** The method lock. */
  RWLock mlock_;
  /** The last happened error. */
  TSD<Error> error_;
  /** The internal logger. */
  Logger* logger_;
  /** The kinds of logged messages. */
  uint32_t logkinds_;
  /** The internal meta operation trigger. */
  MetaTrigger* mtrigger_;
  /** The open mode. */
  uint32_t omode_;
  /** The flag for writer. */
  bool writer_;
  /** The flag for auto transaction. */
  bool autotran_;
  /** The flag for auto synchronization. */
  bool autosync_;
  /** The file for data. */
  File file_;
  /** The cursor objects. */
  CursorList curs_;
  /** The path of the database file. */
  std::string path_;
};


}                                        // common namespace

#endif                                   // duplication check

// END OF FILE