kccachedb.h

/*************************************************************************************************
 * Cache hash database
 *                                                               Copyright (C) 2009-2012 FAL Labs
 *                                                               Copyright (C) 2013-2017 Cloudflare Inc.
 *                                                               Copyright (C) 2018-2019 TrustYou GmbH
 * 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 _KCCACHEDB_H                     // duplication check
#define _KCCACHEDB_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>

namespace kyotocabinet                   // common namespace
{


    /**
     * On-memory hash database with LRU deletion.
     * @note This class is a concrete class to operate a hash database on memory.  This class can be
     * inherited but overwriting methods is forbidden.  Before every database operation, it is
     * necessary to call the CacheDB::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 CacheDB::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 CacheDB : public BasicDB
    {
            friend class PlantDB<CacheDB, BasicDB::TYPEGRASS>;
        public:
            class Cursor;
        private:
            struct Record;
            struct TranLog;
            struct Slot;
            class Repeater;
            class Setter;
            class Remover;
            class ScopedVisitor;
            /** An alias of list of cursors. */
            typedef std::list<Cursor *> CursorList;
            /** An alias of list of transaction logs. */
            typedef std::list<TranLog> TranLogList;
            /** The number of slot tables. */
            static const int32_t SLOTNUM = 16;
            /** The default bucket number. */
            static const size_t DEFBNUM = 1048583LL;
            /** The mininum number of buckets to use mmap. */
            static const size_t ZMAPBNUM = 32768;
            /** The maximum size of each key. */
            static const uint32_t KSIZMAX = 0xfffff;
            /** The size of the record buffer. */
            static const size_t RECBUFSIZ = 48;
            /** The size of the opaque buffer. */
            static const size_t OPAQUESIZ = 16;
            /** The threshold of busy loop and sleep for locking. */
            static const uint32_t LOCKBUSYLOOP = 8192;
        public:
            /**
             * Cursor to indicate a record.
             */
            class Cursor : public BasicDB::Cursor
            {
                    friend class CacheDB;
                public:
                    /**
                     * Constructor.
                     * @param db the container database object.
                     */
                    explicit Cursor ( CacheDB *db ) : db_ ( db ), sidx_ ( -1 ), rec_ ( NULL )
                    {
                        _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 operation for each record 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 accept ( Visitor *visitor, bool writable = true, bool step = false )
                    {
                        _assert_ ( visitor );
                        ScopedRWLock lock ( &db_->mlock_, true );
                        if ( db_->omode_ == 0 )
                        {
                            db_->set_error ( _KCCODELINE_, Error::INVALID, "not opened" );
                            return false;
                        }
                        if ( writable && ! ( db_->omode_ & OWRITER ) )
                        {
                            db_->set_error ( _KCCODELINE_, Error::NOPERM, "permission denied" );
                            return false;
                        }
                        if ( sidx_ < 0 || !rec_ )
                        {
                            db_->set_error ( _KCCODELINE_, Error::NOREC, "no record" );
                            return false;
                        }
                        uint32_t rksiz = rec_->ksiz & KSIZMAX;
                        char *dbuf = ( char * ) rec_ + sizeof ( *rec_ );
                        const char *rvbuf = dbuf + rksiz;
                        size_t rvsiz = rec_->vsiz;
                        char *zbuf = NULL;
                        size_t zsiz = 0;
                        if ( db_->comp_ )
                        {
                            zbuf = db_->comp_->decompress ( rvbuf, rvsiz, &zsiz );
                            if ( zbuf )
                            {
                                rvbuf = zbuf;
                                rvsiz = zsiz;
                            }
                        }
                        size_t vsiz;
                        const char *vbuf = visitor->visit_full ( dbuf, rksiz, rvbuf, rvsiz, &vsiz );
                        delete[] zbuf;
                        if ( vbuf == Visitor::REMOVE )
                        {
                            uint64_t hash = db_->hash_record ( dbuf, rksiz ) / SLOTNUM;
                            Slot *slot = db_->slots_ + sidx_;
                            Repeater repeater ( Visitor::REMOVE, 0 );
                            db_->accept_impl ( slot, hash, dbuf, rksiz, &repeater, db_->comp_, false );
                        }
                        else
                            if ( vbuf == Visitor::NOP )
                            {
                                if ( step )
                                {
                                    step_impl();
                                }
                            }
                            else
                            {
                                uint64_t hash = db_->hash_record ( dbuf, rksiz ) / SLOTNUM;
                                Slot *slot = db_->slots_ + sidx_;
                                Repeater repeater ( vbuf, vsiz );
                                db_->accept_impl ( slot, hash, dbuf, rksiz, &repeater, db_->comp_, false );
                                if ( step )
                                {
                                    step_impl();
                                }
                            }
                        return true;
                    }
                    /**
                     * 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_, true );
                        if ( db_->omode_ == 0 )
                        {
                            db_->set_error ( _KCCODELINE_, Error::INVALID, "not opened" );
                            return false;
                        }
                        for ( int32_t i = 0; i < SLOTNUM; i++ )
                        {
                            Slot *slot = db_->slots_ + i;
                            if ( slot->first )
                            {
                                sidx_ = i;
                                rec_ = slot->first;
                                return true;
                            }
                        }
                        db_->set_error ( _KCCODELINE_, Error::NOREC, "no record" );
                        sidx_ = -1;
                        rec_ = NULL;
                        return false;
                    }
                    /**
                     * 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;
                        }
                        if ( ksiz > KSIZMAX )
                        {
                            ksiz = KSIZMAX;
                        }
                        uint64_t hash = db_->hash_record ( kbuf, ksiz );
                        int32_t sidx = hash % SLOTNUM;
                        hash /= SLOTNUM;
                        Slot *slot = db_->slots_ + sidx;
                        size_t bidx = hash % slot->bnum;
                        Record *rec = slot->buckets[bidx];
                        Record **entp = slot->buckets + bidx;
                        uint32_t fhash = db_->fold_hash ( hash ) & ~KSIZMAX;
                        while ( rec )
                        {
                            uint32_t rhash = rec->ksiz & ~KSIZMAX;
                            uint32_t rksiz = rec->ksiz & KSIZMAX;
                            if ( fhash > rhash )
                            {
                                entp = &rec->left;
                                rec = rec->left;
                            }
                            else
                                if ( fhash < rhash )
                                {
                                    entp = &rec->right;
                                    rec = rec->right;
                                }
                                else
                                {
                                    char *dbuf = ( char * ) rec + sizeof ( *rec );
                                    int32_t kcmp = db_->compare_keys ( kbuf, ksiz, dbuf, rksiz );
                                    if ( kcmp < 0 )
                                    {
                                        entp = &rec->left;
                                        rec = rec->left;
                                    }
                                    else
                                        if ( kcmp > 0 )
                                        {
                                            entp = &rec->right;
                                            rec = rec->right;
                                        }
                                        else
                                        {
                                            sidx_ = sidx;
                                            rec_ = rec;
                                            return true;
                                        }
                                }
                        }
                        db_->set_error ( _KCCODELINE_, Error::NOREC, "no record" );
                        sidx_ = -1;
                        rec_ = NULL;
                        return false;
                    }
                    /**
                     * 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 );
                        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;
                    }
                    /**
                     * Step the cursor to the next record.
                     * @return true on success, or false on failure.
                     */
                    bool step()
                    {
                        _assert_ ( true );
                        ScopedRWLock lock ( &db_->mlock_, true );
                        if ( db_->omode_ == 0 )
                        {
                            db_->set_error ( _KCCODELINE_, Error::INVALID, "not opened" );
                            return false;
                        }
                        if ( sidx_ < 0 || !rec_ )
                        {
                            db_->set_error ( _KCCODELINE_, Error::NOREC, "no record" );
                            return false;
                        }
                        bool err = false;
                        if ( !step_impl() )
                        {
                            err = true;
                        }
                        return !err;
                    }
                    /**
                     * 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.
                     */
                    CacheDB *db()
                    {
                        _assert_ ( true );
                        return db_;
                    }
                private:
                    /**
                     * Step the cursor to the next record.
                     * @return true on success, or false on failure.
                     */
                    bool step_impl()
                    {
                        _assert_ ( true );
                        rec_ = rec_->next;
                        if ( !rec_ )
                        {
                            for ( int32_t i = sidx_ + 1; i < SLOTNUM; i++ )
                            {
                                Slot *slot = db_->slots_ + i;
                                if ( slot->first )
                                {
                                    sidx_ = i;
                                    rec_ = slot->first;
                                    return true;
                                }
                            }
                            db_->set_error ( _KCCODELINE_, Error::NOREC, "no record" );
                            sidx_ = -1;
                            rec_ = NULL;
                            return false;
                        }
                        return true;
                    }
                    /** Dummy constructor to forbid the use. */
                    Cursor ( const Cursor & );
                    /** Dummy Operator to forbid the use. */
                    Cursor &operator = ( const Cursor & );
                    /** The inner database. */
                    CacheDB *db_;
                    /** The index of the current slot. */
                    int32_t sidx_;
                    /** The current record. */
                    Record *rec_;
            };
            /**
             * Tuning options.
             */
            enum Option
            {
                TSMALL = 1 << 0,                     ///< dummy for compatibility
                TLINEAR = 1 << 1,                    ///< dummy for compatibility
                TCOMPRESS = 1 << 2                   ///< compress each record
            };
            /**
             * Status flags.
             */
            enum Flag
            {
                FOPEN = 1 << 0,                      ///< dummy for compatibility
                FFATAL = 1 << 1                      ///< dummy for compatibility
            };
            /**
             * Default constructor.
             */
            explicit CacheDB() :
                mlock_(), flock_(), error_(), logger_ ( NULL ), logkinds_ ( 0 ), mtrigger_ ( NULL ),
                omode_ ( 0 ), curs_(), path_ ( "" ), type_ ( TYPECACHE ),
                opts_ ( 0 ), bnum_ ( DEFBNUM ), capcnt_ ( -1 ), capsiz_ ( -1 ),
                opaque_(), embcomp_ ( ZLIBRAWCOMP ), comp_ ( NULL ), slots_(), rttmode_ ( true ), tran_ ( false )
            {
                _assert_ ( true );
            }
            /**
             * Destructor.
             * @note If the database is not closed, it is closed implicitly.
             */
            virtual ~CacheDB()
            {
                _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 The operation for each record 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 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 && ! ( omode_ & OWRITER ) )
                {
                    set_error ( _KCCODELINE_, Error::NOPERM, "permission denied" );
                    return false;
                }
                if ( ksiz > KSIZMAX )
                {
                    ksiz = KSIZMAX;
                }
                uint64_t hash = hash_record ( kbuf, ksiz );
                int32_t sidx = hash % SLOTNUM;
                hash /= SLOTNUM;
                Slot *slot = slots_ + sidx;
                slot->lock.lock();
                accept_impl ( slot, hash, kbuf, ksiz, visitor, comp_, rttmode_ );
                slot->lock.unlock();
                return true;
            }
            /**
             * 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 The operations for specified records are performed atomically and other threads
             * accessing the same records are blocked.  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_, false );
                if ( omode_ == 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "not opened" );
                    return false;
                }
                if ( writable && ! ( omode_ & OWRITER ) )
                {
                    set_error ( _KCCODELINE_, Error::NOPERM, "permission denied" );
                    return false;
                }
                ScopedVisitor svis ( visitor );
                size_t knum = keys.size();
                if ( knum < 1 )
                {
                    return true;
                }
                struct RecordKey
                {
                    const char *kbuf;
                    size_t ksiz;
                    uint64_t hash;
                    int32_t sidx;
                };
                RecordKey *rkeys = new RecordKey[knum];
                std::set<int32_t> sidxs;
                for ( size_t i = 0; i < knum; i++ )
                {
                    const std::string &key = keys[i];
                    RecordKey *rkey = rkeys + i;
                    rkey->kbuf = key.data();
                    rkey->ksiz = key.size();
                    if ( rkey->ksiz > KSIZMAX )
                    {
                        rkey->ksiz = KSIZMAX;
                    }
                    rkey->hash = hash_record ( rkey->kbuf, rkey->ksiz );
                    rkey->sidx = rkey->hash % SLOTNUM;
                    sidxs.insert ( rkey->sidx );
                    rkey->hash /= SLOTNUM;
                }
                std::set<int32_t>::iterator sit = sidxs.begin();
                std::set<int32_t>::iterator sitend = sidxs.end();
                while ( sit != sitend )
                {
                    Slot *slot = slots_ + *sit;
                    slot->lock.lock();
                    ++sit;
                }
                for ( size_t i = 0; i < knum; i++ )
                {
                    RecordKey *rkey = rkeys + i;
                    Slot *slot = slots_ + rkey->sidx;
                    accept_impl ( slot, rkey->hash, rkey->kbuf, rkey->ksiz, visitor, comp_, rttmode_ );
                }
                sit = sidxs.begin();
                sitend = sidxs.end();
                while ( sit != sitend )
                {
                    Slot *slot = slots_ + *sit;
                    slot->lock.unlock();
                    ++sit;
                }
                delete[] rkeys;
                return true;
            }
            /**
             * 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 whole iteration is performed atomically and other threads are blocked.  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 && ! ( omode_ & OWRITER ) )
                {
                    set_error ( _KCCODELINE_, Error::NOPERM, "permission denied" );
                    return false;
                }
                ScopedVisitor svis ( visitor );
                int64_t allcnt = count_impl();
                if ( checker && !checker->check ( "iterate", "beginning", 0, allcnt ) )
                {
                    set_error ( _KCCODELINE_, Error::LOGIC, "checker failed" );
                    return false;
                }
                int64_t curcnt = 0;
                for ( int32_t i = 0; i < SLOTNUM; i++ )
                {
                    Slot *slot = slots_ + i;
                    Record *rec = slot->first;
                    while ( rec )
                    {
                        Record *next = rec->next;
                        uint32_t rksiz = rec->ksiz & KSIZMAX;
                        char *dbuf = ( char * ) rec + sizeof ( *rec );
                        const char *rvbuf = dbuf + rksiz;
                        size_t rvsiz = rec->vsiz;
                        char *zbuf = NULL;
                        size_t zsiz = 0;
                        if ( comp_ )
                        {
                            zbuf = comp_->decompress ( rvbuf, rvsiz, &zsiz );
                            if ( zbuf )
                            {
                                rvbuf = zbuf;
                                rvsiz = zsiz;
                            }
                        }
                        size_t vsiz;
                        const char *vbuf = visitor->visit_full ( dbuf, rksiz, rvbuf, rvsiz, &vsiz );
                        delete[] zbuf;
                        if ( vbuf == Visitor::REMOVE )
                        {
                            uint64_t hash = hash_record ( dbuf, rksiz ) / SLOTNUM;
                            Repeater repeater ( Visitor::REMOVE, 0 );
                            accept_impl ( slot, hash, dbuf, rksiz, &repeater, comp_, false );
                        }
                        else
                            if ( vbuf != Visitor::NOP )
                            {
                                uint64_t hash = hash_record ( dbuf, rksiz ) / SLOTNUM;
                                Repeater repeater ( vbuf, vsiz );
                                accept_impl ( slot, hash, dbuf, rksiz, &repeater, comp_, false );
                            }
                        rec = next;
                        curcnt++;
                        if ( checker && !checker->check ( "iterate", "processing", curcnt, allcnt ) )
                        {
                            set_error ( _KCCODELINE_, Error::LOGIC, "checker failed" );
                            return false;
                        }
                    }
                }
                if ( checker && !checker->check ( "iterate", "ending", -1, allcnt ) )
                {
                    set_error ( _KCCODELINE_, Error::LOGIC, "checker failed" );
                    return false;
                }
                trigger_meta ( MetaTrigger::ITERATE, "iterate" );
                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.  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.  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;
                }
                thnum = std::pow ( 2.0, ( int32_t ) ( std::log ( thnum * std::sqrt ( 2.0 ) ) / std::log ( 2.0 ) ) );
                if ( thnum > ( size_t ) SLOTNUM )
                {
                    thnum = SLOTNUM;
                }
                ScopedVisitor svis ( visitor );
                int64_t allcnt = count_impl();
                if ( checker && !checker->check ( "scan_parallel", "beginning", -1, allcnt ) )
                {
                    set_error ( _KCCODELINE_, Error::LOGIC, "checker failed" );
                    return false;
                }
                class ThreadImpl : public Thread
                {
                    public:
                        explicit ThreadImpl() :
                            db_ ( NULL ), visitor_ ( NULL ), checker_ ( NULL ), allcnt_ ( 0 ), slots_(), error_() {}
                        void init ( CacheDB *db, Visitor *visitor, ProgressChecker *checker, int64_t allcnt )
                        {
                            db_ = db;
                            visitor_ = visitor;
                            checker_ = checker;
                            allcnt_ = allcnt;
                        }
                        void add_slot ( Slot *slot )
                        {
                            slots_.push_back ( slot );
                        }
                        const Error &error()
                        {
                            return error_;
                        }
                    private:
                        void run()
                        {
                            CacheDB *db = db_;
                            Visitor *visitor = visitor_;
                            ProgressChecker *checker = checker_;
                            int64_t allcnt = allcnt_;
                            Compressor *comp = db->comp_;
                            std::vector<Slot *>::iterator sit = slots_.begin();
                            std::vector<Slot *>::iterator sitend = slots_.end();
                            while ( sit != sitend )
                            {
                                Slot *slot = *sit;
                                Record *rec = slot->first;
                                while ( rec )
                                {
                                    Record *next = rec->next;
                                    uint32_t rksiz = rec->ksiz & KSIZMAX;
                                    char *dbuf = ( char * ) rec + sizeof ( *rec );
                                    const char *rvbuf = dbuf + rksiz;
                                    size_t rvsiz = rec->vsiz;
                                    char *zbuf = NULL;
                                    size_t zsiz = 0;
                                    if ( comp )
                                    {
                                        zbuf = comp->decompress ( rvbuf, rvsiz, &zsiz );
                                        if ( zbuf )
                                        {
                                            rvbuf = zbuf;
                                            rvsiz = zsiz;
                                        }
                                    }
                                    size_t vsiz;
                                    visitor->visit_full ( dbuf, rksiz, rvbuf, rvsiz, &vsiz );
                                    delete[] zbuf;
                                    rec = next;
                                    if ( checker && !checker->check ( "scan_parallel", "processing", -1, allcnt ) )
                                    {
                                        db->set_error ( _KCCODELINE_, Error::LOGIC, "checker failed" );
                                        error_ = db->error();
                                        break;
                                    }
                                }
                                ++sit;
                            }
                        }
                        CacheDB *db_;
                        Visitor *visitor_;
                        ProgressChecker *checker_;
                        int64_t allcnt_;
                        std::vector<Slot *> slots_;
                        Error error_;
                };
                bool err = false;
                bool orttmode = rttmode_;
                rttmode_ = false;
                ThreadImpl *threads = new ThreadImpl[thnum];
                for ( int32_t i = 0; i < SLOTNUM; i++ )
                {
                    ThreadImpl *thread = threads + ( i % thnum );
                    thread->add_slot ( slots_ + i );
                }
                for ( size_t i = 0; i < thnum; i++ )
                {
                    ThreadImpl *thread = threads + i;
                    thread->init ( this, visitor, checker, allcnt );
                    thread->start();
                }
                for ( size_t i = 0; i < thnum; i++ )
                {
                    ThreadImpl *thread = threads + i;
                    thread->join();
                    if ( thread->error() != Error::SUCCESS )
                    {
                        *error_ = thread->error();
                        err = true;
                    }
                }
                delete[] threads;
                rttmode_ = orttmode;
                if ( err )
                {
                    return false;
                }
                if ( checker && !checker->check ( "scan_parallel", "ending", -1, allcnt ) )
                {
                    set_error ( _KCCODELINE_, Error::LOGIC, "checker failed" );
                    return false;
                }
                trigger_meta ( MetaTrigger::ITERATE, "scan_parallel" );
                return true;
            }
            /**
             * 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.  CacheDB::OWRITER as a writer, CacheDB::OREADER as a
             * reader.  The following may be added to the writer mode by bitwise-or: CacheDB::OCREATE,
             * which means it creates a new database if the file does not exist, CacheDB::OTRUNCATE, which
             * means it creates a new database regardless if the file exists, CacheDB::OAUTOTRAN, which
             * means each updating operation is performed in implicit transaction, CacheDB::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: CacheDB::ONOLOCK, which means it opens the database file without file locking,
             * CacheDB::OTRYLOCK, which means locking is performed without blocking, CacheDB::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 CacheDB::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() );
                omode_ = mode;
                path_.append ( path );
                size_t bnum = nearbyprime ( bnum_ / SLOTNUM );
                size_t capcnt = capcnt_ > 0 ? capcnt_ / SLOTNUM + 1 : ( 1ULL << ( sizeof ( capcnt ) * 8 - 1 ) );
                size_t capsiz = capsiz_ > 0 ? capsiz_ / SLOTNUM + 1 : ( 1ULL << ( sizeof ( capsiz ) * 8 - 1 ) );
                if ( capsiz > sizeof ( *this ) / SLOTNUM )
                {
                    capsiz -= sizeof ( *this ) / SLOTNUM;
                }
                if ( capsiz > bnum * sizeof ( Record * ) )
                {
                    capsiz -= bnum * sizeof ( Record * );
                }
                for ( int32_t i = 0; i < SLOTNUM; i++ )
                {
                    initialize_slot ( slots_ + i, bnum, capcnt, capsiz );
                }
                comp_ = ( opts_ & TCOMPRESS ) ? embcomp_ : NULL;
                std::memset ( opaque_, 0, sizeof ( opaque_ ) );
                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() );
                tran_ = false;
                for ( int32_t i = SLOTNUM - 1; i >= 0; i-- )
                {
                    destroy_slot ( slots_ + i );
                }
                path_.clear();
                omode_ = 0;
                trigger_meta ( MetaTrigger::CLOSE, "close" );
                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.
             * @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 ( ( omode_ & OWRITER ) && checker &&
                        !checker->check ( "synchronize", "nothing to be synchronized", -1, -1 ) )
                {
                    set_error ( _KCCODELINE_, Error::LOGIC, "checker failed" );
                    return false;
                }
                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_, count_impl(), size_impl() ) )
                    {
                        set_error ( _KCCODELINE_, Error::LOGIC, "postprocessing failed" );
                        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_, count_impl(), size_impl() ) )
                {
                    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 );
                uint32_t wcnt = 0;
                while ( true )
                {
                    mlock_.lock_writer();
                    if ( omode_ == 0 )
                    {
                        set_error ( _KCCODELINE_, Error::INVALID, "not opened" );
                        mlock_.unlock();
                        return false;
                    }
                    if ( ! ( omode_ & OWRITER ) )
                    {
                        set_error ( _KCCODELINE_, Error::NOPERM, "permission denied" );
                        mlock_.unlock();
                        return false;
                    }
                    if ( !tran_ )
                    {
                        break;
                    }
                    mlock_.unlock();
                    if ( wcnt >= LOCKBUSYLOOP )
                    {
                        Thread::chill();
                    }
                    else
                    {
                        Thread::yield();
                        wcnt++;
                    }
                }
                tran_ = true;
                trigger_meta ( MetaTrigger::BEGINTRAN, "begin_transaction" );
                mlock_.unlock();
                return true;
            }
            /**
             * 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 );
                mlock_.lock_writer();
                if ( omode_ == 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "not opened" );
                    mlock_.unlock();
                    return false;
                }
                if ( ! ( omode_ & OWRITER ) )
                {
                    set_error ( _KCCODELINE_, Error::NOPERM, "permission denied" );
                    mlock_.unlock();
                    return false;
                }
                if ( tran_ )
                {
                    set_error ( _KCCODELINE_, Error::LOGIC, "competition avoided" );
                    mlock_.unlock();
                    return false;
                }
                tran_ = true;
                trigger_meta ( MetaTrigger::BEGINTRAN, "begin_transaction_try" );
                mlock_.unlock();
                return true;
            }
            /**
             * 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;
                }
                if ( !tran_ )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "not in transaction" );
                    return false;
                }
                if ( !commit )
                {
                    disable_cursors();
                }
                for ( int32_t i = 0; i < SLOTNUM; i++ )
                {
                    if ( !commit )
                    {
                        apply_slot_trlogs ( slots_ + i );
                    }
                    slots_[i].trlogs.clear();
                    adjust_slot_capacity ( slots_ + i );
                }
                tran_ = false;
                trigger_meta ( commit ? MetaTrigger::COMMITTRAN : MetaTrigger::ABORTTRAN, "end_transaction" );
                return true;
            }
            /**
             * 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;
                }
                disable_cursors();
                for ( int32_t i = 0; i < SLOTNUM; i++ )
                {
                    Slot *slot = slots_ + i;
                    clear_slot ( slot );
                }
                std::memset ( opaque_, 0, sizeof ( opaque_ ) );
                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;
                }
                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 );
                if ( omode_ == 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "not opened" );
                    return -1;
                }
                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 );
                ScopedRWLock lock ( &mlock_, false );
                if ( omode_ == 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "not opened" );
                    return "";
                }
                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 ) TYPECACHE );
                ( *strmap ) ["realtype"] = strprintf ( "%u", ( unsigned ) type_ );
                ( *strmap ) ["path"] = path_;
                ( *strmap ) ["libver"] = strprintf ( "%u", LIBVER );
                ( *strmap ) ["librev"] = strprintf ( "%u", LIBREV );
                ( *strmap ) ["fmtver"] = strprintf ( "%u", FMTVER );
                ( *strmap ) ["chksum"] = strprintf ( "%u", 0xff );
                ( *strmap ) ["opts"] = strprintf ( "%u", opts_ );
                ( *strmap ) ["bnum"] = strprintf ( "%lld", ( long long ) bnum_ );
                ( *strmap ) ["capcnt"] = strprintf ( "%lld", ( long long ) capcnt_ );
                ( *strmap ) ["capsiz"] = strprintf ( "%lld", ( long long ) capsiz_ );
                ( *strmap ) ["recovered"] = strprintf ( "%d", false );
                ( *strmap ) ["reorganized"] = strprintf ( "%d", false );
                if ( strmap->count ( "opaque" ) > 0 )
                {
                    ( *strmap ) ["opaque"] = std::string ( opaque_, sizeof ( opaque_ ) );
                }
                if ( strmap->count ( "bnum_used" ) > 0 )
                {
                    int64_t cnt = 0;
                    for ( int32_t i = 0; i < SLOTNUM; i++ )
                    {
                        Slot *slot = slots_ + i;
                        Record **buckets = slot->buckets;
                        size_t bnum = slot->bnum;
                        for ( size_t j = 0; j < bnum; j++ )
                        {
                            if ( buckets[j] )
                            {
                                cnt++;
                            }
                        }
                    }
                    ( *strmap ) ["bnum_used"] = strprintf ( "%lld", ( long long ) cnt );
                }
                ( *strmap ) ["count"] = strprintf ( "%lld", ( long long ) count_impl() );
                ( *strmap ) ["size"] = strprintf ( "%lld", ( long long ) size_impl() );
                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;
            }
            /**
             * Set the optional features.
             * @param opts the optional features by bitwise-or: DirDB::TCOMPRESS to compress each record.
             * @return true on success, or false on failure.
             */
            bool tune_options ( int8_t opts )
            {
                _assert_ ( true );
                ScopedRWLock lock ( &mlock_, true );
                if ( omode_ != 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "already opened" );
                    return false;
                }
                opts_ = opts;
                return true;
            }
            /**
             * Set the number of buckets of the hash table.
             * @param bnum the number of buckets of the hash table.
             * @return true on success, or false on failure.
             */
            bool tune_buckets ( int64_t bnum )
            {
                _assert_ ( true );
                ScopedRWLock lock ( &mlock_, true );
                if ( omode_ != 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "already opened" );
                    return false;
                }
                bnum_ = bnum >= 0 ? bnum : DEFBNUM;
                return true;
            }
            /**
             * Set the data compressor.
             * @param comp the data compressor object.
             * @return true on success, or false on failure.
             */
            bool tune_compressor ( Compressor *comp )
            {
                _assert_ ( comp );
                ScopedRWLock lock ( &mlock_, true );
                if ( omode_ != 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "already opened" );
                    return false;
                }
                embcomp_ = comp;
                return true;
            }
            /**
             * Set the capacity by record number.
             * @param count the maximum number of records.
             * @return true on success, or false on failure.
             */
            bool cap_count ( int64_t count )
            {
                _assert_ ( true );
                ScopedRWLock lock ( &mlock_, true );
                if ( omode_ != 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "already opened" );
                    return false;
                }
                capcnt_ = count;
                return true;
            }
            /**
             * Set the capacity by memory usage.
             * @param size the maximum size of memory usage.
             * @return true on success, or false on failure.
             */
            bool cap_size ( int64_t size )
            {
                _assert_ ( true );
                ScopedRWLock lock ( &mlock_, true );
                if ( omode_ != 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "already opened" );
                    return false;
                }
                capsiz_ = size;
                return true;
            }
            /**
             * Switch the mode of LRU rotation.
             * @param rttmode true to enable LRU rotation, false to disable LRU rotation.
             * @return true on success, or false on failure.
             * @note This function can be called while the database is opened.
             */
            bool switch_rotation ( bool rttmode )
            {
                _assert_ ( true );
                ScopedRWLock lock ( &mlock_, true );
                rttmode_ = rttmode;
                return true;
            }
            /**
             * Get the opaque data.
             * @return the pointer to the opaque data region, whose size is 16 bytes.
             */
            char *opaque()
            {
                _assert_ ( true );
                ScopedRWLock lock ( &mlock_, false );
                if ( omode_ == 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "not opened" );
                    return NULL;
                }
                return opaque_;
            }
            /**
             * Synchronize the opaque data.
             * @return true on success, or false on failure.
             */
            bool synchronize_opaque()
            {
                _assert_ ( true );
                ScopedRWLock lock ( &mlock_, true );
                if ( omode_ == 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "not opened" );
                    return false;
                }
                if ( ! ( omode_ & OWRITER ) )
                {
                    set_error ( _KCCODELINE_, Error::NOPERM, "permission denied" );
                    return false;
                }
                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 );
                }
            }
            /**
             * Set the database type.
             * @param type the database type.
             * @return true on success, or false on failure.
             */
            bool tune_type ( int8_t type )
            {
                _assert_ ( true );
                ScopedRWLock lock ( &mlock_, true );
                if ( omode_ != 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "already opened" );
                    return false;
                }
                type_ = type;
                return true;
            }
            /**
             * Get the library version.
             * @return the library version, or 0 on failure.
             */
            uint8_t libver()
            {
                _assert_ ( true );
                ScopedRWLock lock ( &mlock_, false );
                if ( omode_ == 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "not opened" );
                    return 0;
                }
                return LIBVER;
            }
            /**
             * Get the library revision.
             * @return the library revision, or 0 on failure.
             */
            uint8_t librev()
            {
                _assert_ ( true );
                ScopedRWLock lock ( &mlock_, false );
                if ( omode_ == 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "not opened" );
                    return 0;
                }
                return LIBREV;
            }
            /**
             * Get the format version.
             * @return the format version, or 0 on failure.
             */
            uint8_t fmtver()
            {
                _assert_ ( true );
                ScopedRWLock lock ( &mlock_, false );
                if ( omode_ == 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "not opened" );
                    return 0;
                }
                return FMTVER;
            }
            /**
             * Get the module checksum.
             * @return the module checksum, or 0 on failure.
             */
            uint8_t chksum()
            {
                _assert_ ( true );
                ScopedRWLock lock ( &mlock_, false );
                if ( omode_ == 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "not opened" );
                    return 0;
                }
                return 0xff;
            }
            /**
             * Get the database type.
             * @return the database type, or 0 on failure.
             */
            uint8_t type()
            {
                _assert_ ( true );
                ScopedRWLock lock ( &mlock_, false );
                if ( omode_ == 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "not opened" );
                    return 0;
                }
                return type_;
            }
            /**
             * Get the options.
             * @return the options, or 0 on failure.
             */
            uint8_t opts()
            {
                _assert_ ( true );
                ScopedRWLock lock ( &mlock_, false );
                if ( omode_ == 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "not opened" );
                    return 0;
                }
                return opts_;
            }
            /**
             * Get the data compressor.
             * @return the data compressor, or NULL on failure.
             */
            Compressor *comp()
            {
                _assert_ ( true );
                ScopedRWLock lock ( &mlock_, false );
                if ( omode_ == 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "not opened" );
                    return NULL;
                }
                return comp_;
            }
            /**
             * Check whether the database was recovered or not.
             * @return true if recovered, or false if not.
             */
            bool recovered()
            {
                _assert_ ( true );
                ScopedRWLock lock ( &mlock_, false );
                if ( omode_ == 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "not opened" );
                    return false;
                }
                return false;
            }
            /**
             * Check whether the database was reorganized or not.
             * @return true if reorganized, or false if not.
             */
            bool reorganized()
            {
                _assert_ ( true );
                ScopedRWLock lock ( &mlock_, false );
                if ( omode_ == 0 )
                {
                    set_error ( _KCCODELINE_, Error::INVALID, "not opened" );
                    return false;
                }
                return false;
            }
        private:
            /**
             * Set the power of the alignment of record size.
             * @note This is a dummy implementation for compatibility.
             */
            bool tune_alignment ( int8_t apow )
            {
                return true;
            }
            /**
             * Set the power of the capacity of the free block pool.
             * @note This is a dummy implementation for compatibility.
             */
            bool tune_fbp ( int8_t fpow )
            {
                return true;
            }
            /**
             * Set the size of the internal memory-mapped region.
             * @note This is a dummy implementation for compatibility.
             */
            bool tune_map ( int64_t msiz )
            {
                return true;
            }
            /**
             * Set the unit step number of auto defragmentation.
             * @note This is a dummy implementation for compatibility.
             */
            bool tune_defrag ( int64_t dfunit )
            {
                return true;
            }
            /**
             * Perform defragmentation of the file.
             * @note This is a dummy implementation for compatibility.
             */
            bool defrag ( int64_t step = 0 )
            {
                return true;
            }
            /**
             * Get the status flags.
             * @note This is a dummy implementation for compatibility.
             */
            uint8_t flags()
            {
                return 0;
            }
            /**
             * Get the alignment power.
             * @note This is a dummy implementation for compatibility.
             */
            uint8_t apow()
            {
                return 0;
            }
            /**
             * Get the free block pool power.
             * @note This is a dummy implementation for compatibility.
             */
            uint8_t fpow()
            {
                return 0;
            }
            /**
             * Get the bucket number.
             * @note This is a dummy implementation for compatibility.
             */
            int64_t bnum()
            {
                return 1;
            }
            /**
             * Get the size of the internal memory-mapped region.
             * @note This is a dummy implementation for compatibility.
             */
            int64_t msiz()
            {
                return 0;
            }
            /**
             * Get the unit step number of auto defragmentation.
             * @note This is a dummy implementation for compatibility.
             */
            int64_t dfunit()
            {
                return 0;
            }
        private:
            /**
             * Record data.
             */
            struct Record
            {
                uint32_t ksiz;                       ///< size of the key
                uint32_t vsiz;                       ///< size of the value
                Record *left;                        ///< left child record
                Record *right;                       ///< right child record
                Record *prev;                        ///< privious record
                Record *next;                        ///< next record
            };
            /**
             * Transaction log.
             */
            struct TranLog
            {
                bool full;                           ///< flag whether full
                std::string key;                     ///< old key
                std::string value;                   ///< old value
                /** constructor for a full record */
                explicit TranLog ( const char *kbuf, size_t ksiz, const char *vbuf, size_t vsiz ) :
                    full ( true ), key ( kbuf, ksiz ), value ( vbuf, vsiz )
                {
                    _assert_ ( true );
                }
                /** constructor for an empty record */
                explicit TranLog ( const char *kbuf, size_t ksiz ) : full ( false ), key ( kbuf, ksiz )
                {
                    _assert_ ( true );
                }
            };
            /**
             * Slot table.
             */
            struct Slot
            {
                Mutex lock;                       ///< lock
                Record **buckets;                    ///< bucket array
                size_t bnum;                         ///< number of buckets
                size_t capcnt;                       ///< cap of record number
                size_t capsiz;                       ///< cap of memory usage
                Record *first;                       ///< first record
                Record *last;                        ///< last record
                size_t count;                        ///< number of records
                size_t size;                         ///< total size of records
                TranLogList trlogs;                  ///< transaction logs
                size_t trsize;                       ///< size before transaction
            };
            /**
             * Repeating visitor.
             */
            class Repeater : public Visitor
            {
                public:
                    /** constructor */
                    explicit Repeater ( const char *vbuf, size_t vsiz ) : vbuf_ ( vbuf ), vsiz_ ( vsiz ) {}
                private:
                    /** process a full record */
                    const char *visit_full ( const char *kbuf, size_t ksiz,
                                             const char *vbuf, size_t vsiz, size_t *sp )
                    {
                        _assert_ ( kbuf && ksiz <= MEMMAXSIZ && vbuf && vsiz <= MEMMAXSIZ && sp );
                        *sp = vsiz_;
                        return vbuf_;
                    }
                    const char *vbuf_;                   ///< region of the value
                    size_t vsiz_;                        ///< size of the value
            };
            /**
             * Setting visitor.
             */
            class Setter : public Visitor
            {
                public:
                    /** constructor */
                    explicit Setter ( const char *vbuf, size_t vsiz ) : vbuf_ ( vbuf ), vsiz_ ( vsiz ) {}
                private:
                    /** process a full record */
                    const char *visit_full ( const char *kbuf, size_t ksiz,
                                             const char *vbuf, size_t vsiz, size_t *sp )
                    {
                        _assert_ ( kbuf && ksiz <= MEMMAXSIZ && vbuf && vsiz <= MEMMAXSIZ && sp );
                        *sp = vsiz_;
                        return vbuf_;
                    }
                    /** process an empty record */
                    const char *visit_empty ( const char *kbuf, size_t ksiz, size_t *sp )
                    {
                        _assert_ ( kbuf && ksiz <= MEMMAXSIZ && sp );
                        *sp = vsiz_;
                        return vbuf_;
                    }
                    const char *vbuf_;                   ///< region of the value
                    size_t vsiz_;                        ///< size of the value
            };
            /**
             * Removing visitor.
             */
            class Remover : public Visitor
            {
                private:
                    /** visit a record */
                    const char *visit_full ( const char *kbuf, size_t ksiz,
                                             const char *vbuf, size_t vsiz, size_t *sp )
                    {
                        _assert_ ( kbuf && ksiz <= MEMMAXSIZ && vbuf && vsiz <= MEMMAXSIZ && sp );
                        return REMOVE;
                    }
            };
            /**
             * 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 slot the slot of the record.
             * @param hash the hash value of the key.
             * @param kbuf the pointer to the key region.
             * @param ksiz the size of the key region.
             * @param visitor a visitor object.
             * @param comp the data compressor.
             * @param rtt whether to move the record to the last.
             */
            void accept_impl ( Slot *slot, uint64_t hash, const char *kbuf, size_t ksiz, Visitor *visitor,
                               Compressor *comp, bool rtt )
            {
                _assert_ ( slot && kbuf && ksiz <= MEMMAXSIZ && visitor );
                size_t bidx = hash % slot->bnum;
                Record *rec = slot->buckets[bidx];
                Record **entp = slot->buckets + bidx;
                uint32_t fhash = fold_hash ( hash ) & ~KSIZMAX;
                while ( rec )
                {
                    uint32_t rhash = rec->ksiz & ~KSIZMAX;
                    uint32_t rksiz = rec->ksiz & KSIZMAX;
                    if ( fhash > rhash )
                    {
                        entp = &rec->left;
                        rec = rec->left;
                    }
                    else
                        if ( fhash < rhash )
                        {
                            entp = &rec->right;
                            rec = rec->right;
                        }
                        else
                        {
                            char *dbuf = ( char * ) rec + sizeof ( *rec );
                            int32_t kcmp = compare_keys ( kbuf, ksiz, dbuf, rksiz );
                            if ( kcmp < 0 )
                            {
                                entp = &rec->left;
                                rec = rec->left;
                            }
                            else
                                if ( kcmp > 0 )
                                {
                                    entp = &rec->right;
                                    rec = rec->right;
                                }
                                else
                                {
                                    const char *rvbuf = dbuf + rksiz;
                                    size_t rvsiz = rec->vsiz;
                                    char *zbuf = NULL;
                                    size_t zsiz = 0;
                                    if ( comp )
                                    {
                                        zbuf = comp->decompress ( rvbuf, rvsiz, &zsiz );
                                        if ( zbuf )
                                        {
                                            rvbuf = zbuf;
                                            rvsiz = zsiz;
                                        }
                                    }
                                    size_t vsiz;
                                    const char *vbuf = visitor->visit_full ( dbuf, rksiz, rvbuf, rvsiz, &vsiz );
                                    delete[] zbuf;
                                    if ( vbuf == Visitor::REMOVE )
                                    {
                                        if ( tran_ )
                                        {
                                            TranLog log ( kbuf, ksiz, dbuf + rksiz, rec->vsiz );
                                            slot->trlogs.push_back ( log );
                                        }
                                        if ( !curs_.empty() )
                                        {
                                            escape_cursors ( rec );
                                        }
                                        if ( rec == slot->first )
                                        {
                                            slot->first = rec->next;
                                        }
                                        if ( rec == slot->last )
                                        {
                                            slot->last = rec->prev;
                                        }
                                        if ( rec->prev )
                                        {
                                            rec->prev->next = rec->next;
                                        }
                                        if ( rec->next )
                                        {
                                            rec->next->prev = rec->prev;
                                        }
                                        if ( rec->left && !rec->right )
                                        {
                                            *entp = rec->left;
                                        }
                                        else
                                            if ( !rec->left && rec->right )
                                            {
                                                *entp = rec->right;
                                            }
                                            else
                                                if ( !rec->left )
                                                {
                                                    *entp = NULL;
                                                }
                                                else
                                                {
                                                    Record *pivot = rec->left;
                                                    if ( pivot->right )
                                                    {
                                                        Record **pentp = &pivot->right;
                                                        pivot = pivot->right;
                                                        while ( pivot->right )
                                                        {
                                                            pentp = &pivot->right;
                                                            pivot = pivot->right;
                                                        }
                                                        *entp = pivot;
                                                        *pentp = pivot->left;
                                                        pivot->left = rec->left;
                                                        pivot->right = rec->right;
                                                    }
                                                    else
                                                    {
                                                        *entp = pivot;
                                                        pivot->right = rec->right;
                                                    }
                                                }
                                        slot->count--;
                                        slot->size -= sizeof ( Record ) + rksiz + rec->vsiz;
                                        xfree ( rec );
                                    }
                                    else
                                    {
                                        bool adj = false;
                                        if ( vbuf != Visitor::NOP )
                                        {
                                            char *zbuf = NULL;
                                            size_t zsiz = 0;
                                            if ( comp )
                                            {
                                                zbuf = comp->compress ( vbuf, vsiz, &zsiz );
                                                if ( zbuf )
                                                {
                                                    vbuf = zbuf;
                                                    vsiz = zsiz;
                                                }
                                            }
                                            if ( tran_ )
                                            {
                                                TranLog log ( kbuf, ksiz, dbuf + rksiz, rec->vsiz );
                                                slot->trlogs.push_back ( log );
                                            }
                                            else
                                            {
                                                adj = vsiz > rec->vsiz;
                                            }
                                            slot->size -= rec->vsiz;
                                            slot->size += vsiz;
                                            if ( vsiz > rec->vsiz )
                                            {
                                                Record *old = rec;
                                                rec = ( Record * ) xrealloc ( rec, sizeof ( *rec ) + ksiz + vsiz );
                                                if ( rec != old )
                                                {
                                                    if ( !curs_.empty() )
                                                    {
                                                        adjust_cursors ( old, rec );
                                                    }
                                                    if ( slot->first == old )
                                                    {
                                                        slot->first = rec;
                                                    }
                                                    if ( slot->last == old )
                                                    {
                                                        slot->last = rec;
                                                    }
                                                    *entp = rec;
                                                    if ( rec->prev )
                                                    {
                                                        rec->prev->next = rec;
                                                    }
                                                    if ( rec->next )
                                                    {
                                                        rec->next->prev = rec;
                                                    }
                                                    dbuf = ( char * ) rec + sizeof ( *rec );
                                                }
                                            }
                                            std::memcpy ( dbuf + ksiz, vbuf, vsiz );
                                            rec->vsiz = vsiz;
                                            delete[] zbuf;
                                        }
                                        if ( rtt && slot->last != rec )
                                        {
                                            if ( !curs_.empty() )
                                            {
                                                escape_cursors ( rec );
                                            }
                                            if ( slot->first == rec )
                                            {
                                                slot->first = rec->next;
                                            }
                                            if ( rec->prev )
                                            {
                                                rec->prev->next = rec->next;
                                            }
                                            if ( rec->next )
                                            {
                                                rec->next->prev = rec->prev;
                                            }
                                            rec->prev = slot->last;
                                            rec->next = NULL;
                                            slot->last->next = rec;
                                            slot->last = rec;
                                        }
                                        if ( adj )
                                        {
                                            adjust_slot_capacity ( slot );
                                        }
                                    }
                                    return;
                                }
                        }
                }
                size_t vsiz;
                const char *vbuf = visitor->visit_empty ( kbuf, ksiz, &vsiz );
                if ( vbuf != Visitor::NOP && vbuf != Visitor::REMOVE )
                {
                    char *zbuf = NULL;
                    size_t zsiz = 0;
                    if ( comp )
                    {
                        zbuf = comp->compress ( vbuf, vsiz, &zsiz );
                        if ( zbuf )
                        {
                            vbuf = zbuf;
                            vsiz = zsiz;
                        }
                    }
                    if ( tran_ )
                    {
                        TranLog log ( kbuf, ksiz );
                        slot->trlogs.push_back ( log );
                    }
                    slot->size += sizeof ( Record ) + ksiz + vsiz;
                    rec = ( Record * ) xmalloc ( sizeof ( *rec ) + ksiz + vsiz );
                    char *dbuf = ( char * ) rec + sizeof ( *rec );
                    std::memcpy ( dbuf, kbuf, ksiz );
                    rec->ksiz = ksiz | fhash;
                    std::memcpy ( dbuf + ksiz, vbuf, vsiz );
                    rec->vsiz = vsiz;
                    rec->left = NULL;
                    rec->right = NULL;
                    rec->prev = slot->last;
                    rec->next = NULL;
                    *entp = rec;
                    if ( !slot->first )
                    {
                        slot->first = rec;
                    }
                    if ( slot->last )
                    {
                        slot->last->next = rec;
                    }
                    slot->last = rec;
                    slot->count++;
                    if ( !tran_ )
                    {
                        adjust_slot_capacity ( slot );
                    }
                    delete[] zbuf;
                }
            }
            /**
             * Get the number of records.
             * @return the number of records, or -1 on failure.
             */
            int64_t count_impl()
            {
                _assert_ ( true );
                int64_t sum = 0;
                for ( int32_t i = 0; i < SLOTNUM; i++ )
                {
                    Slot *slot = slots_ + i;
                    ScopedMutex lock ( &slot->lock );
                    sum += slot->count;
                }
                return sum;
            }
            /**
             * Get the size of the database file.
             * @return the size of the database file in bytes.
             */
            int64_t size_impl()
            {
                _assert_ ( true );
                int64_t sum = sizeof ( *this );
                for ( int32_t i = 0; i < SLOTNUM; i++ )
                {
                    Slot *slot = slots_ + i;
                    ScopedMutex lock ( &slot->lock );
                    sum += slot->bnum * sizeof ( Record * );
                    sum += slot->size;
                }
                return sum;
            }
            /**
             * Initialize a slot table.
             * @param slot the slot table.
             * @param bnum the number of buckets.
             * @param capcnt the capacity of record number.
             * @param capsiz the capacity of memory usage.
             */
            void initialize_slot ( Slot *slot, size_t bnum, size_t capcnt, size_t capsiz )
            {
                _assert_ ( slot );
                Record **buckets;
                if ( bnum >= ZMAPBNUM )
                {
                    buckets = ( Record ** ) mapalloc ( sizeof ( *buckets ) * bnum );
                }
                else
                {
                    buckets = new Record*[bnum];
                    for ( size_t i = 0; i < bnum; i++ )
                    {
                        buckets[i] = NULL;
                    }
                }
                slot->buckets = buckets;
                slot->bnum = bnum;
                slot->capcnt = capcnt;
                slot->capsiz = capsiz;
                slot->first = NULL;
                slot->last = NULL;
                slot->count = 0;
                slot->size = 0;
            }
            /**
             * Destroy a slot table.
             * @param slot the slot table.
             */
            void destroy_slot ( Slot *slot )
            {
                _assert_ ( slot );
                slot->trlogs.clear();
                Record *rec = slot->last;
                while ( rec )
                {
                    Record *prev = rec->prev;
                    xfree ( rec );
                    rec = prev;
                }
                if ( slot->bnum >= ZMAPBNUM )
                {
                    mapfree ( slot->buckets );
                }
                else
                {
                    delete[] slot->buckets;
                }
            }
            /**
             * Clear a slot table.
             * @param slot the slot table.
             */
            void clear_slot ( Slot *slot )
            {
                _assert_ ( slot );
                Record *rec = slot->last;
                while ( rec )
                {
                    if ( tran_ )
                    {
                        uint32_t rksiz = rec->ksiz & KSIZMAX;
                        char *dbuf = ( char * ) rec + sizeof ( *rec );
                        TranLog log ( dbuf, rksiz, dbuf + rksiz, rec->vsiz );
                        slot->trlogs.push_back ( log );
                    }
                    Record *prev = rec->prev;
                    xfree ( rec );
                    rec = prev;
                }
                Record **buckets = slot->buckets;
                size_t bnum = slot->bnum;
                for ( size_t i = 0; i < bnum; i++ )
                {
                    buckets[i] = NULL;
                }
                slot->first = NULL;
                slot->last = NULL;
                slot->count = 0;
                slot->size = 0;
            }
            /**
             * Apply transaction logs of a slot table.
             * @param slot the slot table.
             */
            void apply_slot_trlogs ( Slot *slot )
            {
                _assert_ ( slot );
                const TranLogList &logs = slot->trlogs;
                TranLogList::const_iterator it = logs.end();
                TranLogList::const_iterator itbeg = logs.begin();
                while ( it != itbeg )
                {
                    --it;
                    const char *kbuf = it->key.c_str();
                    size_t ksiz = it->key.size();
                    const char *vbuf = it->value.c_str();
                    size_t vsiz = it->value.size();
                    uint64_t hash = hash_record ( kbuf, ksiz ) / SLOTNUM;
                    if ( it->full )
                    {
                        Setter setter ( vbuf, vsiz );
                        accept_impl ( slot, hash, kbuf, ksiz, &setter, NULL, false );
                    }
                    else
                    {
                        Remover remover;
                        accept_impl ( slot, hash, kbuf, ksiz, &remover, NULL, false );
                    }
                }
            }
            /**
             * Addjust a slot table to the capacity.
             * @param slot the slot table.
             */
            void adjust_slot_capacity ( Slot *slot )
            {
                _assert_ ( slot );
                if ( ( slot->count > slot->capcnt || slot->size > slot->capsiz ) && slot->first )
                {
                    Record *rec = slot->first;
                    uint32_t rksiz = rec->ksiz & KSIZMAX;
                    char *dbuf = ( char * ) rec + sizeof ( *rec );
                    char stack[RECBUFSIZ];
                    char *kbuf = rksiz > sizeof ( stack ) ? new char[rksiz] : stack;
                    std::memcpy ( kbuf, dbuf, rksiz );
                    uint64_t hash = hash_record ( kbuf, rksiz ) / SLOTNUM;
                    Remover remover;
                    accept_impl ( slot, hash, dbuf, rksiz, &remover, NULL, false );
                    if ( kbuf != stack )
                    {
                        delete[] kbuf;
                    }
                }
            }
            /**
             * Get the hash value of a record.
             * @param kbuf the pointer to the key region.
             * @param ksiz the size of the key region.
             * @return the hash value.
             */
            uint64_t hash_record ( const char *kbuf, size_t ksiz )
            {
                _assert_ ( kbuf && ksiz <= MEMMAXSIZ );
                return hashmurmur ( kbuf, ksiz );
            }
            /**
             * Fold a hash value into a small number.
             * @param hash the hash number.
             * @return the result number.
             */
            uint32_t fold_hash ( uint64_t hash )
            {
                _assert_ ( true );
                return ( ( hash & 0xffffffff00000000ULL ) >> 32 ) ^ ( ( hash & 0x0000ffffffff0000ULL ) >> 16 ) ^
                       ( ( hash & 0x000000000000ffffULL ) << 16 ) ^ ( ( hash & 0x00000000ffff0000ULL ) >> 0 );
            }
            /**
             * Compare two keys in lexical order.
             * @param abuf one key.
             * @param asiz the size of the one key.
             * @param bbuf the other key.
             * @param bsiz the size of the other key.
             * @return positive if the former is big, or negative if the latter is big, or 0 if both are
             * equivalent.
             */
            int32_t compare_keys ( const char *abuf, size_t asiz, const char *bbuf, size_t bsiz )
            {
                _assert_ ( abuf && asiz <= MEMMAXSIZ && bbuf && bsiz <= MEMMAXSIZ );
                if ( asiz != bsiz )
                {
                    return ( int32_t ) asiz - ( int32_t ) bsiz;
                }
                return std::memcmp ( abuf, bbuf, asiz );
            }
            /**
             * Escape cursors on a shifted or removed records.
             * @param rec the record.
             */
            void escape_cursors ( Record *rec )
            {
                _assert_ ( rec );
                ScopedMutex lock ( &flock_ );
                if ( curs_.empty() )
                {
                    return;
                }
                CursorList::const_iterator cit = curs_.begin();
                CursorList::const_iterator citend = curs_.end();
                while ( cit != citend )
                {
                    Cursor *cur = *cit;
                    if ( cur->rec_ == rec )
                    {
                        cur->step_impl();
                    }
                    ++cit;
                }
            }
            /**
             * Adjust cursors on re-allocated records.
             * @param orec the old address.
             * @param nrec the new address.
             */
            void adjust_cursors ( Record *orec, Record *nrec )
            {
                _assert_ ( orec && nrec );
                ScopedMutex lock ( &flock_ );
                if ( curs_.empty() )
                {
                    return;
                }
                CursorList::const_iterator cit = curs_.begin();
                CursorList::const_iterator citend = curs_.end();
                while ( cit != citend )
                {
                    Cursor *cur = *cit;
                    if ( cur->rec_ == orec )
                    {
                        cur->rec_ = nrec;
                    }
                    ++cit;
                }
            }
            /**
             * Disable all cursors.
             */
            void disable_cursors()
            {
                _assert_ ( true );
                ScopedMutex lock ( &flock_ );
                CursorList::const_iterator cit = curs_.begin();
                CursorList::const_iterator citend = curs_.end();
                while ( cit != citend )
                {
                    Cursor *cur = *cit;
                    cur->sidx_ = -1;
                    cur->rec_ = NULL;
                    ++cit;
                }
            }
            /** Dummy constructor to forbid the use. */
            CacheDB ( const CacheDB & );
            /** Dummy Operator to forbid the use. */
            CacheDB &operator = ( const CacheDB & );
            /** The method lock. */
            RWLock mlock_;
            /** The file lock. */
            Mutex flock_;
            /** 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 cursor objects. */
            CursorList curs_;
            /** The path of the database file. */
            std::string path_;
            /** The database type. */
            uint8_t type_;
            /** The options. */
            uint8_t opts_;
            /** The bucket number. */
            int64_t bnum_;
            /** The capacity of record number. */
            int64_t capcnt_;
            /** The capacity of memory usage. */
            int64_t capsiz_;
            /** The opaque data. */
            char opaque_[OPAQUESIZ];
            /** The embedded data compressor. */
            Compressor *embcomp_;
            /** The data compressor. */
            Compressor *comp_;
            /** The slot tables. */
            Slot slots_[SLOTNUM];
            /** The flag whether in LRU rotation. */
            bool rttmode_;
            /** The flag whether in transaction. */
            bool tran_;
    };


    /** An alias of the cache tree database. */
    typedef PlantDB<CacheDB, BasicDB::TYPEGRASS> GrassDB;


}                                        // common namespace

#endif                                   // duplication check

// END OF FILE