added reads and small regions management

src/Makefile.subdir

@@ -63,7 +63,8 @@ src_server_libmobject_server_la_SOURCES = \
   src/server/core/core-write-op.cpp \
   src/server/core/core-read-op.cpp \
   src/server/printer/print-write-op.c \
-  src/server/printer/print-read-op.c
+  src/server/printer/print-read-op.c \
+  src/server/core/fake-kv.cpp
 src_server_libmobject_server_la_CPPFLAGS = ${AM_CPPFLAGS} ${SERVER_CPPFLAGS}
 src_server_libmobject_server_la_CFLAGS   = ${AM_CFLAGS} ${SERVER_CFLAGS}
 src_server_libmobject_server_la_LIBADD = src/omap-iter/libomap-iter.la \

src/client/read-op.c

@@ -64,6 +64,8 @@ void mobject_store_read_op_read(mobject_store_read_op_t read_op,
 	DL_APPEND(read_op->actions, base);
 
 	read_op->num_actions += 1;
+
+    memset(buffer, 0, len);
 }
 
 void mobject_store_read_op_omap_get_keys(mobject_store_read_op_t read_op,

src/server/core/core-read-op.cpp

 #include <map>
 #include <string>
 #include <iostream>
+#include <algorithm>
+#include <list>
+#include <bake-bulk-client.h>
 #include "src/server/core/core-read-op.h"
 #include "src/server/visitor-args.h"
 #include "src/io-chain/read-op-visitor.h"
 #include "src/io-chain/read-resp-impl.h"
 #include "src/omap-iter/omap-iter-impl.h"
+#include "src/server/core/fake-kv.hpp"
+#include "src/server/core/covermap.hpp"
 
 static void read_op_exec_begin(void*);
 static void read_op_exec_stat(void*, uint64_t*, time_t*, int*);
@@ -15,6 +20,16 @@ static void read_op_exec_omap_get_vals(void*, const char*, const char*, uint64_t
 static void read_op_exec_omap_get_vals_by_keys(void*, char const* const*, size_t, mobject_store_omap_iter_t*, int*);
 static void read_op_exec_end(void*);
 
+struct read_request_t {
+    double timestamp;              // timestamp at which the segment was created
+    uint64_t absolute_start_index; // start index within the object
+    uint64_t absolute_end_index;   // end index within the object
+    uint64_t region_start_index;   // where to start within the region
+    uint64_t region_end_index;     // where to end within the region
+    uint64_t client_offset;        // offset within the client's buffer
+    bake_bulk_region_id_t region;  // region id
+};
+
 static struct read_op_visitor read_op_exec = {
 	.visit_begin                 = read_op_exec_begin,
 	.visit_stat                  = read_op_exec_stat,
@@ -43,25 +58,167 @@ void read_op_exec_stat(void* u, uint64_t* psize, time_t* pmtime, int* prval)
 void read_op_exec_read(void* u, uint64_t offset, size_t len, buffer_u buf, size_t* bytes_read, int* prval)
 {
     auto vargs = static_cast<server_visitor_args_t>(u);
+    bake_target_id_t bti = vargs->srv_ctx->bake_id;
+    bake_bulk_region_id_t rid;
+    hg_bulk_t remote_bulk = vargs->bulk_handle;
+    const char* object_name = vargs->object_name;
+    const char* remote_addr_str = vargs->client_addr_str;
+    hg_addr_t   remote_addr     = vargs->client_addr;
+    int ret;
+
+    uint64_t client_start_index = offset;
+    uint64_t client_end_index = offset+len;
+
+    *prval = 0;
+
+    // find oid
+    if(name_map.count(object_name) == 0) {
+        *prval = -1;
+        return;
+    }
+    oid_t oid = name_map[object_name];
+
+    segment_key_t lb;
+    lb.oid = oid;
+    lb.timestamp = std::numeric_limits<double>::max();
+    auto it = segment_map.lower_bound(lb);
+    covermap<uint64_t> coverage(offset, offset+len);
+
+    while(!coverage.full() && it != segment_map.end() && it->first.oid == oid) {
+        const segment_key_t& seg = it->first;
+        const bake_bulk_region_id_t& region = it->second;
+        
+        switch(seg.type) {
+        case seg_type_t::ZERO:
+            coverage.set(seg.start_index, seg.end_index);
+            if(*bytes_read < seg.end_index) *bytes_read = seg.end_index;
+            break;
+        case seg_type_t::TOMBSTONE:
+            coverage.set(seg.start_index, seg.end_index);
+            if(*bytes_read < seg.start_index) *bytes_read = seg.start_index;
+            break;
+        case seg_type_t::BAKE_REGION: {
+            auto ranges = coverage.set(seg.start_index, seg.end_index);
+            for(auto r : ranges) {
+                uint64_t segment_size  = r.end - r.start;
+                uint64_t region_offset = r.start - seg.start_index;
+                uint64_t remote_offset = r.start - offset;
+                bake_bulk_proxy_read(bti, region, region_offset, remote_bulk,
+                        remote_offset, remote_addr_str, segment_size);
+                if(*bytes_read < r.end) *bytes_read = r.end;
+            }
+            break;
+          }
+        case seg_type_t::SMALL_REGION: {
+            auto ranges = coverage.set(seg.start_index, seg.end_index);
+            const char* base = static_cast<const char*>((void*)(&region));
+            margo_instance_id mid = vargs->srv_ctx->mid;
+            for(auto r : ranges) {
+                uint64_t segment_size  = r.end - r.start;
+                uint64_t region_offset = r.start - seg.start_index;
+                uint64_t remote_offset = r.start - offset;
+                void* buf_ptrs[1] = { const_cast<char*>(base + region_offset) };
+                hg_size_t buf_sizes[1] = { segment_size };
+                hg_bulk_t handle;
+                std::cout << "Reading from a small region" << std::endl;
+                ret = margo_bulk_create(mid,1, buf_ptrs, buf_sizes, HG_BULK_READ_ONLY, &handle);
+                ret = margo_bulk_transfer(mid, HG_BULK_PUSH, remote_addr, remote_bulk, buf.as_offset+remote_offset, handle, 0, segment_size);
+                ret = margo_bulk_free(handle);
+                if(*bytes_read < r.end) *bytes_read = r.end;
+            }
+          }
+        }
+        it++;
+    }
 }
 
 void read_op_exec_omap_get_keys(void* u, const char* start_after, uint64_t max_return, 
 				mobject_store_omap_iter_t* iter, int* prval)
 {
     auto vargs = static_cast<server_visitor_args_t>(u);
-//    omap_iter_create(iter);
+    const char* object_name = vargs->object_name;
+
+    *prval = 0;
+
+    // find oid
+    if(name_map.count(object_name) == 0) {
+        *prval = -1;
+        return;
+    }
+    oid_t oid = name_map[object_name];
+
+    omap_iter_create(iter);
+    omap_key_t lb;
+    lb.oid = oid;
+    lb.key = std::string(start_after);
+
+    auto it = omap_map.lower_bound(lb);
+    if(it == omap_map.end()) return;
+    if(it->first.key == lb.key) it++;
+
+    for(uint64_t i=0; it != omap_map.end() && i < max_return; it++, i++) {
+        omap_iter_append(*iter, it->first.key.c_str(), nullptr, 0);
+    }
 }
 
 void read_op_exec_omap_get_vals(void* u, const char* start_after, const char* filter_prefix, uint64_t max_return, mobject_store_omap_iter_t* iter, int* prval)
 {
     auto vargs = static_cast<server_visitor_args_t>(u);
-//    omap_iter_create(iter);
+    const char* object_name = vargs->object_name;
+
+    *prval = 0;
+
+    // find oid
+    if(name_map.count(object_name) == 0) {
+        *prval = -1;
+        return;
+    }
+    oid_t oid = name_map[object_name];
+
+    omap_iter_create(iter);
+    omap_key_t lb;
+    lb.oid = oid;
+    lb.key = std::string(start_after);
+
+    auto it = omap_map.lower_bound(lb);
+    if(it == omap_map.end()) return;
+    if(it->first.key == lb.key) it++;
+
+    std::string prefix(filter_prefix);
+    for(uint64_t i=0; it != omap_map.end() && i < max_return; it++, i++) {
+        const std::string& k = it->first.key;
+        if(k.compare(0, prefix.size(), prefix) == 0) {
+                omap_iter_append(*iter, k.c_str(), &(it->second[0]), it->second.size());
+        }
+    }
 }
 
 void read_op_exec_omap_get_vals_by_keys(void* u, char const* const* keys, size_t num_keys, mobject_store_omap_iter_t* iter, int* prval)
 {
     auto vargs = static_cast<server_visitor_args_t>(u);
-//    omap_iter_create(iter);
+    const char* object_name = vargs->object_name;
+
+    *prval = 0;
+
+    // find oid
+    if(name_map.count(object_name) == 0) {
+        *prval = -1;
+        return;
+    }
+    oid_t oid = name_map[object_name];
+
+    omap_iter_create(iter);
+
+    omap_key_t key;
+    key.oid = oid;
+
+    for(size_t i=0; i<num_keys; i++) {
+        key.key = keys[i];
+        auto it = omap_map.find(key);
+        if(it != omap_map.end()) {
+             omap_iter_append(*iter, keys[i], &(it->second[0]), it->second.size());
+        }
+    }
 }
 
 void read_op_exec_end(void* u)

src/server/core/covermap.hpp

+#ifndef COVERAGE_MAP
+#define COVERAGE_MAP
+
+#include <iostream>
+#include <list>
+#include <map>
+
+template<typename T>
+class covermap {
+
+    const T           m_start;
+    const T           m_end;
+    T                 m_level;
+    std::map<T,T> m_segments;
+
+    static bool intersects(const T& start1, const T& end1, const T& start2, const T& end2) {
+        if(start1 < start2) {
+            return start2 <= end1;
+        } else {
+            return start1 <= end2;
+        }
+    }
+
+    public:
+
+    struct segment {
+        T start;
+        T end;
+        segment() {}
+        segment(T s, T e) 
+            : start(s), end(e) {}
+    };
+
+    covermap(T s, T e)
+        : m_start(s), m_end(e) {}
+
+    std::list<segment> set(T start, T end) {
+        // make start and end match the bounds
+        if(start < m_start) start = m_start;
+        if(end > m_end) end = m_end;
+
+        std::list<segment> result;
+        // easy case: the coverage map is empty
+        if(m_segments.empty()) {
+            m_segments[start] = end;
+            result.emplace_back(start,end);
+            m_level += (end-start);
+            return result;
+        }
+
+        // not easy case
+        // find the first segment intersecting 
+        auto first_seg = m_segments.lower_bound(start);
+        // we need to go back left
+        if(first_seg != m_segments.begin()) first_seg--;
+        // check if the first segment we found intersects
+        if(!intersects(first_seg->first, first_seg->second, start, end)) {
+            // it does not intersect
+            first_seg++;
+        }
+
+        // find the first segment on the right that does NOT intersect
+        auto last_seg = m_segments.lower_bound(end);
+        // if the first_seg and last_seg are the same, then we can
+        // insert the new segment directly
+        if(first_seg == last_seg) {
+            result.emplace_back(start, end);
+            m_level += (end-start);
+            m_segments[start] = end;
+            return result;
+        }
+        // otherwise, we iterate to build the list
+        auto it = first_seg;
+        if(first_seg->first > start) result.emplace_back(start,first_seg->first);
+        for(; it != last_seg; it++) {
+            auto jt = it;
+            jt++;
+            if(jt != last_seg) {
+                result.emplace_back(it->second, jt->first);
+                m_level += (jt->first - it->second);
+            }
+            else if(it->second < end) {
+                result.emplace_back(it->second, end);
+                m_level += (end - it->second);
+            }
+                
+        }
+        start = std::min(first_seg->first, start);
+        auto before_last = last_seg;
+        before_last--;
+        end   = std::max(before_last->second, end);
+        m_segments.erase(first_seg, last_seg);
+        m_segments[start] = end;
+        return result;
+    }
+    void print(std::ostream& ostr) {
+        for(auto& s : m_segments) {
+            ostr << "[" << s.first << "," << s.second << "[";
+        }
+    }
+
+    T level() const {
+        return m_level;
+    }
+
+    T capacity() const {
+        return (m_end - m_start);
+    }
+
+    bool full() const {
+        return capacity() == level();
+    }
+};
+
+#endif

src/server/core/fake-kv.cpp

+#include "src/server/core/fake-kv.hpp"
+
+bool operator<(const segment_key_t& s1, const segment_key_t& s2) {
+    // sort by oid first
+    if(s1.oid != s2.oid)
+        return s1.oid < s2.oid;
+    // larger timestamps will go first
+//    if(s1.timestamp != s2.timestamp)
+        return s1.timestamp > s2.timestamp;
+//    return s1.start_index < s2.start_index;
+}
+
+bool operator<(const omap_key_t& k1, const omap_key_t& k2) {
+    if(k1.oid != k2.oid)
+        return k1.oid < k2.oid;
+    return k1.key < k2.key;
+}
+
+std::map<oid_t, std::string>                    oid_map;
+std::map<std::string, oid_t>                    name_map;
+std::map<segment_key_t, bake_bulk_region_id_t>  segment_map;
+std::map<omap_key_t, std::vector<char>>         omap_map;
+//std::map<oid_t, std::size_t>                    size_map;

src/server/core/fake-kv.hpp

+#include <map>
+#include <vector>
+#include <string>
+#include <bake-bulk-client.h>
+
+typedef uint64_t oid_t;
+
+enum class seg_type_t : std::int32_t {
+    ZERO         = 0,
+    BAKE_REGION  = 1,
+    SMALL_REGION = 2,
+    TOMBSTONE    = 3
+};
+
+struct segment_key_t {
+    oid_t oid;
+    seg_type_t type;
+    double timestamp;
+    uint64_t start_index; // first index, included
+    uint64_t end_index;  // end index is not included
+};
+
+struct omap_key_t {
+    oid_t oid;
+    std::string key;
+};
+
+#define SMALL_REGION_THRESHOLD (sizeof(bake_bulk_region_id_t))
+
+bool operator<(const segment_key_t& s1, const segment_key_t& s2);
+bool operator<(const omap_key_t& k1, const omap_key_t& k2);
+
+extern std::map<oid_t, std::string> oid_map;
+extern std::map<std::string, oid_t> name_map;
+extern std::map<segment_key_t, bake_bulk_region_id_t> segment_map;
+extern std::map<omap_key_t, std::vector<char>> omap_map;
+//extern std::map<oid_t, std::size_t> size_map;

src/server/fake/fake-object.hpp

@@ -35,6 +35,8 @@ public:
 
     void omap_get_keys(const std::string& start_after, size_t max, mobject_store_omap_iter_t iter) const {
         auto it = m_omap.lower_bound(start_after);
+        if(it == m_omap.end()) return;
+        if(it->first == start_after) it++;
         for(size_t i = 0; (it != m_omap.end()) && (i < max); it++, i++) {
             auto& key = it->first;
             omap_iter_append(iter, &key[0], (const char*)0 , 0);
@@ -44,6 +46,8 @@ public:
     void omap_get_vals(const std::string& start_after, const std::string& filter_prefix, 
                        size_t max, mobject_store_omap_iter_t iter) const {
         auto it = m_omap.lower_bound(start_after);
+        if(it == m_omap.end()) return;
+        if(it->first == start_after) it++;
         for(size_t i = 0; (it != m_omap.end()) && (i < max); it++, i++) {
             auto& key = it->first;
             auto& val = it->second;

src/server/fake/fake-read-op.cpp

@@ -60,7 +60,7 @@ void read_op_exec_read(void* u, uint64_t offset, size_t len, buffer_u buf, size_
         return;
     }
     margo_instance_id mid = vargs->srv_ctx->mid;
-    fake_db[name].read(mid, vargs->client_addr.as_handle, vargs->bulk_handle, 
+    fake_db[name].read(mid, vargs->client_addr, vargs->bulk_handle, 
                         buf.as_offset, offset, len, bytes_read);
 	*prval = 0;
 }

src/server/fake/fake-write-op.cpp

@@ -70,7 +70,7 @@ void write_op_exec_write(void* u, buffer_u buf, size_t len, uint64_t offset)
 		std::cerr << "[FAKE-BACKEND-WARNING] (write) Object " << name << " does not exist, it will be created" << std::endl;
 	}
     margo_instance_id mid = vargs->srv_ctx->mid;
-	fake_db[name].write(mid, vargs->client_addr.as_handle, vargs->bulk_handle, buf.as_offset, offset, len);
+	fake_db[name].write(mid, vargs->client_addr, vargs->bulk_handle, buf.as_offset, offset, len);
 }
 
 void write_op_exec_write_full(void* u, buffer_u buf, size_t len)
@@ -81,7 +81,7 @@ void write_op_exec_write_full(void* u, buffer_u buf, size_t len)
 		std::cerr << "[FAKE-BACKEND-WARNING] (write_full) Object " << name << " does not exist, it will be created" << std::endl;
 	}
     margo_instance_id mid = vargs->srv_ctx->mid;
-    fake_db[name].write_full(mid, vargs->client_addr.as_handle, vargs->bulk_handle, buf.as_offset, len);
+    fake_db[name].write_full(mid, vargs->client_addr, vargs->bulk_handle, buf.as_offset, len);
 }
 
 void write_op_exec_writesame(void* u, buffer_u buf, size_t data_len, size_t write_len, uint64_t offset)
@@ -92,7 +92,7 @@ void write_op_exec_writesame(void* u, buffer_u buf, size_t data_len, size_t writ
 		std::cerr << "[FAKE-BACKEND-WARNING] (writesame) Object " << name << " does not exist, it will be created" << std::endl;
 	}
     margo_instance_id mid = vargs->srv_ctx->mid;
-    fake_db[name].writesame(mid, vargs->client_addr.as_handle, vargs->bulk_handle, buf.as_offset, offset, data_len, write_len);
+    fake_db[name].writesame(mid, vargs->client_addr, vargs->bulk_handle, buf.as_offset, offset, data_len, write_len);
 }
 
 void write_op_exec_append(void* u, buffer_u buf, size_t len)
@@ -103,7 +103,7 @@ void write_op_exec_append(void* u, buffer_u buf, size_t len)
 		std::cerr << "[FAKE-BACKEND-WARNING] (append) Object " << name << " does not exist, it will be created" << std::endl;
 	}
     margo_instance_id mid = vargs->srv_ctx->mid;
-	fake_db[name].append(mid, vargs->client_addr.as_handle, vargs->bulk_handle, buf.as_offset, len);
+	fake_db[name].append(mid, vargs->client_addr, vargs->bulk_handle, buf.as_offset, len);
 }
 
 void write_op_exec_remove(void* u)

src/server/mobject-server.c

@@ -214,15 +214,13 @@ static hg_return_t mobject_write_op_ult(hg_handle_t h)
     vargs.object_name = in.object_name;
     vargs.pool_name   = in.pool_name;
     vargs.srv_ctx     = margo_registered_data(mid, info->id);
-    vargs.client_addr_type = MOBJECT_ADDR_STRING;
-    vargs.client_addr.as_string = in.client_addr;
+    vargs.client_addr_str = in.client_addr;
+    vargs.client_addr = info->addr;
     vargs.bulk_handle = in.write_op->bulk_handle;
 
     /* Execute the operation chain */
     //print_write_op(in.write_op, in.object_name);
 #ifdef FAKE_CPP_SERVER
-    vargs.client_addr_type = MOBJECT_ADDR_HANDLE;
-    vargs.client_addr.as_handle = info->addr;
     fake_write_op(in.write_op, &vargs);
 #else
     core_write_op(in.write_op, &vargs);
@@ -267,15 +265,13 @@ static hg_return_t mobject_read_op_ult(hg_handle_t h)
     vargs.object_name = in.object_name;
     vargs.pool_name   = in.pool_name;
     vargs.srv_ctx     = margo_registered_data(mid,info->id);
-    vargs.client_addr_type = MOBJECT_ADDR_STRING;
-    vargs.client_addr.as_string = in.client_addr;
+    vargs.client_addr_str = in.client_addr;
+    vargs.client_addr = info->addr;
     vargs.bulk_handle = in.read_op->bulk_handle;
 
     /* Compute the result. */
     //print_read_op(in.read_op, in.object_name);
 #ifdef FAKE_CPP_SERVER
-    vargs.client_addr_type = MOBJECT_ADDR_HANDLE;
-    vargs.client_addr.as_handle = info->addr;
     fake_read_op(in.read_op, &vargs);
 #else
     core_read_op(in.read_op, &vargs);

src/server/visitor-args.h

@@ -9,21 +9,13 @@
 extern "C" {
 #endif
 
-typedef enum {
-    MOBJECT_ADDR_STRING,
-    MOBJECT_ADDR_HANDLE
-} addr_str_type_t ;
-
 typedef struct {
 	const char*                    object_name;
 	const char*                    pool_name;
     struct mobject_server_context* srv_ctx;
-    union {
-        const char* as_string;
-        hg_addr_t   as_handle;
-    } client_addr;
+    const char*                    client_addr_str;
+    hg_addr_t                      client_addr;
 	hg_bulk_t                      bulk_handle;
-    addr_str_type_t                client_addr_type;
 } server_visitor_args;
 
 typedef server_visitor_args* server_visitor_args_t;

tests/mobject-client-test.c

@@ -63,7 +63,7 @@ int main(int argc, char** argv)
         int prval2;
         mobject_store_read_op_read(read_op, 0, 512, read_buf, &bytes_read, &prval2);
         // Add "omap_get_keys" operation
-        const char* start_after1 = "shane";
+        const char* start_after1 = "rob";
         mobject_store_omap_iter_t iter3;
         int prval3;
         mobject_store_read_op_omap_get_keys(read_op, start_after1, 7, &iter3, &prval3);
@@ -85,6 +85,7 @@ int main(int argc, char** argv)
 
         // print the results of the read operations
         printf("Client received the following results:\n");
+
         printf("stat: psize=%ld pmtime=%lld prval=%d\n", psize, (long long)pmtime, prval1);
         {
             printf("read: bytes_read = %ld, prval=%d content: ", bytes_read, prval2);