text:
/*
 * Copyright (c) 2016 Shanghai Jiao Tong University.
 *     All rights reserved.
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an "AS
 *  IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
 *  express or implied.  See the License for the specific language
 *  governing permissions and limitations under the License.
 *
 *
 */

#pragma once

#include <stdint.h> //uint64_t
#include <vector>
#include <set>
#include <iostream>
#include <pthread.h>

#include <boost/unordered_set.hpp>
#include <tbb/concurrent_hash_map.h>
#include <tbb/concurrent_unordered_set.h>
#include <tbb/concurrent_unordered_map.h>
#include <tbb/concurrent_vector.h>

#include "config.hpp"
#include "rdma_resource.hpp"

#include "mymath.hpp"
#include "timer.hpp"
#include "unit.hpp"

#include "data_statistic.hpp"
#include "rdf_meta.hpp"

using namespace std;

struct triple_t {
    uint64_t s; // subject
    uint64_t p; // predicate
    uint64_t o; // object

    triple_t(): s(0), p(0), o(0) { }

    triple_t(uint64_t _s, uint64_t _p, uint64_t _o): s(_s), p(_p), o(_o) { }
};

struct edge_sort_by_spo {
    /* inline bool operator()(const triple_t &t1, const triple_t &t2) {
     *     if (t1.s < t2.s)
     *         return true;
     *     else if (t1.s == t2.s)
     *         if (t1.p < t2.p)
     *             return true;
     *         else if (t1.p == t2.p && t1.o < t2.o)
     *             return true;
     *     return false;
     * } */
    inline bool operator()(const triple_t &t1, const triple_t &t2) {
        if (t1.p < t2.p)
            return true;
        else if (t1.p == t2.p)
            if (t1.s < t2.s)
                return true;
            else if (t1.s == t2.s && t1.o < t2.o)
                return true;
        return false;
    }
};

struct edge_sort_by_ops {
    /* inline bool operator()(const triple_t &t1, const triple_t &t2) {
     *     if (t1.o < t2.o)
     *         return true;
     *     else if (t1.o == t2.o)
     *         if (t1.p < t2.p)
     *             return true;
     *         else if ((t1.p == t2.p) && (t1.s < t2.s))
     *             return true;
     *     return false;
     * } */
    inline bool operator()(const triple_t &t1, const triple_t &t2) {
        if (t1.p < t2.p)
            return true;
        else if (t1.p == t2.p)
            if (t1.o < t2.o)
                return true;
            else if ((t1.o == t2.o) && (t1.s < t2.s))
                return true;
        return false;
    }
};

enum { NBITS_DIR = 1 };
enum { NBITS_IDX = 17 }; // equal to the size of t/pid
enum { NBITS_VID = (64 - NBITS_IDX - NBITS_DIR) }; // 0: index vertex, ID: normal vertex

enum { PREDICATE_ID = 0, TYPE_ID = 1 }; // reserve two special index IDs
enum dir_t { IN, OUT, CORUN }; // direction: IN=0, OUT=1, and optimization hints

static inline bool is_tpid(int id) { return (id > 1) && (id < (1 << NBITS_IDX)); }
static inline bool is_idx(int vid) { return (vid == 0); }

/**
 * predicate-base key/value store
 * key: vid | t/pid | direction
 * value: v/t/pid list
 */
struct ikey_t {
uint64_t dir : NBITS_DIR; // direction
uint64_t pid : NBITS_IDX; // predicate
uint64_t vid : NBITS_VID; // vertex

    ikey_t(): vid(0), pid(0), dir(0) { }

    ikey_t(uint64_t v, uint64_t p, uint64_t d): vid(v), pid(p), dir(d) {
        assert((vid == v) && (dir == d) && (pid == p)); // no key truncate
    }

    bool operator == (const ikey_t &key) {
        if ((vid == key.vid) && (pid == key.pid) && (dir == key.dir))
            return true;
        return false;
    }

    bool operator != (const ikey_t &key) { return !(operator == (key)); }

    bool is_empty() { return ((vid == 0) && (pid == 0) && (dir == 0)); }

    void print() { cout << "[" << vid << "|" << pid << "|" << dir << "]" << endl; }

    uint64_t hash() {
        uint64_t r = 0;
        r += vid;
        r <<= NBITS_IDX;
        r += pid;
        r <<= NBITS_DIR;
        r += dir;
        return mymath::hash_u64(r); // the standard hash is too slow (i.e., std::hash<uint64_t>()(r))
    }
};

// 64-bit internal pointer (size < 256M and off off < 64GB)
enum { NBITS_SIZE = 28 };
enum { NBITS_PTR = 36 };

/// TODO: add sid and edge type in future
struct iptr_t {
uint64_t size: NBITS_SIZE;
uint64_t off: NBITS_PTR;

    iptr_t(): size(0), off(0) { }

    iptr_t(uint64_t s, uint64_t o): size(s), off(o) {
        // no truncated
        assert ((size == s) && (off == o));
    }

    bool operator == (const iptr_t &ptr) {
        if ((size == ptr.size) && (off == ptr.off))
            return true;
        return false;
    }

    bool operator != (const iptr_t &ptr) {
        return !(operator == (ptr));
    }
};

// 128-bit vertex (key)
struct vertex_t {
    ikey_t key; // 64-bit: vertex | predicate | direction
    iptr_t ptr; // 64-bit: size | offset
};

// 32-bit edge (value)
struct edge_t {
    uint32_t val;  // vertex ID
};

class GStore {
private:
    class RDMA_Cache {
        struct Item {
            pthread_spinlock_t lock;
            vertex_t v;
            Item() {
                pthread_spin_init(&lock, 0);
            }
        };

        static const int NUM_ITEMS = 100000;
        Item items[NUM_ITEMS];

    public:
        /// TODO: use more clever cache structure with lock-free implementation
        bool lookup(ikey_t key, vertex_t &ret) {
            if (!global_enable_caching)
                return false;

            int idx = key.hash() % NUM_ITEMS;
            bool found = false;
            pthread_spin_lock(&(items[idx].lock));
            if (items[idx].v.key == key) {
                ret = items[idx].v;
                found = true;
            }
            pthread_spin_unlock(&(items[idx].lock));
            return found;
        }

        void insert(vertex_t &v) {
            if (!global_enable_caching)
                return;

            int idx = v.key.hash() % NUM_ITEMS;
            pthread_spin_lock(&items[idx].lock);
            items[idx].v = v;
            pthread_spin_unlock(&items[idx].lock);
        }
    };

    static const int NUM_LOCKS = 1024;

    static const int MAIN_RATIO = 80; // the percentage of main headers (e.g., 80%)
    static const int ASSOCIATIVITY = 8;  // the associativity of slots in each bucket

    uint64_t sid;
    Mem *mem;

    vertex_t *vertices;
    edge_t *edges;

    // the size of slot is sizeof(vertex_t)
    // the size of entry is sizeof(edge_t)
    uint64_t num_slots;       // 1 bucket = ASSOCIATIVITY slots
    uint64_t num_buckets;     // main-header region (pre-allocated hash-table)
    uint64_t num_buckets_ext; // indirect-header region (dynamical allocation)
    uint64_t num_entries;     // entry region (dynamical allocation)
    int num_preds;

    // metadata for each predicate
    vector<pred_meta_t> pred_metas;

    // Siyuan: multiple engines will access global_pred_metas
    tbb::concurrent_unordered_map <int, vector<pred_meta_t> > global_pred_metas;
    // predicate metas for each machine
    // map<int, vector<pred_meta_t>> global_pred_metas;

    uint64_t *alloc_table[20];   // [tid][pid]
    pthread_mutex_t mutex;
    pthread_barrier_t barrier;
    pthread_spinlock_t lock;
    pthread_cond_t cv_pred_inserted;
    volatile int num_pred_parts;
    volatile int last_type_pid;
    set<int> pred_set;

    // allocated
    uint64_t last_ext;
    uint64_t last_entry;

    RDMA_Cache rdma_cache;

    pthread_spinlock_t entry_lock;
    pthread_spinlock_t bucket_ext_lock;
    pthread_spinlock_t bucket_locks[NUM_LOCKS]; // lock virtualization (see paper: vLokc CGO'13)

    uint64_t bucket_for_remote(ikey_t key, int dst_sid) {
        // assert(global_pred_metas.find(dst_sid) != global_pred_metas.end());
        vector<pred_meta_t> &remote_pred_metas = global_pred_metas[dst_sid];
        return remote_pred_metas[key.pid].main_hdr_start + key.hash() % remote_pred_metas[key.pid].partition_sz;
    }

    uint64_t bucket_for(ikey_t key) {
        // the smallest pid is 1
        //return (key.pid - 1) * partition_szs[key.pid] + key.hash() % partition_szs[key.pid];
        return pred_metas[key.pid].main_hdr_start + key.hash() % pred_metas[key.pid].partition_sz;
    }

    // cluster chaining hash-table (see paper: DrTM SOSP'15)
    uint64_t insert_key(ikey_t key) {
        uint64_t bucket_id = bucket_for(key);
        uint64_t slot_id = bucket_id * ASSOCIATIVITY;
        uint64_t lock_id = bucket_id % NUM_LOCKS;

        bool found = false;
        pthread_spin_lock(&bucket_locks[lock_id]);
        while (slot_id < num_slots) {
            // the last slot of each bucket is always reserved for pointer to indirect header
            /// TODO: add type info to slot and resue the last slot to store key
            for (uint64_t i = 0; i < ASSOCIATIVITY - 1; i++, slot_id++) {
                //assert(vertices[slot_id].key != key); // no duplicate key
                if (vertices[slot_id].key == key) {
                    key.print();
                    vertices[slot_id].key.print();
                    assert(false);
                }

                // insert to an empty slot
                if (vertices[slot_id].key.is_empty()) {
                    vertices[slot_id].key = key;
                    goto done;
                }
            }

            // whether the bucket_ext (indirect-header region) is used
            if (!vertices[slot_id].key.is_empty()) {
                slot_id = vertices[slot_id].key.vid * ASSOCIATIVITY;
                continue; // continue and jump to next bucket
            }

            // allocate and link a new indirect header
            pthread_spin_lock(&bucket_ext_lock);
            assert(last_ext < num_buckets_ext);
            vertices[slot_id].key.vid = num_buckets + (last_ext++);
            pthread_spin_unlock(&bucket_ext_lock);

            slot_id = vertices[slot_id].key.vid * ASSOCIATIVITY; // move to a new bucket_ext
            vertices[slot_id].key = key; // insert to the first slot
            goto done;
        }
done:
        pthread_spin_unlock(&bucket_locks[lock_id]);
        assert(slot_id < num_slots);
        assert(vertices[slot_id].key == key);
        return slot_id;
    }

    uint64_t sync_fetch_and_alloc_edges(uint64_t n) {
        uint64_t orig;
        pthread_spin_lock(&entry_lock);
        orig = last_entry;
        last_entry += n;
        assert(last_entry < num_entries);
        pthread_spin_unlock(&entry_lock);
        return orig;
    }

    vertex_t get_vertex_remote(int tid, ikey_t key) {
        int dst_sid = mymath::hash_mod(key.vid, global_num_servers);
        uint64_t bucket_id = bucket_for_remote(key, dst_sid);
        vertex_t vert;

        if (rdma_cache.lookup(key, vert))
            return vert; // found

        char *buf = mem->buffer(tid);
        while (true) {
            uint64_t off = bucket_id * ASSOCIATIVITY * sizeof(vertex_t);
            uint64_t sz = ASSOCIATIVITY * sizeof(vertex_t);

            RDMA &rdma = RDMA::get_rdma();
            rdma.dev->RdmaRead(tid, dst_sid, buf, sz, off);
            vertex_t *verts = (vertex_t *)buf;
            for (uint64_t i = 0; i < ASSOCIATIVITY; i++) {
                if (i < ASSOCIATIVITY - 1) {
                    if (verts[i].key == key) {
                        rdma_cache.insert(verts[i]);
                        return verts[i]; // found
                    }
                } else {
                    if (verts[i].key.is_empty())
                        return vertex_t(); // not found

                    bucket_id = verts[i].key.vid; // move to next bucket
                    break; // break for-loop
                }
            }
        }
    }

    vertex_t get_vertex_local(int tid, ikey_t key) {
        uint64_t bucket_id = bucket_for(key);
        while (true) {
            for (uint64_t i = 0; i < ASSOCIATIVITY; i++) {
                uint64_t slot_id = bucket_id * ASSOCIATIVITY + i;
                if (i < ASSOCIATIVITY - 1) {
                    //data part
                    if (vertices[slot_id].key == key) {
                        //we found it
                        return vertices[slot_id];
                    }
                } else {
                    if (vertices[slot_id].key.is_empty())
                        return vertex_t(); // not found

                    bucket_id = vertices[slot_id].key.vid; // move to next bucket
                    break; // break for-loop
                }
            }
        }
    }

    edge_t *get_edges_remote(int tid, int64_t vid, int64_t d, int64_t pid, int *size) {
        int dst_sid = mymath::hash_mod(vid, global_num_servers);
        ikey_t key = ikey_t(vid, pid, d);
        vertex_t v = get_vertex_remote(tid, key);

        if (v.key.is_empty()) {
            *size = 0;
            return NULL; // not found
        }

        char *buf = mem->buffer(tid);
        uint64_t off  = num_slots * sizeof(vertex_t) + v.ptr.off * sizeof(edge_t);
        uint64_t sz = v.ptr.size * sizeof(edge_t);
        RDMA &rdma = RDMA::get_rdma();
        rdma.dev->RdmaRead(tid, dst_sid, buf, sz, off);
        edge_t *result_ptr = (edge_t *)buf;

        *size = v.ptr.size;
        return result_ptr;
    }

    edge_t *get_edges_local(int tid, int64_t vid, int64_t d, int64_t pid, int *size) {
        ikey_t key = ikey_t(vid, pid, d);
        vertex_t v = get_vertex_local(tid, key);

        if (v.key.is_empty()) {
            *size = 0;
            return NULL;
        }

        *size = v.ptr.size;
        uint64_t off = v.ptr.off;
        //std::cout<<"off"<<off<<std::endl;
        //std::cout<<"&(edges[off])->val"<<&(edges[off]).val<<std::endl;

        //for(int i=10000000;i<10000000+43291;++i)
        //    std::cout<<"edges[i].val"<<edges[i].val<<";"<<std::endl;
        //std::cin.get();
        return &(edges[off]);
    }

    typedef tbb::concurrent_hash_map<int64_t, vector< int64_t>> tbb_hash_map;
    typedef tbb::concurrent_unordered_set<int64_t> tbb_unordered_set;

    tbb_hash_map pidx_in_map; // predicate-index (IN)
    tbb_hash_map pidx_out_map; // predicate-index (OUT)
    tbb_hash_map tidx_map; // type-index

    void insert_type_index_map(tbb_hash_map &map, dir_t d) {
        int64_t pid = 0;
        uint64_t off;
        for (auto const &e : map) {
            pid = e.first;
            pred_metas[pid].indrct_hdr_start = num_buckets + last_ext;

            uint64_t sz = e.second.size();
            off = sync_fetch_and_alloc_edges(sz);
            pred_metas[pid].edge_start = off;

            ikey_t key = ikey_t(0, pid, d);
            uint64_t slot_id = insert_key(key);
            iptr_t ptr = iptr_t(sz, off);
            vertices[slot_id].ptr = ptr;

            for (auto const &vid : e.second)
                edges[off++].val = vid;

            record_meta_end(pid, off);
            last_type_pid = pid;
        }

    }

    void record_meta_end(int pid, uint64_t &off) {
        if (num_buckets + last_ext > pred_metas[pid].indrct_hdr_end)
            pred_metas[pid].indrct_hdr_end = num_buckets + last_ext;
        if (off > pred_metas[pid].edge_end)
            pred_metas[pid].edge_end = off;
    }

    void insert_index_map(tbb_hash_map &map, dir_t d, bool &first) {
        if (map.empty())
            return;

        // Siyuan: DBPSB多线程插入的时候map.size()会大于1
        assert(1 == map.size());

        int64_t pid = 0;
        uint64_t off;
        for (auto const &e : map) {
            pid = e.first;

            uint64_t sz = e.second.size();
            off = sync_fetch_and_alloc_edges(sz);

            ikey_t key = ikey_t(0, pid, d);
            uint64_t slot_id = insert_key(key);
            iptr_t ptr = iptr_t(sz, off);
            vertices[slot_id].ptr = ptr;

            for (auto const &vid : e.second)
                edges[off++].val = vid;

            record_meta_end(pid, off);
            pred_set.insert(pid);

        }

        /* edge_start += (off - edge_start);
         * first = true; */
    }

#ifdef VERSATILE
    tbb_unordered_set p_set; // all of predicates
    tbb_unordered_set v_set; // all of vertices (subjects and objects)

    void insert_index_set(tbb_unordered_set &set, dir_t d) {
        uint64_t sz = set.size();
        uint64_t off = sync_fetch_and_alloc_edges(sz);

        ikey_t key = ikey_t(0, TYPE_ID, d);
        uint64_t slot_id = insert_key(key);
        iptr_t ptr = iptr_t(sz, off);
        vertices[slot_id].ptr = ptr;

        for (auto const &e : set)
            edges[off++].val = e;
    }
#endif

public:
    GStore() { }
    // encoding rules of gstore
    // subject/object (vid) >= 2^17, 2^17 > predicate/type (p/tid) > 2^1,
    // TYPE_ID = 1, PREDICATE_ID = 0, OUT = 1, IN = 0
    //
    // NORMAL key/value pair
    //   key = [vid |    predicate | IN/OUT]  value = [vid0, vid1, ..]  i.e., vid's ngbrs w/ predicate
    //   key = [vid |      TYPE_ID |    OUT]  value = [tid0, tid1, ..]  i.e., vid's all types
    //   key = [vid | PREDICATE_ID | IN/OUT]  value = [pid0, pid1, ..]  i.e., vid's all predicates
    // INDEX key/value pair
    //   key = [  0 |          pid | IN/OUT]  value = [vid0, vid1, ..]  i.e., predicate-index
    //   key = [  0 |          tid |     IN]  value = [vid0, vid1, ..]  i.e., type-index
    //   key = [  0 |      TYPE_ID |    OUT]  value = [vid0, vid1, ..]  i.e., all objects/subjects
    //   key = [  0 |      TYPE_ID |    OUT]  value = [vid0, vid1, ..]  i.e., all predicates
    // Empty key
    //   key = [  0 |            0 |      0]  value = [vid0, vid1, ..]  i.e., init

    // GStore: key (main-header and indirect-header region) | value (entry region)
    // The key (head region) is a cluster chaining hash-table (with associativity)
    // The value (entry region) is a varying-size array
    GStore(uint64_t sid, Mem *mem, int num_preds): sid(sid), mem(mem),
        num_preds(num_preds) {
        num_slots = ((uint64_t)global_num_keys_million) * 1000 * 1000;
        num_buckets = (uint64_t)((num_slots / ASSOCIATIVITY) * MAIN_RATIO / 100);
        //num_buckets_ext = (num_slots / ASSOCIATIVITY) / (KEY_RATIO + 1);
        num_buckets_ext = (num_slots / ASSOCIATIVITY) - num_buckets;

        vertices = (vertex_t *)(mem->kvstore());
        edges = (edge_t *)(mem->kvstore() + num_slots * sizeof(vertex_t));

        if (mem->kvstore_size() <= num_slots * sizeof(vertex_t)) {
            cout << "ERROR: " << global_memstore_size_gb
                 << "GB memory store is not enough to store hash table with "
                 << global_num_keys_million << "M keys" << std::endl;
            assert(false);
        }

        num_entries = (mem->kvstore_size() - num_slots * sizeof(vertex_t)) / sizeof(edge_t);
        last_entry = 0;

        pthread_mutex_init(&mutex, 0);
        pthread_barrier_init(&barrier, 0, nthread_parallel_load);
        pthread_spin_init(&lock, 0);
        pthread_cond_init(&cv_pred_inserted, 0);
        num_pred_parts = 0;

        pthread_spin_init(&entry_lock, 0);
        pthread_spin_init(&bucket_ext_lock, 0);
        for (int i = 0; i < NUM_LOCKS; i++)
            pthread_spin_init(&bucket_locks[i], 0);
    }

    // TODO: 在这里初始化partition_szs，统计各个predicate的数据量，
    // 按比例分配num_buckets给各个predicate。
    // 前提：nthread_parallel_load == 1
    void init(const vector<vector<triple_t>> &triple_spo, const vector<vector<triple_t>> &triple_ops) {
        vector<uint64_t> partition_szs; // 最初存的是triple的数量
        // initialize
        for (int i = 0; i <= num_preds; ++i) {
            pred_metas.push_back(pred_meta_t());
            partition_szs.push_back(0);
            pred_metas[i].num_buckets = 0;
        }

        // init alloc_table
        for (int i = 0; i < nthread_parallel_load; ++i) {
            alloc_table[i] = new uint64_t[num_preds + 1];
            memset(alloc_table[i], 0, sizeof(uint64_t) * (num_preds + 1));
        }

        int pid, tid;
        uint64_t num_edges = 0, main_hdr_cnt = 0;
        // count triples for each predicate
        for (tid = 0; tid < nthread_parallel_load; tid++) {
            for (auto spo : triple_spo[tid]) {
                alloc_table[tid][spo.p]++;
                partition_szs[spo.p]++;
                num_edges++;
                if (is_tpid(spo.o)) {
                    partition_szs[spo.o]++;
                    num_edges++;
                }
            }

            for (auto ops: triple_ops[tid]) {
                if (is_tpid(ops.o))
                    continue;

                num_edges++;
                partition_szs[ops.p]++;
                alloc_table[tid][ops.p]++;
            }
        }

        // allocate main headers
        uint64_t pid_num_buckets;
        for (int i = 1; i <= num_preds; ++i) {
            pid_num_buckets = (static_cast<double>(partition_szs[i]) / num_edges) * num_buckets;
            // partition_szs[i] = (static_cast<double>(partition_szs[i]) / num_edges) * num_buckets;
            if (pid_num_buckets == 0 && partition_szs[i] > 0)
                pid_num_buckets = 1;

            pred_metas[i].partition_sz = pid_num_buckets;
            pred_metas[i].main_hdr_start = main_hdr_cnt;
            pred_metas[i].main_hdr_end = main_hdr_cnt + pid_num_buckets;
            main_hdr_cnt += pid_num_buckets;
        }

        assert(main_hdr_cnt <= num_buckets);

        // initiate keys
        #pragma omp parallel for num_threads(global_num_engines)
        for (uint64_t i = 0; i < num_slots; i++)
            vertices[i].key = ikey_t();
    }

    // skip all TYPE triples (e.g., <http://www.Department0.University0.edu> rdf:type ub:University)
    // because Wukong treats all TYPE triples as index vertices. In addition, the triples in triple_ops
    // has been sorted by the vid of object, and IDs of types are always smaller than normal vertex IDs.
    // Consequently, all TYPE triples are aggregated at the beggining of triple_ops
    void insert_normal(int tid, vector<triple_t> &spo, vector<triple_t> &ops) {
        set<int> spo_pid_set;
        set<int> ops_pid_set;

        for (auto triple : spo) {
            spo_pid_set.insert(triple.p);
        }

        for (auto triple : ops) {
            ops_pid_set.insert(triple.p);
        }

        printf("[thread %d] pid in spo [Server %d]:\n", tid, sid);
        for (auto pid : spo_pid_set) {
            printf("%d ", pid);
        }
        printf("\n");

        printf("[thread %d] pid in ops [Server %d]:\n", tid, sid);
        for (auto pid : ops_pid_set) {
            printf("%d ", pid);
        }
        printf("\n");

        // treat type triples as index vertices
        uint64_t type_triples = 0;
        while (type_triples < ops.size() && is_tpid(ops[type_triples].o))
            type_triples++;

#ifdef VERSATILE
        // the number of separate combinations of subject/object and predicate
        uint64_t accum_predicate = 0;
#endif
        // allocate edges in entry region for triples
        /* uint64_t edge_start = sync_fetch_and_alloc_edges(spo.size() + ops.size() - type_triples);
         * uint64_t off = edge_start; */
        uint64_t edge_start, off;

        uint64_t si, sj, ei, ej;
        uint64_t slot_id;
        int pid, current_pid = 0;

        bool finished = false, first = true;

        for (si = 0, sj = type_triples; si < spo.size(); si = ei) {
            // predicate-based key (subject + predicate)
            ei = si + 1;

            pid = spo[si].p;

            if (current_pid == 0 || pid != current_pid) {
                off = sync_fetch_and_alloc_edges(alloc_table[tid][pid]);
                current_pid = pid;
            }

            while ((ei < spo.size())
                    && (spo[si].s == spo[ei].s)
                    && (spo[si].p == spo[ei].p)) {
                    ei++;
            }

#ifdef VERSATILE
            accum_predicate++;
#endif
            // insert vertex
            ikey_t key = ikey_t(spo[si].s, spo[si].p, OUT);
            slot_id = insert_key(key);
            iptr_t ptr = iptr_t(ei - si, off);
            vertices[slot_id].ptr = ptr;

            // insert edges
            for (uint64_t i = si; i < ei; i++)
                edges[off++].val = spo[i].o;

            record_index_info(slot_id);

            // (subject + predicate) part of one predicate has been inserted,
            // insert its (object + predicate) part
            if (ei >= spo.size() || spo[si].p != spo[ei].p) {
                // for predicate type
                if (current_pid == TYPE_ID) {
                    pthread_mutex_lock(&mutex);
                    record_meta_end(key.pid, off);
                    pthread_mutex_unlock(&mutex);
                } else {
                    for ( ; sj < ops.size(); sj = ej) {
                        ej = sj + 1;

                        // sometimes a predicate(e.g telephone) only in vector spo
                        // but not in vector ops
                        if (ops[sj].p != current_pid)
                            break;  // break for

                        // ad-hoc fix for dbpsb_q3
                        // if (ops[sj].p < current_pid) {
                            // pid = ops[sj].p;
                            // assert(alloc_table[tid][pid] > 0);
                            // assert(pred_metas[pid].partition_sz > 0);
                            // off = sync_fetch_and_alloc_edges(alloc_table[tid][pid]);
                            // current_pid = pid;
                        // }

                        while ((ej < ops.size())
                                && (ops[sj].o == ops[ej].o)
                                && (ops[sj].p == ops[ej].p)) { ej++; }
#ifdef VERSATILE
                        accum_predicate++;
#endif

                        // insert vertex
                        ikey_t key = ikey_t(ops[sj].o, ops[sj].p, IN);
                        slot_id = insert_key(key);
                        iptr_t ptr = iptr_t(ej - sj, off);
                        vertices[slot_id].ptr = ptr;

                        // insert edges
                        for (uint64_t i = sj; i < ej; i++)
                            edges[off++].val = ops[i].s;

                        record_index_info(slot_id);

                        // check whether this predicate is finished
                        if (ops[sj].p != ops[ej].p) {
                            sj = ej;
                            break;
                        }
                    }
                }

                pthread_mutex_lock(&mutex);
                ++num_pred_parts;

                if (tid == 0) { // master
                    // wait until all worker finishes its part
                    while (num_pred_parts < nthread_parallel_load)
                        pthread_cond_wait(&cv_pred_inserted, &mutex);

                    insert_type_index_map(tidx_map, IN);
                    // insert_type_index_vec(tidx_vec, IN);
                    insert_index_map(pidx_in_map, IN, first);
                    insert_index_map(pidx_out_map, OUT, first);

                    // tbb_vec().swap(tidx_vec);
                    tbb_hash_map().swap(tidx_map);
                    tbb_hash_map().swap(pidx_in_map);
                    tbb_hash_map().swap(pidx_out_map);

                    num_pred_parts = 0;   // reset
                    pthread_mutex_unlock(&mutex);

                } else {    // slaves
                    if (num_pred_parts == nthread_parallel_load) {
                        pthread_cond_signal(&cv_pred_inserted);
                    }

                    pthread_mutex_unlock(&mutex);
                }

                // wait for all threads finish inserting a predicate
                pthread_barrier_wait(&barrier);

            }   // if

        }   // for

    }

    void record_index_info(uint64_t slot_id) {

        int64_t vid = vertices[slot_id].key.vid;
        int64_t pid = vertices[slot_id].key.pid;

        uint64_t sz = vertices[slot_id].ptr.size;
        uint64_t off = vertices[slot_id].ptr.off;

        if (vertices[slot_id].key.dir == IN) {
            if (pid == PREDICATE_ID) {
            } else if (pid == TYPE_ID) {
                assert(false); // (IN) type triples should be skipped
            } else { // predicate-index (OUT) vid
                tbb_hash_map::accessor a;
                pidx_out_map.insert(a, pid);
                a->second.push_back(vid);
            }
        } else {
            if (pid == PREDICATE_ID) {
            } else if (pid == TYPE_ID) {
                // type-index (IN) vid
                for (uint64_t e = 0; e < sz; e++) {
                    tbb_hash_map::accessor a;
                    tidx_map.insert(a, edges[off + e].val);
                    a->second.push_back(vid);
                    // tidx_vec[edges[off + e].val].push_back(vid);
                }
            } else { // predicate-index (IN) vid

                tbb_hash_map::accessor a;
                pidx_in_map.insert(a, pid);
                a->second.push_back(vid);

            }
        }
    }

    // insert P-index and T-index vertices
    void insert_index() {
        uint64_t t1 = timer::get_usec();

        // scan raw data to generate index data in parallel
        #pragma omp parallel for num_threads(global_num_engines)
        for (int bucket_id = 0; bucket_id < num_buckets + num_buckets_ext; bucket_id++) {
            uint64_t slot_id = bucket_id * ASSOCIATIVITY;
            for (int i = 0; i < ASSOCIATIVITY - 1; i++, slot_id++) {
                // skip empty slot
                if (vertices[slot_id].key.is_empty()) continue;

                int64_t vid = vertices[slot_id].key.vid;
                int64_t pid = vertices[slot_id].key.pid;

                uint64_t sz = vertices[slot_id].ptr.size;
                uint64_t off = vertices[slot_id].ptr.off;

                if (vertices[slot_id].key.dir == IN) {
                    if (pid == PREDICATE_ID) {
#ifdef VERSATILE
                        v_set.insert(vid);
                        for (uint64_t e = 0; e < sz; e++)
                            p_set.insert(edges[off + e].val);
#endif
                    } else if (pid == TYPE_ID) {
                        assert(false); // (IN) type triples should be skipped
                    } else { // predicate-index (OUT) vid
                        tbb_hash_map::accessor a;
                        pidx_out_map.insert(a, pid);
                        a->second.push_back(vid);
                    }
                } else {
                    if (pid == PREDICATE_ID) {
#ifdef VERSATILE
                        v_set.insert(vid);
                        for (uint64_t e = 0; e < sz; e++)
                            p_set.insert(edges[off + e].val);
#endif
                    } else if (pid == TYPE_ID) {
                        // type-index (IN) vid
                        for (uint64_t e = 0; e < sz; e++) {
                            tbb_hash_map::accessor a;
                            tidx_map.insert(a, edges[off + e].val);
                            a->second.push_back(vid);
                        }
                    } else { // predicate-index (IN) vid
                        tbb_hash_map::accessor a;
                        pidx_in_map.insert(a, pid);
                        a->second.push_back(vid);
                    }
                }
            }
        }

        uint64_t t2 = timer::get_usec();
        cout << (t2 - t1) / 1000 << " ms for (parallel) prepare index info" << endl;

        cout << "tidx_map: size: " << tidx_map.size() << endl;
        cout << "pidx_in_map: size: " << pidx_in_map.size() << endl;
        cout << "pidx_out_map: size: " << pidx_out_map.size() << endl;

        // add type/predicate index vertices
        /* insert_index_map(tidx_map, IN);
         * insert_index_map(pidx_in_map, IN);
         * insert_index_map(pidx_out_map, OUT); */

        tbb_hash_map().swap(pidx_in_map);
        tbb_hash_map().swap(pidx_out_map);
        tbb_hash_map().swap(tidx_map);

#ifdef VERSATILE
        insert_index_set(v_set, IN);
        insert_index_set(p_set, OUT);

        tbb_unordered_set().swap(v_set);
        tbb_unordered_set().swap(p_set);
#endif

        uint64_t t3 = timer::get_usec();
        cout << (t3 - t2) / 1000 << " ms for insert index data into gstore" << endl;
    }

    // TODO: 填充非type的predicate的edge_start, indrct_hdr_start
    void finish_pred_metas() {
        int pid, prev_pid;
        // type
        pred_metas[1].indrct_hdr_start = num_buckets;

        set<int>::iterator it = pred_set.begin();
        pid = *it;

        pred_metas[pid].edge_start = pred_metas[last_type_pid].edge_end;
        pred_metas[pid].indrct_hdr_start = pred_metas[last_type_pid].indrct_hdr_end;

        it++;
        for (prev_pid = pid; it != pred_set.end(); ++it) {
            pid = *it;

            pred_metas[pid].edge_start = pred_metas[prev_pid].edge_end;
            pred_metas[pid].indrct_hdr_start = pred_metas[prev_pid].indrct_hdr_end;

            prev_pid = pid;
        }
    }

    void print_pred_metas() {
        for (int pid = 1; pid < pred_metas.size(); ++pid) {
            printf("pid=%d, edge_start=%ld, edge_end=%ld, ind_hdr_start=%ld, ind_hdr_end=%ld\n", pid,
                    pred_metas[pid].edge_start, pred_metas[pid].edge_end,
                    pred_metas[pid].indrct_hdr_start, pred_metas[pid].indrct_hdr_end);
        }
    }

    // 统计predicate实际占用了多少个buckets
    void stat_predicates() {
        for (int bucket_id = 0; bucket_id < num_buckets + num_buckets_ext; bucket_id++) {
                uint64_t slot_id = bucket_id * ASSOCIATIVITY;

            for (int i = 0; i < ASSOCIATIVITY - 1; i++, slot_id++) {
                // skip empty slot
                if (vertices[slot_id].key.is_empty()) continue;

                int64_t vid = vertices[slot_id].key.vid;
                int64_t pid = vertices[slot_id].key.pid;
                uint64_t off = vertices[slot_id].ptr.off;

                if (i == 0)
                    pred_metas[pid].num_buckets++;
            }
        }
    }

    // prepare data for planner
    void generate_statistic(data_statistic& statistic) {
        for (int bucket_id = 0; bucket_id < num_buckets + num_buckets_ext; bucket_id++) {
            uint64_t slot_id = bucket_id * ASSOCIATIVITY;
            for (int i = 0; i < ASSOCIATIVITY - 1; i++, slot_id++) {
                // skip empty slot
                if (vertices[slot_id].key.is_empty()) continue;

                int64_t vid = vertices[slot_id].key.vid;
                int64_t pid = vertices[slot_id].key.pid;
                uint64_t off = vertices[slot_id].ptr.off;

                if (pid == PREDICATE_ID) continue; // skip for index vertex

                unordered_map<int, int>& ptcount = statistic.predicate_to_triple;
                unordered_map<int, int>& pscount = statistic.predicate_to_subject;
                unordered_map<int, int>& pocount = statistic.predicate_to_object;
                unordered_map<int, int>& tyscount = statistic.type_to_subject;
                unordered_map<int, vector<direct_p> >& ipcount = statistic.id_to_predicate;

                if (vertices[slot_id].key.dir == IN) {
                    uint64_t sz = vertices[slot_id].ptr.size;
                    // triples only count from one direction
                    if (ptcount.find(pid) == ptcount.end()) {
                        ptcount[pid] = sz;
                    }
                    else {
                        ptcount[pid] += sz;
                    }
                    // count objects
                    if (pocount.find(pid) == pocount.end()) {
                        pocount[pid] = 1;
                    }
                    else {
                        pocount[pid] ++;
                    }
                    // count in predicates for specific id
                    ipcount[vid].push_back(direct_p(IN, pid));
                }
                else {
                    // count subjects
                    if (pscount.find(pid) == pscount.end()) {
                        pscount[pid] = 1;
                    }
                    else {
                        pscount[pid]++;
                    }
                    // count out predicates for specific id
                    ipcount[vid].push_back(direct_p(OUT, pid));
                    // count type predicate
                    if (pid == TYPE_ID) {
                        uint64_t sz = vertices[slot_id].ptr.size;
                        uint64_t off = vertices[slot_id].ptr.off;
                        for (uint64_t j = 0; j < sz; j++) {
                            //src may belongs to multiple types
                            uint64_t obid = edges[off + j].val;
                            if (tyscount.find(obid) == tyscount.end()) {
                                tyscount[obid] = 1;
                            }
                            else {
                                tyscount[obid] ++;
                            }
                            if (pscount.find(obid) == pscount.end()) {
                                pscount[obid] = 1;
                            } else {
                                pscount[obid] ++;
                            }
                            ipcount[vid].push_back(direct_p(OUT, obid));
                        }
                    }
                }
            }
        }

        //cout<<"sizeof predicate_to_triple = "<<statistic.predicate_to_triple.size()<<endl;
        //cout<<"sizeof predicate_to_subject = "<<statistic.predicate_to_subject.size()<<endl;
        //cout<<"sizeof predicate_to_object = "<<statistic.predicate_to_object.size()<<endl;
        //cout<<"sizeof type_to_subject = "<<statistic.type_to_subject.size()<<endl;
        //cout<<"sizeof id_to_predicate = "<<statistic.id_to_predicate.size()<<endl;

        unordered_map<pair<int, int>, four_num, boost::hash<pair<int, int> > >& ppcount = statistic.correlation;

        // do statistic for correlation
        for (unordered_map<int, vector<direct_p> >::iterator it = statistic.id_to_predicate.begin();
                it != statistic.id_to_predicate.end(); it++ ) {
            int vid = it->first;
            vector<direct_p>& vec = it->second;
            for (int i = 0; i < vec.size(); i++) {
                for (int j = i + 1; j < vec.size(); j++) {
                    int p1, d1, p2, d2;
                    if (vec[i].p < vec[j].p) {
                        p1 = vec[i].p;
                        d1 = vec[i].dir;
                        p2 = vec[j].p;
                        d2 = vec[j].dir;
                    } else {
                        p1 = vec[j].p;
                        d1 = vec[j].dir;
                        p2 = vec[i].p;
                        d2 = vec[i].dir;
                    }
                    if (d1 == OUT && d2 == OUT) {
                        ppcount[make_pair(p1, p2)].out_out++;
                    }
                    if (d1 == OUT && d2 == IN) {
                        ppcount[make_pair(p1, p2)].out_in++;
                    }
                    if (d1 == IN && d2 == IN) {
                        ppcount[make_pair(p1, p2)].in_in++;
                    }
                    if (d1 == IN && d2 == OUT) {
                        ppcount[make_pair(p1, p2)].in_out++;
                    }
                }
            }
        }
        //cout << "sizeof correlation = " << statistic.correlation.size() << endl;
        cout << "INFO#" << sid << ": generating statistics is finished." << endl;
    }

    edge_t *get_edges_global(int tid, int64_t vid, int64_t d, int64_t pid, int *sz) {
        if (mymath::hash_mod(vid, global_num_servers) == sid)
            return get_edges_local(tid, vid, d, pid, sz);
        else
            return get_edges_remote(tid, vid, d, pid, sz);
    }

    edge_t *get_index_edges_local(int tid, int64_t pid, int64_t d, int *sz) {
        // predicate is not important, so we set it 0
        return get_edges_local(tid, 0, d, pid, sz);
    }

    vertex_t *get_vertices_ptr(){return vertices;}
    edge_t *get_edges_ptr(){return edges;}
    uint64_t get_num_slots(){return num_slots;}
    uint64_t get_num_buckets(){return num_buckets;}
    uint64_t get_last_entry(){return last_entry;}
    vector<pred_meta_t> get_pred_metas(){return pred_metas;}
    int get_num_preds() { return num_preds; }

    void set_global_pred_metas(tbb::concurrent_unordered_map <int, vector<pred_meta_t> > &global_pred_metas) {
        this->global_pred_metas = global_pred_metas;
    }

    // analysis and debuging
    void print_mem_usage() {
        uint64_t used_slots = 0;
        for (int x = 0; x < num_buckets; x++) {
            uint64_t slot_id = x * ASSOCIATIVITY;
            for (int y = 0; y < ASSOCIATIVITY - 1; y++, slot_id++) {
                if (vertices[slot_id].key.is_empty())
                    continue;
                used_slots++;
            }
        }

        cout << "main header: " << B2MiB(num_buckets * ASSOCIATIVITY * sizeof(vertex_t))
             << " MB (" << num_buckets * ASSOCIATIVITY << " slots)" << endl;
        cout << "\tused: " << 100.0 * used_slots / (num_buckets * ASSOCIATIVITY)
             << " % (" << used_slots << " slots)" << endl;
        cout << "\tchain: " << 100.0 * num_buckets / (num_buckets * ASSOCIATIVITY)
             << " % (" << num_buckets << " slots)" << endl;

        used_slots = 0;
        for (int x = num_buckets; x < num_buckets + num_buckets_ext; x++) {
            uint64_t slot_id = x * ASSOCIATIVITY;
            for (int y = 0; y < ASSOCIATIVITY - 1; y++, slot_id++) {
                if (vertices[slot_id].key.is_empty())
                    continue;
                used_slots++;
            }
        }

        cout << "indirect header: " << B2MiB(num_buckets_ext * ASSOCIATIVITY * sizeof(vertex_t))
             << " MB (" << num_buckets_ext * ASSOCIATIVITY << " slots)" << endl;
        cout << "\talloced: " << 100.0 * last_ext / num_buckets_ext
             << " % (" << last_ext << " buckets)" << endl;
        cout << "\tused: " << 100.0 * used_slots / (num_buckets_ext * ASSOCIATIVITY)
             << " % (" << used_slots << " slots)" << endl;

        cout << "entry: " << B2MiB(num_entries * sizeof(edge_t))
             << " MB (" << num_entries << " entries)" << endl;
        cout << "\tused: " << 100.0 * last_entry / num_entries
             << " % (" << last_entry << " entries)" << endl;

    }
};

max_stars_repo_name: WukongGPU/WukongGPU

max_issues_repo_name: WukongGPU/WukongGPU

max_forks_repo_name: WukongGPU/WukongGPU


text:
#pragma once

#include "Clients/LogClient.hpp"
#include "Clients/ConfigurationClient.hpp"
#include "Clients/NameClient.hpp"
#include <sw/redis++/redis++.h>
#include <string>
#include <chrono>
#include <functional>
#include <optional>
#include <list>
#include <unordered_map>
#include <shared_mutex>
#include <atomic>
#include <GaiaBackground/GaiaBackground.hpp>
#include <boost/program_options.hpp>
#include <boost/lexical_cast.hpp>

namespace Gaia::Framework
{
    class Service
    {
         template <typename, typename... ArgumentTypes>
         friend void Launch(int, char**, ArgumentTypes... constructor_arguments);

    public:
        using MessageHandler = std::function<void(const std::string&)>;

    private:
        /// Mutex for commands map.
        std::shared_mutex CommandHandlersMutex;
        /// Command handlers map.
        std::unordered_map<std::string, MessageHandler> CommandHandlers;
        /// Mutex for messages map.
        std::shared_mutex MessageHandlersMutex;
        /// Message handlers map.
        std::unordered_multimap<std::string, MessageHandler> MessageHandlers;
        /// Handle a command.
        void HandleCommand(const std::string& name, const std::string& content);
        /// Handle a message.
        void HandleMessage(const std::string& channel, const std::string& content);

        /// Time point of last name service update.
        std::chrono::system_clock::time_point LastHeartBeatTime;

        /**
         * @brief Enable of this service.
         * @details
         *  If false, this service will not be updated frequently,
         *  and message will not be consumed.
         */
        std::atomic_bool Enable {true};

        /// Updater for message pulling loop.
        Gaia::Background::BackgroundWorker MessageUpdater;

    private:
        /// Connection to the Redis server.
        std::shared_ptr<sw::redis::Redis> Connection;
        /// Connection that has a very small overtime tolerance, which may throw timeout exception more frequently.
        std::shared_ptr<sw::redis::Redis> RealtimeConnection;
        /// Communicator for Redis message service.
        std::shared_ptr<sw::redis::Subscriber> Subscriber;
        /// Log service client.
        std::unique_ptr<Clients::LogClient> Logger {nullptr};
        /// Configuration service client.
        std::unique_ptr<Clients::ConfigurationClient> Configurator {nullptr};
        /// Name service client.
        std::unique_ptr<Clients::NameClient> NameResolver {nullptr};

    protected:
        /**
         * @brief Send a command to a service and receive the result.
         * @param service_name Name of the target service.
         * @param command_name Name of the service command.
         * @param content Content for the command request.
         */
        void SendServiceCommand(const std::string& service_name, const std::string& command_name,
                                const std::string& content = "");

        /**
         * @brief Set the value of the given value.
         * @tparam ValueType Type of the value to set.
         * @param name Name of the value.
         * @param value Value to set.
         */
        template <typename ValueType>
        void SetRemoteValue(const std::string& name, ValueType value)
        {
            if (Connection)
            {
                Connection->set(name, std::to_string(value));
            }
        }

        /**
         * @brief Set the value of the given remote value.
         * @tparam ValueType Type of the value to set.
         * @param name Name of the value to set.
         * @param value Value to set.
         * @param lasting_seconds After this amount of seconds, this value will expire.
         */
        template <typename ValueType>
        void SetRemoteValue(const std::string& name, ValueType value, std::chrono::seconds lasting_seconds)
        {
            if (Connection)
            {
                Connection->set(name, std::to_string(value), lasting_seconds);
            }
        }

        /**
         * @brief Check whether the remove value exists or not.
         * @param name Name of the value to check.
         * @retval true Remote value with the given name exists.
         * @retval false Remote value with the given name does not exist.
         */
        bool HasRemoteValue(const std::string& name)
        {
            if (Connection) return Connection->exists(name);
            return false;
        }

        /**
         * @brief Get the value of a remote value.
         * @tparam ValueType Type of the value.
         * @param name Name of the value.
         * @return std::nullopt, if the desired value does not exist or fail to be converted into the given type;
         *         otherwise std::optional that contains the value in the desired type.
         */
        template <typename ValueType>
        std::optional<ValueType> GetRemoteValue(const std::string& name)
        {
            if (Connection)
            {
                auto optional_text = Connection->get(name);
                if (optional_text.has_value())
                    try{
                        return {boost::lexical_cast<ValueType>(optional_text.value())};
                    }catch (boost::bad_lexical_cast& error){}
            }
            return std::nullopt;
        }

        /**
         * @brief Add a command handler to support a command.
         * @param name Name of the command.
         * @param handler Handler functor.
         */
        void AddCommand(const std::string& name, MessageHandler handler);

        /// Remove the handler of the given command.
        void RemoveCommand(const std::string& name);

        /**
         * @brief Add a subscription to a specific channel.
         * @param channel_name Name of the channel to subscribe.
         * @param handler Handler for messages from the channel.
         */
        void AddSubscription(const std::string& channel_name, const MessageHandler& handler);
        /**
         * @brief Remove the subscriptions to the given channel.
         * @param channel_name Name of the channel.
         * @attention All subscriptions to the given channel will be removed.
         */
        void RemoveSubscription(const std::string& channel_name);

        /// Get connection of this service.
        [[nodiscard]] inline const std::shared_ptr<sw::redis::Redis>& GetConnection() const noexcept
        {
            return Connection;
        }
        /// Get communicator of this service.
        [[nodiscard]] inline sw::redis::Subscriber* GetCommunicator() const noexcept
        {
            return Subscriber.get();
        }
        /// Get log client of this service.
        [[nodiscard]] inline Clients::LogClient* GetLogger() const noexcept
        {
            return Logger.get();
        }
        /// Get configuration client of this service.
        [[nodiscard]] inline Clients::ConfigurationClient* GetConfigurator() const noexcept
        {
            return Configurator.get();
        }
        /// Get name client of this service.
        [[nodiscard]] inline Clients::NameClient* GetNameResolver() const noexcept
        {
            return NameResolver.get();
        }

        /**
         * @brief Description for all program options supported by this service.
         * @details
         *  Values passed to this program can be got from OptionVariables in
         *  OnInstall() stage.
         */
        boost::program_options::options_description OptionDescription;
        /// Map that stores values parsed from the program command line.
        boost::program_options::variables_map OptionVariables;

        /// Invoked when this service is connected to a Redis server.
        virtual void OnConnect() {};

        /// Invoked when install this service.
        virtual void OnInstall() {};
        /// Invoked when uninstall this service.
        virtual void OnUninstall() {};

        /// Invoked when this service is paused.
        virtual void OnPause() {};
        /// Invoked when this service is resumed.
        virtual void OnResume() {};

        /**
         * @brief Invoked in every frame.
         * @retval true Launcher should keep program alive.
         * @retval false Launcher should stop the program.
         */
        virtual void OnUpdate() {};

        /// If false, service program will stop in next frame.
        std::atomic_bool LifeFlag {true};

        /// Constructor that will bind the service name.
        explicit Service(std::string name);

    public:
        /// Name of this service.
        const std::string Name;

        /// Check whether this service is enable or not.
        [[nodiscard]] inline bool IsEnable() const noexcept
        {
            return Enable.load();
        }
        /**
         * @brief Pause this service.
         * @details
         *  This method will only effect OnUpdate() method,
         *  and the message loop will keep going on.
         */
        void Pause();
        /// Resume this service.
        void Resume();

        /**
         * @brief Establish a connection to the Redis server.
         * @param port Port of the Redis server.
         * @param ip IP of the Redis server.
         */
        void Connect(unsigned int port = 6379, const std::string& ip = "127.0.0.1");
        /**
         * @brief Install this service.
         */
        void Install();
        /**
         * @brief Update this service and consume Redis messages in parallel.
         * @retval true Launcher should continue main loop.
         * @retval false Launcher should stop the program.
         */
        bool Update();
        /**
         * @brief Uninstall this service.
         */
        void Uninstall();
    };
}

#ifndef CONSTRUCTOR
#define CONSTRUCTOR(Name, Arguments...) public: Name(##Arguments) : Gaia::Framework::Service(#Name)
#endif

ext: hpp

max_stars_repo_name: GaiaCommittee/GaiaFramework

max_issues_repo_name: GaiaCommittee/GaiaFramework

max_forks_repo_name: GaiaCommittee/GaiaFramework


text:
/* -*- Mode: C++; tab-width: 4; indent-tabs-mode: t; c-basic-offset: 4 -*- vim:set ts=4 sw=4 sts=4 noet: */
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#include "filter.h"

#include <cassert>
#include <cstdlib>
#include <cstring>
#include <iostream>
#include <memory>
#include <string>
#include <vector>

#define BOOST_FILESYSTEM_NO_DEPRECATED
#include <boost/filesystem/fstream.hpp>
#include <boost/uuid/uuid_io.hpp>

#include "lo.pb.h"

#include "filter/filter_loader.h"
#include "isdf/isdf.h"

namespace flint {
namespace filter {

namespace {

class Filter {
public:
	Filter(const Filter &) = delete;
	Filter &operator=(const Filter &) = delete;

	Filter() = default;

	void ReadHeader(int /*size*/) const {
		// ignore header
	}

	void ReadColumn(std::unique_ptr<lo::Column> &&column) {
		boost::uuids::uuid u;
		std::memcpy(&u, column->uuid().data(), u.size());
		std::string s;
		if (u.is_nil()) {
			s = column->name();
		} else {
			s = to_string(u) + ":" + column->name();
			if (column->has_label()) s += "@" + column->label();
		}
		int size = column->col() * column->row();
		num_objs_ += size;
		num_bytes_descs_ += (sizeof(std::uint32_t)+s.size()) * size;
		num_bytes_units_ += (sizeof(std::uint32_t)+column->unit().size()) * size;
		for (int i=0;i<size;i++) {
			descriptions_.push_back(s);
			units_.push_back(column->unit());
		}
	}

	void Write(std::ostream *os) const {
		isdf::ISDFHeader header;
		header.num_objs = num_objs_;
		header.num_bytes_comment = 0;
		header.num_bytes_descs = num_bytes_descs_;
		header.num_bytes_units = num_bytes_units_;

		char buf[sizeof(isdf::ISDFHeader)];
		std::memcpy(buf, &header, sizeof(header));
		os->write(buf, sizeof(header));

		for (const auto &desc : descriptions_) {
			std::uint32_t len = desc.size();
			std::memcpy(buf, &len, sizeof(len));
			os->write(buf, sizeof(len));
			os->write(desc.c_str(), len);
		}
		for (const auto &u : units_) {
			std::uint32_t len = u.size();
			std::memcpy(buf, &len, sizeof(len));
			os->write(buf, sizeof(len));
			os->write(u.c_str(), len);
		}
	}

private:
	std::uint32_t num_objs_;
	std::uint32_t num_bytes_descs_;
	std::uint32_t num_bytes_units_;
	std::vector<std::string> descriptions_;
	std::vector<std::string> units_;
};

}

bool Isdh(const boost::filesystem::path &filter_file,
		  const boost::filesystem::path &output_file)
{
	std::unique_ptr<Filter> filter(new Filter());
	{
		std::unique_ptr<FilterLoader> loader(new FilterLoader(filter_file));
		if (!loader->Load(filter.get())) return false;
	}
	boost::filesystem::ofstream ofs(output_file, std::ios::out|std::ios::binary);
	if (!ofs.is_open()) {
		std::cerr << "could not open output file: " << output_file << std::endl;
		return false;
	}
	filter->Write(&ofs);
	ofs.close();
	return true;
}

}
}

ext: cc

max_stars_repo_name: Abhisheknishant/Flint

max_issues_repo_name: Abhisheknishant/Flint

max_forks_repo_name: Abhisheknishant/Flint


text:
/*=============================================================================
    Copyright (c) 2001-2007 Joel de Guzman

    Distributed under the Boost Software License, Version 1.0. (See accompanying
    file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
==============================================================================*/
#ifndef BOOST_PP_IS_ITERATING
#ifndef PHOENIX_CORE_DETAIL_COMPOSITE_HPP
#define PHOENIX_CORE_DETAIL_COMPOSITE_HPP

#include <boost/preprocessor/iterate.hpp>

#define BOOST_PP_ITERATION_PARAMS_1                                             \
    (3, (3, PHOENIX_COMPOSITE_LIMIT,                                            \
    "boost/spirit/home/phoenix/core/detail/composite.hpp"))
#include BOOST_PP_ITERATE()

#endif

///////////////////////////////////////////////////////////////////////////////
//
//  Preprocessor vertical repetition code
//
///////////////////////////////////////////////////////////////////////////////
#else // defined(BOOST_PP_IS_ITERATING)

#define N BOOST_PP_ITERATION()

    template <BOOST_PP_ENUM_PARAMS(N, typename U)>
    composite(BOOST_PP_ENUM_BINARY_PARAMS(N, U, & _))
        : base_type(BOOST_PP_ENUM_PARAMS(N, _)) {}

#undef N
#endif // defined(BOOST_PP_IS_ITERATING)

max_stars_repo_name: importlib/klib

max_issues_repo_name: isuhao/klib

max_forks_repo_name: isuhao/klib


text:
#include <chrono>
#include <cstring> // strncmp
#include <iostream>

#include <boost/asio.hpp>

#include "accelerator/transcoder.hh"

/*------------------------------------------------------------------------------------------------*/

int
main(int argc, char** argv)
{
  using boost::asio::ip::address_v4;
  using boost::asio::ip::udp;

  const auto usage = [&]
  {
    std::cerr << "Usage\n";
    std::cerr << argv[0] << " server server_port app_ip app_port\n";
    std::cerr << argv[0] << " client app_port server_ip server_port\n";
    std::exit(1);
  };

  if (argc != 5)
  {
    usage();
  }
  try
  {
    boost::asio::io_service io;

    // Connection with the proxied application.
    udp::socket app_socket{io};
    udp::endpoint app_endpoint;

    // Encoded tunnel.
    udp::socket socket{io};
    udp::endpoint endpoint;

    // The transcoder is fully symmetric as it handles two-ways communications. Still, UDP
    // requires a client and a server.
    if (std::strncmp(argv[1], "server", 7) == 0)
    {
      const auto server_port = static_cast<unsigned short>(std::atoi(argv[2]));
      const auto app_url = argv[3];
      const auto app_port = argv[4];

      app_socket = udp::socket{io, udp::endpoint(udp::v4(), 0)}; // a client socket
      app_socket.set_option(boost::asio::socket_base::receive_buffer_size{8192*64});
      app_socket.set_option(boost::asio::socket_base::send_buffer_size{8192*64});

      socket = udp::socket{io, udp::endpoint{udp::v4(), server_port}};  // a server socket
      socket.set_option(boost::asio::socket_base::receive_buffer_size{8192*64});
      socket.set_option(boost::asio::socket_base::send_buffer_size{8192*64});

      udp::resolver resolver(io);
      app_endpoint = *resolver.resolve({udp::v4(), app_url, app_port});
    }
    else if (std::strncmp(argv[1], "client", 7) == 0)
    {
      const auto app_port = static_cast<unsigned short>(std::atoi(argv[2]));
      const auto server_url = argv[3];
      const auto server_port = argv[4];

      app_socket = udp::socket{io, udp::endpoint{udp::v4(), app_port}}; // a server socket
      app_socket.set_option(boost::asio::socket_base::receive_buffer_size{8192*64});
      app_socket.set_option(boost::asio::socket_base::send_buffer_size{8192*64});

      socket = udp::socket{io, udp::endpoint(udp::v4(), 0)};            // a client socket
      socket.set_option(boost::asio::socket_base::receive_buffer_size{8192*64});
      socket.set_option(boost::asio::socket_base::send_buffer_size{8192*64});

      udp::resolver resolver(io);
      endpoint = *resolver.resolve({udp::v4(), server_url, server_port});
    }
    else
    {
      usage();
    }

    // Create and configure the transcoder that handles encoding/decoding. The client/server status
    // is completely transparent to the transcoder.
    transcoder t{io, app_socket, app_endpoint, socket, endpoint};

    // Launch the event loop (runs forever).
    io.run();
  }
  catch (const std::exception& e)
  {
    std::cerr << e.what() << '\n';
  }
  return 0;
}

/*------------------------------------------------------------------------------------------------*/

max_stars_repo_name: ahamez/netcode

max_issues_repo_name: ahamez/netcode

max_forks_repo_name: ahamez/netcode


text:
/*
 * (C) Copyright 2017-2019 UCAR
 *
 * This software is licensed under the terms of the Apache Licence Version 2.0
 * which can be obtained at http://www.apache.org/licenses/LICENSE-2.0.
 */

#include "ufo/ObsBias.h"

#include <Eigen/Core>
#include <fstream>
#include <iomanip>
#include <set>

#include "ioda/ObsVector.h"

#include "oops/base/Variables.h"
#include "oops/util/IntSetParser.h"
#include "oops/util/Logger.h"

#include "ufo/ObsBiasIncrement.h"
#include "ufo/ObsDiagnostics.h"

namespace ufo {

// -----------------------------------------------------------------------------

ObsBias::ObsBias(ioda::ObsSpace & odb, const eckit::Configuration & conf)
  : predbases_(0), jobs_(0), odb_(odb), conf_(conf) {
  oops::Log::trace() << "ObsBias::create starting." << std::endl;

  // Get the jobs(channels)
  if (conf_.has("obs bias.jobs")) {
    const std::set<int> jobs = oops::parseIntSet(conf_.getString("obs bias.jobs"));
    jobs_.assign(jobs.begin(), jobs.end());
  }

  // Predictor factory
  if (conf_.has("obs bias.predictors")) {
    std::vector<eckit::LocalConfiguration> confs;
    conf_.get("obs bias.predictors", confs);
    for (std::size_t j = 0; j < confs.size(); ++j) {
      std::shared_ptr<PredictorBase> pred(PredictorFactory::create(confs[j], jobs_));
      predbases_.push_back(pred);
      prednames_.push_back(pred->name());
      geovars_ += pred->requiredGeovars();
      hdiags_ += pred->requiredHdiagnostics();

      // Reserve the space for ObsBiasTerm for predictor
      if (jobs_.size() > 0) {
        for (auto & job : jobs_) {
          hdiags_ += oops::Variables({prednames_.back() + "_" + std::to_string(job)});
        }
      } else {
        hdiags_ += oops::Variables({prednames_.back()});
      }
    }
  }

  if (prednames_.size() * jobs_.size() > 0) {
    // Initialize the biascoeffs to ZERO
    biascoeffs_.resize(prednames_.size() * jobs_.size());
    std::fill(biascoeffs_.begin(), biascoeffs_.end(), 0.0);

    // Read or initialize bias coefficients
    this->read(conf);
  }

  oops::Log::trace() << "ObsBias::create done." << std::endl;
}

// -----------------------------------------------------------------------------

ObsBias::ObsBias(const ObsBias & other, const bool copy)
  : odb_(other.odb_), conf_(other.conf_), predbases_(other.predbases_),
    prednames_(other.prednames_), jobs_(other.jobs_),
    geovars_(other.geovars_), hdiags_(other.hdiags_) {
  oops::Log::trace() << "ObsBias::copy ctor starting." << std::endl;

  // Initialize the biascoeffs
  biascoeffs_.resize(prednames_.size() * jobs_.size());
  std::fill(biascoeffs_.begin(), biascoeffs_.end(), 0.0);

  // Copy the bias coeff data
  if (copy && biascoeffs_.size() > 0) *this = other;

  oops::Log::trace() << "ObsBias::copy ctor done." << std::endl;
}

// -----------------------------------------------------------------------------

ObsBias & ObsBias::operator+=(const ObsBiasIncrement & dx) {
  for (std::size_t jj = 0; jj < biascoeffs_.size(); ++jj)
      biascoeffs_[jj] += dx[jj];
  return *this;
}

// -----------------------------------------------------------------------------

ObsBias & ObsBias::operator=(const ObsBias & rhs) {
  if (rhs.size() > 0 && this->size() == rhs.size()) {
    conf_       = rhs.conf_;
    biascoeffs_ = rhs.biascoeffs_;
    predbases_  = rhs.predbases_;
    prednames_  = rhs.prednames_;
    jobs_       = rhs.jobs_;
    geovars_    = rhs.geovars_;
    hdiags_     = rhs.hdiags_;
  }
  return *this;
}

// -----------------------------------------------------------------------------

void ObsBias::read(const eckit::Configuration & conf) {
  oops::Log::trace() << "ObsBias::read and initialize from file, starting "<< std::endl;

  // Bias coefficients input file name
  std::string input_filename;
  if (conf.has("obs bias.abias_in")) {
    input_filename = conf.getString("obs bias.abias_in");
    oops::Log::debug() << "ObsBias::initialize coefficients from file: "
                       << input_filename <<  std::endl;

    // Default predictor names from GSI
    // temporary solution, we should have a self-explanatory obsbias file
    const std::vector<std::string> gsi_predictors = {"constant",
                                                     "zenith_angle",
                                                     "cloud_liquid_water",
                                                     "lapse_rate_order_2",
                                                     "lapse_rate",
                                                     "cosine_of_latitude_times_orbit_node",
                                                     "sine_of_latitude",
                                                     "emissivity",
                                                     "scan_angle_order_4",
                                                     "scan_angle_order_3",
                                                     "scan_angle_order_2",
                                                     "scan_angle"
                                                     };
    std::ifstream infile(input_filename);

    // get the sensor id
    std::string sensor = conf.getString("obs bias.sensor");
    oops::Log::debug() << "ObsBias::initialize coefficients for sensor: "
                       << sensor <<  std::endl;

    std::size_t ich;     //  sequential number
    std::string nusis;   //  sensor/instrument/satellite
    std::size_t nuchan;  //  channel number
    float tlap, tsum;
    std::size_t ntlapupdate;

    if (infile.is_open())
    {
      oops::Log::debug() << "ObsBias:: prior file is opened" << std::endl;
      float par;
      while (infile >> ich)
      {
        infile >> nusis;
        infile >> nuchan;
        infile >> tlap;
        infile >> tsum;
        infile >> ntlapupdate;
        if (nusis == sensor) {
          auto ijob = std::find(jobs_.begin(), jobs_.end(), nuchan);
          if (ijob != jobs_.end()) {
            int j = std::distance(jobs_.begin(), ijob);

            for (auto & item : gsi_predictors) {
              infile >> par;
              auto ipred = std::find(prednames_.begin(), prednames_.end(), item);
              if (ipred != prednames_.end()) {
                int p = std::distance(prednames_.begin(), ipred);
                biascoeffs_.at(j*prednames_.size() + p) = static_cast<double>(par);
              }
            }
          } else {
            for (auto & item : gsi_predictors)
              infile >> par;
          }
        } else {
          for (auto & item : gsi_predictors)
            infile >> par;
        }
      }
      infile.close();
      oops::Log::debug() << "ObsBias:: read prior from " << input_filename << std::endl;
    } else {
      oops::Log::error() << "Unable to open file : " << input_filename << std::endl;
      ABORT("Unable to open bias prior file ");
    }
  } else {
    oops::Log::warning() << "ObsBias::prior file is NOT available, starting from ZERO"
                         << std::endl;
  }

  oops::Log::trace() << "ObsBias::read and initilization done " << std::endl;
}

// -----------------------------------------------------------------------------

void ObsBias::write(const eckit::Configuration & conf) const {
  oops::Log::trace() << "ObsBias::write to file not implemented" << std::endl;

  // Bias coefficients output file name
  std::string output_filename;
  if (conf.has("ObsBias.abias_out")) {
    output_filename = conf.getString("ObsBias.abias_out");
  }

  oops::Log::trace() << "ObsBias::write to file done" << std::endl;
}

// -----------------------------------------------------------------------------

void ObsBias::computeObsBias(ioda::ObsVector & ybias,
                             ObsDiagnostics & ydiags,
                             const std::vector<ioda::ObsVector> & predData) const {
  oops::Log::trace() << "ObsBias::computeObsBias starting" << std::endl;

  if (this->size() > 0) {
    const std::size_t nlocs  = ybias.nlocs();
    const std::size_t npreds = prednames_.size();
    const std::size_t njobs  = jobs_.size();

    ASSERT(biascoeffs_.size() == npreds*njobs);
    ASSERT(predData.size() == npreds);
    ASSERT(ybias.nvars() == njobs);

    /* predData memory layout (npreds X nlocs X njobs)
     *       Loc     0      1      2       3
     *             --------------------------
     * pred1 Chan1 | 0      3      6       9
     *       Chan2 | 1      4      7      10
     *       ....  | 2      5      8      11
     *
     * pred2 Chan1 |12     15     18      21
     *       Chan2 |13     16     19      22
     *       ....  |14     17     20      23
     */

    ybias.zero();

    /* ybias memory layout (nlocs X njobs)
     *     ch1    ch2    ch3     ch4
     * Loc --------------------------
     *  0 | 0      1      2       3
     *  1 | 4      5      6       7
     *  2 | 8      9     10      11
     *  3 |12     13     14      15
     *  4 |16     17     18      19
     * ...|
     */

    /* map bias coeff to eigen matrix npreds X njobs (read only)
     * bias coeff memory layout (npreds X njobs)
     *        ch1    ch2    ch3     ch4
     *       --------------------------
     * pred1 | 0      1      2       3
     * pred2 | 4      5      6       7
     * pred3 | 8      9     10      11
     * ....  |
     */
    Eigen::Map<const Eigen::MatrixXd> coeffs(biascoeffs_.data(), npreds, njobs);

    std::vector<double> biasTerm(nlocs, 0.0);
    //  For each channel: ( nlocs X 1 ) =  ( nlocs X npreds ) * (  npreds X 1 )
    for (std::size_t jch = 0; jch < njobs; ++jch) {
      for (std::size_t jp = 0; jp < npreds; ++jp) {
        // axpy
        const double beta = coeffs(jp, jch);
        for (std::size_t jl = 0; jl < nlocs; ++jl) {
          biasTerm[jl] = predData[jp][jl*njobs+jch] * beta;
          ybias[jl*njobs+jch] += biasTerm[jl];
        }
        // Save ObsBiasTerms (bias_coeff * predictor) for QC
        const std::string varname = predbases_[jp]->name() + "_" + std::to_string(jobs_[jch]);
        if (ydiags.has(varname)) {
          ydiags.save(biasTerm, varname, 1);
        } else {
          oops::Log::error() << varname << " is not reserved in ydiags !" << std::endl;
          ABORT("ObsBiasTerm variable is not reserved in ydiags");
        }
      }
    }
  }

  oops::Log::trace() << "ObsBias::computeObsBias done." << std::endl;
}

// -----------------------------------------------------------------------------
std::vector<ioda::ObsVector> ObsBias::computePredictors(const GeoVaLs & geovals,
                                                        const ObsDiagnostics & ydiags) const {
  const std::size_t npreds = predbases_.size();

  std::vector<ioda::ObsVector> predData(npreds, ioda::ObsVector(odb_));

  for (std::size_t p = 0; p < npreds; ++p) {
    predbases_[p]->compute(odb_, geovals, ydiags, predData[p]);
    predData[p].save(predbases_[p]->name() + "Predictor");
  }

  oops::Log::trace() << "ObsBias::computePredictors done." << std::endl;
  return predData;
}

// -----------------------------------------------------------------------------

double ObsBias::norm() const {
  oops::Log::trace() << "ObsBias::norm starting." << std::endl;
  double zz = 0.0;
  for (std::size_t jj = 0; jj < biascoeffs_.size(); ++jj) {
    zz += biascoeffs_[jj] * biascoeffs_[jj];
  }
  if (biascoeffs_.size() > 0) zz = std::sqrt(zz/biascoeffs_.size());
  oops::Log::trace() << "ObsBias::norm done." << std::endl;
  return zz;
}

// -----------------------------------------------------------------------------

void ObsBias::print(std::ostream & os) const {
  if (this->size() > 0) {
    // map bias coeffs to eigen matrix (writable)
    Eigen::Map<const Eigen::MatrixXd>
      coeffs(biascoeffs_.data(), prednames_.size(), jobs_.size());

    os << "ObsBias::print " << std::endl;
    os << "---------------------------------------------------------------" << std::endl;
    auto njobs = jobs_.size();
    for (std::size_t p = 0; p < prednames_.size(); ++p) {
      os << std::fixed << std::setw(20) << prednames_[p]
         << ":  Min= " << std::setw(15) << std::setprecision(8)
         << coeffs.row(p).minCoeff()
         << ",  Max= " << std::setw(15) << std::setprecision(8)
         << coeffs.row(p).maxCoeff()
         << ",  Norm= " << std::setw(15) << std::setprecision(8)
         << coeffs.row(p).norm()
         << std::endl;
    }
    os << "---------------------------------------------------------------" << std::endl;
    os << "ObsBias::print done" << std::endl;
  }
}

// -----------------------------------------------------------------------------

}  // namespace ufo

max_stars_repo_name: guoqing-noaa/ufo

max_issues_repo_name: guoqing-noaa/ufo

max_forks_repo_name: guoqing-noaa/ufo


text:
// This file is part of MLDB. Copyright 2015 mldb.ai inc. All rights reserved.

/* sparse_features.cc
   Jeremy Barnes, 25 June 2003
   Copyright (c) 2003 Jeremy Barnes.  All rights reserved.
   $Source$

   Implementation of a sparse feature space.
*/

#include "sparse_features.h"
#include "mldb/plugins/jml/jml/registry.h"
#include "mldb/base/parse_context.h"
#include "mldb/utils/file_functions.h"
#include <boost/utility.hpp>

using namespace std;
using namespace ML;
using namespace MLDB::DB;

namespace ML {

/*****************************************************************************/
/* SPARSE_FEATURE_SPACE                                                      */
/*****************************************************************************/

Sparse_Feature_Space::Sparse_Feature_Space()
{
}

Sparse_Feature_Space::Sparse_Feature_Space(DB::Store_Reader & store)
{
    reconstitute(store);
}

Sparse_Feature_Space::~Sparse_Feature_Space()
{
}

void Sparse_Feature_Space::init()
{
}

void Sparse_Feature_Space::
set_info(const Feature & feature, const Feature_Info & info)
{
    if (feature == MISSING_FEATURE)
        throw Exception("attempt to set feature info for missing feature");
    //cerr << this << endl;
    //cerr << "setting info for feature " << feature << endl;
    //cerr << "string_lookup.size() = " << string_lookup.size() << endl;
    Mutable_Feature_Info & infoa = info_array.at(feature.type());
    infoa = info;
}

Feature_Info
Sparse_Feature_Space::
info(const ML::Feature & feature) const
{
    if (feature == MISSING_FEATURE)
        return MISSING_FEATURE_INFO;

    int id = feature.type();
    if (id < 0 || id >= info_array.size())
        throw Exception("Feature with unknown ID " + ostream_format(id)
                        + " requested from Sparse_Feature_Space: "
                        + feature.print());

    return info_array[id];
}

std::string Sparse_Feature_Space::
print(const ML::Feature & feature) const
{
    if (feature == MISSING_FEATURE)
        return "MISSING";
    return get_name(feature);
}

bool Sparse_Feature_Space::
parse(ParseContext & context, ML::Feature & feature) const
{
    throw Exception("unimplemented");
}

void Sparse_Feature_Space::
expect(ParseContext & context, ML::Feature & feature) const
{
    throw Exception("unimplemented");
}

void Sparse_Feature_Space::
serialize(DB::Store_Writer & store, const ML::Feature & feature) const
{
    store << get_name(feature);
}

void Sparse_Feature_Space::
reconstitute(DB::Store_Reader & store, ML::Feature & feature) const
{
    std::string name;
    store >> name;
    feature = get_feature(name);
}

Sparse_Feature_Space * Sparse_Feature_Space::make_copy() const
{
    return new Sparse_Feature_Space(*this);
}

namespace {
static const std::string ID = "SPARSE_FEATURE_SPACE";
static const compact_size_t VERSION(0);
} // file scope

std::string Sparse_Feature_Space::class_id() const
{
    return ID;
}

void Sparse_Feature_Space::serialize(DB::Store_Writer & store) const
{
    store << ID << VERSION;
}

void Sparse_Feature_Space::
reconstitute(DB::Store_Reader & store,
             const std::shared_ptr<const Feature_Space> & feature_space)
{
    string type;
    compact_size_t version;
    store >> type >> version;
    if (type != ID)
        throw Exception("Attempt to reconstitute " + type + "; expected "
                        + ID);
    if (version > VERSION)
        throw Exception(format("Attemp to reconstitute sparse fs version %zd, "
                               "only <= %d supported", version.size_,
                               0));
}

void Sparse_Feature_Space::
reconstitute(DB::Store_Reader & store)
{
    //cerr << "sparse_feature_space reconstitution" << endl;
    string type;
    compact_size_t version;
    store >> type >> version;
    if (type != ID)
        throw Exception("Attempt to reconstitute " + type + "; expected "
                        + ID);
    if (version > VERSION)
        throw Exception(format("Attemp to reconstitute sparse fs version %zd, "
                               "only <= %d supported", version.size_,
                               0));
}

Feature
Sparse_Feature_Space::
make_feature(const std::string & name, const Feature_Info & info)
{
    //cerr << "make_feature(\"" << name << "\")" << endl;

    string_lookup_type::const_iterator it = string_lookup.find(name);
    if (it == string_lookup.end()) {
        string_lookup[name] = info_array.size();
        info_array.push_back(info);
        name_array.push_back(name);
        return Feature(info_array.size() - 1);
    }

    return Feature(it->second);
}

std::string Sparse_Feature_Space::get_name(const Feature & feature) const
{
    int id = feature.type();
    if (id < 0 || id >= name_array.size())
        throw Exception("Feature with unknown ID " + ostream_format(id)
                        + " requested from Sparse_Feature_Space: "
                        + feature.print());

    return name_array[id];
}

Feature Sparse_Feature_Space::get_feature(const std::string & name) const
{
    string_lookup_type::const_iterator it = string_lookup.find(name);
    if (it == string_lookup.end()) {
        return const_cast<Sparse_Feature_Space *>(this)->make_feature(name);
        //throw Exception("Feature with name " + name + " not found in "
        //                "Sparse_Feature_Space");
    }
    return Feature(it->second);
}

/*****************************************************************************/
/* SPARSE_TRAINING_DATA                                                      */
/*****************************************************************************/

Sparse_Training_Data::Sparse_Training_Data()
{
}

Sparse_Training_Data::
Sparse_Training_Data(const std::string & filename)
{
    init(filename);
}

Sparse_Training_Data::
Sparse_Training_Data(const std::string & filename,
                     const std::shared_ptr<Sparse_Feature_Space> & fs)
{
    init(filename, fs);
}

Sparse_Training_Data::~Sparse_Training_Data()
{
}

void
Sparse_Training_Data::
expect_feature(ParseContext & c, Mutable_Feature_Set & features,
               Sparse_Feature_Space & feature_space,
               bool & guessed_wrong)
{
    //__builtin_prefetch(c.get_pos() + 128, 0, 0);

    string name = expect_feature_name(c);
    //cerr << "name = " << name << endl;
    Feature feature = feature_space.make_feature(name);
    //cerr << "feature = " << feature << endl;
    int num = feature.type();

    //cerr << "num = " << num << endl;

    Mutable_Feature_Info & info
        = feature_space.info_array[num];

    bool categorical = !!info.categorical();

    if (categorical && (!c || *c != ':')) {
        /* A categorical feature can't have a missing label. */
        c.exception("categorical feature " + name + " missing value");
    }

    if (!categorical) {
        float value = 1.0;

        if (c && *c == ':') {
            /* It has a value attached. */
            ++c;
            if (c.match_float(value))
                features.add(feature, value);
            else {
                /* We guessed wrong... this is categorical */
                cerr << "guessed_wrong" << endl;
                cerr << "name = " << name << endl;
                cerr << "line = " << c.get_line() << endl;
                guessed_wrong = true;
                info.make_categorical();
                categorical = true;
            }

            /* Allow further values to be specified. */
            while (!categorical && c && *c == ',') {
                ++c;  // skip the ','
                if (c.match_float(value))
                    features.add(feature, value);
                else {
                    guessed_wrong = true;
                    info.make_categorical();
                    categorical = true;
                }
            }
        }
        else if (!c || *c == '\n' || isspace(*c)) {
            /* Feature with defaulted value */
            features.add(feature, 1.0);
        }
    }

    if (categorical) {
        if (c && c.match_literal(':')) {
            /* It has a value attached. */
            string category = expect_feature_name(c);
            float value = info.mutable_categorical()->parse_or_add(category);
            features.add(feature, value);
        }

        /* Allow further values to be specified. */
        while (c && c.match_literal(',')) {
            string category = expect_feature_name(c);
            float value = info.mutable_categorical()->parse_or_add(category);
            features.add(feature, value);
        }
    }
}

namespace {

bool match_label(ParseContext & c, Mutable_Feature_Set & features,
                 Sparse_Feature_Space & feature_space)
{
    /* If the first part is numeric and doesn't include a ":", we assume that
       it's a label.  This is for compatibility with old-format files.

       The new files will have LABEL:xxx, so this will not be necessary.
    */
    float val;
    if (c.match_float(val) && isspace(*c)) {
        features.add(feature_space.make_feature("LABEL"), val);
        return true;
    }
    return false;
}

} // file scope

void Sparse_Training_Data::
init(const std::string & filename,
     std::shared_ptr<Sparse_Feature_Space> feature_space)
{
    vector<string> filenames(1, filename);
    init(filenames, feature_space);
}

void Sparse_Training_Data::
init(const std::vector<std::string> & filenames,
     std::shared_ptr<Sparse_Feature_Space> feature_space)
{
    //cerr << "sparse data init from file" << endl;

    Training_Data::init(feature_space);
    sparse_fs = feature_space;

    /* Did we guess wrongly about which are categorical?  If so, we need to
       reparse the array.  We keep on going until we are right about which
       are categorical. */
    bool guessed_wrong = false;

    /* Keep on going until we get all the categorical values correct. */
    do {
        guessed_wrong = false;

        clear();

        for (unsigned i = 0;  i < filenames.size();  ++i) {
            string filename = filenames[i];
            //cerr << "loading data from " << filename << endl;
            /* We load the training data from the file. */
            File_Read_Buffer file(filename);

            ParseContext c(filename, file.start(), file.end());

            /* Skip over the header. */
            c.skip_line();

            while (c) {
                std::shared_ptr<Mutable_Feature_Set>
                    features(new Mutable_Feature_Set());

                try {
                    c.skip_whitespace();
                    if (c.match_eol()) continue;   // skip blank lines
                    if (*c == '#') {
                        c.skip_line();  // skip comments
                        continue;
                    }

                    /* Allow a label to be specified, with the implicit feature
                       name "LABEL", in the first position. */
                    if (match_label(c, *features, *feature_space))
                        c.skip_whitespace();

                    while (!c.match_eol()) {
                        if (*c == '#') {
                            // skip comments after the data
                            c.skip_line();
                            break;
                        }
                        else {
                            expect_feature(c, *features, *feature_space,
                                           guessed_wrong);
                            c.skip_whitespace();
                        }
                    }
                }
                catch (const Exception & exc) {
                    cerr << "warning: " << exc.what() << ": skipping line"
                         << endl;
                    c.skip_line();
                }

                //cerr << "features are " << feature_space->print(*features) << endl;

                add_example(features);
            }
        }

    } while (guessed_wrong);
}

void Sparse_Training_Data::
init(const std::string & filename)
{
    init(filename,
         std::shared_ptr<Sparse_Feature_Space>(new Sparse_Feature_Space()));
}

Sparse_Training_Data * Sparse_Training_Data::make_copy() const
{
    return new Sparse_Training_Data(*this);
}

Sparse_Training_Data * Sparse_Training_Data::make_type() const
{
    Sparse_Training_Data * result = new Sparse_Training_Data();
    result->sparse_fs = sparse_fs;
    return result;
}

/*****************************************************************************/
/* REGISTRATION                                                              */
/*****************************************************************************/

namespace {
Register_Factory<Feature_Space, Sparse_Feature_Space> SFS_REG(ID);
} // file scope

} // namespace ML

max_stars_repo_name: tomzhang/mldb

max_issues_repo_name: tomzhang/mldb

max_forks_repo_name: tomzhang/mldb


text:
#ifndef VIENNAGRID_IO_SERIALIZATION_HPP
#define VIENNAGRID_IO_SERIALIZATION_HPP

/* =======================================================================
   Copyright (c) 2011-2014, Institute for Microelectronics,
                            Institute for Analysis and Scientific Computing,
                            TU Wien.

                            -----------------
                     ViennaGrid - The Vienna Grid Library
                            -----------------

   License:      MIT (X11), see file LICENSE in the base directory
======================================================================= */

/** @file serialization.hpp
 *  @brief    A mesh wrapper class which models the Boost Serialization concept: http://www.boost.org/libs/serialization/
 */

#include "viennagrid/forwards.hpp"
#include "viennagrid/mesh/mesh.hpp"

#include <boost/archive/text_iarchive.hpp>
#include <boost/archive/text_oarchive.hpp>
#include <boost/serialization/base_object.hpp>
#include <boost/serialization/utility.hpp>
#include <boost/serialization/list.hpp>
#include <boost/serialization/assume_abstract.hpp>
#include <boost/shared_ptr.hpp>

/** @file viennagrid/io/serialization.hpp
    @brief Provides a serialization mechanism for use with Boost
*/

namespace viennagrid
{
  namespace io
  {

    /** @brief Provides an exception for the bad serialization state */
    class bad_serialization_state_exception : public std::exception
    {
      public:
        virtual const char* what() const throw()
        {
          std::stringstream ss;
          ss << "* ViennaGrid: Bad serialization state: " << msg_ << "!";
          return ss.str().c_str();
        }

        bad_serialization_state_exception(std::string msg) : msg_(msg) {}

        virtual ~bad_serialization_state_exception() throw() {}

      private:
        std::string msg_;
    };

    /** @brief Mesh wrapper which models the Boost serialization concept
     *
     */
    template<typename MeshT>
    struct mesh_serializer
    {
    private:
      typedef typename viennagrid::result_of::cell_tag<MeshT>::type                        CellTag;
      typedef typename viennagrid::result_of::const_vertex_range<MeshT>::type              ConstVertexRange;
      typedef typename viennagrid::result_of::iterator<ConstVertexRange>::type             ConstVertexIterator;
      typedef typename viennagrid::result_of::point<MeshT>::type                           PointType;
      typedef typename viennagrid::result_of::vertex<MeshT>::type                          VertexType;
      typedef typename viennagrid::result_of::vertex_handle<MeshT>::type                   VertexHandleType;
      typedef typename viennagrid::result_of::id<VertexType>::type                         VertexIDType;
      typedef typename viennagrid::result_of::cell<MeshT>::type                            CellType;
      typedef typename viennagrid::result_of::const_cell_range<MeshT>::type                ConstCellRange;
      typedef typename viennagrid::result_of::iterator<ConstCellRange>::type               ConstCellIterator;
      typedef typename viennagrid::result_of::const_vertex_range<CellType>::type           ConstVertexOnCellRange;
      typedef typename viennagrid::result_of::iterator<ConstVertexOnCellRange>::type       ConstVertexOnCellIterator;

      static const int DIMG = PointType::dim;

      friend class boost::serialization::access;

      /** @brief Save function is used to 'read' the data from the mesh */
      template<class Archive>
      void save(Archive & ar, const unsigned int version) const
      {
        if (!mesh_pointer)
          throw bad_serialization_state_exception( "Mesh not loaded into serialzer object" );

        MeshT const & mesh_obj = *mesh_pointer;

        // -----------------------------------------------
        // the geometry is read and transmitted
        //
        std::size_t point_size = viennagrid::vertices(mesh_obj).size();
        ar & point_size;
        ConstVertexRange vertices(mesh_obj);
        for (ConstVertexIterator vit = vertices.begin();
             vit != vertices.end(); ++vit)
        {
          for(int d = 0; d < DIMG; d++)
            ar & viennagrid::point(*vit)[d];
        }

        // -----------------------------------------------
        // the specific cells are read and transmitted
        //
        ConstCellRange cells(mesh_obj);

        std::size_t cell_size = viennagrid::cells(mesh_obj).size();
        ar & cell_size;

        for (ConstCellIterator cit = cells.begin();
        cit != cells.end(); ++cit)
        {
          ConstVertexOnCellRange vertices_on_cell(*cit);

          for (ConstVertexOnCellIterator vocit = vertices_on_cell.begin();
          vocit != vertices_on_cell.end();
          ++vocit)
          {
            std::size_t id = vocit->id().get();
            ar & id;
          }
        }
        // -----------------------------------------------
      }

      /** @brief Load function is used to 'write' the data to the mesh */
      template<class Archive>
      void load(Archive & ar, const unsigned int version)
      {
        if (!mesh_pointer)
          throw bad_serialization_state_exception( "Mesh not loaded into serialzer object" );

        MeshT & mesh_obj = *mesh_pointer;

        // -----------------------------------------------
        // the geometry is received and stored
        //
        std::size_t point_size;
        ar & point_size;

        for(std::size_t i = 0; i < point_size; i++)
        {
          PointType p;
          for(int d = 0; d < DIMG; d++)
            ar & p[d];

          viennagrid::make_vertex( *mesh_pointer, p );
        }
        // -----------------------------------------------

        // -----------------------------------------------
        // the specific cells are received and stored
        //

        std::size_t cell_size;
        ar & cell_size;
        for(std::size_t ci = 0; ci < cell_size; ci++)
        {
          const int num_vertices = viennagrid::boundary_elements<CellTag, vertex_tag>::num;
          VertexHandleType vertices[num_vertices];

          for (int j=0; j< num_vertices; ++j)
          {
            std::size_t id;
            ar & id;
            vertices[j] = viennagrid::find_by_id( mesh_obj, VertexIDType(id) ).handle();
          }

          viennagrid::make_cell( mesh_obj, vertices, vertices + num_vertices);
        }
        // -----------------------------------------------
      }

      // -----------------------------------------------
      // notify Boost Serialization that the save and load
      // procedures differ
      //
      BOOST_SERIALIZATION_SPLIT_MEMBER()
      // -----------------------------------------------
    public:
      /** @brief The default constructor*/
      mesh_serializer() : mesh_pointer(0) {}

      /** @brief The constructor expects a shared pointer on a mesh object and sets the state */
      mesh_serializer(MeshT & mesh_obj) : mesh_pointer(mesh_obj) {}

      /** @brief The load function enables to associate a mesh with the serialzer after
      a serializer object has been constructed. */
      inline void load(MeshT & mesh_obj) { mesh_pointer = &mesh;  }

      /** @brief The get function enables to retrieve the mesh pointer
      */
      inline MeshT & get() { return *mesh_pointer; }
      inline MeshT const & get() const { return *mesh_pointer; }

      /** @brief The functor returns a reference to the mesh state*/
      MeshT & operator()() { return *mesh_pointer; }
      MeshT const & operator()() const { return *mesh_pointer; }

      /** @brief The state holds a shared pointer to the mesh */

    private:
      MeshT * mesh_pointer;
    };

  } //namespace io
} //namespace viennagrid

#endif

max_stars_repo_name: viennagrid/viennagrid-dev

max_issues_repo_name: viennagrid/viennagrid-dev

max_forks_repo_name: viennagrid/viennagrid-dev


text:
#include "vtk_solution_reader.hpp"

#include "data_structures/solution_state.hpp"
#include "logger/logger.hpp"
#include "tinyxml2.h"

#include <boost/algorithm/string.hpp>
#include <boost/algorithm/string/split.hpp>
#include <boost/lexical_cast.hpp>

#include <iostream>

CartesianDomainDefinition
VTKSolutionReader::getCartesianDomain()
{
  tinyxml2::XMLDocument doc;
  auto                  timer = Logger::get().timer("VTKSolutionReader::getCartesianDomain() reading " + filename_);

  doc.LoadFile(filename_.c_str());

  const char* value_attribute = doc.FirstChildElement("VTKFile")
                                    ->FirstChildElement("StructuredGrid")
                                    ->FirstChildElement("Piece")
                                    ->FirstAttribute()
                                    ->Value();
  std::string attribute_string(value_attribute);
  // remove unwanted leading or trailing whitespace
  boost::trim_if(attribute_string, boost::is_any_of("\t "));

  // split by spsaces
  std::vector<std::string> tokens;
  boost::algorithm::split(tokens, attribute_string, boost::is_any_of("\t "), boost::token_compress_on);

  // cast to ints, will throw on failure
  const int nxbeg = boost::lexical_cast<int>(tokens[0]);
  const int nxend = boost::lexical_cast<int>(tokens[1]);
  const int nybeg = boost::lexical_cast<int>(tokens[2]);
  const int nyend = boost::lexical_cast<int>(tokens[3]);
  const int nzbeg = boost::lexical_cast<int>(tokens[4]);
  const int nzend = boost::lexical_cast<int>(tokens[5]);

  Logger::get().FatalAssert(nxbeg == 0, "VTKSolutionReader nxbeg == 0");
  Logger::get().FatalAssert(nybeg == 0, "VTKSolutionReader nybeg == 0");
  Logger::get().FatalAssert(nzbeg == 0, "VTKSolutionReader nzbeg == 0");

  CartesianDomainDefinition domain;
  domain.nx  = nxend;
  domain.ny  = nyend;
  dimension_ = 2;
  if (nzend == nzbeg) {
    domain.nz = 0;
  }
  else {
    domain.nz  = nzend;
    dimension_ = 3;
  }

  auto dataarray = doc.FirstChildElement("VTKFile")
                       ->FirstChildElement("StructuredGrid")
                       ->FirstChildElement("Piece")
                       ->FirstChildElement("CellData")
                       ->FirstChildElement("DataArray")
                       ->GetText();

  //  printf("Points GetText: %s\n", dataarray);
  std::string cell_data_string(dataarray);
  // remove unwanted leading or trailing whitespace
  boost::trim_if(cell_data_string, boost::is_any_of("\t \n"));

  std::vector<std::string> cell_data_tokens;
  boost::algorithm::split(cell_data_tokens, cell_data_string, boost::is_any_of("\t\n "), boost::token_compress_on);

  //  for (auto it = c
  auto points_array = doc.FirstChildElement("VTKFile")
                          ->FirstChildElement("StructuredGrid")
                          ->FirstChildElement("Piece")
                          ->FirstChildElement("Points")
                          ->FirstChildElement("DataArray")
                          ->GetText();

  std::string points_data_string(points_array);
  boost::trim_if(points_data_string, boost::is_any_of("\t \n"));
  std::vector<std::string> point_data_tokens;
  boost::algorithm::split(point_data_tokens, points_data_string, boost::is_any_of("\t\n "), boost::token_compress_on);

  const int npoints = point_data_tokens.size() / 3;
  Logger::get().FatalAssert(npoints == pow(nxend + 1, dimension_), "VTKSolutionReader npoints == (nxend+1)^dim");

  std::vector<double> x, y, z;
  for (size_t i = 0; i < point_data_tokens.size(); i += 3) {
    x.push_back(boost::lexical_cast<double>(point_data_tokens[i]));
    y.push_back(boost::lexical_cast<double>(point_data_tokens[i + 1]));
    z.push_back(boost::lexical_cast<double>(point_data_tokens[i + 2]));
  }

  domain.xbeg = *min_element(x.begin(), x.end());
  domain.xend = *max_element(x.begin(), x.end());
  domain.ybeg = *min_element(y.begin(), y.end());
  domain.zbeg = *min_element(z.begin(), z.end());

  return domain;
}

void
VTKSolutionReader::setSolution(const IndexArray& interior_indices, SolutionState& state)
{
  tinyxml2::XMLDocument doc;
  auto                  timer = Logger::get().timer("VTKSolutionReader::setSolution() reading " + filename_);
  doc.LoadFile(filename_.c_str());

  auto solnarray = doc.FirstChildElement("VTKFile")
                       ->FirstChildElement("StructuredGrid")
                       ->FirstChildElement("Piece")
                       ->FirstChildElement("CellData")
                       ->FirstChildElement("DataArray")
                       ->GetText();

  std::string cell_data_string(solnarray);
  // remove unwanted leading or trailing whitespace
  boost::trim_if(cell_data_string, boost::is_any_of("\t \n"));

  std::vector<std::string> cell_data_tokens;
  boost::algorithm::split(cell_data_tokens, cell_data_string, boost::is_any_of("\t\n "), boost::token_compress_on);

  std::vector<double> solution(cell_data_tokens.size());
  for (size_t i = 0; i < solution.size(); ++i) {
    solution[i] = boost::lexical_cast<double>(cell_data_tokens[i]);
  }

  auto idx = read_access_host(interior_indices);
  auto sol = write_access_host(state.getVec(0));

  for (int i = 0; i < idx.size(); ++i) {
    sol[idx[i]] = solution[i];
  }
}

ext: cpp

max_stars_repo_name: mlohry/maDGiCart-CH

max_issues_repo_name: mlohry/maDGiCart-CH

max_forks_repo_name: mlohry/maDGiCart-CH


text:
#include <boost/random/random_device.hpp>

int main()
{
	boost::random::random_device gen;
}