hyper_stub_base.h

/*
 * ===============================================================
 *    Description:  Base class for Hyperdex client, used by both
 *                  db::hyper_stub and coordinator::hyper_stub.
 *
 *        Created:  2014-02-26 15:23:54
 *
 *         Author:  Ayush Dubey, dubey@cs.cornell.edu
 *
 * Copyright (C) 2013-2014, Cornell University, see the LICENSE
 *                     file for licensing agreement
 * ===============================================================
 */

#ifndef weaver_common_hyper_stub_base_h_
#define weaver_common_hyper_stub_base_h_

#include <memory>
#include <algorithm>
#include <unordered_map>
#include <unordered_set>
#include <random>
#include <e/endian.h>
#include <e/buffer.h>
#include <hyperdex/client.h>
#include <hyperdex/datastructures.h>

#include "common/stl_serialization.h"
#include "common/transaction.h"
#include "db/types.h"
#include "db/node.h"

#define NUM_NODE_ATTRS 8
#define NUM_EDGE_ATTRS 4
#define NUM_EDGE_ID_ATTRS 1
#define NUM_TX_ATTRS 2

enum persist_node_state
{
    STABLE = 0,
    MOVING
};

enum async_call_type
{
    PUT_NODE,
    PUT_EDGE_SET,
    PUT_EDGE,
    PUT_EDGE_ID,
    ADD_INDEX,
    DEL,
    GET_NODE,
    GET_EDGE_HANDLE,
    GET_EDGE
};

const char*
async_call_type_to_string(async_call_type);

class hyper_stub_base
{
    protected:
        struct aux_edge_data
        {
            node_handle_t node_handle;
            uint64_t shard, edge_id;
        };

        struct async_call
        {
            async_call_type type;
            hyperdex_client_returncode status;
            uint64_t exec_time;
            size_t packed_sz;
            int64_t op_id;
            bool putine, tx;

            async_call() : status(HYPERDEX_CLIENT_GARBAGE), exec_time(42), packed_sz(42) { }
        };

        struct async_put_node : public async_call
        {
            std::string handle;
            hyperdex_client_attribute attrs[NUM_NODE_ATTRS];
            std::unique_ptr<e::buffer> creat_clk_buf;
            std::unique_ptr<e::buffer> props_buf;
            std::unique_ptr<e::buffer> out_edges_buf;
            std::unique_ptr<e::buffer> aliases_buf;
            size_t num_attrs;

            async_put_node() { type = PUT_NODE; }
        };

        struct async_put_edge_set : public async_call
        {
            std::string node_handle;
            int64_t max_edge_id;
            hyperdex_client_attribute set_attr[2];
            std::unique_ptr<e::buffer> set_buf;

            async_put_edge_set() { type = PUT_EDGE_SET; }
        };

        struct async_put_edge : public async_call
        {
            std::string edge_handle, node_handle;
            hyperdex_client_attribute attrs[NUM_EDGE_ATTRS];
            std::unique_ptr<e::buffer> buf;
            db::edge *e;
            bool del_after_call;
            uint64_t edge_id, shard;

            async_put_edge() : del_after_call(false) { type = PUT_EDGE; }
        };

        struct async_add_index : public async_call
        {
            std::string node_handle, alias;
            hyperdex_client_attribute index_attrs[2];

            async_add_index() { type = ADD_INDEX; }
        };

        struct async_put_edge_id : public async_call
        {
            uint64_t edge_id;
            std::string edge_handle;
            hyperdex_client_attribute edgeid_attrs[NUM_EDGE_ID_ATTRS];

            async_put_edge_id() { type = PUT_EDGE_ID; }
        };

        struct async_del : public async_call
        {
            const char *key;

            async_del() { type = DEL; }
        };

        struct async_get_node : public async_call
        {
            db::node *n;
            const hyperdex_client_attribute *cl_attr;
            size_t num_attrs;
            bool recreate_edges;

            async_get_node() { type = GET_NODE; }
        };

        struct async_get_edge_handle : public async_call
        {
            db::node *n;
            int64_t edge_id;
            const hyperdex_client_attribute *cl_attr;
            size_t num_attrs;

            async_get_edge_handle() { type = GET_EDGE_HANDLE; }
        };

        struct async_get_edge : public async_call
        {
            db::node *n;
            edge_handle_t handle;
            const hyperdex_client_attribute *cl_attr;
            size_t num_attrs;

            async_get_edge() { type = GET_EDGE; }
        };

        using async_call_ptr_t  = std::shared_ptr<hyper_stub_base::async_call>;
        using apn_ptr_t         = std::shared_ptr<hyper_stub_base::async_put_node>;
        using apes_ptr_t        = std::shared_ptr<hyper_stub_base::async_put_edge_set>;
        using ape_ptr_t         = std::shared_ptr<hyper_stub_base::async_put_edge>;
        using apei_ptr_t        = std::shared_ptr<hyper_stub_base::async_put_edge_id>;
        using aai_ptr_t         = std::shared_ptr<hyper_stub_base::async_add_index>;
        using ad_ptr_t          = std::shared_ptr<hyper_stub_base::async_del>;
        using agn_ptr_t         = std::shared_ptr<hyper_stub_base::async_get_node>;
        using ageh_ptr_t        = std::shared_ptr<hyper_stub_base::async_get_edge_handle>;
        using age_ptr_t         = std::shared_ptr<hyper_stub_base::async_get_edge>;

        void debug_print_async_call(async_call_ptr_t);

    protected:
        // node handle -> node data
        const char *node_space = "weaver_node_data";
        const char *node_attrs[NUM_NODE_ATTRS];
        const char *node_key = "node_handle";
        const enum hyperdatatype node_dtypes[NUM_NODE_ATTRS];
        // edge handle -> node handle, shard, edge id, edge data
        const char *edge_space = "weaver_edge_data";
        const char *edge_attrs[NUM_EDGE_ATTRS];
        const char *edge_key = "edge_handle";
        const enum hyperdatatype edge_dtypes[NUM_EDGE_ATTRS];
        // edge id -> edge handle
        const char *edge_id_space = "weaver_edge_id_data";
        const char *edge_id_attrs[NUM_EDGE_ID_ATTRS];
        const char *edge_id_key = "edge_id";
        const enum hyperdatatype edge_id_dtypes[NUM_EDGE_ID_ATTRS];
        // tx id -> vt id, tx data
        const char *tx_space = "weaver_tx_data";
        const char *tx_attrs[NUM_TX_ATTRS];
        const char *tx_key = "tx_id";
        const enum hyperdatatype tx_dtypes[NUM_TX_ATTRS];

        using hyper_func = int64_t (*) (struct hyperdex_client *client,
            const char*,
            const char*,
            size_t,
            const struct hyperdex_client_attribute*,
            size_t,
            hyperdex_client_returncode*);
        using hyper_map_func = int64_t (*) (struct hyperdex_client *client,
            const char*,
            const char*,
            size_t,
            const struct hyperdex_client_map_attribute*,
            size_t,
            hyperdex_client_returncode*);
        using hyper_tx_func = int64_t (*) (struct hyperdex_client_transaction *client,
            const char*,
            const char*,
            size_t,
            const struct hyperdex_client_attribute*,
            size_t,
            hyperdex_client_returncode*);
        using hyper_map_tx_func = int64_t (*) (struct hyperdex_client_transaction *client,
            const char*,
            const char*,
            size_t,
            const struct hyperdex_client_map_attribute*,
            size_t,
            hyperdex_client_returncode*);

        hyperdex_client *m_cl;
        hyperdex_client_transaction *m_hyper_tx;

        void check_op_id(int64_t op_id,
                         hyperdex_client_returncode status,
                         bool &success, int &success_calls);
        void begin_tx();
        void commit_tx(hyperdex_client_returncode &commit_status);
        void abort_tx();
        bool call(hyper_func h,
            const char *space,
            const char *key, size_t key_sz,
            hyperdex_client_attribute *cl_attr, size_t num_attrs);
        bool call(hyper_tx_func h,
            const char *space,
            const char *key, size_t key_sz,
            hyperdex_client_attribute *cl_attr, size_t num_attrs,
            hyperdex_client_returncode *ret_code=nullptr);
        bool call_no_loop(hyper_func h,
            const char *space,
            const char *key, size_t key_sz,
            hyperdex_client_attribute *cl_attr, size_t num_attrs,
            int64_t &op_id, hyperdex_client_returncode &status);
        bool call_no_loop(hyper_tx_func h,
            const char *space,
            const char *key, size_t key_sz,
            hyperdex_client_attribute *cl_attr, size_t num_attrs,
            int64_t &op_id, hyperdex_client_returncode &status);
        bool del_no_loop(const char *space,
                         const char *key, size_t key_sz,
                         int64_t &op_id, hyperdex_client_returncode &status);
        bool get_no_loop(const char* space,
                         const char *key, size_t key_sz,
                         const hyperdex_client_attribute **cl_attr, size_t *num_attrs,
                         int64_t &op_id, hyperdex_client_returncode &status,
                         bool tx);
        bool map_call(hyper_map_tx_func h,
            const char *space,
            const char *key, size_t key_sz,
            hyperdex_client_map_attribute *map_attr, size_t num_attrs);
        bool map_call_no_loop(hyper_map_func h,
            const char *space,
            const char *key, size_t key_sz,
            hyperdex_client_map_attribute *map_attr, size_t num_attrs,
            int64_t &op_id, hyperdex_client_returncode &status);
        bool loop(bool timeout, int64_t &op_id, hyperdex_client_returncode &loop_code);

        bool multiple_call(std::vector<hyper_func> &funcs,
            std::vector<const char*> &spaces,
            std::vector<const char*> &keys, std::vector<size_t> &key_szs,
            std::vector<hyperdex_client_attribute*> &attrs, std::vector<size_t> &num_attrs);
        bool multiple_call(std::vector<hyper_tx_func> &funcs,
            std::vector<const char*> &spaces,
            std::vector<const char*> &keys, std::vector<size_t> &key_szs,
            std::vector<hyperdex_client_attribute*> &attrs, std::vector<size_t> &num_attrs);
        bool multiple_call(std::vector<hyper_tx_func> &funcs,
            std::vector<const char*> &spaces,
            std::vector<const char*> &keys, std::vector<size_t> &key_szs,
            std::vector<hyperdex_client_attribute*> &attrs, std::vector<size_t> &num_attrs,
            std::vector<hyper_map_tx_func> &map_funcs,
            std::vector<const char*> &map_spaces,
            std::vector<const char*> &map_keys, std::vector<size_t> &map_key_szs,
            std::vector<hyperdex_client_map_attribute*> &map_attrs, std::vector<size_t> &map_num_attrs);

        bool get(const char *space,
            const char *key, size_t key_sz,
            const hyperdex_client_attribute **cl_attr, size_t *num_attrs,
            bool tx);
        bool get_partial(const char *space,
            const char *key, size_t key_sz,
            const char** attrnames, size_t attrnames_sz,
            const hyperdex_client_attribute **cl_attr, size_t *num_attrs,
            bool tx);
        bool multiple_get(std::vector<const char*> &spaces,
            std::vector<const char*> &keys, std::vector<size_t> &key_szs,
            std::vector<const hyperdex_client_attribute**> &cl_attrs, std::vector<size_t*> &num_attrs,
            bool tx);
        bool multiple_get_partial(std::vector<const char*> &spaces,
            std::vector<const char*> &keys, std::vector<size_t> &key_szs,
            const char** attrnames, size_t attrnames_sz,
            std::vector<const hyperdex_client_attribute**> &cl_attrs, std::vector<size_t*> &num_attrs,
            bool tx);
        bool del(const char* space,
            const char *key, size_t key_sz);
        bool multiple_del(std::vector<const char*> &spaces,
            std::vector<const char*> &keys, std::vector<size_t> &key_szs);

        // graph data functions
        bool get_edge(const edge_handle_t &edge_handle,
                      db::edge **e,
                      aux_edge_data &aux_data,
                      bool tx);
        bool get_edges(std::unordered_map<edge_handle_t, db::edge*> &edges,
                       std::unordered_map<edge_handle_t, aux_edge_data> &aux_data,
                       bool tx);
        bool get_node_edges(db::node &n, bool tx);
        bool get_node(db::node &n, bool recreate_edges);
        bool get_nodes(std::unordered_map<node_handle_t, db::node*> &nodes,
                       bool recreate_edges,
                       bool tx);
        hyperdex_client_returncode put_new_edge(uint64_t edge_id,
                                                db::edge *e,
                                                const node_handle_t &handle,
                                                uint64_t shard);
        bool put_edge(const db::edge *e, bool if_not_exist);
        bool put_edges(const db::data_map<std::vector<db::edge*>> &edges);
        bool put_node(const db::node *e, bool if_not_exist);
        bool put_nodes(std::unordered_map<node_handle_t, db::node*> &nodes, bool if_not_exist);
        bool del_edge(const edge_handle_t&, uint64_t edge_id);
        bool del_edges(const db::node&);
        bool del_node(const db::node&);
        bool del_nodes(std::vector<db::node*> &nodes);
        bool recreate_edge(const hyperdex_client_attribute *cl_attr,
                           node_handle_t &handle,
                           uint64_t &shard,
                           uint64_t &edge_id,
                           db::edge **e);
        bool recreate_node(const hyperdex_client_attribute *cl_attr, db::node &n);

        // node map functions
        bool update_nmap(const node_handle_t &handle, uint64_t loc);
        std::unordered_map<node_handle_t, uint64_t> get_nmap(std::unordered_set<node_handle_t> &toGet, bool tx);
        uint64_t get_nmap(node_handle_t &handle);

        // auxiliary index functions
    private:
        bool recreate_index(const hyperdex_client_attribute *cl_attr, std::pair<node_handle_t, uint64_t> &value);
        void sort_and_pack_as_set(std::vector<std::string>&, std::unique_ptr<e::buffer>&);
    public:
        bool add_indices(std::unordered_map<std::string, db::node*> &indices, bool tx, bool if_not_exist);
        bool get_indices(std::unordered_map<std::string, std::pair<node_handle_t, uint64_t>> &indices, bool tx);
        bool del_indices(std::vector<std::string> &indices);

        // async calls
        bool put_node_async(apn_ptr_t apn,
                            db::node *n,
                            std::unique_ptr<e::buffer> &lastupd_clk_buf,
                            std::unique_ptr<e::buffer> &restore_clk_buf,
                            std::unordered_map<int64_t, async_call_ptr_t> &async_calls,
                            bool put_if_not_exist,
                            bool tx);
        bool put_edge_id_async(apei_ptr_t apei,
                               uint64_t edge_id,
                               const edge_handle_t &edge_handle,
                               std::unordered_map<int64_t, async_call_ptr_t> &async_calls,
                               bool tx);
        bool put_edge_async(ape_ptr_t ape,
                            const node_handle_t &node_handle,
                            db::edge *e,
                            uint64_t edge_id,
                            uint64_t shard,
                            bool del_after_call,
                            std::unordered_map<int64_t, async_call_ptr_t> &async_calls,
                            bool put_if_not_exist,
                            bool tx);
        bool add_index_async(aai_ptr_t aai,
                             const node_handle_t &node_handle,
                             const std::string &alias,
                             const uint64_t &shard,
                             std::unordered_map<int64_t, async_call_ptr_t> &async_calls,
                             bool put_if_not_exist,
                             bool tx);
        bool del_async(ad_ptr_t ad,
                       const char *key, size_t key_sz,
                       const char *space,
                       std::unordered_map<int64_t, async_call_ptr_t> &async_calls);
        bool get_node_async(agn_ptr_t agn,
                            db::node *n,
                            std::unordered_map<int64_t, async_call_ptr_t>&,
                            bool recreate_edges,
                            bool tx);
        bool get_edge_handle_async(ageh_ptr_t ageh,
                                   db::node *n,
                                   uint64_t edge_id,
                                   std::unordered_map<int64_t, async_call_ptr_t>&,
                                   bool tx);
        bool get_edge_async(age_ptr_t age,
                            db::node *n,
                            const edge_handle_t &edge_handle,
                            std::unordered_map<int64_t, async_call_ptr_t>&,
                            bool tx);

        template <typename T> void prepare_buffer(const T &t, std::unique_ptr<e::buffer> &buf);
        template <typename T> void unpack_buffer(const char *buf, uint64_t buf_sz, T &t);
        template <typename T> void prepare_buffer(const std::unordered_map<std::string, T> &map, std::unique_ptr<e::buffer> &buf);
        template <typename T> void unpack_buffer(const char *buf, uint64_t buf_sz, std::unordered_map<std::string, T> &map);
        template <typename T> void prepare_buffer(const db::data_map<T> &map, std::unique_ptr<e::buffer> &buf);
        template <typename T> void unpack_buffer(const char *buf, uint64_t buf_sz, db::data_map<T> &map);
        void prepare_buffer(const std::unordered_set<std::string> &set, std::unique_ptr<e::buffer> &buf);
        void unpack_buffer(const char *buf, uint64_t buf_sz, std::unordered_set<std::string> &set);
        void prepare_buffer(const db::string_set&, std::unique_ptr<e::buffer> &buf);
        void unpack_buffer(const char *buf, uint64_t buf_sz, db::string_set&);
        // properties
        void prepare_buffer(const std::vector<std::shared_ptr<db::property>>&, std::unique_ptr<e::buffer>&);
        void unpack_buffer(const char *buf, uint64_t buf_sz, std::vector<std::shared_ptr<db::property>>&);
        // edge ids
        void prepare_buffer(const std::set<int64_t>&, std::unique_ptr<e::buffer>&);
        void unpack_buffer(const char *buf, uint64_t buf_sz, std::set<int64_t>&);

    protected:
        // pseudo random number gen for edge ids, seeded by /dev/urandom
        uint64_t m_gen_seed;
        std::mt19937_64 m_mt64_gen;
        std::uniform_int_distribution<uint64_t> m_uint64max_dist;

        void prepare_node(hyperdex_client_attribute *attr,
            db::node &n,
            std::unique_ptr<e::buffer>&,
            std::unique_ptr<e::buffer>&,
            std::unique_ptr<e::buffer>&,
            std::unique_ptr<e::buffer>&,
            std::unique_ptr<e::buffer>&,
            std::unique_ptr<e::buffer>&,
            size_t &num_attrs,
            size_t &packed_node_sz);
        void prepare_edge(hyperdex_client_attribute *attrs,
            const node_handle_t &handle,
            const uint64_t &shard,
            db::edge &e,
            const uint64_t &edge_id,
            std::unique_ptr<e::buffer> &buf,
            size_t &packed_sz);
        void pack_uint64(e::packer &packer, uint64_t num);
        void unpack_uint64(e::unpacker &unpacker, uint64_t &num);
        void pack_uint32(e::packer &packer, uint32_t num);
        void unpack_uint32(e::unpacker &unpacker, uint32_t &num);
        void pack_string(e::packer &packer, const std::string &t);
        void unpack_string(e::unpacker &unpacker, std::string &t, const uint32_t sz);

    public:
        hyper_stub_base();
};

// store the given t as a HYPERDATATYPE_STRING
template <typename T>
inline void
hyper_stub_base :: prepare_buffer(const T &t, std::unique_ptr<e::buffer> &buf)
{
    uint64_t buf_sz = message::size(nullptr, t);
    buf.reset(e::buffer::create(buf_sz));
    e::packer packer = buf->pack_at(0);
    message::pack_buffer(packer, nullptr, t);
}

// unpack the HYPERDATATYPE_STRING in to the given object
template <typename T>
inline void
hyper_stub_base :: unpack_buffer(const char *buf, uint64_t buf_sz, T &t)
{
    std::unique_ptr<e::buffer> ebuf(e::buffer::create(buf, buf_sz));
    e::unpacker unpacker = ebuf->unpack_from(0);
    message::unpack_buffer(unpacker, nullptr, t);
}

// store the given unordered_map as a HYPERDATATYPE_MAP_STRING_STRING
template <typename T>
inline void
hyper_stub_base :: prepare_buffer(const std::unordered_map<std::string, T> &map, std::unique_ptr<e::buffer> &buf)
{
    uint64_t buf_sz = 0;
    std::vector<std::string> sorted;
    sorted.reserve(map.size());

    for (const auto &p: map) {
        sorted.emplace_back(p.first);
    }
    std::sort(sorted.begin(), sorted.end());

    std::vector<uint32_t> val_sz(map.size(), UINT32_MAX);
    // now iterate in sorted order
    for (uint64_t i = 0; i < sorted.size(); i++) {
        val_sz[i] = message::size(nullptr, map.at(sorted[i]));
        buf_sz += sizeof(uint32_t) // map key encoding sz
                + sorted[i].size()
                + sizeof(uint32_t) // map val encoding sz
                + val_sz[i]; // map val encoding
    }

    buf.reset(e::buffer::create(buf_sz));
    e::packer packer = buf->pack();

    for (uint64_t i = 0; i < sorted.size(); i++) {
        pack_uint32(packer, sorted[i].size());
        pack_string(packer, sorted[i]);

        pack_uint32(packer, val_sz[i]);
        message::pack_buffer(packer, nullptr, map.at(sorted[i]));
    }
}

// unpack the HYPERDATATYPE_MAP_STRING_STRING in to the given map
template <typename T>
inline void
hyper_stub_base :: unpack_buffer(const char *buf, uint64_t buf_sz, std::unordered_map<std::string, T> &map)
{
    std::unique_ptr<e::buffer> ebuf(e::buffer::create(buf, buf_sz));
    e::unpacker unpacker = ebuf->unpack_from(0);
    std::string key;
    uint32_t sz;

    while (unpacker.remain() > 0) {
        key.erase();

        unpack_uint32(unpacker, sz);
        unpack_string(unpacker, key, sz);

        unpack_uint32(unpacker, sz);
        message::unpack_buffer(unpacker, nullptr, map[key]);
    }
}

// store the given unordered_map as a HYPERDATATYPE_MAP_STRING_STRING
template <typename T>
inline void
hyper_stub_base :: prepare_buffer(const db::data_map<T> &map, std::unique_ptr<e::buffer> &buf)
{
    uint64_t buf_sz = 0;
    std::vector<std::string> sorted;
    sorted.reserve(map.size());

    for (const auto &p: map) {
        sorted.emplace_back(p.first);
    }
    std::sort(sorted.begin(), sorted.end());

    std::vector<uint32_t> val_sz(map.size(), UINT32_MAX);
    // now iterate in sorted order
    for (uint64_t i = 0; i < sorted.size(); i++) {
        val_sz[i] = message::size(nullptr, map.find(sorted[i])->second);
        buf_sz += sizeof(uint32_t) // map key encoding sz
                + sorted[i].size()
                + sizeof(uint32_t) // map val encoding sz
                + val_sz[i]; // map val encoding
    }

    buf.reset(e::buffer::create(buf_sz));
    e::packer packer = buf->pack();

    for (uint64_t i = 0; i < sorted.size(); i++) {
        pack_uint32(packer, sorted[i].size());
        pack_string(packer, sorted[i]);

        pack_uint32(packer, val_sz[i]);
        message::pack_buffer(packer, nullptr, map.find(sorted[i])->second);
    }
}

// unpack the HYPERDATATYPE_MAP_STRING_STRING in to the given map
template <typename T>
inline void
hyper_stub_base :: unpack_buffer(const char *buf, uint64_t buf_sz, db::data_map<T> &map)
{
    std::unique_ptr<e::buffer> ebuf(e::buffer::create(buf, buf_sz));
    e::unpacker unpacker = ebuf->unpack_from(0);
    std::string key;
    uint32_t sz;

    while (unpacker.remain() > 0) {
        key.erase();

        unpack_uint32(unpacker, sz);
        unpack_string(unpacker, key, sz);

        unpack_uint32(unpacker, sz);
        message::unpack_buffer(unpacker, nullptr, map[key]);
    }
}

#endif