sdskv-server.cc 32.5 KB
Newer Older
Matthieu Dorier's avatar
Matthieu Dorier committed
1 2 3 4 5 6 7
/*
 * (C) 2015 The University of Chicago
 * 
 * See COPYRIGHT in top-level directory.
 */

#include <map>
Matthieu Dorier's avatar
Matthieu Dorier committed
8
#include <iostream>
Matthieu Dorier's avatar
Matthieu Dorier committed
9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27
#include <unordered_map>
#define SDSKV
#include "datastore/datastore_factory.h"
#include "sdskv-rpc-types.h"
#include "sdskv-server.h"

struct sdskv_server_context_t
{
    std::unordered_map<sdskv_database_id_t, AbstractDataStore*> databases;
    std::map<std::string, sdskv_database_id_t> name2id;
    std::map<sdskv_database_id_t, std::string> id2name;
};

DECLARE_MARGO_RPC_HANDLER(sdskv_put_ult)
DECLARE_MARGO_RPC_HANDLER(sdskv_length_ult)
DECLARE_MARGO_RPC_HANDLER(sdskv_get_ult)
DECLARE_MARGO_RPC_HANDLER(sdskv_open_ult)
DECLARE_MARGO_RPC_HANDLER(sdskv_bulk_put_ult)
DECLARE_MARGO_RPC_HANDLER(sdskv_bulk_get_ult)
28
DECLARE_MARGO_RPC_HANDLER(sdskv_list_keys_ult)
29
DECLARE_MARGO_RPC_HANDLER(sdskv_list_keyvals_ult)
Matthieu Dorier's avatar
Matthieu Dorier committed
30
DECLARE_MARGO_RPC_HANDLER(sdskv_erase_ult)
Matthieu Dorier's avatar
Matthieu Dorier committed
31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48

static void sdskv_server_finalize_cb(void *data);

extern "C" int sdskv_provider_register(
        margo_instance_id mid,
        uint8_t mplex_id,
        ABT_pool abt_pool,
        sdskv_provider_t* provider)
{
    sdskv_server_context_t *tmp_svr_ctx;

    /* check if a provider with the same multiplex id already exists */
    {
        hg_id_t id;
        hg_bool_t flag;
        margo_registered_name_mplex(mid, "sdskv_put_rpc", mplex_id, &id, &flag);
        if(flag == HG_TRUE) {
            fprintf(stderr, "sdskv_provider_register(): a provider with the same mplex id (%d) already exists\n", mplex_id);
49
            return SDSKV_ERR_MERCURY;
Matthieu Dorier's avatar
Matthieu Dorier committed
50 51 52 53
        }
    }

    /* allocate the resulting structure */    
Matthieu Dorier's avatar
Matthieu Dorier committed
54
    tmp_svr_ctx = new sdskv_server_context_t;
Matthieu Dorier's avatar
Matthieu Dorier committed
55
    if(!tmp_svr_ctx)
56
        return SDSKV_ERR_ALLOCATION;
Matthieu Dorier's avatar
Matthieu Dorier committed
57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83

    /* register RPCs */
    hg_id_t rpc_id;
    rpc_id = MARGO_REGISTER_MPLEX(mid, "sdskv_put_rpc",
            put_in_t, put_out_t,
            sdskv_put_ult, mplex_id, abt_pool);
    margo_register_data_mplex(mid, rpc_id, mplex_id, (void*)tmp_svr_ctx, NULL);
    rpc_id = MARGO_REGISTER_MPLEX(mid, "sdskv_bulk_put_rpc",
            bulk_put_in_t, bulk_put_out_t,
            sdskv_bulk_put_ult, mplex_id, abt_pool);
    margo_register_data_mplex(mid, rpc_id, mplex_id, (void*)tmp_svr_ctx, NULL);
    rpc_id = MARGO_REGISTER_MPLEX(mid, "sdskv_get_rpc",
            get_in_t, get_out_t,
            sdskv_get_ult, mplex_id, abt_pool);
    margo_register_data_mplex(mid, rpc_id, mplex_id, (void*)tmp_svr_ctx, NULL);
    rpc_id = MARGO_REGISTER_MPLEX(mid, "sdskv_length_rpc",
            length_in_t, length_out_t,
            sdskv_length_ult, mplex_id, abt_pool);
    margo_register_data_mplex(mid, rpc_id, mplex_id, (void*)tmp_svr_ctx, NULL);
    rpc_id = MARGO_REGISTER_MPLEX(mid, "sdskv_bulk_get_rpc",
            bulk_get_in_t, bulk_get_out_t,
            sdskv_bulk_get_ult, mplex_id, abt_pool);
    margo_register_data_mplex(mid, rpc_id, mplex_id, (void*)tmp_svr_ctx, NULL);
    rpc_id = MARGO_REGISTER_MPLEX(mid, "sdskv_open_rpc",
            open_in_t, open_out_t,
            sdskv_open_ult, mplex_id, abt_pool);
    margo_register_data_mplex(mid, rpc_id, mplex_id, (void*)tmp_svr_ctx, NULL);
84
    rpc_id = MARGO_REGISTER_MPLEX(mid, "sdskv_list_keys_rpc",
85
            list_keys_in_t, list_keys_out_t,
86
            sdskv_list_keys_ult, mplex_id, abt_pool);
Matthieu Dorier's avatar
Matthieu Dorier committed
87
    margo_register_data_mplex(mid, rpc_id, mplex_id, (void*)tmp_svr_ctx, NULL);
88 89 90 91
    rpc_id = MARGO_REGISTER_MPLEX(mid, "sdskv_list_keyvals_rpc",
            list_keyvals_in_t, list_keyvals_out_t,
            sdskv_list_keyvals_ult, mplex_id, abt_pool);
    margo_register_data_mplex(mid, rpc_id, mplex_id, (void*)tmp_svr_ctx, NULL);
Matthieu Dorier's avatar
Matthieu Dorier committed
92 93 94 95
    rpc_id = MARGO_REGISTER_MPLEX(mid, "sdskv_erase_rpc",
            erase_in_t, erase_out_t,
            sdskv_erase_ult, mplex_id, abt_pool);
    margo_register_data_mplex(mid, rpc_id, mplex_id, (void*)tmp_svr_ctx, NULL);
Matthieu Dorier's avatar
Matthieu Dorier committed
96 97 98 99 100 101 102

    /* install the bake server finalize callback */
    margo_push_finalize_callback(mid, &sdskv_server_finalize_cb, tmp_svr_ctx);

    if(provider != SDSKV_PROVIDER_IGNORE)
        *provider = tmp_svr_ctx;

103
    return SDSKV_SUCCESS;
Matthieu Dorier's avatar
Matthieu Dorier committed
104 105 106 107 108 109
}

extern "C" int sdskv_provider_add_database(
        sdskv_provider_t provider,
        const char *db_name,
        sdskv_db_type_t db_type,
110
        sdskv_compare_fn comp_fn,
Matthieu Dorier's avatar
Matthieu Dorier committed
111 112
        sdskv_database_id_t* db_id)
{
Matthieu Dorier's avatar
Matthieu Dorier committed
113
    auto db = datastore_factory::create_datastore(db_type, std::string(db_name));
114
    if(db == nullptr) return SDSKV_ERR_DB_CREATE;
115
    db->set_comparison_function(comp_fn);
Matthieu Dorier's avatar
Matthieu Dorier committed
116 117 118 119 120 121
    sdskv_database_id_t id = (sdskv_database_id_t)(db);

    provider->name2id[std::string(db_name)] = id;
    provider->id2name[id] = std::string(db_name);
    provider->databases[id] = db;

122
    return SDSKV_SUCCESS;
Matthieu Dorier's avatar
Matthieu Dorier committed
123 124 125 126 127 128 129 130 131 132 133 134 135
}

extern "C" int sdskv_provider_remove_database(
        sdskv_provider_t provider,
        sdskv_database_id_t db_id)
{
    if(provider->databases.count(db_id)) {
        auto dbname = provider->id2name[db_id];
        provider->id2name.erase(db_id);
        provider->name2id.erase(dbname);
        auto db = provider->databases[db_id];
        delete db;
        provider->databases.erase(db_id);
136
        return SDSKV_SUCCESS;
Matthieu Dorier's avatar
Matthieu Dorier committed
137
    } else {
138
        return SDSKV_ERR_UNKNOWN_DB;
Matthieu Dorier's avatar
Matthieu Dorier committed
139 140 141 142 143 144 145 146 147 148 149 150 151
    }
}

extern "C" int sdskv_provider_remove_all_databases(
        sdskv_provider_t provider)
{
    for(auto db : provider->databases) {
        delete db.second;
    }
    provider->databases.clear();
    provider->name2id.clear();
    provider->id2name.clear();

152
    return SDSKV_SUCCESS;
Matthieu Dorier's avatar
Matthieu Dorier committed
153 154 155 156 157 158 159
}

extern "C" int sdskv_provider_count_databases(
        sdskv_provider_t provider,
        uint64_t* num_db)
{
    *num_db = provider->databases.size();
160
    return SDSKV_SUCCESS;
Matthieu Dorier's avatar
Matthieu Dorier committed
161 162 163 164 165 166 167 168 169 170 171
}

extern "C" int sdskv_provider_list_databases(
        sdskv_provider_t provider,
        sdskv_database_id_t* targets)
{
    unsigned i = 0;
    for(auto p : provider->name2id) {
        targets[i] = p.second;
        i++;
    }
172
    return SDSKV_SUCCESS;
Matthieu Dorier's avatar
Matthieu Dorier committed
173 174 175 176 177 178 179 180 181 182 183 184 185 186 187
}

static void sdskv_put_ult(hg_handle_t handle)
{
    hg_return_t hret;
    put_in_t in;
    put_out_t out;

    margo_instance_id mid = margo_hg_handle_get_instance(handle);
    assert(mid);
    const struct hg_info* info = margo_get_info(handle);
    sdskv_provider_t svr_ctx = 
        (sdskv_provider_t)margo_registered_data_mplex(mid, info->id, info->target_id);
    if(!svr_ctx) {
        fprintf(stderr, "Error: SDSKV could not find provider\n"); 
188
        out.ret = SDSKV_ERR_UNKNOWN_PR;
Matthieu Dorier's avatar
Matthieu Dorier committed
189 190 191 192 193 194 195
        margo_respond(handle, &out);
        margo_destroy(handle);
        return;
    }
    
    hret = margo_get_input(handle, &in);
    if(hret != HG_SUCCESS) {
196
        out.ret = SDSKV_ERR_MERCURY;
Matthieu Dorier's avatar
Matthieu Dorier committed
197 198 199 200 201 202 203
        margo_respond(handle, &out);
        margo_destroy(handle);
        return;
    }

    auto it = svr_ctx->databases.find(in.db_id);
    if(it == svr_ctx->databases.end()) {
204
        out.ret = SDSKV_ERR_UNKNOWN_DB;
Matthieu Dorier's avatar
Matthieu Dorier committed
205 206 207 208 209 210
        margo_respond(handle, &out);
        margo_free_input(handle, &in);
        margo_destroy(handle);
        return;
    }

211 212
    ds_bulk_t kdata(in.key.data, in.key.data+in.key.size);
    ds_bulk_t vdata(in.value.data, in.value.data+in.value.size);
Matthieu Dorier's avatar
Matthieu Dorier committed
213 214

    if(it->second->put(kdata, vdata)) {
215
        out.ret = SDSKV_SUCCESS;
Matthieu Dorier's avatar
Matthieu Dorier committed
216
    } else {
217
        out.ret = SDSKV_ERR_PUT;
Matthieu Dorier's avatar
Matthieu Dorier committed
218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239
    }

    margo_respond(handle, &out);
    margo_free_input(handle, &in);
    margo_destroy(handle);
}
DEFINE_MARGO_RPC_HANDLER(sdskv_put_ult)

static void sdskv_length_ult(hg_handle_t handle)
{

    hg_return_t hret;
    length_in_t in;
    length_out_t out;

    margo_instance_id mid = margo_hg_handle_get_instance(handle);
    assert(mid);
    const struct hg_info* info = margo_get_info(handle);
    sdskv_provider_t svr_ctx = 
        (sdskv_provider_t)margo_registered_data_mplex(mid, info->id, info->target_id);
    if(!svr_ctx) {
        fprintf(stderr, "Error: SDSKV could not find provider\n"); 
240
        out.ret = SDSKV_ERR_UNKNOWN_PR;
Matthieu Dorier's avatar
Matthieu Dorier committed
241 242 243 244 245 246 247
        margo_respond(handle, &out);
        margo_destroy(handle);
        return;
    }

    hret = margo_get_input(handle, &in);
    if(hret != HG_SUCCESS) {
248
        out.ret = SDSKV_ERR_MERCURY;
Matthieu Dorier's avatar
Matthieu Dorier committed
249 250 251 252 253 254 255
        margo_respond(handle, &out);
        margo_destroy(handle);
        return;
    }

    auto it = svr_ctx->databases.find(in.db_id);
    if(it == svr_ctx->databases.end()) {
256
        out.ret = SDSKV_ERR_UNKNOWN_DB;
Matthieu Dorier's avatar
Matthieu Dorier committed
257 258 259 260 261 262
        margo_respond(handle, &out);
        margo_free_input(handle, &in);
        margo_destroy(handle);
        return;
    }
    
263
    ds_bulk_t kdata(in.key.data, in.key.data+in.key.size);
Matthieu Dorier's avatar
Matthieu Dorier committed
264 265 266 267

    ds_bulk_t vdata;
    if(it->second->get(kdata, vdata)) {
        out.size = vdata.size();
268
        out.ret  = SDSKV_SUCCESS;
Matthieu Dorier's avatar
Matthieu Dorier committed
269 270
    } else {
        out.size = 0;
271
        out.ret  = SDSKV_ERR_UNKNOWN_KEY;
Matthieu Dorier's avatar
Matthieu Dorier committed
272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295
    }

    margo_respond(handle, &out);
    margo_free_input(handle, &in);
    margo_destroy(handle); 

    return;
}
DEFINE_MARGO_RPC_HANDLER(sdskv_length_ult)

static void sdskv_get_ult(hg_handle_t handle)
{

    hg_return_t hret;
    get_in_t in;
    get_out_t out;

    margo_instance_id mid = margo_hg_handle_get_instance(handle);
    assert(mid);
    const struct hg_info* info = margo_get_info(handle);
    sdskv_provider_t svr_ctx = 
        (sdskv_provider_t)margo_registered_data_mplex(mid, info->id, info->target_id);
    if(!svr_ctx) {
        fprintf(stderr, "Error: SDSKV could not find provider\n"); 
296
        out.ret = SDSKV_ERR_UNKNOWN_PR;
Matthieu Dorier's avatar
Matthieu Dorier committed
297 298 299 300 301 302 303
        margo_respond(handle, &out);
        margo_destroy(handle);
        return;
    }

    hret = margo_get_input(handle, &in);
    if(hret != HG_SUCCESS) {
304
        out.ret = SDSKV_ERR_MERCURY;
305 306
        out.value.data = nullptr;
        out.value.size = 0;
Matthieu Dorier's avatar
Matthieu Dorier committed
307 308 309 310 311 312 313
        margo_respond(handle, &out);
        margo_destroy(handle);
        return;
    }

    auto it = svr_ctx->databases.find(in.db_id);
    if(it == svr_ctx->databases.end()) {
314
        out.ret = SDSKV_ERR_UNKNOWN_DB;
315 316
        out.value.data = nullptr;
        out.value.size = 0;
Matthieu Dorier's avatar
Matthieu Dorier committed
317 318 319 320 321 322
        margo_respond(handle, &out);
        margo_free_input(handle, &in);
        margo_destroy(handle);
        return;
    }
    
323
    ds_bulk_t kdata(in.key.data, in.key.data+in.key.size);
Matthieu Dorier's avatar
Matthieu Dorier committed
324 325 326 327

    ds_bulk_t vdata;
    if(it->second->get(kdata, vdata)) {
        if(vdata.size() <= in.vsize) {
328 329
            out.value.size = vdata.size();
            out.value.data = vdata.data();
330
            out.ret = SDSKV_SUCCESS;
Matthieu Dorier's avatar
Matthieu Dorier committed
331
        } else {
332 333
            out.value.size = 0;
            out.value.data = nullptr;
334
            out.ret = SDSKV_ERR_SIZE;
Matthieu Dorier's avatar
Matthieu Dorier committed
335 336
        }
    } else {
337 338
        out.value.size = 0;
        out.value.data = nullptr;
339
        out.ret   = SDSKV_ERR_UNKNOWN_KEY;
Matthieu Dorier's avatar
Matthieu Dorier committed
340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363
    }

    margo_respond(handle, &out);
    margo_free_input(handle, &in);
    margo_destroy(handle); 

    return;
}
DEFINE_MARGO_RPC_HANDLER(sdskv_get_ult)

static void sdskv_open_ult(hg_handle_t handle)
{

    hg_return_t hret;
    open_in_t in;
    open_out_t out;

    margo_instance_id mid = margo_hg_handle_get_instance(handle);
    assert(mid);
    const struct hg_info* info = margo_get_info(handle);
    sdskv_provider_t svr_ctx = 
        (sdskv_provider_t)margo_registered_data_mplex(mid, info->id, info->target_id);
    if(!svr_ctx) {
        fprintf(stderr, "Error: SDSKV could not find provider\n"); 
364
        out.ret = SDSKV_ERR_UNKNOWN_PR;
Matthieu Dorier's avatar
Matthieu Dorier committed
365 366 367 368 369 370 371
        margo_respond(handle, &out);
        margo_destroy(handle);
        return;
    }

    hret = margo_get_input(handle, &in);
    if(hret != HG_SUCCESS) {
372
        out.ret = SDSKV_ERR_MERCURY;
Matthieu Dorier's avatar
Matthieu Dorier committed
373 374 375 376 377 378 379
        margo_respond(handle, &out);
        margo_destroy(handle);
        return;
    }

    auto it = svr_ctx->name2id.find(std::string(in.name));
    if(it == svr_ctx->name2id.end()) {
380
        out.ret = SDSKV_ERR_DB_NAME;
Matthieu Dorier's avatar
Matthieu Dorier committed
381 382 383 384 385 386 387
        margo_respond(handle, &out);
        margo_free_input(handle, &in);
        margo_destroy(handle);
        return;
    }

    out.db_id = it->second;
388
    out.ret  = SDSKV_SUCCESS;
Matthieu Dorier's avatar
Matthieu Dorier committed
389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411

    margo_respond(handle, &out);
    margo_free_input(handle, &in);
    margo_destroy(handle);
    return;
}
DEFINE_MARGO_RPC_HANDLER(sdskv_open_ult)

static void sdskv_bulk_put_ult(hg_handle_t handle)
{

    hg_return_t hret;
    bulk_put_in_t in;
    bulk_put_out_t out;
    hg_bulk_t bulk_handle;

    margo_instance_id mid = margo_hg_handle_get_instance(handle);
    assert(mid);
    const struct hg_info* info = margo_get_info(handle);
    sdskv_provider_t svr_ctx = 
        (sdskv_provider_t)margo_registered_data_mplex(mid, info->id, info->target_id);
    if(!svr_ctx) {
        fprintf(stderr, "Error: SDSKV could not find provider\n"); 
412
        out.ret = SDSKV_ERR_UNKNOWN_PR;
Matthieu Dorier's avatar
Matthieu Dorier committed
413 414 415 416 417 418 419
        margo_respond(handle, &out);
        margo_destroy(handle);
        return;
    }

    hret = margo_get_input(handle, &in);
    if(hret != HG_SUCCESS) {
420
        out.ret = SDSKV_ERR_MERCURY;
Matthieu Dorier's avatar
Matthieu Dorier committed
421 422 423 424 425
        margo_respond(handle, &out);
        margo_destroy(handle);
        return;
    }

426
    auto it = svr_ctx->databases.find(in.db_id);
Matthieu Dorier's avatar
Matthieu Dorier committed
427
    if(it == svr_ctx->databases.end()) {
428
        out.ret = SDSKV_ERR_UNKNOWN_DB;
Matthieu Dorier's avatar
Matthieu Dorier committed
429 430 431 432 433 434
        margo_respond(handle, &out);
        margo_free_input(handle, &in);
        margo_destroy(handle);
        return;
    }
   
435
    ds_bulk_t vdata(in.vsize);
Matthieu Dorier's avatar
Matthieu Dorier committed
436 437 438 439 440
    void *buffer = (void*)vdata.data();
    hg_size_t size = vdata.size();
    hret = margo_bulk_create(mid, 1, (void**)&buffer, &size,
            HG_BULK_WRITE_ONLY, &bulk_handle);
    if(hret != HG_SUCCESS) {
441
        out.ret = SDSKV_ERR_MERCURY;
Matthieu Dorier's avatar
Matthieu Dorier committed
442 443 444 445 446 447
        margo_respond(handle, &out);
        margo_free_input(handle, &in);
        margo_destroy(handle);
        return;
    }

448
    hret = margo_bulk_transfer(mid, HG_BULK_PULL, info->addr, in.handle, 0,
Matthieu Dorier's avatar
Matthieu Dorier committed
449 450
            bulk_handle, 0, vdata.size());
    if(hret != HG_SUCCESS) {
451
        out.ret = SDSKV_ERR_MERCURY;
Matthieu Dorier's avatar
Matthieu Dorier committed
452 453 454 455 456 457 458
        margo_respond(handle, &out);
        margo_free_input(handle, &in);
        margo_bulk_free(bulk_handle);
        margo_destroy(handle);
        return;
    }

459
    ds_bulk_t kdata(in.key.data, in.key.data+in.key.size);
Matthieu Dorier's avatar
Matthieu Dorier committed
460 461 462 463

    auto b = it->second->put(kdata, vdata);

    if(b) {
464
        out.ret = SDSKV_SUCCESS;
Matthieu Dorier's avatar
Matthieu Dorier committed
465
    } else {
466
        out.ret = SDSKV_ERR_PUT;
Matthieu Dorier's avatar
Matthieu Dorier committed
467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492
    }

    margo_respond(handle, &out);
    margo_free_input(handle, &in);
    margo_bulk_free(bulk_handle);
    margo_destroy(handle);

    return;
}
DEFINE_MARGO_RPC_HANDLER(sdskv_bulk_put_ult)

static void sdskv_bulk_get_ult(hg_handle_t handle)
{

    hg_return_t hret;
    bulk_get_in_t in;
    bulk_get_out_t out;
    hg_bulk_t bulk_handle;

    margo_instance_id mid = margo_hg_handle_get_instance(handle);
    assert(mid);
    const struct hg_info* info = margo_get_info(handle);
    sdskv_provider_t svr_ctx = 
        (sdskv_provider_t)margo_registered_data_mplex(mid, info->id, info->target_id);
    if(!svr_ctx) {
        fprintf(stderr, "Error: SDSKV could not find provider\n"); 
493
        out.ret = SDSKV_ERR_UNKNOWN_PR;
Matthieu Dorier's avatar
Matthieu Dorier committed
494 495 496 497 498 499 500
        margo_respond(handle, &out);
        margo_destroy(handle);
        return;
    }

    hret = margo_get_input(handle, &in);
    if(hret != HG_SUCCESS) {
501
        out.ret = SDSKV_ERR_MERCURY;
Matthieu Dorier's avatar
Matthieu Dorier committed
502 503 504 505 506
        margo_respond(handle, &out);
        margo_destroy(handle);
        return;
    }

507
    auto it = svr_ctx->databases.find(in.db_id);
Matthieu Dorier's avatar
Matthieu Dorier committed
508
    if(it == svr_ctx->databases.end()) {
509
        out.ret = SDSKV_ERR_UNKNOWN_DB;
Matthieu Dorier's avatar
Matthieu Dorier committed
510 511 512 513 514 515
        margo_respond(handle, &out);
        margo_free_input(handle, &in);
        margo_destroy(handle);
        return;
    }
    
516
    ds_bulk_t kdata(in.key.data, in.key.data+in.key.size);
Matthieu Dorier's avatar
Matthieu Dorier committed
517 518 519 520

    ds_bulk_t vdata;
    auto b = it->second->get(kdata, vdata);

521
    if(!b || vdata.size() > in.vsize) {
Matthieu Dorier's avatar
Matthieu Dorier committed
522
        out.size = 0;
523
        out.ret = SDSKV_ERR_SIZE;
Matthieu Dorier's avatar
Matthieu Dorier committed
524 525 526 527 528 529 530 531 532 533 534 535
        margo_respond(handle, &out);
        margo_free_input(handle, &in);
        margo_destroy(handle);
        return;
    }

    void *buffer = (void*)vdata.data();
    hg_size_t size = vdata.size();
    hret = margo_bulk_create(mid, 1, (void**)&buffer, &size,
            HG_BULK_READ_ONLY, &bulk_handle);
    if(hret != HG_SUCCESS) {
        out.size = 0;
536
        out.ret = SDSKV_ERR_MERCURY;
Matthieu Dorier's avatar
Matthieu Dorier committed
537 538 539 540 541 542
        margo_respond(handle, &out);
        margo_free_input(handle, &in);
        margo_destroy(handle);
        return;
    }

543
    hret = margo_bulk_transfer(mid, HG_BULK_PUSH, info->addr, in.handle, 0,
Matthieu Dorier's avatar
Matthieu Dorier committed
544 545 546
            bulk_handle, 0, vdata.size());
    if(hret != HG_SUCCESS) {
        out.size = 0;
547
        out.ret = SDSKV_ERR_MERCURY;
Matthieu Dorier's avatar
Matthieu Dorier committed
548 549 550 551 552 553 554 555
        margo_respond(handle, &out);
        margo_free_input(handle, &in);
        margo_bulk_free(bulk_handle);
        margo_destroy(handle);
        return;
    }

    out.size = vdata.size();
556
    out.ret  = SDSKV_SUCCESS;
Matthieu Dorier's avatar
Matthieu Dorier committed
557 558 559 560 561 562 563 564 565 566

    margo_respond(handle, &out);
    margo_free_input(handle, &in);
    margo_bulk_free(bulk_handle);
    margo_destroy(handle);

    return;
}
DEFINE_MARGO_RPC_HANDLER(sdskv_bulk_get_ult)

Matthieu Dorier's avatar
Matthieu Dorier committed
567 568 569 570 571 572 573 574 575 576 577 578 579 580
static void sdskv_erase_ult(hg_handle_t handle)
{

    hg_return_t hret;
    erase_in_t in;
    erase_out_t out;

    margo_instance_id mid = margo_hg_handle_get_instance(handle);
    assert(mid);
    const struct hg_info* info = margo_get_info(handle);
    sdskv_provider_t svr_ctx = 
        (sdskv_provider_t)margo_registered_data_mplex(mid, info->id, info->target_id);
    if(!svr_ctx) {
        fprintf(stderr, "Error: SDSKV could not find provider\n"); 
581
        out.ret = SDSKV_ERR_UNKNOWN_PR;
Matthieu Dorier's avatar
Matthieu Dorier committed
582 583 584 585 586 587 588
        margo_respond(handle, &out);
        margo_destroy(handle);
        return;
    }

    hret = margo_get_input(handle, &in);
    if(hret != HG_SUCCESS) {
589
        out.ret = SDSKV_ERR_MERCURY;
Matthieu Dorier's avatar
Matthieu Dorier committed
590 591 592 593 594 595 596
        margo_respond(handle, &out);
        margo_destroy(handle);
        return;
    }

    auto it = svr_ctx->databases.find(in.db_id);
    if(it == svr_ctx->databases.end()) {
597
        out.ret = SDSKV_ERR_UNKNOWN_DB;
Matthieu Dorier's avatar
Matthieu Dorier committed
598 599 600 601 602 603
        margo_respond(handle, &out);
        margo_free_input(handle, &in);
        margo_destroy(handle);
        return;
    }
    
604
    ds_bulk_t kdata(in.key.data, in.key.data+in.key.size);
Matthieu Dorier's avatar
Matthieu Dorier committed
605 606

    if(it->second->erase(kdata)) {
607
        out.ret   = SDSKV_SUCCESS;
Matthieu Dorier's avatar
Matthieu Dorier committed
608
    } else {
609
        out.ret   = SDSKV_ERR_ERASE;
Matthieu Dorier's avatar
Matthieu Dorier committed
610 611 612 613 614 615 616 617 618 619
    }

    margo_respond(handle, &out);
    margo_free_input(handle, &in);
    margo_destroy(handle); 

    return;
}
DEFINE_MARGO_RPC_HANDLER(sdskv_erase_ult)

620
static void sdskv_list_keys_ult(hg_handle_t handle)
Matthieu Dorier's avatar
Matthieu Dorier committed
621 622 623
{

    hg_return_t hret;
624 625
    list_keys_in_t in;
    list_keys_out_t out;
Matthieu Dorier's avatar
Matthieu Dorier committed
626 627
    hg_bulk_t ksizes_local_bulk = HG_BULK_NULL;
    hg_bulk_t keys_local_bulk   = HG_BULK_NULL;
Matthieu Dorier's avatar
Matthieu Dorier committed
628

629
    out.ret     = SDSKV_SUCCESS;
630 631
    out.nkeys   = 0;

Matthieu Dorier's avatar
Matthieu Dorier committed
632
    /* get the provider handling this request */
Matthieu Dorier's avatar
Matthieu Dorier committed
633 634 635 636 637 638
    margo_instance_id mid = margo_hg_handle_get_instance(handle);
    assert(mid);
    const struct hg_info* info = margo_get_info(handle);
    sdskv_provider_t svr_ctx = 
        (sdskv_provider_t)margo_registered_data_mplex(mid, info->id, info->target_id);
    if(!svr_ctx) {
639 640
        std::cerr << "Error: SDSKV list_keys could not find provider" << std::endl;
        out.ret = SDSKV_ERR_UNKNOWN_PR;
Matthieu Dorier's avatar
Matthieu Dorier committed
641 642 643 644 645
        margo_respond(handle, &out);
        margo_destroy(handle);
        return;
    }

Matthieu Dorier's avatar
Matthieu Dorier committed
646
    /* get the input */
Matthieu Dorier's avatar
Matthieu Dorier committed
647 648
    hret = margo_get_input(handle, &in);
    if(hret != HG_SUCCESS) {
Matthieu Dorier's avatar
Matthieu Dorier committed
649
        std::cerr << "Error: SDSKV list_keys could not get RPC input" << std::endl;
650
        out.ret = SDSKV_ERR_MERCURY;
Matthieu Dorier's avatar
Matthieu Dorier committed
651 652 653 654 655
        margo_respond(handle, &out);
        margo_destroy(handle);
        return;
    }

Matthieu Dorier's avatar
Matthieu Dorier committed
656 657 658 659 660 661
    try {

        /* find the database targeted */
        auto it = svr_ctx->databases.find(in.db_id);
        if(it == svr_ctx->databases.end()) {
            std::cerr << "Error: SDSKV list_keys could not get database with id " << in.db_id << std::endl;
662
            throw SDSKV_ERR_UNKNOWN_DB;
Matthieu Dorier's avatar
Matthieu Dorier committed
663 664 665 666 667 668 669 670 671 672 673 674
        }
        auto db = it->second;

        /* create a bulk handle to receive and send key sizes from client */
        std::vector<hg_size_t> ksizes(in.max_keys);
        std::vector<void*> ksizes_addr(1);
        ksizes_addr[0] = (void*)ksizes.data();
        hg_size_t ksizes_bulk_size = ksizes.size()*sizeof(hg_size_t);
        hret = margo_bulk_create(mid, 1, ksizes_addr.data(), 
                &ksizes_bulk_size, HG_BULK_READWRITE, &ksizes_local_bulk);
        if(hret != HG_SUCCESS) {
            std::cerr << "Error: SDSKV list_keys could not create bulk handle (ksizes_local_bulk)" << std::endl;
675
            throw SDSKV_ERR_MERCURY;
Matthieu Dorier's avatar
Matthieu Dorier committed
676 677
        }

678

Matthieu Dorier's avatar
Matthieu Dorier committed
679 680 681 682 683 684 685
        /* receive the key sizes from the client */
        hg_addr_t origin_addr = info->addr;
        hret = margo_bulk_transfer(mid, HG_BULK_PULL, origin_addr,
                in.ksizes_bulk_handle, 0, ksizes_local_bulk, 0, ksizes_bulk_size);
        if(hret != HG_SUCCESS) {
            std::cerr << "Error: SDSKV list_keys could not issue bulk transfer " 
                << "(pull from in.ksizes_bulk_handle to ksizes_local_bulk)" << std::endl;
686
            throw SDSKV_ERR_MERCURY;
Matthieu Dorier's avatar
Matthieu Dorier committed
687 688
        }

689 690 691
        /* make a copy of the remote key sizes */
        std::vector<hg_size_t> remote_ksizes(ksizes.begin(), ksizes.end());

Matthieu Dorier's avatar
Matthieu Dorier committed
692
        /* get the keys from the underlying database */    
693
        ds_bulk_t start_kdata(in.start_key.data, in.start_key.data+in.start_key.size);
694 695
        ds_bulk_t prefix(in.prefix.data, in.prefix.data+in.prefix.size);
        auto keys = db->list_keys(start_kdata, in.max_keys, prefix);
Matthieu Dorier's avatar
Matthieu Dorier committed
696 697 698 699 700 701 702 703 704
        hg_size_t num_keys = std::min(keys.size(), in.max_keys);

        /* create the array of actual sizes */
        std::vector<hg_size_t> true_ksizes(num_keys);
        hg_size_t keys_bulk_size = 0;
        for(unsigned i = 0; i < num_keys; i++) {
            true_ksizes[i] = keys[i].size();
            if(true_ksizes[i] > ksizes[i]) {
                // this key has a size that exceeds the allocated size on client
705
                throw SDSKV_ERR_SIZE;
Matthieu Dorier's avatar
Matthieu Dorier committed
706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722
            } else {
                ksizes[i] = true_ksizes[i];
                keys_bulk_size += ksizes[i];
            }
        }

        /* create an array of addresses pointing to keys */
        std::vector<void*> keys_addr(num_keys);
        for(unsigned i=0; i < num_keys; i++) {
            keys_addr[i] = (void*)(keys[i].data());
        }

        /* expose the keys for bulk transfer */
        hret = margo_bulk_create(mid, num_keys, keys_addr.data(),
                true_ksizes.data(), HG_BULK_READ_ONLY, &keys_local_bulk);
        if(hret != HG_SUCCESS) {
            std::cerr << "Error: SDSKV list_keys could not create bulk handle (keys_local_bulk)" << std::endl;
723
            throw SDSKV_ERR_MERCURY;
Matthieu Dorier's avatar
Matthieu Dorier committed
724 725 726 727 728 729 730 731
        }

        /* transfer the ksizes back to the client */
        hret = margo_bulk_transfer(mid, HG_BULK_PUSH, origin_addr, 
                in.ksizes_bulk_handle, 0, ksizes_local_bulk, 0, ksizes_bulk_size);
        if(hret != HG_SUCCESS) {
            std::cerr << "Error: SDSKV list_keys could not issue bulk transfer "
                << "(push from ksizes_local_bulk to in.ksizes_bulk_handle)" << std::endl;
732
            throw SDSKV_ERR_MERCURY;
Matthieu Dorier's avatar
Matthieu Dorier committed
733 734 735
        }

        /* transfer the keys to the client */
736 737 738 739 740 741 742 743
        uint64_t remote_offset = 0;
        uint64_t local_offset  = 0;
        for(unsigned i = 0; i < num_keys; i++) {

            hret = margo_bulk_transfer(mid, HG_BULK_PUSH, origin_addr,
                    in.keys_bulk_handle, remote_offset, keys_local_bulk, local_offset, true_ksizes[i]);
            if(hret != HG_SUCCESS) {
                std::cerr << "Error: SDSKV list_keys could not issue bulk transfer (keys_local_bulk)" << std::endl;
744
                throw SDSKV_ERR_MERCURY;
745 746 747 748 749 750 751
            }

            remote_offset += remote_ksizes[i];
            local_offset  += true_ksizes[i];
        }

        out.nkeys = num_keys;
752
        out.ret = SDSKV_SUCCESS;
753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778

    } catch(int exc_no) {
        out.ret = exc_no;
    }

    margo_bulk_free(ksizes_local_bulk);
    margo_bulk_free(keys_local_bulk);
    margo_respond(handle, &out);
    margo_free_input(handle, &in);
    margo_destroy(handle); 

    return;
}
DEFINE_MARGO_RPC_HANDLER(sdskv_list_keys_ult)

static void sdskv_list_keyvals_ult(hg_handle_t handle)
{

    hg_return_t hret;
    list_keyvals_in_t in;
    list_keyvals_out_t out;
    hg_bulk_t ksizes_local_bulk = HG_BULK_NULL;
    hg_bulk_t keys_local_bulk   = HG_BULK_NULL;
    hg_bulk_t vsizes_local_bulk = HG_BULK_NULL;
    hg_bulk_t vals_local_bulk   = HG_BULK_NULL;

779
    out.ret     = SDSKV_SUCCESS;
780 781 782 783 784 785 786 787 788
    out.nkeys   = 0;

    /* get the provider handling this request */
    margo_instance_id mid = margo_hg_handle_get_instance(handle);
    assert(mid);
    const struct hg_info* info = margo_get_info(handle);
    sdskv_provider_t svr_ctx = 
        (sdskv_provider_t)margo_registered_data_mplex(mid, info->id, info->target_id);
    if(!svr_ctx) {
789 790
        std::cerr << "Error: SDSKV list_keyvals could not find provider" << std::endl;
        out.ret = SDSKV_ERR_UNKNOWN_PR;
791 792 793 794 795 796 797 798 799
        margo_respond(handle, &out);
        margo_destroy(handle);
        return;
    }

    /* get the input */
    hret = margo_get_input(handle, &in);
    if(hret != HG_SUCCESS) {
        std::cerr << "Error: SDSKV list_keyvals could not get RPC input" << std::endl;
800
        out.ret = SDSKV_ERR_MERCURY;
801 802 803 804 805 806 807 808 809 810 811
        margo_respond(handle, &out);
        margo_destroy(handle);
        return;
    }

    try {

        /* find the database targeted */
        auto it = svr_ctx->databases.find(in.db_id);
        if(it == svr_ctx->databases.end()) {
            std::cerr << "Error: SDSKV list_keyvals could not get database with id " << in.db_id << std::endl;
812
            throw SDSKV_ERR_UNKNOWN_DB;
813 814 815 816 817 818 819 820 821 822 823 824
        }
        auto db = it->second;

        /* create a bulk handle to receive and send key sizes from client */
        std::vector<hg_size_t> ksizes(in.max_keys);
        std::vector<void*> ksizes_addr(1);
        ksizes_addr[0] = (void*)ksizes.data();
        hg_size_t ksizes_bulk_size = ksizes.size()*sizeof(hg_size_t);
        hret = margo_bulk_create(mid, 1, ksizes_addr.data(), 
                &ksizes_bulk_size, HG_BULK_READWRITE, &ksizes_local_bulk);
        if(hret != HG_SUCCESS) {
            std::cerr << "Error: SDSKV list_keyvals could not create bulk handle (ksizes_local_bulk)" << std::endl;
825
            throw SDSKV_ERR_MERCURY;
826 827 828 829 830 831 832 833 834 835 836
        }

        /* create a bulk handle to receive and send value sizes from client */
        std::vector<hg_size_t> vsizes(in.max_keys);
        std::vector<void*> vsizes_addr(1);
        vsizes_addr[0] = (void*)vsizes.data();
        hg_size_t vsizes_bulk_size = vsizes.size()*sizeof(hg_size_t);
        hret = margo_bulk_create(mid, 1, vsizes_addr.data(), 
                &vsizes_bulk_size, HG_BULK_READWRITE, &vsizes_local_bulk);
        if(hret != HG_SUCCESS) {
            std::cerr << "Error: SDSKV list_keyvals could not create bulk handle (vsizes_local_bulk)" << std::endl;
837
            throw SDSKV_ERR_MERCURY;
838 839 840 841 842 843 844 845 846
        }

        /* receive the key sizes from the client */
        hg_addr_t origin_addr = info->addr;
        hret = margo_bulk_transfer(mid, HG_BULK_PULL, origin_addr,
                in.ksizes_bulk_handle, 0, ksizes_local_bulk, 0, ksizes_bulk_size);
        if(hret != HG_SUCCESS) {
            std::cerr << "Error: SDSKV list_keyvals could not issue bulk transfer " 
                << "(pull from in.ksizes_bulk_handle to ksizes_local_bulk)" << std::endl;
847
            throw SDSKV_ERR_MERCURY;
848 849 850 851 852
        }

        /* receive the values sizes from the client */
        hret = margo_bulk_transfer(mid, HG_BULK_PULL, origin_addr,
                in.vsizes_bulk_handle, 0, vsizes_local_bulk, 0, vsizes_bulk_size);
Matthieu Dorier's avatar
Matthieu Dorier committed
853
        if(hret != HG_SUCCESS) {
854 855
            std::cerr << "Error: SDSKV list_keyvals could not issue bulk transfer " 
                << "(pull from in.vsizes_bulk_handle to vsizes_local_bulk)" << std::endl;
856
            throw SDSKV_ERR_MERCURY;
Matthieu Dorier's avatar
Matthieu Dorier committed
857 858
        }

859 860 861 862 863
        /* make a copy of the remote key sizes and value sizes */
        std::vector<hg_size_t> remote_ksizes(ksizes.begin(), ksizes.end());
        std::vector<hg_size_t> remote_vsizes(vsizes.begin(), vsizes.end());

        /* get the keys and values from the underlying database */    
864
        ds_bulk_t start_kdata(in.start_key.data, in.start_key.data+in.start_key.size);
865 866
        ds_bulk_t prefix(in.prefix.data, in.prefix.data+in.prefix.size);
        auto keyvals = db->list_keyvals(start_kdata, in.max_keys, prefix);
867 868 869 870 871 872 873 874 875
        hg_size_t num_keys = std::min(keyvals.size(), in.max_keys);

        /* create the array of actual key sizes */
        std::vector<hg_size_t> true_ksizes(num_keys);
        hg_size_t keys_bulk_size = 0;
        for(unsigned i = 0; i < num_keys; i++) {
            true_ksizes[i] = keyvals[i].first.size();
            if(true_ksizes[i] > ksizes[i]) {
                // this key has a size that exceeds the allocated size on client
876
                throw SDSKV_ERR_SIZE;
877 878 879 880 881 882 883 884 885 886 887 888 889
            } else {
                ksizes[i] = true_ksizes[i];
                keys_bulk_size += ksizes[i];
            }
        }

        /* create the array of actual value sizes */
        std::vector<hg_size_t> true_vsizes(num_keys);
        hg_size_t vals_bulk_size = 0;
        for(unsigned i = 0; i < num_keys; i++) {
            true_vsizes[i] = keyvals[i].second.size();
            if(true_vsizes[i] > vsizes[i]) {
                // this value has a size that exceeds the allocated size on client
890
                throw SDSKV_ERR_SIZE;
891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913
            } else {
                vsizes[i] = true_vsizes[i];
                vals_bulk_size += vsizes[i];
            }
        }

        /* create an array of addresses pointing to keys */
        std::vector<void*> keys_addr(num_keys);
        for(unsigned i=0; i < num_keys; i++) {
            keys_addr[i] = (void*)(keyvals[i].first.data());
        }

        /* create an array of addresses pointing to values */
        std::vector<void*> vals_addr(num_keys);
        for(unsigned i=0; i < num_keys; i++) {
            vals_addr[i] = (void*)(keyvals[i].second.data());
        }

        /* expose the keys for bulk transfer */
        hret = margo_bulk_create(mid, num_keys, keys_addr.data(),
                true_ksizes.data(), HG_BULK_READ_ONLY, &keys_local_bulk);
        if(hret != HG_SUCCESS) {
            std::cerr << "Error: SDSKV list_keyvals could not create bulk handle (keys_local_bulk)" << std::endl;
914
            throw SDSKV_ERR_MERCURY;
915 916 917 918 919 920 921
        }

        /* expose the values for bulk transfer */
        hret = margo_bulk_create(mid, num_keys, vals_addr.data(),
                true_vsizes.data(), HG_BULK_READ_ONLY, &vals_local_bulk);
        if(hret != HG_SUCCESS) {
            std::cerr << "Error: SDSKV list_keyvals could not create bulk handle (vals_local_bulk)" << std::endl;
922
            throw SDSKV_ERR_MERCURY;
923 924 925 926 927 928 929 930
        }

        /* transfer the ksizes back to the client */
        hret = margo_bulk_transfer(mid, HG_BULK_PUSH, origin_addr, 
                in.ksizes_bulk_handle, 0, ksizes_local_bulk, 0, ksizes_bulk_size);
        if(hret != HG_SUCCESS) {
            std::cerr << "Error: SDSKV list_keyvals could not issue bulk transfer "
                << "(push from ksizes_local_bulk to in.ksizes_bulk_handle)" << std::endl;
931
            throw SDSKV_ERR_MERCURY;
932 933 934 935 936 937 938 939
        }

        /* transfer the vsizes back to the client */
        hret = margo_bulk_transfer(mid, HG_BULK_PUSH, origin_addr, 
                in.vsizes_bulk_handle, 0, vsizes_local_bulk, 0, vsizes_bulk_size);
        if(hret != HG_SUCCESS) {
            std::cerr << "Error: SDSKV list_keyvals could not issue bulk transfer "
                << "(push from vsizes_local_bulk to in.vsizes_bulk_handle)" << std::endl;
940
            throw SDSKV_ERR_MERCURY;
941 942 943 944 945 946 947 948 949 950 951
        }

        uint64_t remote_offset = 0;
        uint64_t local_offset  = 0;

        /* transfer the keys to the client */
        for(unsigned i=0; i < num_keys; i++) {
            hret = margo_bulk_transfer(mid, HG_BULK_PUSH, origin_addr,
                    in.keys_bulk_handle, remote_offset, keys_local_bulk, local_offset, true_ksizes[i]);
            if(hret != HG_SUCCESS) {
                std::cerr << "Error: SDSKV list_keyvals could not issue bulk transfer (keys_local_bulk)" << std::endl;
952
                throw SDSKV_ERR_MERCURY;
953 954 955 956 957 958 959 960 961 962 963 964 965 966
            }
            remote_offset += remote_ksizes[i];
            local_offset  += true_ksizes[i];
        }

        remote_offset = 0;
        local_offset  = 0;

        /* transfer the values to the client */
        for(unsigned i=0; i < num_keys; i++) {
            hret = margo_bulk_transfer(mid, HG_BULK_PUSH, origin_addr,
                    in.vals_bulk_handle, remote_offset, vals_local_bulk, local_offset, true_vsizes[i]);
            if(hret != HG_SUCCESS) {
                std::cerr << "Error: SDSKV list_keyvals could not issue bulk transfer (vals_local_bulk)" << std::endl;
967
                throw SDSKV_ERR_MERCURY;
968 969 970 971 972
            }
            remote_offset += remote_vsizes[i];
            local_offset  += true_vsizes[i];
        }

Matthieu Dorier's avatar
Matthieu Dorier committed
973
        out.nkeys = num_keys;
974
        out.ret = SDSKV_SUCCESS;
Matthieu Dorier's avatar
Matthieu Dorier committed
975 976 977

    } catch(int exc_no) {
        out.ret = exc_no;
Matthieu Dorier's avatar
Matthieu Dorier committed
978 979
    }

Matthieu Dorier's avatar
Matthieu Dorier committed
980 981
    margo_bulk_free(ksizes_local_bulk);
    margo_bulk_free(keys_local_bulk);
982 983
    margo_bulk_free(vsizes_local_bulk);
    margo_bulk_free(vals_local_bulk);
Matthieu Dorier's avatar
Matthieu Dorier committed
984 985 986 987 988 989
    margo_respond(handle, &out);
    margo_free_input(handle, &in);
    margo_destroy(handle); 

    return;
}
990
DEFINE_MARGO_RPC_HANDLER(sdskv_list_keyvals_ult)
Matthieu Dorier's avatar
Matthieu Dorier committed
991 992 993 994 995 996 997 998

static void sdskv_server_finalize_cb(void *data)
{
    sdskv_provider_t svr_ctx = (sdskv_provider_t)data;
    assert(svr_ctx);

    sdskv_provider_remove_all_databases(svr_ctx);

Matthieu Dorier's avatar
Matthieu Dorier committed
999
    delete svr_ctx;
Matthieu Dorier's avatar
Matthieu Dorier committed
1000 1001 1002 1003

    return;
}