<commit_before>// Copyright (c) 2012, Cornell University
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are met:
//
//     * Redistributions of source code must retain the above copyright notice,
//       this list of conditions and the following disclaimer.
//     * Redistributions in binary form must reproduce the above copyright
//       notice, this list of conditions and the following disclaimer in the
//       documentation and/or other materials provided with the distribution.
//     * Neither the name of HyperDex nor the names of its contributors may be
//       used to endorse or promote products derived from this software without
//       specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
// AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
// ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
// LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
// CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
// SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
// CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
// ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
// POSSIBILITY OF SUCH DAMAGE.

#define __STDC_LIMIT_MACROS

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

// STL
#include <string>

// Google Log
#include <glog/logging.h>

// LevelDB
#include <leveldb/write_batch.h>
#include <leveldb/filter_policy.h>

// e
#include <e/endian.h>

// HyperDex
#include "common/macros.h"
#include "common/serialization.h"
#include "daemon/daemon.h"
#include "daemon/datalayer.h"
#include "daemon/index_encode.h"
#include "datatypes/apply.h"
#include "datatypes/microerror.h"

using std::tr1::placeholders::_1;
using hyperdex::datalayer;
using hyperdex::reconfigure_returncode;

class datalayer::search_filter
{
    public:
        static bool sort_by_size(const datalayer::search_filter& lhs, const datalayer::search_filter& rhs);

    public:
        search_filter();
        search_filter(const search_filter&);
        ~search_filter() throw ();

    public:
        search_filter& operator = (const search_filter&);

    public:
        leveldb::Range range;
        void (*parse)(const leveldb::Slice& in, e::slice* out);
        uint64_t size;
        leveldb::Iterator* iter;
};

bool
datalayer :: search_filter :: sort_by_size(const datalayer::search_filter& lhs,
                                           const datalayer::search_filter& rhs)
{
    return lhs.size < rhs.size;
}

datalayer :: search_filter :: search_filter()
    : range()
    , parse(NULL)
    , size(0)
    , iter(NULL)
{
}

datalayer :: search_filter :: search_filter(const search_filter& other)
    : range(other.range)
    , parse(other.parse)
    , size(other.size)
    , iter(other.iter)
{
}

datalayer :: search_filter :: ~search_filter() throw ()
{
}

datalayer::search_filter&
datalayer :: search_filter :: operator = (const search_filter& rhs)
{
    range = rhs.range;
    size = rhs.size;
    iter = rhs.iter;
    return *this;
}

static void
parse_index_empty(const leveldb::Slice&, e::slice* k)
{
    *k = e::slice("", 0);
}

static void
parse_index_string(const leveldb::Slice& s, e::slice* k)
{
    const uint8_t* ptr = reinterpret_cast<const uint8_t*>(s.data());
    const uint8_t* end = ptr + s.size();
    uint32_t sz;
    e::unpack32be(end - sizeof(uint32_t), &sz);
    ptr = end - sizeof(uint32_t) - sz;
    *k = e::slice(ptr, sz);
}

static void
parse_index_sizeof8(const leveldb::Slice& s, e::slice* k)
{
    const uint8_t* ptr = reinterpret_cast<const uint8_t*>(s.data());
    const uint8_t* end = ptr + s.size();
    ptr += sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint32_t)
         + sizeof(uint16_t) + sizeof(uint8_t) + sizeof(uint16_t)
         + sizeof(uint64_t);
    *k = e::slice(ptr, end - ptr);
}

datalayer :: datalayer(daemon* d)
    : m_daemon(d)
    , m_db(NULL)
    , m_counters()
    , m_cleaner(std::tr1::bind(&datalayer::cleaner, this))
    , m_block_cleaner()
    , m_wakeup_cleaner(&m_block_cleaner)
    , m_need_cleaning(false)
    , m_shutdown(false)
{
    m_cleaner.start();
}

datalayer :: ~datalayer() throw ()
{
    // Intentionally leak m_db if destructed.  It must be released from
    // "teardown" to ensure we clean up outstanding snapshots first.
    shutdown();
}

bool
datalayer :: setup(const po6::pathname& path,
                   bool* saved,
                   server_id* saved_us,
                   po6::net::location* saved_bind_to,
                   po6::net::hostname* saved_coordinator)
{
    leveldb::Options opts;
    opts.create_if_missing = true;
    opts.filter_policy = leveldb::NewBloomFilterPolicy(10);
    std::string name(path.get());
    leveldb::Status st = leveldb::DB::Open(opts, name, &m_db);

    if (!st.ok())
    {
        LOG(ERROR) << "could not open LevelDB: " << st.ToString();
        return false;
    }

    leveldb::ReadOptions ropts;
    ropts.fill_cache = true;
    ropts.verify_checksums = true;
    leveldb::WriteOptions wopts;
    wopts.sync = false;

    leveldb::Slice rk("hyperdex", 8);
    std::string rbacking;
    st = m_db->Get(ropts, rk, &rbacking);
    bool first_time = false;

    if (st.ok())
    {
        first_time = false;

        if (rbacking != PACKAGE_VERSION)
        {
            LOG(ERROR) << "could not restore from LevelDB because "
                       << "the existing data was created by "
                       << "HyperDex " << rbacking << " but "
                       << "this is version " << PACKAGE_VERSION;
            return false;
        }
    }
    else if (st.IsNotFound())
    {
        first_time = true;
        leveldb::Slice k("hyperdex", 8);
        leveldb::Slice v(PACKAGE_VERSION, strlen(PACKAGE_VERSION));
        st = m_db->Put(wopts, k, v);

        if (st.ok())
        {
            // fall through
        }
        else if (st.IsNotFound())
        {
            LOG(ERROR) << "could not restore from LevelDB because Put returned NotFound:  "
                       << st.ToString();
            return false;
        }
        else if (st.IsCorruption())
        {
            LOG(ERROR) << "could not restore from LevelDB because of corruption:  "
                       << st.ToString();
            return false;
        }
        else if (st.IsIOError())
        {
            LOG(ERROR) << "could not restore from LevelDB because of an IO error:  "
                       << st.ToString();
            return false;
        }
        else
        {
            LOG(ERROR) << "could not restore from LevelDB because it returned an "
                       << "unknown error that we don't know how to handle:  "
                       << st.ToString();
            return false;
        }
    }
    else if (st.IsCorruption())
    {
        LOG(ERROR) << "could not restore from LevelDB because of corruption:  "
                   << st.ToString();
        return false;
    }
    else if (st.IsIOError())
    {
        LOG(ERROR) << "could not restore from LevelDB because of an IO error:  "
                   << st.ToString();
        return false;
    }
    else
    {
        LOG(ERROR) << "could not restore from LevelDB because it returned an "
                   << "unknown error that we don't know how to handle:  "
                   << st.ToString();
        return false;
    }

    leveldb::Slice sk("state", 5);
    std::string sbacking;
    st = m_db->Get(ropts, sk, &sbacking);

    if (st.ok())
    {
        if (first_time)
        {
            LOG(ERROR) << "could not restore from LevelDB because a previous "
                       << "execution crashed and the database was tampered with; "
                       << "you're on your own with this one";
            return false;
        }
    }
    else if (st.IsNotFound())
    {
        if (!first_time)
        {
            LOG(ERROR) << "could not restore from LevelDB because a previous "
                       << "execution crashed; run the recovery program and try again";
            return false;
        }
    }
    else if (st.IsCorruption())
    {
        LOG(ERROR) << "could not restore from LevelDB because of corruption:  "
                   << st.ToString();
        return false;
    }
    else if (st.IsIOError())
    {
        LOG(ERROR) << "could not restore from LevelDB because of an IO error:  "
                   << st.ToString();
        return false;
    }
    else
    {
        LOG(ERROR) << "could not restore from LevelDB because it returned an "
                   << "unknown error that we don't know how to handle:  "
                   << st.ToString();
        return false;
    }

    if (first_time)
    {
        *saved = false;
        return true;
    }

    uint64_t us;
    *saved = true;
    e::unpacker up(sbacking.data(), sbacking.size());
    up = up >> us >> *saved_bind_to >> *saved_coordinator;
    *saved_us = server_id(us);

    if (up.error())
    {
        LOG(ERROR) << "could not restore from LevelDB because a previous "
                   << "execution saved invalid state; run the recovery program and try again";
        return false;
    }

    return true;
}

void
datalayer :: teardown()
{
    shutdown();

    if (m_db)
    {
        delete m_db;
        m_db = NULL;
    }
}

reconfigure_returncode
datalayer :: prepare(const configuration&,
                     const configuration&,
                     const server_id&)
{
    m_block_cleaner.lock();
    return RECONFIGURE_SUCCESS;
}

reconfigure_returncode
datalayer :: reconfigure(const configuration&,
                         const configuration& new_config,
                         const server_id&)
{
    std::vector<region_id> regions;
    new_config.captured_regions(m_daemon->m_us, &regions);
    std::sort(regions.begin(), regions.end());
    m_counters.adopt(regions);
    return RECONFIGURE_SUCCESS;
}

reconfigure_returncode
datalayer :: cleanup(const configuration&,
                     const configuration&,
                     const server_id&)
{
    m_wakeup_cleaner.broadcast();
    m_need_cleaning = true;
    m_block_cleaner.unlock();
    return RECONFIGURE_SUCCESS;
}

datalayer::returncode
datalayer :: get(const region_id& ri,
                 const e::slice& key,
                 std::vector<e::slice>* value,
                 uint64_t* version,
                 reference* ref)
{
    leveldb::ReadOptions opts;
    opts.fill_cache = true;
    opts.verify_checksums = true;
    std::vector<char> kbacking;
    leveldb::Slice lkey;
    encode_key(ri, key, &kbacking, &lkey);
    leveldb::Status st = m_db->Get(opts, lkey, &ref->m_backing);

    if (st.ok())
    {
        e::slice v(ref->m_backing.data(), ref->m_backing.size());
        return decode_value(v, value, version);
    }
    else if (st.IsNotFound())
    {
        return NOT_FOUND;
    }
    else if (st.IsCorruption())
    {
        LOG(ERROR) << "corruption at the disk layer: region=" << ri
                   << " key=0x" << key.hex() << " desc=" << st.ToString();
        return CORRUPTION;
    }
    else if (st.IsIOError())
    {
        LOG(ERROR) << "IO error at the disk layer: region=" << ri
                   << " key=0x" << key.hex() << " desc=" << st.ToString();
        return IO_ERROR;
    }
    else
    {
        LOG(ERROR) << "LevelDB returned an unknown error that we don't know how to handle";
        return LEVELDB_ERROR;
    }
}

datalayer::returncode
datalayer :: put(const region_id& ri,
                 uint64_t seq_id,
                 const e::slice& key,
                 const std::vector<e::slice>& value,
                 uint64_t version)
{
    const schema* sc = m_daemon->m_config.get_schema(ri);
    assert(sc);
    leveldb::WriteOptions opts;
    opts.sync = false;
    std::vector<char> kbacking;
    leveldb::Slice lkey;
    encode_key(ri, key, &kbacking, &lkey);
    std::vector<char> vbacking;
    leveldb::Slice lvalue;
    encode_value(value, version, &vbacking, &lvalue);
    backing_t backing;
    leveldb::WriteBatch updates;
    updates.Put(lkey, lvalue);
    returncode rc = create_index_changes(sc, ri, key, lkey, value, &backing, &updates);

    if (rc != SUCCESS)
    {
        return rc;
    }

    // Mark acked as part of this batch write
    if (seq_id != 0)
    {
        char abacking[sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t)];
        abacking[0] = 'a';
        e::pack64be(ri.get(), abacking + sizeof(uint8_t));
        e::pack64be(seq_id, abacking + sizeof(uint8_t) + sizeof(uint64_t));
        leveldb::Slice akey(abacking, sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t));
        leveldb::Slice aval("", 0);
        updates.Put(akey, aval);
    }

    uint64_t count;

    // If this is a captured region, then we must log this transfer
    if (m_counters.lookup(ri, &count))
    {
        capture_id cid = m_daemon->m_config.capture_for(ri);
        assert(cid != capture_id());
        backing.push_back(std::vector<char>());
        leveldb::Slice tkey;
        encode_transfer(cid, count, &backing.back(), &tkey);
        backing.push_back(std::vector<char>());
        leveldb::Slice tvalue;
        encode_key_value(key, &value, version, &backing.back(), &tvalue);
        updates.Put(tkey, tvalue);
    }

    // Perform the write
    leveldb::Status st = m_db->Write(opts, &updates);

    if (st.ok())
    {
        return SUCCESS;
    }
    else if (st.IsNotFound())
    {
        LOG(ERROR) << "put returned NOT_FOUND at the disk layer: region=" << ri
                   << " key=0x" << key.hex() << " desc=" << st.ToString();
        return NOT_FOUND;
    }
    else if (st.IsCorruption())
    {
        LOG(ERROR) << "corruption at the disk layer: region=" << ri
                   << " key=0x" << key.hex() << " desc=" << st.ToString();
        return CORRUPTION;
    }
    else if (st.IsIOError())
    {
        LOG(ERROR) << "IO error at the disk layer: region=" << ri
                   << " key=0x" << key.hex() << " desc=" << st.ToString();
        return IO_ERROR;
    }
    else
    {
        LOG(ERROR) << "LevelDB returned an unknown error that we don't know how to handle";
        return LEVELDB_ERROR;
    }
}

datalayer::returncode
datalayer :: del(const region_id& ri,
                 uint64_t seq_id,
                 const e::slice& key)
{
    const schema* sc = m_daemon->m_config.get_schema(ri);
    assert(sc);
    leveldb::WriteOptions opts;
    opts.sync = false;
    std::vector<char> kbacking;
    leveldb::Slice lkey;
    encode_key(ri, key, &kbacking, &lkey);
    backing_t backing;
    leveldb::WriteBatch updates;
    updates.Delete(lkey);
    returncode rc = create_index_changes(sc, ri, key, lkey, &backing, &updates);

    if (rc != SUCCESS)
    {
        return rc;
    }

    // Mark acked as part of this batch write
    if (seq_id != 0)
    {
        char abacking[sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t)];
        abacking[0] = 'a';
        e::pack64be(ri.get(), abacking + sizeof(uint8_t));
        e::pack64be(seq_id, abacking + sizeof(uint8_t) + sizeof(uint64_t));
        leveldb::Slice akey(abacking, sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t));
        leveldb::Slice aval("", 0);
        updates.Put(akey, aval);
    }

    uint64_t count;

    // If this is a captured region, then we must log this transfer
    if (m_counters.lookup(ri, &count))
    {
        capture_id cid = m_daemon->m_config.capture_for(ri);
        assert(cid != capture_id());
        backing.push_back(std::vector<char>());
        leveldb::Slice tkey;
        encode_transfer(cid, count, &backing.back(), &tkey);
        backing.push_back(std::vector<char>());
        leveldb::Slice tvalue;
        encode_key_value(key, NULL, 0, &backing.back(), &tvalue);
        updates.Put(tkey, tvalue);
    }

    // Perform the write
    leveldb::Status st = m_db->Write(opts, &updates);

    if (st.ok())
    {
        return SUCCESS;
    }
    else if (st.IsNotFound())
    {
        return NOT_FOUND;
    }
    else if (st.IsCorruption())
    {
        LOG(ERROR) << "corruption at the disk layer: region=" << ri
                   << " key=0x" << key.hex() << " desc=" << st.ToString();
        return CORRUPTION;
    }
    else if (st.IsIOError())
    {
        LOG(ERROR) << "IO error at the disk layer: region=" << ri
                   << " key=0x" << key.hex() << " desc=" << st.ToString();
        return IO_ERROR;
    }
    else
    {
        LOG(ERROR) << "LevelDB returned an unknown error that we don't know how to handle";
        return LEVELDB_ERROR;
    }
}

datalayer::returncode
datalayer :: make_snapshot(const region_id& ri,
                           attribute_check* checks, size_t checks_sz,
                           snapshot* snap)
{
    std::stable_sort(checks, checks + checks_sz);
    attribute_check* check_ptr = checks;
    attribute_check* check_end = checks + checks_sz;

    while (check_ptr < check_end)
    {
        attribute_check* check_tmp = check_ptr;

        while (check_tmp < check_end && check_tmp->attr == check_ptr->attr)
        {
            ++check_tmp;
        }

        generate_search_filters(ri, check_ptr, check_tmp, &snap->m_backing, &snap->m_sfs);
        check_ptr = check_tmp;
    }

    std::vector<leveldb::Range> ranges;
    ranges.reserve(snap->m_sfs.size() + 1);
    std::vector<uint64_t> sizes(snap->m_sfs.size() + 1, 0);

    for (size_t i = 0; i < snap->m_sfs.size(); ++i)
    {
        ranges.push_back(snap->m_sfs[i].range);
    }

    ranges.push_back(leveldb::Range());
    generate_object_range(ri, &snap->m_backing, &ranges[snap->m_sfs.size()]);
    m_db->GetApproximateSizes(&ranges.front(), ranges.size(), &sizes.front());
    size_t sum = 0;

    for (size_t i = 0; i < snap->m_sfs.size(); ++i)
    {
        sum += sizes[i];
        snap->m_sfs[i].size = sizes[i];
    }

    std::sort(snap->m_sfs.begin(), snap->m_sfs.end(), search_filter::sort_by_size);

    while (!snap->m_sfs.empty() && sum > sizes[snap->m_sfs.size()] / 10)
    {
        sum -= snap->m_sfs.back().size;
        snap->m_sfs.pop_back();
    }

    snap->m_dl = this;
    snap->m_snap = m_db->GetSnapshot();
    snap->m_checks = checks;
    snap->m_checks_sz = checks_sz;
    leveldb::ReadOptions opts;
    opts.fill_cache = false;
    opts.verify_checksums = true;
    opts.snapshot = snap->m_snap;
    snap->m_ri = ri;
    snap->m_obj_range = ranges.back();
    snap->m_iter = m_db->NewIterator(opts);
    snap->m_iter->Seek(snap->m_obj_range.start);

    snap->m_iter = m_db->NewIterator(opts);
    snap->m_iter->Seek(snap->m_obj_range.start);

    for (size_t i = 0; i < snap->m_sfs.size(); ++i)
    {
        snap->m_sfs[i].iter = m_db->NewIterator(opts);
        snap->m_sfs[i].iter->Seek(snap->m_sfs[i].range.start);
    }

    return SUCCESS;
}

std::tr1::shared_ptr<leveldb::Snapshot>
datalayer :: make_raw_snapshot()
{
    std::tr1::function<void (const leveldb::Snapshot*)> dtor;
    dtor = std::tr1::bind(&leveldb::DB::ReleaseSnapshot, m_db, _1);
    std::tr1::shared_ptr<leveldb::Snapshot> ret(m_db->GetSnapshot(), dtor);
    return ret;
}

void
datalayer :: make_region_iterator(region_iterator* riter,
                                  std::tr1::shared_ptr<leveldb::Snapshot> snap,
                                  const region_id& ri)
{
    riter->m_dl = this;
    riter->m_snap = snap;
    riter->m_region = ri;
    leveldb::ReadOptions opts;
    opts.fill_cache = true;
    opts.verify_checksums = true;
    opts.snapshot = riter->m_snap.get();
    riter->m_iter.reset(m_db->NewIterator(opts));
    char backing[sizeof(uint8_t) + sizeof(uint64_t)];
    char* ptr = backing;
    ptr = e::pack8be('o', ptr);
    ptr = e::pack64be(riter->m_region.get(), ptr);
    riter->m_iter->Seek(leveldb::Slice(backing, sizeof(uint8_t) + sizeof(uint64_t)));
}

datalayer::returncode
datalayer :: get_transfer(const region_id& ri,
                          uint64_t seq_no,
                          bool* has_value,
                          e::slice* key,
                          std::vector<e::slice>* value,
                          uint64_t* version,
                          reference* ref)
{
    leveldb::ReadOptions opts;
    opts.fill_cache = true;
    opts.verify_checksums = true;
    std::vector<char> kbacking;
    leveldb::Slice lkey;
    capture_id cid = m_daemon->m_config.capture_for(ri);
    assert(cid != capture_id());
    encode_transfer(cid, seq_no, &kbacking, &lkey);
    leveldb::Status st = m_db->Get(opts, lkey, &ref->m_backing);

    if (st.ok())
    {
        e::slice v(ref->m_backing.data(), ref->m_backing.size());
        return decode_key_value(v, has_value, key, value, version);
    }
    else if (st.IsNotFound())
    {
        return NOT_FOUND;
    }
    else if (st.IsCorruption())
    {
        LOG(ERROR) << "corruption at the disk layer: region=" << ri
                   << " seq_no=" << seq_no << " desc=" << st.ToString();
        return CORRUPTION;
    }
    else if (st.IsIOError())
    {
        LOG(ERROR) << "IO error at the disk layer: region=" << ri
                   << " seq_no=" << seq_no << " desc=" << st.ToString();
        return IO_ERROR;
    }
    else
    {
        LOG(ERROR) << "LevelDB returned an unknown error that we don't know how to handle";
        return LEVELDB_ERROR;
    }
}

bool
datalayer :: check_acked(const region_id& reg_id, uint64_t seq_id)
{
    // make it so that increasing seq_ids are ordered in reverse in the KVS
    seq_id = UINT64_MAX - seq_id;
    leveldb::ReadOptions opts;
    opts.fill_cache = true;
    opts.verify_checksums = true;
    char backing[sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t)];
    backing[0] = 'a';
    e::pack64be(reg_id.get(), backing + sizeof(uint8_t));
    e::pack64be(seq_id, backing + sizeof(uint8_t) + sizeof(uint64_t));
    leveldb::Slice key(backing, sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t));
    std::string val;
    leveldb::Status st = m_db->Get(opts, key, &val);

    if (st.ok())
    {
        return true;
    }
    else if (st.IsNotFound())
    {
        return false;
    }
    else if (st.IsCorruption())
    {
        LOG(ERROR) << "corruption at the disk layer: region=" << reg_id
                   << " seq_id=" << seq_id << " desc=" << st.ToString();
        return false;
    }
    else if (st.IsIOError())
    {
        LOG(ERROR) << "IO error at the disk layer: region=" << reg_id
                   << " seq_id=" << seq_id << " desc=" << st.ToString();
        return false;
    }
    else
    {
        LOG(ERROR) << "LevelDB returned an unknown error that we don't know how to handle";
        return false;
    }
}

void
datalayer :: mark_acked(const region_id& reg_id, uint64_t seq_id)
{
    // make it so that increasing seq_ids are ordered in reverse in the KVS
    seq_id = UINT64_MAX - seq_id;
    leveldb::WriteOptions opts;
    opts.sync = false;
    char backing[sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t)];
    backing[0] = 'a';
    e::pack64be(reg_id.get(), backing + sizeof(uint8_t));
    e::pack64be(seq_id, backing + sizeof(uint8_t) + sizeof(uint64_t));
    leveldb::Slice key(backing, sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t));
    leveldb::Slice val("", 0);
    leveldb::Status st = m_db->Put(opts, key, val);

    if (st.ok())
    {
        // Yay!
    }
    else if (st.IsNotFound())
    {
        LOG(ERROR) << "mark_acked returned NOT_FOUND at the disk layer: region=" << reg_id
                   << " seq_id=" << seq_id << " desc=" << st.ToString();
    }
    else if (st.IsCorruption())
    {
        LOG(ERROR) << "corruption at the disk layer: region=" << reg_id
                   << " seq_id=" << seq_id << " desc=" << st.ToString();
    }
    else if (st.IsIOError())
    {
        LOG(ERROR) << "IO error at the disk layer: region=" << reg_id
                   << " seq_id=" << seq_id << " desc=" << st.ToString();
    }
    else
    {
        LOG(ERROR) << "LevelDB returned an unknown error that we don't know how to handle";
    }
}

void
datalayer :: max_seq_id(const region_id& reg_id, uint64_t* seq_id)
{
    leveldb::ReadOptions opts;
    opts.fill_cache = false;
    opts.verify_checksums = true;
    opts.snapshot = NULL;
    std::auto_ptr<leveldb::Iterator> it(m_db->NewIterator(opts));
    char backing[sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t)];
    backing[0] = 'a';
    e::pack64be(reg_id.get(), backing + sizeof(uint8_t));
    e::pack64be(0, backing + sizeof(uint8_t) + sizeof(uint64_t));
    leveldb::Slice key(backing, sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t));
    it->Seek(key);

    if (!it->Valid())
    {
        *seq_id = 0;
        return;
    }

    key = it->key();
    e::unpacker up(key.data(), key.size());
    uint8_t p = '\0';
    uint64_t r;
    uint64_t s;
    up = up >> p >> r >> s;

    if (up.error() || p != 'a')
    {
        *seq_id = 0;
        return;
    }

    *seq_id = s;
}

void
datalayer :: encode_key(const region_id& ri,
                        const e::slice& key,
                        std::vector<char>* kbacking,
                        leveldb::Slice* lkey)
{
    size_t sz = sizeof(uint8_t) + sizeof(uint64_t) + key.size();
    kbacking->resize(sz);
    char* ptr = &kbacking->front();
    ptr = e::pack8be('o', ptr);
    ptr = e::pack64be(ri.get(), ptr);
    memmove(ptr, key.data(), key.size());
    *lkey = leveldb::Slice(&kbacking->front(), sz);
}

datalayer::returncode
datalayer :: decode_key(const e::slice& lkey,
                        region_id* ri,
                        e::slice* key)
{
    const uint8_t* ptr = lkey.data();
    const uint8_t* end = ptr + lkey.size();

    if (ptr >= end || *ptr != 'o')
    {
        return BAD_ENCODING;
    }

    ++ptr;
    uint64_t rid;

    if (ptr + sizeof(uint64_t) <= end)
    {
        ptr = e::unpack64be(ptr, &rid);
    }
    else
    {
        return BAD_ENCODING;
    }

    *ri  = region_id(rid);
    *key = e::slice(ptr, end - ptr);
    return SUCCESS;
}

void
datalayer :: encode_value(const std::vector<e::slice>& attrs,
                          uint64_t version,
                          std::vector<char>* backing,
                          leveldb::Slice* lvalue)
{
    assert(attrs.size() < 65536);
    size_t sz = sizeof(uint64_t) + sizeof(uint16_t);

    for (size_t i = 0; i < attrs.size(); ++i)
    {
        sz += sizeof(uint32_t) + attrs[i].size();
    }

    backing->resize(sz);
    char* ptr = &backing->front();
    ptr = e::pack64be(version, ptr);
    ptr = e::pack16be(attrs.size(), ptr);

    for (size_t i = 0; i < attrs.size(); ++i)
    {
        ptr = e::pack32be(attrs[i].size(), ptr);
        memmove(ptr, attrs[i].data(), attrs[i].size());
        ptr += attrs[i].size();
    }

    *lvalue = leveldb::Slice(&backing->front(), sz);
}

datalayer::returncode
datalayer :: decode_value(const e::slice& value,
                          std::vector<e::slice>* attrs,
                          uint64_t* version)
{
    const uint8_t* ptr = value.data();
    const uint8_t* end = ptr + value.size();

    if (ptr + sizeof(uint64_t) <= end)
    {
        ptr = e::unpack64be(ptr, version);
    }
    else
    {
        return BAD_ENCODING;
    }

    uint16_t num_attrs;

    if (ptr + sizeof(uint64_t) <= end)
    {
        ptr = e::unpack16be(ptr, &num_attrs);
    }
    else
    {
        return BAD_ENCODING;
    }

    attrs->clear();

    for (size_t i = 0; i < num_attrs; ++i)
    {
        uint32_t sz = 0;

        if (ptr + sizeof(uint32_t) <= end)
        {
            ptr = e::unpack32be(ptr, &sz);
        }
        else
        {
            return BAD_ENCODING;
        }

        e::slice s(reinterpret_cast<const uint8_t*>(ptr), sz);
        ptr += sz;
        attrs->push_back(s);
    }

    return SUCCESS;
}

void
datalayer :: encode_transfer(const capture_id& ci,
                             uint64_t count,
                             std::vector<char>* backing,
                             leveldb::Slice* tkey)
{
    backing->resize(sizeof(uint8_t) + 2 * sizeof(uint64_t));
    char* tmp = &backing->front();
    tmp = e::pack8be('t', tmp);
    tmp = e::pack64be(ci.get(), tmp);
    tmp = e::pack64be(count, tmp);
    *tkey = leveldb::Slice(&backing->front(), sizeof(uint8_t) + 2 * sizeof(uint64_t));
}

void
datalayer :: encode_key_value(const e::slice& key,
                              const std::vector<e::slice>* value,
                              uint64_t version,
                              std::vector<char>* backing,
                              leveldb::Slice* slice)
{
    size_t sz = sizeof(uint32_t) + key.size() + sizeof(uint64_t) + sizeof(uint16_t);

    for (size_t i = 0; value && i < value->size(); ++i)
    {
        sz += sizeof(uint32_t) + (*value)[i].size();
    }

    backing->resize(sz);
    char* ptr = &backing->front();
    *slice = leveldb::Slice(ptr, sz);
    ptr = e::pack32be(key.size(), ptr);
    memmove(ptr, key.data(), key.size());
    ptr += key.size();
    ptr = e::pack64be(version, ptr);

    if (value)
    {
        ptr = e::pack16be(value->size(), ptr);

        for (size_t i = 0; i < value->size(); ++i)
        {
            ptr = e::pack32be((*value)[i].size(), ptr);
            memmove(ptr, (*value)[i].data(), (*value)[i].size());
            ptr += (*value)[i].size();
        }
    }
}

datalayer::returncode
datalayer :: decode_key_value(const e::slice& slice,
                              bool* has_value,
                              e::slice* key,
                              std::vector<e::slice>* value,
                              uint64_t* version)
{
    const uint8_t* ptr = slice.data();
    const uint8_t* end = ptr + slice.size();

    if (ptr >= end)
    {
        return BAD_ENCODING;
    }

    uint32_t key_sz;

    if (ptr + sizeof(uint32_t) <= end)
    {
        ptr = e::unpack32be(ptr, &key_sz);
    }
    else
    {
        return BAD_ENCODING;
    }

    if (ptr + key_sz <= end)
    {
        *key = e::slice(ptr, key_sz);
        ptr += key_sz;
    }
    else
    {
        return BAD_ENCODING;
    }

    if (ptr + sizeof(uint64_t) <= end)
    {
        ptr = e::unpack64be(ptr, version);
    }
    else
    {
        return BAD_ENCODING;
    }

    if (ptr == end)
    {
        *has_value = false;
        return SUCCESS;
    }

    uint16_t num_attrs;

    if (ptr + sizeof(uint16_t) <= end)
    {
        ptr = e::unpack16be(ptr, &num_attrs);
    }
    else
    {
        return BAD_ENCODING;
    }

    value->clear();

    for (size_t i = 0; i < num_attrs; ++i)
    {
        uint32_t sz = 0;

        if (ptr + sizeof(uint32_t) <= end)
        {
            ptr = e::unpack32be(ptr, &sz);
        }
        else
        {
            return BAD_ENCODING;
        }

        e::slice s(ptr, sz);
        ptr += sz;
        value->push_back(s);
    }

    return SUCCESS;
}

void
datalayer :: generate_object_range(const region_id& ri,
                                   backing_t* backing,<commit_msg>Set has_value to true in decode_key_value<commit_after>// Copyright (c) 2012, Cornell University
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are met:
//
//     * Redistributions of source code must retain the above copyright notice,
//       this list of conditions and the following disclaimer.
//     * Redistributions in binary form must reproduce the above copyright
//       notice, this list of conditions and the following disclaimer in the
//       documentation and/or other materials provided with the distribution.
//     * Neither the name of HyperDex nor the names of its contributors may be
//       used to endorse or promote products derived from this software without
//       specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
// AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
// ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
// LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
// CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
// SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
// CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
// ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
// POSSIBILITY OF SUCH DAMAGE.

#define __STDC_LIMIT_MACROS

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

// STL
#include <string>

// Google Log
#include <glog/logging.h>

// LevelDB
#include <leveldb/write_batch.h>
#include <leveldb/filter_policy.h>

// e
#include <e/endian.h>

// HyperDex
#include "common/macros.h"
#include "common/serialization.h"
#include "daemon/daemon.h"
#include "daemon/datalayer.h"
#include "daemon/index_encode.h"
#include "datatypes/apply.h"
#include "datatypes/microerror.h"

using std::tr1::placeholders::_1;
using hyperdex::datalayer;
using hyperdex::reconfigure_returncode;

class datalayer::search_filter
{
    public:
        static bool sort_by_size(const datalayer::search_filter& lhs, const datalayer::search_filter& rhs);

    public:
        search_filter();
        search_filter(const search_filter&);
        ~search_filter() throw ();

    public:
        search_filter& operator = (const search_filter&);

    public:
        leveldb::Range range;
        void (*parse)(const leveldb::Slice& in, e::slice* out);
        uint64_t size;
        leveldb::Iterator* iter;
};

bool
datalayer :: search_filter :: sort_by_size(const datalayer::search_filter& lhs,
                                           const datalayer::search_filter& rhs)
{
    return lhs.size < rhs.size;
}

datalayer :: search_filter :: search_filter()
    : range()
    , parse(NULL)
    , size(0)
    , iter(NULL)
{
}

datalayer :: search_filter :: search_filter(const search_filter& other)
    : range(other.range)
    , parse(other.parse)
    , size(other.size)
    , iter(other.iter)
{
}

datalayer :: search_filter :: ~search_filter() throw ()
{
}

datalayer::search_filter&
datalayer :: search_filter :: operator = (const search_filter& rhs)
{
    range = rhs.range;
    size = rhs.size;
    iter = rhs.iter;
    return *this;
}

static void
parse_index_empty(const leveldb::Slice&, e::slice* k)
{
    *k = e::slice("", 0);
}

static void
parse_index_string(const leveldb::Slice& s, e::slice* k)
{
    const uint8_t* ptr = reinterpret_cast<const uint8_t*>(s.data());
    const uint8_t* end = ptr + s.size();
    uint32_t sz;
    e::unpack32be(end - sizeof(uint32_t), &sz);
    ptr = end - sizeof(uint32_t) - sz;
    *k = e::slice(ptr, sz);
}

static void
parse_index_sizeof8(const leveldb::Slice& s, e::slice* k)
{
    const uint8_t* ptr = reinterpret_cast<const uint8_t*>(s.data());
    const uint8_t* end = ptr + s.size();
    ptr += sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint32_t)
         + sizeof(uint16_t) + sizeof(uint8_t) + sizeof(uint16_t)
         + sizeof(uint64_t);
    *k = e::slice(ptr, end - ptr);
}

datalayer :: datalayer(daemon* d)
    : m_daemon(d)
    , m_db(NULL)
    , m_counters()
    , m_cleaner(std::tr1::bind(&datalayer::cleaner, this))
    , m_block_cleaner()
    , m_wakeup_cleaner(&m_block_cleaner)
    , m_need_cleaning(false)
    , m_shutdown(false)
{
    m_cleaner.start();
}

datalayer :: ~datalayer() throw ()
{
    // Intentionally leak m_db if destructed.  It must be released from
    // "teardown" to ensure we clean up outstanding snapshots first.
    shutdown();
}

bool
datalayer :: setup(const po6::pathname& path,
                   bool* saved,
                   server_id* saved_us,
                   po6::net::location* saved_bind_to,
                   po6::net::hostname* saved_coordinator)
{
    leveldb::Options opts;
    opts.create_if_missing = true;
    opts.filter_policy = leveldb::NewBloomFilterPolicy(10);
    std::string name(path.get());
    leveldb::Status st = leveldb::DB::Open(opts, name, &m_db);

    if (!st.ok())
    {
        LOG(ERROR) << "could not open LevelDB: " << st.ToString();
        return false;
    }

    leveldb::ReadOptions ropts;
    ropts.fill_cache = true;
    ropts.verify_checksums = true;
    leveldb::WriteOptions wopts;
    wopts.sync = false;

    leveldb::Slice rk("hyperdex", 8);
    std::string rbacking;
    st = m_db->Get(ropts, rk, &rbacking);
    bool first_time = false;

    if (st.ok())
    {
        first_time = false;

        if (rbacking != PACKAGE_VERSION)
        {
            LOG(ERROR) << "could not restore from LevelDB because "
                       << "the existing data was created by "
                       << "HyperDex " << rbacking << " but "
                       << "this is version " << PACKAGE_VERSION;
            return false;
        }
    }
    else if (st.IsNotFound())
    {
        first_time = true;
        leveldb::Slice k("hyperdex", 8);
        leveldb::Slice v(PACKAGE_VERSION, strlen(PACKAGE_VERSION));
        st = m_db->Put(wopts, k, v);

        if (st.ok())
        {
            // fall through
        }
        else if (st.IsNotFound())
        {
            LOG(ERROR) << "could not restore from LevelDB because Put returned NotFound:  "
                       << st.ToString();
            return false;
        }
        else if (st.IsCorruption())
        {
            LOG(ERROR) << "could not restore from LevelDB because of corruption:  "
                       << st.ToString();
            return false;
        }
        else if (st.IsIOError())
        {
            LOG(ERROR) << "could not restore from LevelDB because of an IO error:  "
                       << st.ToString();
            return false;
        }
        else
        {
            LOG(ERROR) << "could not restore from LevelDB because it returned an "
                       << "unknown error that we don't know how to handle:  "
                       << st.ToString();
            return false;
        }
    }
    else if (st.IsCorruption())
    {
        LOG(ERROR) << "could not restore from LevelDB because of corruption:  "
                   << st.ToString();
        return false;
    }
    else if (st.IsIOError())
    {
        LOG(ERROR) << "could not restore from LevelDB because of an IO error:  "
                   << st.ToString();
        return false;
    }
    else
    {
        LOG(ERROR) << "could not restore from LevelDB because it returned an "
                   << "unknown error that we don't know how to handle:  "
                   << st.ToString();
        return false;
    }

    leveldb::Slice sk("state", 5);
    std::string sbacking;
    st = m_db->Get(ropts, sk, &sbacking);

    if (st.ok())
    {
        if (first_time)
        {
            LOG(ERROR) << "could not restore from LevelDB because a previous "
                       << "execution crashed and the database was tampered with; "
                       << "you're on your own with this one";
            return false;
        }
    }
    else if (st.IsNotFound())
    {
        if (!first_time)
        {
            LOG(ERROR) << "could not restore from LevelDB because a previous "
                       << "execution crashed; run the recovery program and try again";
            return false;
        }
    }
    else if (st.IsCorruption())
    {
        LOG(ERROR) << "could not restore from LevelDB because of corruption:  "
                   << st.ToString();
        return false;
    }
    else if (st.IsIOError())
    {
        LOG(ERROR) << "could not restore from LevelDB because of an IO error:  "
                   << st.ToString();
        return false;
    }
    else
    {
        LOG(ERROR) << "could not restore from LevelDB because it returned an "
                   << "unknown error that we don't know how to handle:  "
                   << st.ToString();
        return false;
    }

    if (first_time)
    {
        *saved = false;
        return true;
    }

    uint64_t us;
    *saved = true;
    e::unpacker up(sbacking.data(), sbacking.size());
    up = up >> us >> *saved_bind_to >> *saved_coordinator;
    *saved_us = server_id(us);

    if (up.error())
    {
        LOG(ERROR) << "could not restore from LevelDB because a previous "
                   << "execution saved invalid state; run the recovery program and try again";
        return false;
    }

    return true;
}

void
datalayer :: teardown()
{
    shutdown();

    if (m_db)
    {
        delete m_db;
        m_db = NULL;
    }
}

reconfigure_returncode
datalayer :: prepare(const configuration&,
                     const configuration&,
                     const server_id&)
{
    m_block_cleaner.lock();
    return RECONFIGURE_SUCCESS;
}

reconfigure_returncode
datalayer :: reconfigure(const configuration&,
                         const configuration& new_config,
                         const server_id&)
{
    std::vector<region_id> regions;
    new_config.captured_regions(m_daemon->m_us, &regions);
    std::sort(regions.begin(), regions.end());
    m_counters.adopt(regions);
    return RECONFIGURE_SUCCESS;
}

reconfigure_returncode
datalayer :: cleanup(const configuration&,
                     const configuration&,
                     const server_id&)
{
    m_wakeup_cleaner.broadcast();
    m_need_cleaning = true;
    m_block_cleaner.unlock();
    return RECONFIGURE_SUCCESS;
}

datalayer::returncode
datalayer :: get(const region_id& ri,
                 const e::slice& key,
                 std::vector<e::slice>* value,
                 uint64_t* version,
                 reference* ref)
{
    leveldb::ReadOptions opts;
    opts.fill_cache = true;
    opts.verify_checksums = true;
    std::vector<char> kbacking;
    leveldb::Slice lkey;
    encode_key(ri, key, &kbacking, &lkey);
    leveldb::Status st = m_db->Get(opts, lkey, &ref->m_backing);

    if (st.ok())
    {
        e::slice v(ref->m_backing.data(), ref->m_backing.size());
        return decode_value(v, value, version);
    }
    else if (st.IsNotFound())
    {
        return NOT_FOUND;
    }
    else if (st.IsCorruption())
    {
        LOG(ERROR) << "corruption at the disk layer: region=" << ri
                   << " key=0x" << key.hex() << " desc=" << st.ToString();
        return CORRUPTION;
    }
    else if (st.IsIOError())
    {
        LOG(ERROR) << "IO error at the disk layer: region=" << ri
                   << " key=0x" << key.hex() << " desc=" << st.ToString();
        return IO_ERROR;
    }
    else
    {
        LOG(ERROR) << "LevelDB returned an unknown error that we don't know how to handle";
        return LEVELDB_ERROR;
    }
}

datalayer::returncode
datalayer :: put(const region_id& ri,
                 uint64_t seq_id,
                 const e::slice& key,
                 const std::vector<e::slice>& value,
                 uint64_t version)
{
    const schema* sc = m_daemon->m_config.get_schema(ri);
    assert(sc);
    leveldb::WriteOptions opts;
    opts.sync = false;
    std::vector<char> kbacking;
    leveldb::Slice lkey;
    encode_key(ri, key, &kbacking, &lkey);
    std::vector<char> vbacking;
    leveldb::Slice lvalue;
    encode_value(value, version, &vbacking, &lvalue);
    backing_t backing;
    leveldb::WriteBatch updates;
    updates.Put(lkey, lvalue);
    returncode rc = create_index_changes(sc, ri, key, lkey, value, &backing, &updates);

    if (rc != SUCCESS)
    {
        return rc;
    }

    // Mark acked as part of this batch write
    if (seq_id != 0)
    {
        char abacking[sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t)];
        abacking[0] = 'a';
        e::pack64be(ri.get(), abacking + sizeof(uint8_t));
        e::pack64be(seq_id, abacking + sizeof(uint8_t) + sizeof(uint64_t));
        leveldb::Slice akey(abacking, sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t));
        leveldb::Slice aval("", 0);
        updates.Put(akey, aval);
    }

    uint64_t count;

    // If this is a captured region, then we must log this transfer
    if (m_counters.lookup(ri, &count))
    {
        capture_id cid = m_daemon->m_config.capture_for(ri);
        assert(cid != capture_id());
        backing.push_back(std::vector<char>());
        leveldb::Slice tkey;
        encode_transfer(cid, count, &backing.back(), &tkey);
        backing.push_back(std::vector<char>());
        leveldb::Slice tvalue;
        encode_key_value(key, &value, version, &backing.back(), &tvalue);
        updates.Put(tkey, tvalue);
    }

    // Perform the write
    leveldb::Status st = m_db->Write(opts, &updates);

    if (st.ok())
    {
        return SUCCESS;
    }
    else if (st.IsNotFound())
    {
        LOG(ERROR) << "put returned NOT_FOUND at the disk layer: region=" << ri
                   << " key=0x" << key.hex() << " desc=" << st.ToString();
        return NOT_FOUND;
    }
    else if (st.IsCorruption())
    {
        LOG(ERROR) << "corruption at the disk layer: region=" << ri
                   << " key=0x" << key.hex() << " desc=" << st.ToString();
        return CORRUPTION;
    }
    else if (st.IsIOError())
    {
        LOG(ERROR) << "IO error at the disk layer: region=" << ri
                   << " key=0x" << key.hex() << " desc=" << st.ToString();
        return IO_ERROR;
    }
    else
    {
        LOG(ERROR) << "LevelDB returned an unknown error that we don't know how to handle";
        return LEVELDB_ERROR;
    }
}

datalayer::returncode
datalayer :: del(const region_id& ri,
                 uint64_t seq_id,
                 const e::slice& key)
{
    const schema* sc = m_daemon->m_config.get_schema(ri);
    assert(sc);
    leveldb::WriteOptions opts;
    opts.sync = false;
    std::vector<char> kbacking;
    leveldb::Slice lkey;
    encode_key(ri, key, &kbacking, &lkey);
    backing_t backing;
    leveldb::WriteBatch updates;
    updates.Delete(lkey);
    returncode rc = create_index_changes(sc, ri, key, lkey, &backing, &updates);

    if (rc != SUCCESS)
    {
        return rc;
    }

    // Mark acked as part of this batch write
    if (seq_id != 0)
    {
        char abacking[sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t)];
        abacking[0] = 'a';
        e::pack64be(ri.get(), abacking + sizeof(uint8_t));
        e::pack64be(seq_id, abacking + sizeof(uint8_t) + sizeof(uint64_t));
        leveldb::Slice akey(abacking, sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t));
        leveldb::Slice aval("", 0);
        updates.Put(akey, aval);
    }

    uint64_t count;

    // If this is a captured region, then we must log this transfer
    if (m_counters.lookup(ri, &count))
    {
        capture_id cid = m_daemon->m_config.capture_for(ri);
        assert(cid != capture_id());
        backing.push_back(std::vector<char>());
        leveldb::Slice tkey;
        encode_transfer(cid, count, &backing.back(), &tkey);
        backing.push_back(std::vector<char>());
        leveldb::Slice tvalue;
        encode_key_value(key, NULL, 0, &backing.back(), &tvalue);
        updates.Put(tkey, tvalue);
    }

    // Perform the write
    leveldb::Status st = m_db->Write(opts, &updates);

    if (st.ok())
    {
        return SUCCESS;
    }
    else if (st.IsNotFound())
    {
        return NOT_FOUND;
    }
    else if (st.IsCorruption())
    {
        LOG(ERROR) << "corruption at the disk layer: region=" << ri
                   << " key=0x" << key.hex() << " desc=" << st.ToString();
        return CORRUPTION;
    }
    else if (st.IsIOError())
    {
        LOG(ERROR) << "IO error at the disk layer: region=" << ri
                   << " key=0x" << key.hex() << " desc=" << st.ToString();
        return IO_ERROR;
    }
    else
    {
        LOG(ERROR) << "LevelDB returned an unknown error that we don't know how to handle";
        return LEVELDB_ERROR;
    }
}

datalayer::returncode
datalayer :: make_snapshot(const region_id& ri,
                           attribute_check* checks, size_t checks_sz,
                           snapshot* snap)
{
    std::stable_sort(checks, checks + checks_sz);
    attribute_check* check_ptr = checks;
    attribute_check* check_end = checks + checks_sz;

    while (check_ptr < check_end)
    {
        attribute_check* check_tmp = check_ptr;

        while (check_tmp < check_end && check_tmp->attr == check_ptr->attr)
        {
            ++check_tmp;
        }

        generate_search_filters(ri, check_ptr, check_tmp, &snap->m_backing, &snap->m_sfs);
        check_ptr = check_tmp;
    }

    std::vector<leveldb::Range> ranges;
    ranges.reserve(snap->m_sfs.size() + 1);
    std::vector<uint64_t> sizes(snap->m_sfs.size() + 1, 0);

    for (size_t i = 0; i < snap->m_sfs.size(); ++i)
    {
        ranges.push_back(snap->m_sfs[i].range);
    }

    ranges.push_back(leveldb::Range());
    generate_object_range(ri, &snap->m_backing, &ranges[snap->m_sfs.size()]);
    m_db->GetApproximateSizes(&ranges.front(), ranges.size(), &sizes.front());
    size_t sum = 0;

    for (size_t i = 0; i < snap->m_sfs.size(); ++i)
    {
        sum += sizes[i];
        snap->m_sfs[i].size = sizes[i];
    }

    std::sort(snap->m_sfs.begin(), snap->m_sfs.end(), search_filter::sort_by_size);

    while (!snap->m_sfs.empty() && sum > sizes[snap->m_sfs.size()] / 10)
    {
        sum -= snap->m_sfs.back().size;
        snap->m_sfs.pop_back();
    }

    snap->m_dl = this;
    snap->m_snap = m_db->GetSnapshot();
    snap->m_checks = checks;
    snap->m_checks_sz = checks_sz;
    leveldb::ReadOptions opts;
    opts.fill_cache = false;
    opts.verify_checksums = true;
    opts.snapshot = snap->m_snap;
    snap->m_ri = ri;
    snap->m_obj_range = ranges.back();
    snap->m_iter = m_db->NewIterator(opts);
    snap->m_iter->Seek(snap->m_obj_range.start);

    snap->m_iter = m_db->NewIterator(opts);
    snap->m_iter->Seek(snap->m_obj_range.start);

    for (size_t i = 0; i < snap->m_sfs.size(); ++i)
    {
        snap->m_sfs[i].iter = m_db->NewIterator(opts);
        snap->m_sfs[i].iter->Seek(snap->m_sfs[i].range.start);
    }

    return SUCCESS;
}

std::tr1::shared_ptr<leveldb::Snapshot>
datalayer :: make_raw_snapshot()
{
    std::tr1::function<void (const leveldb::Snapshot*)> dtor;
    dtor = std::tr1::bind(&leveldb::DB::ReleaseSnapshot, m_db, _1);
    std::tr1::shared_ptr<leveldb::Snapshot> ret(m_db->GetSnapshot(), dtor);
    return ret;
}

void
datalayer :: make_region_iterator(region_iterator* riter,
                                  std::tr1::shared_ptr<leveldb::Snapshot> snap,
                                  const region_id& ri)
{
    riter->m_dl = this;
    riter->m_snap = snap;
    riter->m_region = ri;
    leveldb::ReadOptions opts;
    opts.fill_cache = true;
    opts.verify_checksums = true;
    opts.snapshot = riter->m_snap.get();
    riter->m_iter.reset(m_db->NewIterator(opts));
    char backing[sizeof(uint8_t) + sizeof(uint64_t)];
    char* ptr = backing;
    ptr = e::pack8be('o', ptr);
    ptr = e::pack64be(riter->m_region.get(), ptr);
    riter->m_iter->Seek(leveldb::Slice(backing, sizeof(uint8_t) + sizeof(uint64_t)));
}

datalayer::returncode
datalayer :: get_transfer(const region_id& ri,
                          uint64_t seq_no,
                          bool* has_value,
                          e::slice* key,
                          std::vector<e::slice>* value,
                          uint64_t* version,
                          reference* ref)
{
    leveldb::ReadOptions opts;
    opts.fill_cache = true;
    opts.verify_checksums = true;
    std::vector<char> kbacking;
    leveldb::Slice lkey;
    capture_id cid = m_daemon->m_config.capture_for(ri);
    assert(cid != capture_id());
    encode_transfer(cid, seq_no, &kbacking, &lkey);
    leveldb::Status st = m_db->Get(opts, lkey, &ref->m_backing);

    if (st.ok())
    {
        e::slice v(ref->m_backing.data(), ref->m_backing.size());
        return decode_key_value(v, has_value, key, value, version);
    }
    else if (st.IsNotFound())
    {
        return NOT_FOUND;
    }
    else if (st.IsCorruption())
    {
        LOG(ERROR) << "corruption at the disk layer: region=" << ri
                   << " seq_no=" << seq_no << " desc=" << st.ToString();
        return CORRUPTION;
    }
    else if (st.IsIOError())
    {
        LOG(ERROR) << "IO error at the disk layer: region=" << ri
                   << " seq_no=" << seq_no << " desc=" << st.ToString();
        return IO_ERROR;
    }
    else
    {
        LOG(ERROR) << "LevelDB returned an unknown error that we don't know how to handle";
        return LEVELDB_ERROR;
    }
}

bool
datalayer :: check_acked(const region_id& reg_id, uint64_t seq_id)
{
    // make it so that increasing seq_ids are ordered in reverse in the KVS
    seq_id = UINT64_MAX - seq_id;
    leveldb::ReadOptions opts;
    opts.fill_cache = true;
    opts.verify_checksums = true;
    char backing[sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t)];
    backing[0] = 'a';
    e::pack64be(reg_id.get(), backing + sizeof(uint8_t));
    e::pack64be(seq_id, backing + sizeof(uint8_t) + sizeof(uint64_t));
    leveldb::Slice key(backing, sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t));
    std::string val;
    leveldb::Status st = m_db->Get(opts, key, &val);

    if (st.ok())
    {
        return true;
    }
    else if (st.IsNotFound())
    {
        return false;
    }
    else if (st.IsCorruption())
    {
        LOG(ERROR) << "corruption at the disk layer: region=" << reg_id
                   << " seq_id=" << seq_id << " desc=" << st.ToString();
        return false;
    }
    else if (st.IsIOError())
    {
        LOG(ERROR) << "IO error at the disk layer: region=" << reg_id
                   << " seq_id=" << seq_id << " desc=" << st.ToString();
        return false;
    }
    else
    {
        LOG(ERROR) << "LevelDB returned an unknown error that we don't know how to handle";
        return false;
    }
}

void
datalayer :: mark_acked(const region_id& reg_id, uint64_t seq_id)
{
    // make it so that increasing seq_ids are ordered in reverse in the KVS
    seq_id = UINT64_MAX - seq_id;
    leveldb::WriteOptions opts;
    opts.sync = false;
    char backing[sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t)];
    backing[0] = 'a';
    e::pack64be(reg_id.get(), backing + sizeof(uint8_t));
    e::pack64be(seq_id, backing + sizeof(uint8_t) + sizeof(uint64_t));
    leveldb::Slice key(backing, sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t));
    leveldb::Slice val("", 0);
    leveldb::Status st = m_db->Put(opts, key, val);

    if (st.ok())
    {
        // Yay!
    }
    else if (st.IsNotFound())
    {
        LOG(ERROR) << "mark_acked returned NOT_FOUND at the disk layer: region=" << reg_id
                   << " seq_id=" << seq_id << " desc=" << st.ToString();
    }
    else if (st.IsCorruption())
    {
        LOG(ERROR) << "corruption at the disk layer: region=" << reg_id
                   << " seq_id=" << seq_id << " desc=" << st.ToString();
    }
    else if (st.IsIOError())
    {
        LOG(ERROR) << "IO error at the disk layer: region=" << reg_id
                   << " seq_id=" << seq_id << " desc=" << st.ToString();
    }
    else
    {
        LOG(ERROR) << "LevelDB returned an unknown error that we don't know how to handle";
    }
}

void
datalayer :: max_seq_id(const region_id& reg_id, uint64_t* seq_id)
{
    leveldb::ReadOptions opts;
    opts.fill_cache = false;
    opts.verify_checksums = true;
    opts.snapshot = NULL;
    std::auto_ptr<leveldb::Iterator> it(m_db->NewIterator(opts));
    char backing[sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t)];
    backing[0] = 'a';
    e::pack64be(reg_id.get(), backing + sizeof(uint8_t));
    e::pack64be(0, backing + sizeof(uint8_t) + sizeof(uint64_t));
    leveldb::Slice key(backing, sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t));
    it->Seek(key);

    if (!it->Valid())
    {
        *seq_id = 0;
        return;
    }

    key = it->key();
    e::unpacker up(key.data(), key.size());
    uint8_t p = '\0';
    uint64_t r;
    uint64_t s;
    up = up >> p >> r >> s;

    if (up.error() || p != 'a')
    {
        *seq_id = 0;
        return;
    }

    *seq_id = s;
}

void
datalayer :: encode_key(const region_id& ri,
                        const e::slice& key,
                        std::vector<char>* kbacking,
                        leveldb::Slice* lkey)
{
    size_t sz = sizeof(uint8_t) + sizeof(uint64_t) + key.size();
    kbacking->resize(sz);
    char* ptr = &kbacking->front();
    ptr = e::pack8be('o', ptr);
    ptr = e::pack64be(ri.get(), ptr);
    memmove(ptr, key.data(), key.size());
    *lkey = leveldb::Slice(&kbacking->front(), sz);
}

datalayer::returncode
datalayer :: decode_key(const e::slice& lkey,
                        region_id* ri,
                        e::slice* key)
{
    const uint8_t* ptr = lkey.data();
    const uint8_t* end = ptr + lkey.size();

    if (ptr >= end || *ptr != 'o')
    {
        return BAD_ENCODING;
    }

    ++ptr;
    uint64_t rid;

    if (ptr + sizeof(uint64_t) <= end)
    {
        ptr = e::unpack64be(ptr, &rid);
    }
    else
    {
        return BAD_ENCODING;
    }

    *ri  = region_id(rid);
    *key = e::slice(ptr, end - ptr);
    return SUCCESS;
}

void
datalayer :: encode_value(const std::vector<e::slice>& attrs,
                          uint64_t version,
                          std::vector<char>* backing,
                          leveldb::Slice* lvalue)
{
    assert(attrs.size() < 65536);
    size_t sz = sizeof(uint64_t) + sizeof(uint16_t);

    for (size_t i = 0; i < attrs.size(); ++i)
    {
        sz += sizeof(uint32_t) + attrs[i].size();
    }

    backing->resize(sz);
    char* ptr = &backing->front();
    ptr = e::pack64be(version, ptr);
    ptr = e::pack16be(attrs.size(), ptr);

    for (size_t i = 0; i < attrs.size(); ++i)
    {
        ptr = e::pack32be(attrs[i].size(), ptr);
        memmove(ptr, attrs[i].data(), attrs[i].size());
        ptr += attrs[i].size();
    }

    *lvalue = leveldb::Slice(&backing->front(), sz);
}

datalayer::returncode
datalayer :: decode_value(const e::slice& value,
                          std::vector<e::slice>* attrs,
                          uint64_t* version)
{
    const uint8_t* ptr = value.data();
    const uint8_t* end = ptr + value.size();

    if (ptr + sizeof(uint64_t) <= end)
    {
        ptr = e::unpack64be(ptr, version);
    }
    else
    {
        return BAD_ENCODING;
    }

    uint16_t num_attrs;

    if (ptr + sizeof(uint64_t) <= end)
    {
        ptr = e::unpack16be(ptr, &num_attrs);
    }
    else
    {
        return BAD_ENCODING;
    }

    attrs->clear();

    for (size_t i = 0; i < num_attrs; ++i)
    {
        uint32_t sz = 0;

        if (ptr + sizeof(uint32_t) <= end)
        {
            ptr = e::unpack32be(ptr, &sz);
        }
        else
        {
            return BAD_ENCODING;
        }

        e::slice s(reinterpret_cast<const uint8_t*>(ptr), sz);
        ptr += sz;
        attrs->push_back(s);
    }

    return SUCCESS;
}

void
datalayer :: encode_transfer(const capture_id& ci,
                             uint64_t count,
                             std::vector<char>* backing,
                             leveldb::Slice* tkey)
{
    backing->resize(sizeof(uint8_t) + 2 * sizeof(uint64_t));
    char* tmp = &backing->front();
    tmp = e::pack8be('t', tmp);
    tmp = e::pack64be(ci.get(), tmp);
    tmp = e::pack64be(count, tmp);
    *tkey = leveldb::Slice(&backing->front(), sizeof(uint8_t) + 2 * sizeof(uint64_t));
}

void
datalayer :: encode_key_value(const e::slice& key,
                              const std::vector<e::slice>* value,
                              uint64_t version,
                              std::vector<char>* backing,
                              leveldb::Slice* slice)
{
    size_t sz = sizeof(uint32_t) + key.size() + sizeof(uint64_t) + sizeof(uint16_t);

    for (size_t i = 0; value && i < value->size(); ++i)
    {
        sz += sizeof(uint32_t) + (*value)[i].size();
    }

    backing->resize(sz);
    char* ptr = &backing->front();
    *slice = leveldb::Slice(ptr, sz);
    ptr = e::pack32be(key.size(), ptr);
    memmove(ptr, key.data(), key.size());
    ptr += key.size();
    ptr = e::pack64be(version, ptr);

    if (value)
    {
        ptr = e::pack16be(value->size(), ptr);

        for (size_t i = 0; i < value->size(); ++i)
        {
            ptr = e::pack32be((*value)[i].size(), ptr);
            memmove(ptr, (*value)[i].data(), (*value)[i].size());
            ptr += (*value)[i].size();
        }
    }
}

datalayer::returncode
datalayer :: decode_key_value(const e::slice& slice,
                              bool* has_value,
                              e::slice* key,
                              std::vector<e::slice>* value,
                              uint64_t* version)
{
    const uint8_t* ptr = slice.data();
    const uint8_t* end = ptr + slice.size();

    if (ptr >= end)
    {
        return BAD_ENCODING;
    }

    uint32_t key_sz;

    if (ptr + sizeof(uint32_t) <= end)
    {
        ptr = e::unpack32be(ptr, &key_sz);
    }
    else
    {
        return BAD_ENCODING;
    }

    if (ptr + key_sz <= end)
    {
        *key = e::slice(ptr, key_sz);
        ptr += key_sz;
    }
    else
    {
        return BAD_ENCODING;
    }

    if (ptr + sizeof(uint64_t) <= end)
    {
        ptr = e::unpack64be(ptr, version);
    }
    else
    {
        return BAD_ENCODING;
    }

    if (ptr == end)
    {
        *has_value = false;
        return SUCCESS;
    }

    uint16_t num_attrs;

    if (ptr + sizeof(uint16_t) <= end)
    {
        ptr = e::unpack16be(ptr, &num_attrs);
    }
    else
    {
        return BAD_ENCODING;
    }

    value->clear();

    for (size_t i = 0; i < num_attrs; ++i)
    {
        uint32_t sz = 0;

        if (ptr + sizeof(uint32_t) <= end)
        {
            ptr = e::unpack32be(ptr, &sz);
        }
        else
        {
            return BAD_ENCODING;
        }

        e::slice s(ptr, sz);
        ptr += sz;
        value->push_back(s);
    }

    *has_value = true;
    return SUCCESS;
}

void
datalayer :: generate_object_range(const region_id& ri,
                                   backing_t* backing,
{'fraction_non_alphanumeric': 0.13940997402100633, 'fraction_numerical': 0.010774110259311796, 'mean_word_length': 2.2125339204341814, 'pattern_counts': {'":': 0, '<': 635, '<?xml version=': 0, '>': 297, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2181367', 'n_tokens_mistral': 21437, 'n_tokens_neox': 20714, 'n_words': 4812}