include/stx-cbtreedb.h

Go to the documentation of this file.

// -*- mode: c++; fill-column: 79 -*-
// $Id: stx-cbtreedb.h 7 2010-04-14 11:24:20Z tb $

/*
 * STX Constant B-Tree Database Template Classes v0.7.0
 * Copyright (C) 2010 Timo Bingmann
 *
 * This library is free software; you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as published by the
 * Free Software Foundation; either version 2.1 of the License, or (at your
 * option) any later version.
 *
 * This library 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 Lesser General Public License
 * for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this library; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 */

#ifndef _STX_CBTREEDB_H_
#define _STX_CBTREEDB_H_

#include <string.h>
#include <stdint.h>

#include <stdexcept>
#include <vector>
#include <map>
#include <iostream>

namespace stx {

// *** Compile-time assertion macros borrowed from Boost

#ifndef STX_STATIC_ASSERT

template <bool> struct STATIC_ASSERTION_FAILURE;
template <> struct STATIC_ASSERTION_FAILURE<true> { enum { value = 1 }; };
template <int x> struct static_assert_test {};

#define STX_MACRO_JOIN(X,Y) STX_MACRO_DO_JOIN(X,Y)
#define STX_MACRO_DO_JOIN(X,Y) STX_MACRO_DO_JOIN2(X,Y)
#define STX_MACRO_DO_JOIN2(X,Y) X##Y

#define STX_STATIC_ASSERT(A) \
    typedef static_assert_test<sizeof(STATIC_ASSERTION_FAILURE< static_cast<bool>(A) >)> \
        STX_MACRO_JOIN(static_assert_typedef_, __LINE__)

#endif

template < typename _Key = uint32_t,
           typename _Compare = std::less<_Key>,
           unsigned int _BTreePageSize = 1024,
           uint32_t _AppVersionId = 0>
class CBTreeDB
{
public:
    // *** Template Parameters

    typedef _Key        key_type;

    typedef _Compare    key_compare;

    static const unsigned int BTreePageSize = _BTreePageSize;

    static const uint32_t AppVersionId = _AppVersionId;

    // *** Structure parameters calculated from page size

    static const unsigned int LeafNodeHead = 2 * sizeof(uint16_t) + sizeof(uint64_t) + sizeof(uint32_t);

    static const unsigned int LeafNodeNumKeys = (BTreePageSize - LeafNodeHead) / (sizeof(key_type) + sizeof(uint32_t));

    STX_STATIC_ASSERT( LeafNodeNumKeys >= 1 );

    static const unsigned int LeafNodeFiller = BTreePageSize - LeafNodeHead - LeafNodeNumKeys * (sizeof(key_type) + sizeof(uint32_t));

    static const unsigned int InnerNodeHead = 2 * sizeof(uint16_t) + sizeof(uint32_t);

    static const unsigned int InnerNodeNumKeys = (BTreePageSize - InnerNodeHead) / sizeof(key_type);

    STX_STATIC_ASSERT( InnerNodeNumKeys >= 2 );

    static const unsigned int InnerNodeFiller = BTreePageSize - InnerNodeHead - InnerNodeNumKeys * sizeof(key_type);

public:
    class Exception : public std::runtime_error
    {
    public:
        explicit Exception(const std::string& str)
            : std::runtime_error("CBTreeDB: " + str)
        {
        }
    };

#define CBTREEDB_ASSERT(x) do { if (!(x)) throw(Exception("Assertion failed: " #x)); } while(0)

#define CBTREEDB_CHECK(x,msg) do { if (!(x)) throw(Exception(msg)); } while(0)

public:
    struct SignaturePage
    {
        char            signature[8];   
        uint32_t        header_crc32;   
        uint32_t        version;        
        uint32_t        app_version_id; 

        uint32_t        items;          
        uint32_t        key_size;       

        uint64_t        btree_offset;   
        uint64_t        btree_size;     
        uint64_t        btree_firstleaf; 
        uint32_t        btree_pagesize; 
        uint32_t        btree_levels;   
        uint32_t        btree_leaves;   
        uint8_t         btree_sha256[32]; 

        uint64_t        value_offset;   
        uint64_t        value_size;     
        uint8_t         value_sha256[32]; 
    }
        __attribute__((packed));

    static const unsigned int SignaturePageSize = BTreePageSize;

    STX_STATIC_ASSERT( sizeof(SignaturePage) <= BTreePageSize );

protected:
    struct LeafNode
    {
        uint16_t        level;

        uint16_t        slots;

        uint64_t        baseoffset;

        union
        {
            struct
            {
                key_type        keys[LeafNodeNumKeys];

                uint32_t        offsets[LeafNodeNumKeys+1];
            }
                __attribute__((packed));

            char        filler[BTreePageSize - LeafNodeHead + sizeof(uint32_t)];
        };

        inline explicit LeafNode()
            : level(0), slots(0), baseoffset(0)
        {
            memset(filler, 0, sizeof(filler));
        }

        inline bool IsFull() const
        {
            return (slots >= LeafNodeNumKeys);
        }

        inline const key_type& LastKey() const
        {
            CBTREEDB_ASSERT(slots > 0 && slots < LeafNodeNumKeys+1);
            return keys[slots-1];
        }
    }
        __attribute__((packed));

    STX_STATIC_ASSERT( sizeof(LeafNode) == BTreePageSize );

    struct InnerNode
    {
        uint16_t        level;

        uint16_t        slots;

        uint32_t        childrenoffset;

        union
        {
            key_type    keys[InnerNodeNumKeys];

            char        filler[BTreePageSize - InnerNodeHead];
        };

        inline explicit InnerNode(uint16_t level_)
            : level(level_), slots(0), childrenoffset(0)
        {
            memset(filler, 0, sizeof(filler));
        }

        inline bool IsFull() const
        {
            return (slots >= InnerNodeNumKeys);
        }

        inline const key_type& LastKey() const
        {
            CBTREEDB_ASSERT(slots > 0 && slots < InnerNodeNumKeys+1);
            return keys[slots-1];
        }
    }
        __attribute__((packed));

    STX_STATIC_ASSERT( sizeof(InnerNode) == BTreePageSize );

protected:
    class CRC32
    {
    private:
        uint32_t        m_crc;

    public:
        CRC32()
        {
            clear();
        }

        void clear()
        {
            m_crc = 0xFFFFFFFF;
        }

        CRC32& update(const unsigned char* input, uint32_t length)
        {
            static const uint32_t table[256] = {
                0x00000000, 0x77073096, 0xEE0E612C, 0x990951BA, 0x076DC419,
                0x706AF48F, 0xE963A535, 0x9E6495A3, 0x0EDB8832, 0x79DCB8A4,
                0xE0D5E91E, 0x97D2D988, 0x09B64C2B, 0x7EB17CBD, 0xE7B82D07,
                0x90BF1D91, 0x1DB71064, 0x6AB020F2, 0xF3B97148, 0x84BE41DE,
                0x1ADAD47D, 0x6DDDE4EB, 0xF4D4B551, 0x83D385C7, 0x136C9856,
                0x646BA8C0, 0xFD62F97A, 0x8A65C9EC, 0x14015C4F, 0x63066CD9,
                0xFA0F3D63, 0x8D080DF5, 0x3B6E20C8, 0x4C69105E, 0xD56041E4,
                0xA2677172, 0x3C03E4D1, 0x4B04D447, 0xD20D85FD, 0xA50AB56B,
                0x35B5A8FA, 0x42B2986C, 0xDBBBC9D6, 0xACBCF940, 0x32D86CE3,
                0x45DF5C75, 0xDCD60DCF, 0xABD13D59, 0x26D930AC, 0x51DE003A,
                0xC8D75180, 0xBFD06116, 0x21B4F4B5, 0x56B3C423, 0xCFBA9599,
                0xB8BDA50F, 0x2802B89E, 0x5F058808, 0xC60CD9B2, 0xB10BE924,
                0x2F6F7C87, 0x58684C11, 0xC1611DAB, 0xB6662D3D, 0x76DC4190,
                0x01DB7106, 0x98D220BC, 0xEFD5102A, 0x71B18589, 0x06B6B51F,
                0x9FBFE4A5, 0xE8B8D433, 0x7807C9A2, 0x0F00F934, 0x9609A88E,
                0xE10E9818, 0x7F6A0DBB, 0x086D3D2D, 0x91646C97, 0xE6635C01,
                0x6B6B51F4, 0x1C6C6162, 0x856530D8, 0xF262004E, 0x6C0695ED,
                0x1B01A57B, 0x8208F4C1, 0xF50FC457, 0x65B0D9C6, 0x12B7E950,
                0x8BBEB8EA, 0xFCB9887C, 0x62DD1DDF, 0x15DA2D49, 0x8CD37CF3,
                0xFBD44C65, 0x4DB26158, 0x3AB551CE, 0xA3BC0074, 0xD4BB30E2,
                0x4ADFA541, 0x3DD895D7, 0xA4D1C46D, 0xD3D6F4FB, 0x4369E96A,
                0x346ED9FC, 0xAD678846, 0xDA60B8D0, 0x44042D73, 0x33031DE5,
                0xAA0A4C5F, 0xDD0D7CC9, 0x5005713C, 0x270241AA, 0xBE0B1010,
                0xC90C2086, 0x5768B525, 0x206F85B3, 0xB966D409, 0xCE61E49F,
                0x5EDEF90E, 0x29D9C998, 0xB0D09822, 0xC7D7A8B4, 0x59B33D17,
                0x2EB40D81, 0xB7BD5C3B, 0xC0BA6CAD, 0xEDB88320, 0x9ABFB3B6,
                0x03B6E20C, 0x74B1D29A, 0xEAD54739, 0x9DD277AF, 0x04DB2615,
                0x73DC1683, 0xE3630B12, 0x94643B84, 0x0D6D6A3E, 0x7A6A5AA8,
                0xE40ECF0B, 0x9309FF9D, 0x0A00AE27, 0x7D079EB1, 0xF00F9344,
                0x8708A3D2, 0x1E01F268, 0x6906C2FE, 0xF762575D, 0x806567CB,
                0x196C3671, 0x6E6B06E7, 0xFED41B76, 0x89D32BE0, 0x10DA7A5A,
                0x67DD4ACC, 0xF9B9DF6F, 0x8EBEEFF9, 0x17B7BE43, 0x60B08ED5,
                0xD6D6A3E8, 0xA1D1937E, 0x38D8C2C4, 0x4FDFF252, 0xD1BB67F1,
                0xA6BC5767, 0x3FB506DD, 0x48B2364B, 0xD80D2BDA, 0xAF0A1B4C,
                0x36034AF6, 0x41047A60, 0xDF60EFC3, 0xA867DF55, 0x316E8EEF,
                0x4669BE79, 0xCB61B38C, 0xBC66831A, 0x256FD2A0, 0x5268E236,
                0xCC0C7795, 0xBB0B4703, 0x220216B9, 0x5505262F, 0xC5BA3BBE,
                0xB2BD0B28, 0x2BB45A92, 0x5CB36A04, 0xC2D7FFA7, 0xB5D0CF31,
                0x2CD99E8B, 0x5BDEAE1D, 0x9B64C2B0, 0xEC63F226, 0x756AA39C,
                0x026D930A, 0x9C0906A9, 0xEB0E363F, 0x72076785, 0x05005713,
                0x95BF4A82, 0xE2B87A14, 0x7BB12BAE, 0x0CB61B38, 0x92D28E9B,
                0xE5D5BE0D, 0x7CDCEFB7, 0x0BDBDF21, 0x86D3D2D4, 0xF1D4E242,
                0x68DDB3F8, 0x1FDA836E, 0x81BE16CD, 0xF6B9265B, 0x6FB077E1,
                0x18B74777, 0x88085AE6, 0xFF0F6A70, 0x66063BCA, 0x11010B5C,
                0x8F659EFF, 0xF862AE69, 0x616BFFD3, 0x166CCF45, 0xA00AE278,
                0xD70DD2EE, 0x4E048354, 0x3903B3C2, 0xA7672661, 0xD06016F7,
                0x4969474D, 0x3E6E77DB, 0xAED16A4A, 0xD9D65ADC, 0x40DF0B66,
                0x37D83BF0, 0xA9BCAE53, 0xDEBB9EC5, 0x47B2CF7F, 0x30B5FFE9,
                0xBDBDF21C, 0xCABAC28A, 0x53B39330, 0x24B4A3A6, 0xBAD03605,
                0xCDD70693, 0x54DE5729, 0x23D967BF, 0xB3667A2E, 0xC4614AB8,
                0x5D681B02, 0x2A6F2B94, 0xB40BBE37, 0xC30C8EA1, 0x5A05DF1B,
                0x2D02EF8D };

            register uint32_t tmp = m_crc;

            while(length >= 16)
            {
                tmp = table[(tmp ^ input[ 0]) & 0xFF] ^ (tmp >> 8);
                tmp = table[(tmp ^ input[ 1]) & 0xFF] ^ (tmp >> 8);
                tmp = table[(tmp ^ input[ 2]) & 0xFF] ^ (tmp >> 8);
                tmp = table[(tmp ^ input[ 3]) & 0xFF] ^ (tmp >> 8);
                tmp = table[(tmp ^ input[ 4]) & 0xFF] ^ (tmp >> 8);
                tmp = table[(tmp ^ input[ 5]) & 0xFF] ^ (tmp >> 8);
                tmp = table[(tmp ^ input[ 6]) & 0xFF] ^ (tmp >> 8);
                tmp = table[(tmp ^ input[ 7]) & 0xFF] ^ (tmp >> 8);
                tmp = table[(tmp ^ input[ 8]) & 0xFF] ^ (tmp >> 8);
                tmp = table[(tmp ^ input[ 9]) & 0xFF] ^ (tmp >> 8);
                tmp = table[(tmp ^ input[10]) & 0xFF] ^ (tmp >> 8);
                tmp = table[(tmp ^ input[11]) & 0xFF] ^ (tmp >> 8);
                tmp = table[(tmp ^ input[12]) & 0xFF] ^ (tmp >> 8);
                tmp = table[(tmp ^ input[13]) & 0xFF] ^ (tmp >> 8);
                tmp = table[(tmp ^ input[14]) & 0xFF] ^ (tmp >> 8);
                tmp = table[(tmp ^ input[15]) & 0xFF] ^ (tmp >> 8);
                input += 16;
                length -= 16;
            }

            for(uint32_t j = 0; j != length; ++j)
                tmp = table[(tmp ^ input[j]) & 0xFF] ^ (tmp >> 8);

            m_crc = tmp;

            return *this;
        }

        CRC32& update(const void* input, uint32_t length)
        {
            return update(reinterpret_cast<const unsigned char*>(input), length);
        }

        uint32_t final() const
        {
            return (m_crc ^ 0xFFFFFFFF);
        }

        static uint32_t digest(const void* input, uint32_t length)
        {
            return CRC32().update(input, length).final();
        }

        static uint32_t digest(const std::string& input)
        {
            return digest(input.data(), input.size());
        }
    };

protected:
    class SHA256
    {
    private:
        typedef uint8_t byte;

        typedef uint32_t u32bit;

        typedef uint64_t u64bit;

        static const u32bit OUTPUT_LENGTH = 32;

        static const u32bit HASH_BLOCK_SIZE = 64;

        static const u32bit COUNT_SIZE = 8;

    private:
        u32bit          W[64];
        u32bit          digest[8];

        byte            buffer[HASH_BLOCK_SIZE];

        u64bit          count;
        u32bit          position;

    private:
        template<typename T> inline T rotate_left(T input, u32bit rot)
        { return static_cast<T>((input << rot) | (input >> (8*sizeof(T)-rot))); }

        template<typename T> inline T rotate_right(T input, u32bit rot)
        { return static_cast<T>((input >> rot) | (input << (8*sizeof(T)-rot))); }

        template<typename T> inline byte get_byte(u32bit byte_num, T input)
        { return static_cast<byte>(input >> ((sizeof(T)-1-(byte_num&(sizeof(T)-1))) << 3)); }

        inline u32bit make_u32bit(byte input0, byte input1, byte input2, byte input3)
        { return static_cast<u32bit>((static_cast<u32bit>(input0) << 24) |
                                     (static_cast<u32bit>(input1) << 16) |
                                     (static_cast<u32bit>(input2) <<  8) | input3); }

        inline u32bit rho(u32bit X, u32bit rot1, u32bit rot2, u32bit rot3)
        {
            return (rotate_right(X, rot1) ^ rotate_right(X, rot2) ^
                    rotate_right(X, rot3));
        }

        inline u32bit sigma(u32bit X, u32bit rot1, u32bit rot2, u32bit shift)
        {
            return (rotate_right(X, rot1) ^ rotate_right(X, rot2) ^ (X >> shift));
        }

        inline void F1(u32bit A, u32bit B, u32bit C, u32bit& D,
                       u32bit E, u32bit F, u32bit G, u32bit& H,
                       u32bit msg, u32bit magic)
        {
            magic += rho(E, 6, 11, 25) + ((E & F) ^ (~E & G)) + msg;
            D += magic + H;
            H += magic + rho(A, 2, 13, 22) + ((A & B) ^ (A & C) ^ (B & C));
        }

        void hash(const byte input[])
        {
            for(u32bit j = 0; j != 16; ++j)
                W[j] = make_u32bit(input[4*j], input[4*j+1], input[4*j+2], input[4*j+3]);
            for(u32bit j = 16; j != 64; ++j)
                W[j] = sigma(W[j- 2], 17, 19, 10) + W[j- 7] +
                    sigma(W[j-15],  7, 18,  3) + W[j-16];

            u32bit A = digest[0], B = digest[1], C = digest[2],
                D = digest[3], E = digest[4], F = digest[5],
                G = digest[6], H = digest[7];

            F1(A,B,C,D,E,F,G,H,W[ 0],0x428A2F98);
            F1(H,A,B,C,D,E,F,G,W[ 1],0x71374491);
            F1(G,H,A,B,C,D,E,F,W[ 2],0xB5C0FBCF);
            F1(F,G,H,A,B,C,D,E,W[ 3],0xE9B5DBA5);
            F1(E,F,G,H,A,B,C,D,W[ 4],0x3956C25B);
            F1(D,E,F,G,H,A,B,C,W[ 5],0x59F111F1);
            F1(C,D,E,F,G,H,A,B,W[ 6],0x923F82A4);
            F1(B,C,D,E,F,G,H,A,W[ 7],0xAB1C5ED5);
            F1(A,B,C,D,E,F,G,H,W[ 8],0xD807AA98);
            F1(H,A,B,C,D,E,F,G,W[ 9],0x12835B01);
            F1(G,H,A,B,C,D,E,F,W[10],0x243185BE);
            F1(F,G,H,A,B,C,D,E,W[11],0x550C7DC3);
            F1(E,F,G,H,A,B,C,D,W[12],0x72BE5D74);
            F1(D,E,F,G,H,A,B,C,W[13],0x80DEB1FE);
            F1(C,D,E,F,G,H,A,B,W[14],0x9BDC06A7);
            F1(B,C,D,E,F,G,H,A,W[15],0xC19BF174);
            F1(A,B,C,D,E,F,G,H,W[16],0xE49B69C1);
            F1(H,A,B,C,D,E,F,G,W[17],0xEFBE4786);
            F1(G,H,A,B,C,D,E,F,W[18],0x0FC19DC6);
            F1(F,G,H,A,B,C,D,E,W[19],0x240CA1CC);
            F1(E,F,G,H,A,B,C,D,W[20],0x2DE92C6F);
            F1(D,E,F,G,H,A,B,C,W[21],0x4A7484AA);
            F1(C,D,E,F,G,H,A,B,W[22],0x5CB0A9DC);
            F1(B,C,D,E,F,G,H,A,W[23],0x76F988DA);
            F1(A,B,C,D,E,F,G,H,W[24],0x983E5152);
            F1(H,A,B,C,D,E,F,G,W[25],0xA831C66D);
            F1(G,H,A,B,C,D,E,F,W[26],0xB00327C8);
            F1(F,G,H,A,B,C,D,E,W[27],0xBF597FC7);
            F1(E,F,G,H,A,B,C,D,W[28],0xC6E00BF3);
            F1(D,E,F,G,H,A,B,C,W[29],0xD5A79147);
            F1(C,D,E,F,G,H,A,B,W[30],0x06CA6351);
            F1(B,C,D,E,F,G,H,A,W[31],0x14292967);
            F1(A,B,C,D,E,F,G,H,W[32],0x27B70A85);
            F1(H,A,B,C,D,E,F,G,W[33],0x2E1B2138);
            F1(G,H,A,B,C,D,E,F,W[34],0x4D2C6DFC);
            F1(F,G,H,A,B,C,D,E,W[35],0x53380D13);
            F1(E,F,G,H,A,B,C,D,W[36],0x650A7354);
            F1(D,E,F,G,H,A,B,C,W[37],0x766A0ABB);
            F1(C,D,E,F,G,H,A,B,W[38],0x81C2C92E);
            F1(B,C,D,E,F,G,H,A,W[39],0x92722C85);
            F1(A,B,C,D,E,F,G,H,W[40],0xA2BFE8A1);
            F1(H,A,B,C,D,E,F,G,W[41],0xA81A664B);
            F1(G,H,A,B,C,D,E,F,W[42],0xC24B8B70);
            F1(F,G,H,A,B,C,D,E,W[43],0xC76C51A3);
            F1(E,F,G,H,A,B,C,D,W[44],0xD192E819);
            F1(D,E,F,G,H,A,B,C,W[45],0xD6990624);
            F1(C,D,E,F,G,H,A,B,W[46],0xF40E3585);
            F1(B,C,D,E,F,G,H,A,W[47],0x106AA070);
            F1(A,B,C,D,E,F,G,H,W[48],0x19A4C116);
            F1(H,A,B,C,D,E,F,G,W[49],0x1E376C08);
            F1(G,H,A,B,C,D,E,F,W[50],0x2748774C);
            F1(F,G,H,A,B,C,D,E,W[51],0x34B0BCB5);
            F1(E,F,G,H,A,B,C,D,W[52],0x391C0CB3);
            F1(D,E,F,G,H,A,B,C,W[53],0x4ED8AA4A);
            F1(C,D,E,F,G,H,A,B,W[54],0x5B9CCA4F);
            F1(B,C,D,E,F,G,H,A,W[55],0x682E6FF3);
            F1(A,B,C,D,E,F,G,H,W[56],0x748F82EE);
            F1(H,A,B,C,D,E,F,G,W[57],0x78A5636F);
            F1(G,H,A,B,C,D,E,F,W[58],0x84C87814);
            F1(F,G,H,A,B,C,D,E,W[59],0x8CC70208);
            F1(E,F,G,H,A,B,C,D,W[60],0x90BEFFFA);
            F1(D,E,F,G,H,A,B,C,W[61],0xA4506CEB);
            F1(C,D,E,F,G,H,A,B,W[62],0xBEF9A3F7);
            F1(B,C,D,E,F,G,H,A,W[63],0xC67178F2);

            digest[0] += A; digest[1] += B; digest[2] += C;
            digest[3] += D; digest[4] += E; digest[5] += F;
            digest[6] += G; digest[7] += H;
        }

        void copy_out(byte output[])
        {
            for(u32bit j = 0; j != OUTPUT_LENGTH; ++j)
                output[j] = get_byte(j % 4, digest[j/4]);
        }

    protected:

        void add_data(const byte input[], u32bit length)
        {
            count += length;

            if (position)
            {
                memcpy(buffer + position, input,
                       std::min<u32bit>(length, sizeof(buffer) - position));

                if(position + length >= HASH_BLOCK_SIZE)
                {
                    hash(buffer);
                    input += (HASH_BLOCK_SIZE - position);
                    length -= (HASH_BLOCK_SIZE - position);
                    position = 0;
                }
            }

            while(length >= HASH_BLOCK_SIZE)
            {
                hash(input);
                input += HASH_BLOCK_SIZE;
                length -= HASH_BLOCK_SIZE;
            }

            memcpy(buffer + position, input,
                   std::min<u32bit>(length, sizeof(buffer) - position));

            position += length;
        }

        void write_count(byte out[])
        {
            for(u32bit j = 0; j != 8; ++j)
            {
                out[j+COUNT_SIZE-8] = get_byte(j % 8, 8 * count);
            }
        }

        void final_result(byte output[OUTPUT_LENGTH])
        {
            buffer[position] = 0x80;
            for(u32bit j = position+1; j != HASH_BLOCK_SIZE; ++j)
                buffer[j] = 0;
            if(position >= HASH_BLOCK_SIZE - COUNT_SIZE)
            {
                hash(buffer);
                memset(buffer, 0, sizeof(buffer));
            }
            write_count(buffer + HASH_BLOCK_SIZE - COUNT_SIZE);

            hash(buffer);
            copy_out(output);
            clear();
        }

    public:
        SHA256()
        {
            clear();
        }

        void clear() throw()
        {
            memset(buffer, 0, sizeof(buffer));
            count = position = 0;

            memset(W, 0, sizeof(W));
            digest[0] = 0x6A09E667;
            digest[1] = 0xBB67AE85;
            digest[2] = 0x3C6EF372;
            digest[3] = 0xA54FF53A;
            digest[4] = 0x510E527F;
            digest[5] = 0x9B05688C;
            digest[6] = 0x1F83D9AB;
            digest[7] = 0x5BE0CD19;
        }

        SHA256& update(const void* input, uint32_t length)
        {
            add_data(reinterpret_cast<const unsigned char*>(input), length);
            return *this;
        }

        void final(byte output[OUTPUT_LENGTH])
        {
            final_result(output);
        }

        std::string final()
        {
            byte result[OUTPUT_LENGTH];
            final_result(result);
            return std::string(reinterpret_cast<char*>(result), OUTPUT_LENGTH);
        }

        bool final_equals(byte compare[OUTPUT_LENGTH])
        {
            byte result[OUTPUT_LENGTH];
            final_result(result);
            return (memcmp(compare, result, OUTPUT_LENGTH) == 0);
        }

        std::string final_hex()
        {
            byte result[OUTPUT_LENGTH];
            final_result(result);

            std::string out(OUTPUT_LENGTH*2, '\0');

            static const char xdigits[16] = {
                '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'A', 'B', 'C', 'D', 'E', 'F'
            };

            std::string::iterator oi = out.begin();
            for (unsigned int i = 0; i < OUTPUT_LENGTH; ++i)
            {
                *oi++ = xdigits[ (result[i] & 0xF0) >> 4 ];
                *oi++ = xdigits[ (result[i] & 0x0F) ];
            }

            return out;
        }

        static std::string digest_bin(const void* input, uint32_t length)
        {
            return SHA256().update(input, length).final();
        }

        static std::string digest_bin(const std::string& input)
        {
            return digest_bin(input.data(), input.size());
        }

        static std::string digest_hex(const void* input, uint32_t length)
        {
            return SHA256().update(input, length).final_hex();
        }

        static std::string digest_hex(const std::string& input)
        {
            return digest_hex(input.data(), input.size());
        }
    };

protected:
    class BTreePage
    {
    protected:
        struct Impl
        {
            unsigned int refs;

            char        data[BTreePageSize];
        };

        struct Impl*    m_impl;

    public:
        BTreePage()
            : m_impl(NULL)
        {
        }

        BTreePage(const BTreePage& btp)
            : m_impl(btp.m_impl)
        {
            if (m_impl)
                ++m_impl->refs;
        }

        ~BTreePage()
        {
            if (m_impl && --m_impl->refs == 0)
                delete m_impl;
        }

        BTreePage& operator=(const BTreePage& btp)
        {
            if (this != &btp)
            {
                if (m_impl && --m_impl->refs == 0)
                    delete m_impl;

                m_impl = btp.m_impl;

                if (m_impl)
                    ++m_impl->refs;
            }

            return *this;
        }

        bool IsValid() const
        {
            return (m_impl != NULL);
        }

        void Create()
        {
            if (m_impl && --m_impl->refs == 0)
                delete m_impl;

            m_impl = new Impl;
            m_impl->refs = 1;
        }

        char* GetBuffer()
        {
            CBTREEDB_ASSERT(m_impl);
            return m_impl->data;
        }

        uint16_t GetLevel() const
        {
            CBTREEDB_ASSERT(m_impl);
            return reinterpret_cast<InnerNode*>(m_impl->data)->level;
        }

        bool IsLeafNode() const
        {
            return (GetLevel() == 0);
        }

        InnerNode* GetAsInnerNode() const
        {
            CBTREEDB_ASSERT(m_impl && !IsLeafNode());
            return reinterpret_cast<InnerNode*>(m_impl->data);
        }

        LeafNode* GetAsLeafNode() const
        {
            CBTREEDB_ASSERT(m_impl && IsLeafNode());
            return reinterpret_cast<LeafNode*>(m_impl->data);
        }
    };

protected:
    class PageCacheImpl
    {
    protected:

        unsigned int    m_refs;

        unsigned int    m_maxsize;

        unsigned int    m_size;

#ifdef CBTREEDB_SELF_VERIFY

        uint32_t        m_lrutime;
#endif

        struct HashCell
        {
            struct HashCell     *bucket_next;

            struct HashCell     *bucket_prev;

            struct HashCell     *list_next;

            struct HashCell     *list_prev;

#ifdef CBTREEDB_SELF_VERIFY

            uint32_t    lrutime;
#endif

            void*       btreeid;

            uint32_t    pageid;

            BTreePage   page;
        };

        std::vector<struct HashCell*> m_hasharray;

        struct HashCell m_sentinel;

        inline unsigned int hashfunc(void* btreeid, uint32_t pageid)
        {
            // since hot pageids are usually ascending, I guess this is a
            // pretty good hash function.
            return (reinterpret_cast<uintptr_t>(btreeid) + pageid) % m_hasharray.size();
        }

    public:
        explicit PageCacheImpl(unsigned int maxsize)
            : m_refs(0), m_maxsize(maxsize), m_size(0)
        {
            m_hasharray.resize(m_maxsize / 2, NULL);

            m_sentinel.list_prev = m_sentinel.list_next = &m_sentinel;
#ifdef CBTREEDB_SELF_VERIFY
            m_lrutime = 0;
            m_sentinel.lrutime = 0;
#endif
        }

        ~PageCacheImpl()
        {
            Clear();
        }

        void RefInc()
        {
            ++m_refs;
        }

        unsigned int RefDec()
        {
            return --m_refs;
        }

        void Clear()
        {
            // free up hash cells
            struct HashCell* hc = m_sentinel.list_next;
            while (hc != &m_sentinel)
            {
                struct HashCell* nc = hc->list_next;
                delete hc;
                hc = nc;
            }

            // zero hash array
            for (unsigned int i = 0; i < m_hasharray.size(); ++i)
                m_hasharray[i] = NULL;

            m_sentinel.list_prev = m_sentinel.list_next = &m_sentinel;
            m_size = 0;
#ifdef CBTREEDB_SELF_VERIFY
            m_sentinel.lrutime = 0;
            m_lrutime = 0;
#endif
        }

        void Store(void* btreeid, uint32_t pageid, const BTreePage& page)
        {
            // check whether its already in the cache

            unsigned int h = hashfunc(btreeid, pageid);

            struct HashCell* hc = m_hasharray[h];

            while( hc && ! (hc->btreeid == btreeid && hc->pageid == pageid) )
            {
                // advance in bucket list
                hc = hc->bucket_next;
            }

            if ( hc ) // found in cache: unlink from LRU list and place in front
            {
                if (hc != m_sentinel.list_next)
                {
                    // remove cell wherever it is
                    hc->list_next->list_prev = hc->list_prev;
                    hc->list_prev->list_next = hc->list_next;

                    // place at head
                    hc->list_prev = &m_sentinel;
                    hc->list_next = m_sentinel.list_next;
                    m_sentinel.list_next->list_prev = hc;
                    m_sentinel.list_next = hc;

                    // copy page, it may have changed
                    hc->page = page;

#ifdef CBTREEDB_SELF_VERIFY
                    hc->lrutime = ++m_lrutime;
#endif
                }
                // else hc is the head -> do nothing.
            }
            else // not found in cache.
            {
                // remove last page in LRU list if necessary
                while (m_size >= m_maxsize)
                {
                    struct HashCell* lc = m_sentinel.list_prev;

                    CBTREEDB_ASSERT( lc != &m_sentinel );

                    // unlink from bucket list
                    if (reinterpret_cast<uintptr_t>(lc->bucket_prev) > m_hasharray.size())
                    {
                        if (lc->bucket_next)
                            lc->bucket_next->bucket_prev = lc->bucket_prev;

                        lc->bucket_prev->bucket_next = lc->bucket_next;
                    }
                    else // at first place in bucket list
                    {
                        if (lc->bucket_next)
                            lc->bucket_next->bucket_prev = lc->bucket_prev;

                        m_hasharray[ reinterpret_cast<uintptr_t>(lc->bucket_prev) ] = lc->bucket_next;
                    }

                    // unlink from LRU list
                    lc->list_next->list_prev = lc->list_prev;
                    lc->list_prev->list_next = lc->list_next;

                    delete lc;

                    --m_size;
                }

                // create new hash cell
                hc = new HashCell;

#ifdef CBTREEDB_SELF_VERIFY
                hc->lrutime = ++m_lrutime;
#endif
                hc->btreeid = btreeid;
                hc->pageid = pageid;
                hc->page = page;

                // set hash cell as head of LRU list
                hc->list_prev = &m_sentinel;
                hc->list_next = m_sentinel.list_next;
                m_sentinel.list_next->list_prev = hc;
                m_sentinel.list_next = hc;

                // insert new hash cell to correct bucket
                hc->bucket_prev = reinterpret_cast<HashCell*>( h );
                hc->bucket_next = m_hasharray[h];

                if (m_hasharray[h])
                    m_hasharray[h]->bucket_prev = hc;

                m_hasharray[h] = hc;

                ++m_size;
            }

#ifdef CBTREEDB_SELF_VERIFY
            CBTREEDB_ASSERT( Verify() );
#endif
        }

        bool Retrieve(void* btreeid, uint32_t pageid, BTreePage& outpage)
        {
            // check whether its in the cache

            unsigned int h = hashfunc(btreeid, pageid);

            struct HashCell* hc = m_hasharray[h];

            while( hc && ! (hc->btreeid == btreeid && hc->pageid == pageid) )
            {
                // advance in bucket list
                hc = hc->bucket_next;
            }

            if ( hc ) // found in cache: unlink from LRU list and place in front
            {
                if (hc != m_sentinel.list_next)
                {
                    // remove cell wherever it is
                    hc->list_next->list_prev = hc->list_prev;
                    hc->list_prev->list_next = hc->list_next;

                    // place at head
                    hc->list_prev = &m_sentinel;
                    hc->list_next = m_sentinel.list_next;
                    m_sentinel.list_next->list_prev = hc;
                    m_sentinel.list_next = hc;

#ifdef CBTREEDB_SELF_VERIFY
                    hc->lrutime = ++m_lrutime;
#endif
                }
                // else hc is the head -> do nothing.

                outpage = hc->page;

#ifdef CBTREEDB_SELF_VERIFY
                CBTREEDB_ASSERT( Verify() );
#endif
                return true;
            }
            else
            {
#ifdef CBTREEDB_SELF_VERIFY
                CBTREEDB_ASSERT( Verify() );
#endif
                return false;
            }
        }

        void SetMaxSize(unsigned int maxsize)
        {
            m_maxsize = maxsize;
        }

        std::vector< std::pair<void*, uint32_t> > GetPagelist() const
        {
            std::vector< std::pair<void*, uint32_t> > v;

            struct HashCell* hc = m_sentinel.list_next;

            while (hc != &m_sentinel)
            {
                v.push_back( std::make_pair(hc->btreeid, hc->pageid) );
                hc = hc->list_next;
            }

            return v;
        }

        bool Verify() const
        {
            { // traverse LRU list forwards

                unsigned int size = 0;
                struct HashCell* hc = m_sentinel.list_next;

                while (hc != &m_sentinel)
                {
                    if (!(hc->list_prev->list_next == hc)) return false;
                    if (!(hc->list_next->list_prev == hc)) return false;
#ifdef CBTREEDB_SELF_VERIFY
                    if (!(hc->lrutime > hc->list_next->lrutime)) return false;
#endif

                    ++size;
                    hc = hc->list_next;
                }

                if (size != m_size) return false;
            }

            { // traverse LRU list backwards

                unsigned int size = 0;
                struct HashCell* hc = m_sentinel.list_prev;

                while (hc != &m_sentinel)
                {
                    if (!(hc->list_prev->list_next == hc)) return false;
                    if (!(hc->list_next->list_prev == hc)) return false;
#ifdef CBTREEDB_SELF_VERIFY
                    if (!(hc->lrutime < hc->list_prev->lrutime || hc->list_prev == &m_sentinel)) return false;
#endif

                    ++size;
                    hc = hc->list_prev;
                }

                if (size != m_size) return false;
            }

            { // check and count hash cells in buckets

                unsigned int size = 0;

                for (unsigned int b = 0; b < m_hasharray.size(); ++b)
                {
                    struct HashCell* hc = m_hasharray[b];

                    if (!hc) continue;

                    if (!(reinterpret_cast<uintptr_t>(hc->bucket_prev) == b)) return false;

                    ++size;

                    while (hc->bucket_next != NULL)
                    {
                        if (!(hc->bucket_next->bucket_prev == hc)) return false;

                        hc = hc->bucket_next;

                        ++size;
                    }
                }

                if (size != m_size) return false;
            }

            return true;
        }
    };

public:
    class PageCache
    {
    protected:

        PageCacheImpl*  m_impl;

    public:
        explicit PageCache(unsigned int maxpages)
            : m_impl(new PageCacheImpl(maxpages))
        {
            m_impl->RefInc();
        }

        PageCache(const PageCache& pc)
            : m_impl(pc.m_impl)
        {
            m_impl->RefInc();
        }

        ~PageCache()
        {
            if (m_impl->RefDec() == 0)
                delete m_impl;
        }

        PageCache& operator=(const PageCache& pc)
        {
            if (this != &pc)
            {
                if (m_impl->RefDec() == 0)
                    delete m_impl;

                m_impl = pc.m_impl;
                m_impl->RefInc();
            }

            return *this;
        }

        void Clear()
        {
            return m_impl->Clear();
        }

        void Store(void* btreeid, uint32_t pageid, const BTreePage& page)
        {
            return m_impl->Store(btreeid, pageid, page);
        }

        bool Retrieve(void* btreeid, uint32_t pageid, BTreePage& outpage)
        {
            return m_impl->Retrieve(btreeid, pageid, outpage);
        }

        void SetMaxSize(unsigned int maxsize)
        {
            return m_impl->SetMaxSize(maxsize);
        }

        std::vector< std::pair<void*, uint32_t> > GetPagelist() const
        {
            return m_impl->GetPagelist();
        }

        bool Verify() const
        {
            return m_impl->Verify();
        }
    };

protected:
    class ReaderImpl
    {
    protected:

        unsigned int    m_refs;

        key_compare     m_key_less;

        char            m_signaturestr[8];

        std::istream*   m_istream;

        SignaturePage   m_signature;

        PageCache*      m_pagecache;

        BTreePage ReadIndexPage(uint32_t pageoffset)
        {
            CBTREEDB_CHECK(pageoffset + m_signature.btree_pagesize <= m_signature.btree_size,
                           "Invalid B-tree page offset to retrieve.");

            CBTREEDB_ASSERT(m_istream);

            BTreePage page;

            if (m_pagecache)
            {
                if (m_pagecache->Retrieve(this, pageoffset, page))
                    return page;
            }

            m_istream->seekg(m_signature.btree_offset + pageoffset);
            CBTREEDB_CHECK(m_istream->good(), "Could not read B-tree page.");

            page.Create();

            m_istream->read(page.GetBuffer(), BTreePageSize);
            CBTREEDB_CHECK(m_istream->good(), "Could not read B-tree page.");

            if (m_pagecache)
            {
                m_pagecache->Store(this, pageoffset, page);
            }

            return page;
        }

        bool ReadValueRange(uint64_t offset, void* data, uint32_t size)
        {
            CBTREEDB_ASSERT(m_istream);

            if (offset + size > m_signature.value_size) return false;

            m_istream->seekg(m_signature.value_offset + offset);
            if (m_istream->bad()) return false;

            m_istream->read(reinterpret_cast<char*>(data), size);
            if (m_istream->bad()) return false;

            return true;
        }

        bool KeyEqual(const key_type& a, const key_type& b)
        {
            return !m_key_less(a,b) && !m_key_less(b,a);
        }

        bool KeyUnequal(const key_type& a, const key_type& b)
        {
            return m_key_less(a,b) || m_key_less(b,a);
        }

        template <typename NodeType>
        int BinarySearch(const NodeType* node, key_type key)
        {
            register int lo = 0, hi = node->slots;

            while (lo < hi)
            {
                register int mid = (hi + lo) >> 1;

                if (m_key_less(node->keys[mid], key)) {
                    lo = mid + 1;
                }
                else {
                    hi = mid;
                }
            }

#ifdef CBTREEDB_SELF_VERIFY
            // verify result using simple linear search
            {
                int i = 0;
                while(i < node->slots && m_key_less(node->keys[i], key))
                    ++i;

                CBTREEDB_ASSERT(i == lo);
            }
#endif
            return lo;
        }

    public:
        ReaderImpl(const key_compare& key_less)
            : m_refs(0), m_key_less(key_less), m_istream(NULL), m_pagecache(NULL)
        {
            memcpy(m_signaturestr, "cbtreedb", 8);
        }

        void RefInc()
        {
            ++m_refs;
        }

        unsigned int RefDec()
        {
            return --m_refs;
        }

        void SetSignature(const char* newsignature)
        {
            unsigned int i = 0;
            for(; i < 8 && newsignature[i]; ++i)
                m_signaturestr[i] = newsignature[i];

            for(; i < 8; ++i)
                m_signaturestr[i] = 0;
        }

        bool Open(std::istream& file, std::string* errortext = NULL)
        {
            m_istream = NULL;

            file.seekg(0, std::ios::beg);
            if (file.bad()) {
                if (errortext) *errortext = "Could not open database.";
                return false;
            }

            file.read(reinterpret_cast<char*>(&m_signature), sizeof(m_signature));
            if (file.bad()) {
                if (errortext) *errortext = "Could not read signature.";
                return false;
            }

            if (memcmp(m_signature.signature, m_signaturestr, 8) != 0) {
                if (errortext) *errortext = "Could not verify signature.";
                return false;
            }

            if (m_signature.version != 0x00010000) {
                if (errortext) *errortext = "Signature contains unknown version.";
                return false;
            }
            
            if (m_signature.app_version_id != AppVersionId) {
                if (errortext) *errortext = "Signature mismatches application version identifier.";
                return false;
            }
            
            uint32_t crc = CRC32::digest(reinterpret_cast<char*>(&m_signature)+16, sizeof(m_signature)-16);

            if (m_signature.header_crc32 != crc) {
                if (errortext) *errortext = "Header checksum mismatches.";
                return false;
            }

            if (m_signature.key_size != sizeof(key_type)) {
                if (errortext) *errortext  = "Database not compatible with this reader: key sizes mismatch.";
                return false;
            }

            if (m_signature.btree_pagesize != BTreePageSize) {
                if (errortext) *errortext  = "Database not compatible with this reader: page sizes mismatch.";
                return false;
            }

            // test database compatibility with order relation by checking root
            // node's key sequence.

            m_istream = &file;

            BTreePage root = ReadIndexPage(0);

            if ( root.IsLeafNode() )
            {
                LeafNode* leaf = root.GetAsLeafNode();

                for(uint16_t s = 0; s < leaf->slots - 1; ++s)
                {
                    if (!m_key_less(leaf->keys[s], leaf->keys[s+1])) {
                        m_istream = NULL;
                        if (errortext) *errortext  = "Database not compatible with this reader: root keys order mismatches.";
                        return false;
                    }
                }
            }
            else
            {
                InnerNode* inner = root.GetAsInnerNode();

                for(uint16_t s = 0; s < inner->slots - 1; ++s)
                {
                    if (!m_key_less(inner->keys[s], inner->keys[s+1])) {
                        m_istream = NULL;
                        if (errortext) *errortext  = "Database not compatible with this reader: root keys order mismatches.";
                        return false;
                    }
                }
            }

            return true;
        }

        void Close()
        {
            if (m_istream)
                m_istream = NULL;
        }

        void SetPageCache(PageCache* newpagecache)
        {
            m_pagecache = newpagecache;
        }

        uint32_t Size() const
        {
            CBTREEDB_CHECK(m_istream, "No database loaded.");

            return m_signature.items;
        }

        const SignaturePage& GetSignature() const
        {
            return m_signature;
        }

    protected:
        bool FindKey(const key_type& key, uint64_t& outoffset, uint32_t& outsize)
        {
            if (m_signature.btree_size == 0) return false;

            BTreePage page = ReadIndexPage(0);

            bool checklastkey = false;
            key_type lastkey = key_type();

            while( ! page.IsLeafNode() )
            {
                InnerNode* inner = page.GetAsInnerNode();

                CBTREEDB_CHECK(!checklastkey || !m_key_less(lastkey, inner->LastKey()),
                               "BTree corrupt (lastkey does not match).");

                int slot = BinarySearch(inner, key);

                uint32_t next = inner->childrenoffset;
                if (inner->level > 1)
                    next += slot * sizeof(InnerNode);
                else
                    next += slot * sizeof(LeafNode);

                int oldlevel = inner->level;

                if (slot < inner->slots) {
                    checklastkey = true;
                    lastkey = inner->keys[slot];
                }

                page = ReadIndexPage(next);

                CBTREEDB_CHECK(page.GetLevel() == oldlevel-1,
                               "BTree corrupt (level order mismatch).");
            }

            LeafNode* leaf = page.GetAsLeafNode();

            CBTREEDB_CHECK(!checklastkey || KeyEqual(leaf->LastKey(), lastkey),
                           "BTree corrupt (lastkey in leaf does not match).");

            int slot = BinarySearch(leaf, key);

            if (slot >= leaf->slots || KeyUnequal(leaf->keys[slot],  key))
                return false;

            // Return offset and size via pointers.

            outoffset = leaf->baseoffset + leaf->offsets[slot];

            // figure out value size of this slot: compute it from the offsets
            CBTREEDB_CHECK(leaf->offsets[slot] <= leaf->offsets[slot+1],
                           "BTree corrupt (offsets are not ascending).");

            outsize = leaf->offsets[slot+1] - leaf->offsets[slot];

            return true;
        }

    public:
        bool Exists(const key_type& key)
        {
            CBTREEDB_CHECK(m_istream, "No database loaded.");

            uint64_t offset;
            uint32_t size;

            return FindKey(key, offset, size);
        }

        bool Lookup(const key_type& key, void* outvalue, uint32_t maxsize)
        {
            CBTREEDB_CHECK(m_istream, "No database loaded.");

            uint64_t offset;
            uint32_t size;

            if (!FindKey(key, offset, size))
                return false;

            uint32_t readsize = size;
            if (readsize > maxsize) readsize = maxsize;

            return ReadValueRange(offset, outvalue, readsize);
        }

        bool Lookup(const key_type& key, std::string& outvalue)
        {
            CBTREEDB_CHECK(m_istream, "No database loaded.");

            uint64_t offset;
            uint32_t size;

            if (!FindKey(key, offset, size))
                return false;

            outvalue.resize(size);

            return ReadValueRange(offset, const_cast<char*>(outvalue.data()), size);
        }

        std::string operator[](const key_type& key)
        {
            CBTREEDB_CHECK(m_istream, "No database loaded.");

            uint64_t offset;
            uint32_t size;

            if (!FindKey(key, offset, size))
                return std::string();

            std::string outvalue;
            outvalue.resize(size);

            if (!ReadValueRange(offset, const_cast<char*>(outvalue.data()), size))
                return std::string();

            return outvalue;
        }
        
    protected:
        bool FindIndex(uint32_t index, key_type& outkey, uint64_t& outoffset, uint32_t& outsize)
        {
            if (index >= m_signature.items) return false;

            // directly compute offset of leaf containing to the key index

            uint32_t offset = index / LeafNodeNumKeys * BTreePageSize;
            unsigned int slot = index % LeafNodeNumKeys;

            BTreePage page = ReadIndexPage(m_signature.btree_firstleaf + offset);

            CBTREEDB_CHECK(page.IsLeafNode(),
                           "BTree corrupt (expecting leaf node).");

            LeafNode* leaf = page.GetAsLeafNode();

            CBTREEDB_CHECK(slot < leaf->slots,
                           "BTree corrupt (index beyond range in leaf node).");

            // copy key and offset
            outkey = leaf->keys[slot];
            outoffset = leaf->baseoffset + leaf->offsets[slot];

            // figure out value size of this slot: compute it from the offsets
            CBTREEDB_CHECK(leaf->offsets[slot] <= leaf->offsets[slot+1],
                           "BTree corrupt (offsets are not ascending).");

            outsize = leaf->offsets[slot+1] - leaf->offsets[slot];

            return true;
        }

    public:
        uint32_t GetIndex(uint32_t index, key_type& outkey)
        {
            CBTREEDB_CHECK(m_istream, "No database loaded.");

            uint64_t offset;
            uint32_t size;

            if (!FindIndex(index, outkey, offset, size))
                return 0;

            return size;
        }

        uint32_t GetIndex(uint32_t index, key_type& outkey, void* outvalue, uint32_t maxsize)
        {
            CBTREEDB_CHECK(m_istream, "No database loaded.");

            uint64_t offset;
            uint32_t size;

            if (!FindIndex(index, outkey, offset, size))
                return 0;

            uint32_t outsize = size;
            if (outsize > maxsize) outsize = maxsize;

            if (!ReadValueRange(offset, outvalue, outsize))
                return 0;

            return size;
        }

        uint32_t GetIndex(uint32_t index, key_type& outkey, std::string& outvalue)
        {
            CBTREEDB_CHECK(m_istream, "No database loaded.");

            uint64_t offset;
            uint32_t size;

            if (!FindIndex(index, outkey, offset, size))
                return 0;

            outvalue.resize(size);

            if (!ReadValueRange(offset, const_cast<char*>(outvalue.data()), size))
                return 0;

            return size;
        }

        bool Verify()
        {
            CBTREEDB_CHECK(m_istream, "No database loaded.");

            if (!VerifyBTree()) return false;

            if (!VerifyBTreeChecksum()) return false;

            if (!VerifyValueChecksum()) return false;

            return true;
        }

        bool VerifyBTree()
        {
            CBTREEDB_CHECK(m_istream, "No database loaded.");

            if (m_signature.btree_size == 0) return true;

            key_type minkey = key_type(), maxkey = key_type();
            uint64_t lastoffset = 0;
            return VerifyBTreeNode(0, &minkey, &maxkey, &lastoffset);
        }

    protected:
        bool VerifyBTreeNode(uint32_t offset, key_type* minkey, key_type* maxkey, uint64_t* lastoffset)
        {
            BTreePage page = ReadIndexPage(offset);

            if ( page.IsLeafNode() )
            {
                LeafNode* leaf = page.GetAsLeafNode();

                if (*lastoffset != leaf->baseoffset + leaf->offsets[0]) return false;

                for(uint16_t s = 0; s < leaf->slots - 1; ++s)
                {
                    if (!m_key_less(leaf->keys[s], leaf->keys[s+1])) return false;

                    if (!(leaf->offsets[s] <= leaf->offsets[s+1])) return false;
                }

                *minkey = leaf->keys[0];
                *maxkey = leaf->keys[leaf->slots - 1];
                *lastoffset = leaf->baseoffset + leaf->offsets[leaf->slots];
            }
            else
            {
                InnerNode* inner = page.GetAsInnerNode();

                for(uint16_t s = 0; s < inner->slots - 1; ++s)
                {
                    if (!m_key_less(inner->keys[s], inner->keys[s+1])) return false;
                }

                for(uint16_t s = 0; s <= inner->slots; ++s)
                {
                    uint32_t childoffset = inner->childrenoffset;

                    if (inner->level > 1)
                        childoffset += s * sizeof(InnerNode);
                    else
                        childoffset += s * sizeof(LeafNode);

                    key_type subminkey = key_type(), submaxkey = key_type();

                    if (!VerifyBTreeNode(childoffset, &subminkey, &submaxkey, lastoffset)) return false;

                    if (s == 0)
                        *minkey = subminkey;
                    else
                        if (!m_key_less(inner->keys[s-1], subminkey)) return false;

                    if (s == inner->slots)
                        *maxkey = submaxkey;
                    else
                        if (!KeyEqual(inner->keys[s], submaxkey)) return false;
                }
            }

            return true;
        }

    public:
        bool VerifyBTreeChecksum()
        {
            CBTREEDB_CHECK(m_istream, "No database loaded.");

            SHA256 sha;

            for(uint32_t offset = 0; offset < m_signature.btree_size; offset += BTreePageSize)
            {
                BTreePage page = ReadIndexPage(offset);

                sha.update(page.GetBuffer(), BTreePageSize);
            }

            return ( sha.final_equals(m_signature.btree_sha256) );
        }

        bool VerifyValueChecksum()
        {
            CBTREEDB_CHECK(m_istream, "No database loaded.");

            SHA256 sha;

            char buffer[64*1024];

            for(uint64_t offset = 0; offset < m_signature.value_size; offset += sizeof(buffer))
            {
                uint64_t remsize = std::min<uint64_t>(sizeof(buffer), m_signature.value_size - offset);

                if (!ReadValueRange(offset, buffer, remsize))
                    return false;

                sha.update(buffer, remsize);
            }

            return( sha.final_equals(m_signature.value_sha256) );
        }
    };

public:
    class Reader
    {
    protected:

        ReaderImpl*     m_impl;

    public:
        Reader(const key_compare& key_less=key_compare())
            : m_impl(new ReaderImpl(key_less))
        {
            m_impl->RefInc();
        }

        Reader(const Reader& rd)
            : m_impl(rd.m_impl)
        {
            m_impl->RefInc();
        }

        ~Reader()
        {
            if (m_impl->RefDec() == 0)
                delete m_impl;
        }

        Reader& operator=(const Reader& rd)
        {
            if (this != &rd)
            {
                if (m_impl->RefDec() == 0)
                    delete m_impl;

                m_impl = rd.m_impl;
                m_impl->RefInc();
            }

            return *this;
        }

        void SetSignature(const char* newsignature)
        {
            return m_impl->SetSignature(newsignature);
        }

        bool Open(std::istream& file, std::string* errortext = NULL)
        {
            return m_impl->Open(file, errortext);
        }

        void Close()
        {
            return m_impl->Close();
        }

        void SetPageCache(PageCache* newpagecache)
        {
            return m_impl->SetPageCache(newpagecache);
        }

        uint32_t Size() const
        {
            return m_impl->Size();
        }

        const SignaturePage& GetSignature() const
        {
            return m_impl->GetSignature();
        }

        bool Exists(const key_type& key)
        {
            return m_impl->Exists(key);
        }

        bool Lookup(const key_type& key, void* outvalue, uint32_t maxsize)
        {
            return m_impl->Lookup(key, outvalue, maxsize);
        }

        bool Lookup(const key_type& key, std::string& outvalue)
        {
            return m_impl->Lookup(key, outvalue);
        }

        std::string operator[](const key_type& key)
        {
            return (*m_impl)[key];
        }

        uint32_t GetIndex(uint32_t index, key_type& outkey)
        {
            return m_impl->GetIndex(index, outkey);
        }

        uint32_t GetIndex(uint32_t index, key_type& outkey, void* outvalue, uint32_t maxsize)
        {
            return m_impl->GetIndex(index, outkey, outvalue, maxsize);
        }

        uint32_t GetIndex(uint32_t index, key_type& outkey, std::string& outvalue)
        {
            return m_impl->GetIndex(index, outkey, outvalue);
        }

        bool Verify()
        {
            return m_impl->Verify();
        }

        bool VerifyBTree()
        {
            return m_impl->VerifyBTree();
        }

        bool VerifyBTreeChecksum()
        {
            return m_impl->VerifyBTreeChecksum();
        }

        bool VerifyValueChecksum()
        {
            return m_impl->VerifyValueChecksum();
        }
    };

protected:

    class BTreeBuilder
    {
    protected:

        key_compare     m_key_less;

        unsigned int    m_size;

        std::vector<LeafNode> m_leaves;

        typedef std::vector<InnerNode> innerlevel_type;

        std::vector<innerlevel_type> m_inners;

    public:
        BTreeBuilder(const key_compare& key_less)
            : m_key_less(key_less), m_size(0)
        {
        }

    protected:
        void AddInner(uint16_t level, const key_type& key)
        {
            if (m_inners.size() < level)
            {
                CBTREEDB_ASSERT(m_inners.size()+1 == level);

                // Create vector for this level
                m_inners.push_back(innerlevel_type());
            }

            if (m_inners[level-1].size() == 0)
            {
                // Create first inner node on this level
                m_inners[level-1].push_back(InnerNode(level));
            }

            InnerNode* inner = &m_inners[level-1].back();

            CBTREEDB_ASSERT(inner->slots == 0 || m_key_less(inner->LastKey(), key));

            if (inner->IsFull())
            {
                CBTREEDB_ASSERT(m_key_less(inner->LastKey(), key));

                // Put last key of leaf key into inner node(s) of higher level
                AddInner(inner->level + 1, key);

                // Create new inner node
                m_inners[level-1].push_back(InnerNode(level));
                inner = &m_inners[level-1].back();
            }
            else
            {
                // Append Key
                inner->keys[inner->slots++] = key;
            }
        }

    public:
        void Add(const key_type& key, uint64_t offset, uint32_t size)
        {
            if (m_leaves.size() == 0) {
                // create first leaf node
                m_leaves.push_back(LeafNode());
            }

            LeafNode* leaf = &m_leaves.back();

            if (leaf->slots > 0) {
                CBTREEDB_ASSERT(m_key_less(leaf->LastKey(), key));
            }

            if (leaf->IsFull())
            {
                // put last key into inner node(s)
                AddInner(1, leaf->LastKey());

                // create new leaf
                m_leaves.push_back(LeafNode());
                leaf = &m_leaves.back();
                leaf->baseoffset = offset;
            }

            // append key + value items relative offset
            leaf->keys[leaf->slots] = key;

            CBTREEDB_ASSERT(offset >= leaf->baseoffset);
            leaf->offsets[leaf->slots] = offset - leaf->baseoffset;
            leaf->offsets[leaf->slots+1] = leaf->offsets[leaf->slots] + size;

            ++leaf->slots;
            ++m_size;
        }

        unsigned int Size() const
        {
            return m_size;
        }

        const key_type& GetLastKey() const
        {
            CBTREEDB_CHECK(m_size > 0, "No keys inserted in the tree yet");

            return m_leaves.back().LastKey();
        }

        void GetIndex(unsigned int index, key_type& outkey, uint32_t& outsize) const
        {
            CBTREEDB_CHECK(index < m_size, "Attempting to retrieve out of bounds index.");

            unsigned int leafnum = index / LeafNodeNumKeys;
            unsigned int slot = index % LeafNodeNumKeys;

            CBTREEDB_ASSERT(leafnum < m_leaves.size());

            const LeafNode* leaf = &m_leaves[leafnum];

            CBTREEDB_ASSERT(slot < leaf->slots);

            outkey = leaf->keys[slot];
            outsize = leaf->offsets[slot+1] - leaf->offsets[slot];
        }

#if 0 // extra debug code for printing the tree

        void Print(std::ostream& os) const
        {
            os << "Leaves:" << std::endl;
            for (unsigned int i = 0; i < m_leaves.size(); ++i)
            {
                os << i << ":";
                for (unsigned int j = 0; j < m_leaves[i].slots; ++j)
                {
                    os << " " << m_leaves[i].keys[j];
                }
                os << std::endl;
            }

            for (unsigned int l = 0; l < m_inners.size(); ++l)
            {
                os << "Level " << (l+1) << std::endl;

                for (unsigned int i = 0; i < m_inners[l].size(); ++i)
                {
                    os << i << ":";
                    for (unsigned int j = 0; j < m_inners[l][i].slots; ++j)
                    {
                        os << " " << m_inners[l][i].keys[j];
                    }
                    os << std::endl;
                }
            }
        }

#endif // extra debug code for printing the tree

        void Write(std::ostream& os, SignaturePage& signature)
        {
            signature.btree_pagesize = BTreePageSize;
            signature.btree_levels = m_inners.size() + 1;
            signature.btree_leaves = m_leaves.size();

            // Fill in childrenoffset field by precomputing the offsets.
            {
                // start with offset after the root (inner) node.
                uint32_t offset = sizeof(InnerNode);

                for (int l = m_inners.size()-1; l >= 0; --l)
                {
                    for (unsigned int i = 0; i < m_inners[l].size(); ++i)
                    {
                        InnerNode& inner = m_inners[l][i];

                        inner.childrenoffset = offset;

                        // add all children node sizes to offset.
                        offset += (inner.slots + 1) * sizeof(InnerNode);
                    }
                }
            }

            // Write out inner nodes

            uint64_t writesize = 0;
            SHA256 sha;

            for (int l = m_inners.size()-1; l >= 0; --l)
            {
                for (unsigned int i = 0; i < m_inners[l].size(); ++i)
                {
                    os.write(reinterpret_cast<char*>(&m_inners[l][i]), sizeof(m_inners[l][i]));
                    CBTREEDB_CHECK(os.good(), "Error writing B-tree inner node page to output stream.");

                    sha.update(&m_inners[l][i], sizeof(m_inners[l][i]));

                    writesize += sizeof(m_inners[l][i]);
                }
            }

            // Write out leaf nodes

            signature.btree_firstleaf = writesize;

            for (unsigned int i = 0; i < m_leaves.size(); ++i)
            {
                os.write(reinterpret_cast<char*>(&m_leaves[i]), sizeof(m_leaves[i]));
                CBTREEDB_CHECK(os.good(), "Error writing B-tree leaf node page to output stream.");

                sha.update(&m_leaves[i], sizeof(m_leaves[i]));

                writesize += sizeof(m_leaves[i]);
            }

            sha.final(signature.btree_sha256);
            signature.btree_size = writesize;
        }
    };

public:
    class Writer
    {
    protected:
        typedef std::map<key_type, std::string, key_compare> datamap_type;

        datamap_type    m_datamap;

        key_compare     m_key_less;

        char            m_signaturestr[8];

    public:
        Writer(const key_compare& key_less=key_compare())
            : m_datamap(key_less),
              m_key_less(key_less)
        {
            memcpy(m_signaturestr, "cbtreedb", 8);
        }

        void Add(const key_type& key, const void* data, size_t size)
        {
            m_datamap.insert( typename datamap_type::value_type(key, std::string(reinterpret_cast<const char*>(data), size)) );
        }

        void Add(const key_type& key, const std::string& data)
        {
            Add(key, data.data(), data.size());
        }

        size_t Size() const
        {
            return m_datamap.size();
        }

        void SetSignature(const char* newsignature)
        {
            unsigned int i = 0;
            for(; i < 8 && newsignature[i]; ++i)
                m_signaturestr[i] = newsignature[i];

            for(; i < 8; ++i)
                m_signaturestr[i] = 0;
        }

        void Write(std::ostream& os) const
        {
            os.clear();
            os.seekp(0, std::ios::beg);

            // Write zeroed signature block to be overwritten when the file is
            // finialized.

            SignaturePage signature;
            memset(&signature, 0, sizeof(signature));
            os.write(reinterpret_cast<char*>(&signature), sizeof(signature));

            char signature_padding[SignaturePageSize - sizeof(SignaturePage)];
            memset(signature_padding, 0, SignaturePageSize - sizeof(SignaturePage));
            os.write(signature_padding, SignaturePageSize - sizeof(SignaturePage));

            CBTREEDB_CHECK(os.good(), "Error writing signature block out output stream.");

            // Prepare signature for data

            memcpy(signature.signature, m_signaturestr, 8);
            signature.version = 0x00010000;
            signature.app_version_id = AppVersionId;
            signature.items = m_datamap.size();
            signature.key_size = sizeof(key_type);

            // Construct a and write a constant B-Tree

            BTreeBuilder btree(m_key_less);
            uint64_t dataoffset = 0;

            for (typename datamap_type::const_iterator di = m_datamap.begin();
                 di != m_datamap.end(); ++di)
            {
                btree.Add(di->first, dataoffset, di->second.size());
                dataoffset += di->second.size();
            }

            signature.btree_offset = BTreePageSize;
            btree.Write(os, signature);

            CBTREEDB_CHECK(os.good(), "Error writing B-tree pages to output stream.");
            CBTREEDB_ASSERT(os.tellp() == std::ostream::pos_type(SignaturePageSize + signature.btree_size));

            // Write all value blobs to file

            SHA256 sha;

            for (typename datamap_type::const_iterator di = m_datamap.begin();
                 di != m_datamap.end(); ++di)
            {
                os.write(di->second.data(), di->second.size());
                CBTREEDB_CHECK(os.good(), "Error writing data block to output stream.");

                sha.update(di->second.data(), di->second.size());
            }

            CBTREEDB_ASSERT(os.tellp() == std::ostream::pos_type(SignaturePageSize + signature.btree_size + dataoffset));

            // Fill in signature page

            signature.value_offset = SignaturePageSize + signature.btree_size;
            signature.value_size = dataoffset;
            sha.final(signature.value_sha256);

            // Calculate header checksum

            signature.header_crc32 = CRC32::digest(reinterpret_cast<char*>(&signature)+16, sizeof(signature)-16);

            os.seekp(0, std::ios::beg);
            CBTREEDB_CHECK(os.good(), "Error seeking back to signature page in output stream.");
            CBTREEDB_ASSERT(os.tellp() == std::ostream::pos_type(0));

            os.write(reinterpret_cast<char*>(&signature), sizeof(signature));
            CBTREEDB_CHECK(os.good(), "Error writing signature page to output stream.");
        }
    };

public:
    class WriterSequential
    {
    protected:
        key_compare     m_key_less;

        BTreeBuilder    m_btree;

        uint64_t        m_dataoffset;

        char            m_signaturestr[8];

        SignaturePage   m_signature;

        std::ostream*   m_ostream;

        uint32_t        m_currpos;

        uint64_t        m_curroffset;

        SHA256          m_datasha;

    public:
        WriterSequential(const key_compare& key_less=key_compare())
            : m_key_less(key_less),
              m_btree(key_less),
              m_dataoffset(0),
              m_ostream(NULL),
              m_currpos(-1)
        {
            memcpy(m_signaturestr, "cbtreedb", 8);
        }

        void Add(const key_type& key, uint32_t size)
        {
            CBTREEDB_CHECK(m_btree.Size() == 0 || m_key_less(m_btree.GetLastKey(), key),
                           "Key sequence for Add() must be ascending.");
            CBTREEDB_CHECK(m_ostream == NULL,
                           "Cannot declare keys after starting phase 2.");

            m_btree.Add(key, m_dataoffset, size);
            m_dataoffset += size;
        }

        size_t Size() const
        {
            return m_btree.Size();
        }

        void SetSignature(const char* newsignature)
        {
            unsigned int i = 0;
            for(; i < 8 && newsignature[i]; ++i)
                m_signaturestr[i] = newsignature[i];

            for(; i < 8; ++i)
                m_signaturestr[i] = 0;
        }

        void WriteHeader(std::ostream& os)
        {
            CBTREEDB_CHECK(m_ostream == NULL,
                           "Cannot write header again in phase 2.");

            os.seekp(0, std::ios::beg);

            // write zeroed signature page to be overwritten when the file is
            // finialized.
            memset(&m_signature, 0, sizeof(m_signature));
            os.write(reinterpret_cast<char*>(&m_signature), sizeof(m_signature));

            char signature_padding[SignaturePageSize - sizeof(SignaturePage)];
            memset(signature_padding, 0, SignaturePageSize - sizeof(SignaturePage));
            os.write(signature_padding, SignaturePageSize - sizeof(SignaturePage));

            CBTREEDB_CHECK(os.good(), "Error writing signature page to output stream.");

            // prepare signature for data

            memcpy(m_signature.signature, m_signaturestr, 8);
            m_signature.version = 0x00010000;
            m_signature.app_version_id = AppVersionId;
            m_signature.btree_offset = SignaturePageSize;
            m_signature.items = m_btree.Size();
            m_signature.key_size = sizeof(key_type);
            m_signature.value_size = m_dataoffset;

            // write constant B-tree

            m_btree.Write(os, m_signature);
            CBTREEDB_CHECK(os.good(), "Error writing B-tree pages to output stream.");
            CBTREEDB_ASSERT(os.tellp() == std::ostream::pos_type(SignaturePageSize + m_signature.btree_size));

            // prepare for writing value area

            m_ostream = &os;
            m_currpos = 0;
            m_curroffset = 0;
            m_datasha.clear();
        }

        void WriteValue(const key_type& key, const void* data, uint32_t size)
        {
            CBTREEDB_CHECK(m_ostream != NULL,
                           "Cannot write data, because phase 2 was not started.");

            CBTREEDB_CHECK(m_currpos < m_btree.Size(),
                           "Invalid key in WriteData() beyond end of predeclaration.");

            CBTREEDB_CHECK(m_ostream->tellp() == std::ostream::pos_type(SignaturePageSize + m_signature.btree_size + m_curroffset),
                           "Output stream data position is incorrect.");

            key_type expectedkey;
            uint32_t expectedsize;

            m_btree.GetIndex(m_currpos, expectedkey, expectedsize);

            CBTREEDB_CHECK(!m_key_less(key, expectedkey) && !m_key_less(expectedkey, key), // test equality
                           "Key in WriteData() mismatches predeclared sequence.");

            CBTREEDB_CHECK(size == expectedsize,
                           "Value data size in WriteData() mismatches predeclared sequence.");

            m_ostream->write(reinterpret_cast<const char*>(data), size);
            CBTREEDB_CHECK(m_ostream->good(), "Error writing data blocks to output stream.");

            m_datasha.update(data, size);

            ++m_currpos;
            m_curroffset += size;
        }

        void WriteValue(const key_type& key, const std::string& data)
        {
            return WriteValue(key, data.data(), data.size());
        }

        void WriteFinalize()
        {
            CBTREEDB_CHECK(m_ostream != NULL,
                           "Cannot write data, because phase 2 was not started.");

            CBTREEDB_CHECK(m_currpos == m_btree.Size(),
                           "WriteFinalize() called before end of predeclared sequence.");

            CBTREEDB_CHECK(m_ostream->tellp() == std::ostream::pos_type(SignaturePageSize + m_signature.btree_size + m_signature.value_size),
                           "Output stream data position is incorrect.");

            // Fill in signature page

            m_signature.value_offset = SignaturePageSize + m_signature.btree_size;
            m_datasha.final(m_signature.value_sha256);

            // Calculate header checksum

            m_signature.header_crc32 = CRC32::digest(reinterpret_cast<char*>(&m_signature)+16, sizeof(m_signature)-16);

            m_ostream->seekp(0, std::ios::beg);
            CBTREEDB_CHECK(m_ostream->good(), "Error seeking back to signature page in output stream.");

            m_ostream->write(reinterpret_cast<char*>(&m_signature), sizeof(m_signature));
            CBTREEDB_CHECK(m_ostream->good(), "Error writing signature page to output stream.");

            m_ostream->flush();
            m_ostream = NULL;
        }
    };
};

} // namespace stx

#endif // _STX_CBTREEDB_H_

---