margo.c 42.1 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11

/*
 * (C) 2015 The University of Chicago
 * 
 * See COPYRIGHT in top-level directory.
 */

#include <assert.h>
#include <unistd.h>
#include <errno.h>
#include <abt.h>
Philip Carns's avatar
Philip Carns committed
12
#include <stdlib.h>
Matthieu Dorier's avatar
Matthieu Dorier committed
13 14 15

#include <margo-config.h>
#ifdef HAVE_ABT_SNOOZER
16
#include <abt-snoozer.h>
Matthieu Dorier's avatar
Matthieu Dorier committed
17
#endif
18
#include <time.h>
Philip Carns's avatar
bug fix  
Philip Carns committed
19
#include <math.h>
20 21

#include "margo.h"
22
#include "margo-timer.h"
Philip Carns's avatar
Philip Carns committed
23
#include "utlist.h"
Philip Carns's avatar
Philip Carns committed
24
#include "uthash.h"
25

26
#define DEFAULT_MERCURY_PROGRESS_TIMEOUT_UB 100 /* 100 milliseconds */
Shane Snyder's avatar
Shane Snyder committed
27
#define DEFAULT_MERCURY_HANDLE_CACHE_SIZE 32
28

29
struct provider_element
Philip Carns's avatar
Philip Carns committed
30 31 32
{
    hg_id_t id;
    ABT_pool pool;
33 34
    void* user_data;
    void(*user_free_callback)(void*);
Philip Carns's avatar
Philip Carns committed
35 36 37
    UT_hash_handle hh;
};

Philip Carns's avatar
Philip Carns committed
38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53
struct diag_data
{
    double min;
    double max;
    double cumulative;
    int count;
};

#define __DIAG_UPDATE(__data, __time)\
do {\
    __data.count++; \
    __data.cumulative += (__time); \
    if((__time) > __data.max) __data.max = (__time); \
    if((__time) < __data.min) __data.min = (__time); \
} while(0)

Shane Snyder's avatar
Shane Snyder committed
54 55 56 57 58 59 60
struct margo_handle_cache_el
{
    hg_handle_t handle;
    UT_hash_handle hh; /* in-use hash link */
    struct margo_handle_cache_el *next; /* free list link */
};

61 62 63 64 65 66 67
struct margo_finalize_cb
{
    void(*callback)(void*);
    void* uargs;
    struct margo_finalize_cb* next;
};

68 69
struct margo_timer_list; /* defined in margo-timer.c */

70 71
struct margo_instance
{
Shane Snyder's avatar
Shane Snyder committed
72
    /* mercury/argobots state */
73 74
    hg_context_t *hg_context;
    hg_class_t *hg_class;
75 76 77
    ABT_pool handler_pool;
    ABT_pool progress_pool;

78
    /* internal to margo for this particular instance */
Shane Snyder's avatar
Shane Snyder committed
79
    int margo_init;
80
    int abt_init;
81 82
    ABT_thread hg_progress_tid;
    int hg_progress_shutdown_flag;
Jonathan Jenkins's avatar
Jonathan Jenkins committed
83
    ABT_xstream progress_xstream;
Jonathan Jenkins's avatar
Jonathan Jenkins committed
84 85 86
    int owns_progress_pool;
    ABT_xstream *rpc_xstreams;
    int num_handler_pool_threads;
87
    unsigned int hg_progress_timeout_ub;
88 89 90

    /* control logic for callers waiting on margo to be finalized */
    int finalize_flag;
Jonathan Jenkins's avatar
Jonathan Jenkins committed
91
    int refcount;
92 93
    ABT_mutex finalize_mutex;
    ABT_cond finalize_cond;
94
    struct margo_finalize_cb* finalize_cb;
95

Matthieu Dorier's avatar
Matthieu Dorier committed
96 97 98 99
    /* control logic for shutting down */
    hg_id_t shutdown_rpc_id;
    int enable_remote_shutdown;

100 101 102
    /* timer data */
    struct margo_timer_list* timer_list;

103 104
    /* hash table to track provider IDs registered with margo */
    struct provider_element *provider_table;
Philip Carns's avatar
Philip Carns committed
105

Shane Snyder's avatar
Shane Snyder committed
106 107 108
    /* linked list of free hg handles and a hash of in-use handles */
    struct margo_handle_cache_el *free_handle_list;
    struct margo_handle_cache_el *used_handle_hash;
109
    ABT_mutex handle_cache_mtx; /* mutex protecting access to above caches */
Shane Snyder's avatar
Shane Snyder committed
110

Philip Carns's avatar
Philip Carns committed
111 112 113 114 115 116 117 118 119 120 121
    /* optional diagnostics data tracking */
    /* NOTE: technically the following fields are subject to races if they
     * are updated from more than one thread at a time.  We will be careful
     * to only update the counters from the progress_fn,
     * which will serialize access.
     */
    int diag_enabled;
    struct diag_data diag_trigger_elapsed;
    struct diag_data diag_progress_elapsed_zero_timeout;
    struct diag_data diag_progress_elapsed_nonzero_timeout;
    struct diag_data diag_progress_timeout_value;
122 123
};

124 125 126 127 128 129
struct margo_rpc_data
{
	margo_instance_id mid;
	void* user_data;
	void (*user_free_callback)(void *);
};
130

Matthieu Dorier's avatar
Matthieu Dorier committed
131 132
MERCURY_GEN_PROC(margo_shutdown_out_t, ((int32_t)(ret)))

133
static void hg_progress_fn(void* foo);
134
static void margo_rpc_data_free(void* ptr);
Matthieu Dorier's avatar
Matthieu Dorier committed
135 136
static void remote_shutdown_ult(hg_handle_t handle);
DECLARE_MARGO_RPC_HANDLER(remote_shutdown_ult);
137

138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173
static inline void demux_id(hg_id_t in, hg_id_t* base_id, uint16_t *provider_id)
{
    /* retrieve low bits for provider */
    *provider_id = 0;
    *provider_id += (in & (((1<<(__MARGO_PROVIDER_ID_SIZE*8))-1)));

    /* clear low order bits */
    *base_id = (in >> (__MARGO_PROVIDER_ID_SIZE*8)) <<
        (__MARGO_PROVIDER_ID_SIZE*8);

    return;
}

static inline hg_id_t mux_id(hg_id_t base_id, uint16_t provider_id)
{
    hg_id_t id;

    id = (base_id >> (__MARGO_PROVIDER_ID_SIZE*8)) <<
       (__MARGO_PROVIDER_ID_SIZE*8);
    id |= provider_id;

    return id;
}

static inline hg_id_t gen_id(const char* func_name, uint16_t provider_id)
{
    hg_id_t id;
    unsigned hashval;

    HASH_JEN(func_name, strlen(func_name), hashval);
    id = hashval << (__MARGO_PROVIDER_ID_SIZE*8);
    id |= provider_id;

    return id;
}

Shane Snyder's avatar
Shane Snyder committed
174 175 176 177 178 179
static hg_return_t margo_handle_cache_init(margo_instance_id mid);
static void margo_handle_cache_destroy(margo_instance_id mid);
static hg_return_t margo_handle_cache_get(margo_instance_id mid,
    hg_addr_t addr, hg_id_t id, hg_handle_t *handle);
static hg_return_t margo_handle_cache_put(margo_instance_id mid,
    hg_handle_t handle);
180 181
static void delete_provider_hash(margo_instance_id mid);
static int margo_lookup_provider(margo_instance_id mid, hg_id_t id, uint16_t provider_id, ABT_pool *pool);
182 183
static hg_id_t margo_register_internal(margo_instance_id mid, hg_id_t id,
    hg_proc_cb_t in_proc_cb, hg_proc_cb_t out_proc_cb, hg_rpc_cb_t rpc_cb);
Shane Snyder's avatar
Shane Snyder committed
184

Shane Snyder's avatar
Shane Snyder committed
185
margo_instance_id margo_init(const char *addr_str, int mode,
Shane Snyder's avatar
Shane Snyder committed
186
    int use_progress_thread, int rpc_thread_count)
187
{
Jonathan Jenkins's avatar
Jonathan Jenkins committed
188 189 190 191 192
    ABT_xstream progress_xstream = ABT_XSTREAM_NULL;
    ABT_pool progress_pool = ABT_POOL_NULL;
    ABT_xstream *rpc_xstreams = NULL;
    ABT_xstream rpc_xstream = ABT_XSTREAM_NULL;
    ABT_pool rpc_pool = ABT_POOL_NULL;
Shane Snyder's avatar
Shane Snyder committed
193 194
    hg_class_t *hg_class = NULL;
    hg_context_t *hg_context = NULL;
Shane Snyder's avatar
Shane Snyder committed
195
    int listen_flag = (mode == MARGO_CLIENT_MODE) ? HG_FALSE : HG_TRUE;
196
    int abt_init = 0;
Jonathan Jenkins's avatar
Jonathan Jenkins committed
197
    int i;
Shane Snyder's avatar
Shane Snyder committed
198 199 200
    int ret;
    struct margo_instance *mid = MARGO_INSTANCE_NULL;

Shane Snyder's avatar
Shane Snyder committed
201
    if(mode != MARGO_CLIENT_MODE && mode != MARGO_SERVER_MODE) goto err;
Shane Snyder's avatar
Shane Snyder committed
202

Philip Carns's avatar
Philip Carns committed
203 204 205 206 207 208 209 210 211 212 213 214
    /* NOTE: Margo is very likely to create a single producer (the
     * progress function), multiple consumer usage pattern that
     * causes excess memory consumption in some versions of
     * Argobots.  See
     * https://xgitlab.cels.anl.gov/sds/margo/issues/40 for details.
     * We therefore manually set the ABT_MEM_MAX_NUM_STACKS parameter 
     * for Argobots to a low value so that RPC handler threads do not
     * queue large numbers of stacks for reuse in per-ES data 
     * structures.
     */
    putenv("ABT_MEM_MAX_NUM_STACKS=8");

215 216 217 218 219 220
    if (ABT_initialized() == ABT_ERR_UNINITIALIZED)
    {
        ret = ABT_init(0, NULL); /* XXX: argc/argv not currently used by ABT ... */
        if(ret != 0) goto err;
        abt_init = 1;
    }
Shane Snyder's avatar
Shane Snyder committed
221

222
    /* set caller (self) ES to idle without polling */
Matthieu Dorier's avatar
Matthieu Dorier committed
223
#ifdef HAVE_ABT_SNOOZER
Shane Snyder's avatar
Shane Snyder committed
224 225
    ret = ABT_snoozer_xstream_self_set();
    if(ret != 0) goto err;
Matthieu Dorier's avatar
Matthieu Dorier committed
226
#endif
Jonathan Jenkins's avatar
Jonathan Jenkins committed
227 228 229

    if (use_progress_thread)
    {
Matthieu Dorier's avatar
Matthieu Dorier committed
230
#ifdef HAVE_ABT_SNOOZER
Jonathan Jenkins's avatar
Jonathan Jenkins committed
231
        ret = ABT_snoozer_xstream_create(1, &progress_pool, &progress_xstream);
Matthieu Dorier's avatar
Matthieu Dorier committed
232 233 234 235 236 237 238
		if (ret != ABT_SUCCESS) goto err;
#else
		ret = ABT_xstream_create(ABT_SCHED_NULL, &progress_xstream);
		if (ret != ABT_SUCCESS) goto err;
		ret = ABT_xstream_get_main_pools(progress_xstream, 1, &progress_pool);
		if (ret != ABT_SUCCESS) goto err;
#endif
Jonathan Jenkins's avatar
Jonathan Jenkins committed
239 240 241 242 243 244 245 246 247
    }
    else
    {
        ret = ABT_xstream_self(&progress_xstream);
        if (ret != ABT_SUCCESS) goto err;
        ret = ABT_xstream_get_main_pools(progress_xstream, 1, &progress_pool);
        if (ret != ABT_SUCCESS) goto err;
    }

248
    if (rpc_thread_count > 0)
Jonathan Jenkins's avatar
Jonathan Jenkins committed
249
    {
250 251
        rpc_xstreams = calloc(rpc_thread_count, sizeof(*rpc_xstreams));
        if (rpc_xstreams == NULL) goto err;
Matthieu Dorier's avatar
Matthieu Dorier committed
252
#ifdef HAVE_ABT_SNOOZER
253 254 255
        ret = ABT_snoozer_xstream_create(rpc_thread_count, &rpc_pool,
                rpc_xstreams);
        if (ret != ABT_SUCCESS) goto err;
Matthieu Dorier's avatar
Matthieu Dorier committed
256
#else
257 258 259 260 261
        int j;
        ret = ABT_pool_create_basic(ABT_POOL_FIFO, ABT_POOL_ACCESS_MPMC, ABT_TRUE, &rpc_pool);
        if (ret != ABT_SUCCESS) goto err;
        for(j=0; j<rpc_thread_count; j++) {
            ret = ABT_xstream_create(ABT_SCHED_NULL, rpc_xstreams+j);
Shane Snyder's avatar
Shane Snyder committed
262 263
            if (ret != ABT_SUCCESS) goto err;
        }
264 265 266 267 268 269 270 271 272 273 274 275
#endif
    }
    else if (rpc_thread_count == 0)
    {
        ret = ABT_xstream_self(&rpc_xstream);
        if (ret != ABT_SUCCESS) goto err;
        ret = ABT_xstream_get_main_pools(rpc_xstream, 1, &rpc_pool);
        if (ret != ABT_SUCCESS) goto err;
    }
    else
    {
        rpc_pool = progress_pool;
Jonathan Jenkins's avatar
Jonathan Jenkins committed
276 277
    }

Shane Snyder's avatar
Shane Snyder committed
278 279 280 281 282 283
    hg_class = HG_Init(addr_str, listen_flag);
    if(!hg_class) goto err;

    hg_context = HG_Context_create(hg_class);
    if(!hg_context) goto err;

Jonathan Jenkins's avatar
Jonathan Jenkins committed
284 285 286
    mid = margo_init_pool(progress_pool, rpc_pool, hg_context);
    if (mid == MARGO_INSTANCE_NULL) goto err;

Shane Snyder's avatar
Shane Snyder committed
287
    mid->margo_init = 1;
288
    mid->abt_init = abt_init;
Jonathan Jenkins's avatar
Jonathan Jenkins committed
289 290 291
    mid->owns_progress_pool = use_progress_thread;
    mid->progress_xstream = progress_xstream;
    mid->num_handler_pool_threads = rpc_thread_count < 0 ? 0 : rpc_thread_count;
Jonathan Jenkins's avatar
Jonathan Jenkins committed
292
    mid->rpc_xstreams = rpc_xstreams;
293

Jonathan Jenkins's avatar
Jonathan Jenkins committed
294 295 296
    return mid;

err:
Shane Snyder's avatar
Shane Snyder committed
297 298
    if(mid)
    {
299
        margo_timer_list_free(mid->timer_list);
Shane Snyder's avatar
Shane Snyder committed
300 301 302 303
        ABT_mutex_free(&mid->finalize_mutex);
        ABT_cond_free(&mid->finalize_cond);
        free(mid);
    }
Jonathan Jenkins's avatar
Jonathan Jenkins committed
304 305 306 307 308 309 310 311 312 313 314 315 316 317
    if (use_progress_thread && progress_xstream != ABT_XSTREAM_NULL)
    {
        ABT_xstream_join(progress_xstream);
        ABT_xstream_free(&progress_xstream);
    }
    if (rpc_thread_count > 0 && rpc_xstreams != NULL)
    {
        for (i = 0; i < rpc_thread_count; i++)
        {
            ABT_xstream_join(rpc_xstreams[i]);
            ABT_xstream_free(&rpc_xstreams[i]);
        }
        free(rpc_xstreams);
    }
Shane Snyder's avatar
Shane Snyder committed
318 319 320 321
    if(hg_context)
        HG_Context_destroy(hg_context);
    if(hg_class)
        HG_Finalize(hg_class);
322 323
    if(abt_init)
        ABT_finalize();
Jonathan Jenkins's avatar
Jonathan Jenkins committed
324 325 326 327
    return MARGO_INSTANCE_NULL;
}

margo_instance_id margo_init_pool(ABT_pool progress_pool, ABT_pool handler_pool,
Jonathan Jenkins's avatar
Jonathan Jenkins committed
328
    hg_context_t *hg_context)
329 330
{
    int ret;
Shane Snyder's avatar
Shane Snyder committed
331
    hg_return_t hret;
332 333
    struct margo_instance *mid;

Matthieu Dorier's avatar
Matthieu Dorier committed
334
    mid = calloc(1,sizeof(*mid));
Shane Snyder's avatar
Shane Snyder committed
335
    if(!mid) goto err;
336
    memset(mid, 0, sizeof(*mid));
337

338 339 340
    ABT_mutex_create(&mid->finalize_mutex);
    ABT_cond_create(&mid->finalize_cond);

341 342
    mid->progress_pool = progress_pool;
    mid->handler_pool = handler_pool;
Jonathan Jenkins's avatar
Jonathan Jenkins committed
343
    mid->hg_class = HG_Context_get_class(hg_context);
344
    mid->hg_context = hg_context;
345
    mid->hg_progress_timeout_ub = DEFAULT_MERCURY_PROGRESS_TIMEOUT_UB;
346
    mid->provider_table = NULL;
Jonathan Jenkins's avatar
Jonathan Jenkins committed
347
    mid->refcount = 1;
348
    mid->finalize_cb = NULL;
Matthieu Dorier's avatar
Matthieu Dorier committed
349
    mid->enable_remote_shutdown = 0;
350

351 352
    mid->timer_list = margo_timer_list_create();
    if(mid->timer_list == NULL) goto err;
353

Shane Snyder's avatar
Shane Snyder committed
354 355 356 357
    /* initialize the handle cache */
    hret = margo_handle_cache_init(mid);
    if(hret != HG_SUCCESS) goto err;

358
    ret = ABT_thread_create(mid->progress_pool, hg_progress_fn, mid, 
359
        ABT_THREAD_ATTR_NULL, &mid->hg_progress_tid);
Shane Snyder's avatar
Shane Snyder committed
360 361
    if(ret != 0) goto err;

Matthieu Dorier's avatar
Matthieu Dorier committed
362 363 364
    mid->shutdown_rpc_id = MARGO_REGISTER(mid, "__shutdown__", 
            void, margo_shutdown_out_t, remote_shutdown_ult);

Shane Snyder's avatar
Shane Snyder committed
365 366
    return mid;

Shane Snyder's avatar
Shane Snyder committed
367 368
err:
    if(mid)
369
    {
Shane Snyder's avatar
Shane Snyder committed
370
        margo_handle_cache_destroy(mid);
371
        margo_timer_list_free(mid->timer_list);
Shane Snyder's avatar
Shane Snyder committed
372 373
        ABT_mutex_free(&mid->finalize_mutex);
        ABT_cond_free(&mid->finalize_cond);
374
        free(mid);
375
    }
Shane Snyder's avatar
Shane Snyder committed
376
    return MARGO_INSTANCE_NULL;
377 378
}

Jonathan Jenkins's avatar
Jonathan Jenkins committed
379 380 381 382
static void margo_cleanup(margo_instance_id mid)
{
    int i;

383 384 385 386 387 388 389 390 391
    /* call finalize callbacks */
    struct margo_finalize_cb* fcb = mid->finalize_cb;
    while(fcb) {
        (fcb->callback)(fcb->uargs);
        struct margo_finalize_cb* tmp = fcb;
        fcb = fcb->next;
        free(tmp);
    }

392
    margo_timer_list_free(mid->timer_list);
Jonathan Jenkins's avatar
Jonathan Jenkins committed
393

394 395
    /* delete the hash used for provider IDs */
    delete_provider_hash(mid);
396

Jonathan Jenkins's avatar
Jonathan Jenkins committed
397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415
    ABT_mutex_free(&mid->finalize_mutex);
    ABT_cond_free(&mid->finalize_cond);

    if (mid->owns_progress_pool)
    {
        ABT_xstream_join(mid->progress_xstream);
        ABT_xstream_free(&mid->progress_xstream);
    }

    if (mid->num_handler_pool_threads > 0)
    {
        for (i = 0; i < mid->num_handler_pool_threads; i++)
        {
            ABT_xstream_join(mid->rpc_xstreams[i]);
            ABT_xstream_free(&mid->rpc_xstreams[i]);
        }
        free(mid->rpc_xstreams);
    }

Shane Snyder's avatar
Shane Snyder committed
416 417
    margo_handle_cache_destroy(mid);

Shane Snyder's avatar
Shane Snyder committed
418 419 420 421 422 423
    if (mid->margo_init)
    {
        if (mid->hg_context)
            HG_Context_destroy(mid->hg_context);
        if (mid->hg_class)
            HG_Finalize(mid->hg_class);
424 425
        if (mid->abt_init)
            ABT_finalize();
Shane Snyder's avatar
Shane Snyder committed
426 427
    }

Jonathan Jenkins's avatar
Jonathan Jenkins committed
428 429 430
    free(mid);
}

431
void margo_finalize(margo_instance_id mid)
432
{
Jonathan Jenkins's avatar
Jonathan Jenkins committed
433
    int do_cleanup;
434

435
    /* tell progress thread to wrap things up */
436
    mid->hg_progress_shutdown_flag = 1;
437 438

    /* wait for it to shutdown cleanly */
439 440
    ABT_thread_join(mid->hg_progress_tid);
    ABT_thread_free(&mid->hg_progress_tid);
441

442 443 444 445
    ABT_mutex_lock(mid->finalize_mutex);
    mid->finalize_flag = 1;
    ABT_cond_broadcast(mid->finalize_cond);

Jonathan Jenkins's avatar
Jonathan Jenkins committed
446 447
    mid->refcount--;
    do_cleanup = mid->refcount == 0;
448

Jonathan Jenkins's avatar
Jonathan Jenkins committed
449 450 451 452 453 454 455
    ABT_mutex_unlock(mid->finalize_mutex);

    /* if there was noone waiting on the finalize at the time of the finalize
     * broadcast, then we're safe to clean up. Otherwise, let the finalizer do
     * it */
    if (do_cleanup)
        margo_cleanup(mid);
456 457 458 459 460 461

    return;
}

void margo_wait_for_finalize(margo_instance_id mid)
{
Jonathan Jenkins's avatar
Jonathan Jenkins committed
462
    int do_cleanup;
463 464 465

    ABT_mutex_lock(mid->finalize_mutex);

Jonathan Jenkins's avatar
Jonathan Jenkins committed
466
        mid->refcount++;
467 468 469 470
            
        while(!mid->finalize_flag)
            ABT_cond_wait(mid->finalize_cond, mid->finalize_mutex);

Jonathan Jenkins's avatar
Jonathan Jenkins committed
471 472 473
        mid->refcount--;
        do_cleanup = mid->refcount == 0;

474
    ABT_mutex_unlock(mid->finalize_mutex);
Jonathan Jenkins's avatar
Jonathan Jenkins committed
475 476 477 478

    if (do_cleanup)
        margo_cleanup(mid);

479 480 481
    return;
}

482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498
void margo_push_finalize_callback(
            margo_instance_id mid,
            void(*cb)(void*),                  
            void* uargs)
{
    if(cb == NULL) return;

    struct margo_finalize_cb* fcb = 
        (struct margo_finalize_cb*)malloc(sizeof(*fcb));
    fcb->callback = cb;
    fcb->uargs = uargs;

    struct margo_finalize_cb* next = mid->finalize_cb;
    fcb->next = next;
    mid->finalize_cb = fcb;
}

Matthieu Dorier's avatar
Matthieu Dorier committed
499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514
void margo_enable_remote_shutdown(margo_instance_id mid)
{
    mid->enable_remote_shutdown = 1;
}

int margo_shutdown_remote_instance(
        margo_instance_id mid,
        hg_addr_t remote_addr)
{
    hg_return_t hret;
    hg_handle_t handle;

    hret = margo_create(mid, remote_addr,
                        mid->shutdown_rpc_id, &handle);
    if(hret != HG_SUCCESS) return -1;

515
    hret = margo_forward(handle, NULL);
Matthieu Dorier's avatar
Matthieu Dorier committed
516 517 518 519 520 521 522 523 524 525 526 527 528 529
    if(hret != HG_SUCCESS)
    {
        margo_destroy(handle);
        return -1;
    }

    margo_shutdown_out_t out;
    hret = margo_get_output(handle, &out);
    if(hret != HG_SUCCESS)
    {
        margo_destroy(handle);
        return -1;
    }

530
    margo_free_output(handle, &out);
Matthieu Dorier's avatar
Matthieu Dorier committed
531 532 533 534 535
    margo_destroy(handle);

    return out.ret;
}

536
hg_id_t margo_provider_register_name(margo_instance_id mid, const char *func_name,
537
    hg_proc_cb_t in_proc_cb, hg_proc_cb_t out_proc_cb, hg_rpc_cb_t rpc_cb,
538
    uint16_t provider_id, ABT_pool pool)
539
{
540
    struct provider_element *element;
541
    hg_id_t id;
542 543 544 545 546
    int ret;

    assert(provider_id <= MARGO_MAX_PROVIDER_ID);
    
    id = gen_id(func_name, provider_id);
547

548 549
    ret = margo_register_internal(mid, id, in_proc_cb, out_proc_cb, rpc_cb);
    if(ret == 0)
550 551 552
        return(0);

    /* nothing to do, we'll let the handler pool take this directly */
553
    if(provider_id == MARGO_DEFAULT_PROVIDER_ID)
554 555
        return(id);

556
    HASH_FIND(hh, mid->provider_table, &id, sizeof(id), element);
557 558 559
    if(element)
        return(id);

560
    element = calloc(1,sizeof(*element));
561 562
    if(!element)
        return(0);
563
    element->id = id;
564 565
    element->pool = pool;

566
    HASH_ADD(hh, mid->provider_table, id, sizeof(id), element);
567 568

    return(id);
569 570
}

571 572
hg_return_t margo_registered_name(margo_instance_id mid, const char *func_name,
    hg_id_t *id, hg_bool_t *flag)
573
{
574 575
    *id = gen_id(func_name, 0);
    return(HG_Registered(mid->hg_class, *id, flag));
576 577
}

578 579
hg_return_t margo_registered_name_provider(margo_instance_id mid, const char *func_name,
    uint16_t provider_id, hg_id_t *id, hg_bool_t *flag)
580 581
{
    hg_bool_t b;
582 583 584 585 586
    hg_return_t ret;

    *id = gen_id(func_name, provider_id);

    ret = HG_Registered(mid->hg_class, *id, &b);
587 588
    if(ret != HG_SUCCESS) 
        return ret;
589
    if((!b) || (!provider_id)) {
590 591 592 593
        *flag = b;
        return ret;
    }

594
    struct provider_element *element;
595

596
    HASH_FIND(hh, mid->provider_table, id, sizeof(*id), element);
597 598 599 600 601
    if(!element) {
        *flag = 0;
        return HG_SUCCESS;
    }

602
    assert(element->id == *id);
603 604 605 606 607

    *flag = 1;
    return HG_SUCCESS;
}

608 609 610 611 612 613 614
hg_return_t margo_register_data(
    margo_instance_id mid,
    hg_id_t id,
    void *data,
    void (*free_callback)(void *)) 
{
	struct margo_rpc_data* margo_data 
615
		= (struct margo_rpc_data*) HG_Registered_data(mid->hg_class, id);
616
	if(!margo_data) return HG_OTHER_ERROR;
617 618 619
    if(margo_data->user_data && margo_data->user_free_callback) {
        (margo_data->user_free_callback)(margo_data->user_data);
    }
620 621 622 623 624 625 626 627 628 629 630 631 632
	margo_data->user_data = data;
	margo_data->user_free_callback = free_callback;
	return HG_SUCCESS;
}

void* margo_registered_data(margo_instance_id mid, hg_id_t id)
{
	struct margo_rpc_data* data
		= (struct margo_rpc_data*) HG_Registered_data(margo_get_class(mid), id);
	if(!data) return NULL;
	else return data->user_data;
}

633 634 635 636
hg_return_t margo_registered_disable_response(
    margo_instance_id mid,
    hg_id_t id,
    int disable_flag)
637
{
638
    return(HG_Registered_disable_response(mid->hg_class, id, disable_flag));
639
}
640

641
struct lookup_cb_evt
642
{
643
    hg_return_t hret;
644 645 646 647 648 649
    hg_addr_t addr;
};

static hg_return_t margo_addr_lookup_cb(const struct hg_cb_info *info)
{
    struct lookup_cb_evt evt;
650
    evt.hret = info->ret;
651
    evt.addr = info->info.lookup.addr;
Matthieu Dorier's avatar
Matthieu Dorier committed
652
    ABT_eventual eventual = (ABT_eventual)(info->arg);
653 654

    /* propagate return code out through eventual */
Matthieu Dorier's avatar
Matthieu Dorier committed
655
    ABT_eventual_set(eventual, &evt, sizeof(evt));
656

657 658 659
    return(HG_SUCCESS);
}

660 661 662 663
hg_return_t margo_addr_lookup(
    margo_instance_id mid,
    const char   *name,
    hg_addr_t    *addr)
664
{
665
    hg_return_t hret;
666 667 668
    struct lookup_cb_evt *evt;
    ABT_eventual eventual;
    int ret;
669

670 671 672 673 674 675
    ret = ABT_eventual_create(sizeof(*evt), &eventual);
    if(ret != 0)
    {
        return(HG_NOMEM_ERROR);        
    }

676
    hret = HG_Addr_lookup(mid->hg_context, margo_addr_lookup_cb,
Matthieu Dorier's avatar
Matthieu Dorier committed
677
        (void*)eventual, name, HG_OP_ID_IGNORE);
678
    if(hret == HG_SUCCESS)
679 680 681
    {
        ABT_eventual_wait(eventual, (void**)&evt);
        *addr = evt->addr;
682
        hret = evt->hret;
683 684 685 686
    }

    ABT_eventual_free(&eventual);

687
    return(hret);
688 689 690 691 692
}

hg_return_t margo_addr_free(
    margo_instance_id mid,
    hg_addr_t addr)
693
{
694 695
    return(HG_Addr_free(mid->hg_class, addr));
}
696

697 698 699 700 701
hg_return_t margo_addr_self(
    margo_instance_id mid,
    hg_addr_t *addr)
{
    return(HG_Addr_self(mid->hg_class, addr));
702 703
}

704 705 706 707 708 709 710 711 712
hg_return_t margo_addr_dup(
    margo_instance_id mid,
    hg_addr_t addr,
    hg_addr_t *new_addr)
{
    return(HG_Addr_dup(mid->hg_class, addr, new_addr));
}

hg_return_t margo_addr_to_string(
713
    margo_instance_id mid,
714 715 716 717 718 719 720 721 722 723
    char *buf,
    hg_size_t *buf_size,
    hg_addr_t addr)
{
    return(HG_Addr_to_string(mid->hg_class, buf, buf_size, addr));
}

hg_return_t margo_create(margo_instance_id mid, hg_addr_t addr,
    hg_id_t id, hg_handle_t *handle)
{
724
    hg_return_t hret = HG_OTHER_ERROR;
Shane Snyder's avatar
Shane Snyder committed
725 726 727 728 729 730 731 732

    /* look for a handle to reuse */
    hret = margo_handle_cache_get(mid, addr, id, handle);
    if(hret != HG_SUCCESS)
    {
        /* else try creating a new handle */
        hret = HG_Create(mid->hg_context, addr, id, handle);
    }
733

Shane Snyder's avatar
Shane Snyder committed
734
    return hret;
735 736
}

737
hg_return_t margo_destroy(hg_handle_t handle)
738
{
739
    margo_instance_id mid;
740
    hg_return_t hret = HG_OTHER_ERROR;
Shane Snyder's avatar
Shane Snyder committed
741

742 743 744
    /* use the handle to get the associated mid */
    mid = margo_hg_handle_get_instance(handle);

Shane Snyder's avatar
Shane Snyder committed
745 746 747 748 749 750 751
    /* recycle this handle if it came from the handle cache */
    hret = margo_handle_cache_put(mid, handle);
    if(hret != HG_SUCCESS)
    {
        /* else destroy the handle manually */
        hret = HG_Destroy(handle);
    }
752

Shane Snyder's avatar
Shane Snyder committed
753
    return hret;
754 755 756 757 758
}

static hg_return_t margo_cb(const struct hg_cb_info *info)
{
    hg_return_t hret = info->ret;
Matthieu Dorier's avatar
Matthieu Dorier committed
759
    ABT_eventual eventual = (ABT_eventual)(info->arg);
760 761

    /* propagate return code out through eventual */
Matthieu Dorier's avatar
Matthieu Dorier committed
762
    ABT_eventual_set(eventual, &hret, sizeof(hret));
763 764 765 766
    
    return(HG_SUCCESS);
}

767
hg_return_t margo_provider_forward(
768
    uint16_t provider_id,
769 770
    hg_handle_t handle,
    void *in_struct)
Matthieu Dorier's avatar
Matthieu Dorier committed
771 772 773
{
	hg_return_t hret;
	margo_request req;
774
	hret = margo_provider_iforward(provider_id, handle, in_struct, &req);
Matthieu Dorier's avatar
Matthieu Dorier committed
775 776 777 778 779
	if(hret != HG_SUCCESS) 
		return hret;
	return margo_wait(req);
}

780
hg_return_t margo_provider_iforward(
781
    uint16_t provider_id,
Matthieu Dorier's avatar
Matthieu Dorier committed
782 783 784
    hg_handle_t handle,
    void *in_struct,
    margo_request* req)
785 786
{
    hg_return_t hret = HG_TIMEOUT;
787
    ABT_eventual eventual;
788
    int ret;
789 790 791 792 793 794 795 796
    const struct hg_info* hgi; 
    hg_id_t id;
    hg_proc_cb_t in_cb, out_cb;
    hg_bool_t flag;

    assert(provider_id <= MARGO_MAX_PROVIDER_ID);

    hgi = HG_Get_info(handle);
Philip Carns's avatar
Philip Carns committed
797
    id = mux_id(hgi->id, provider_id);
798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815

    /* TODO: if we reset the handle here, is there any reason to do so in
     * the handle cache?
     */
    ret = HG_Reset(handle, hgi->addr, id);
    if(ret == HG_NO_MATCH)
    {
        /* if Mercury does not recognize this ID (with provider id included)
         * then register it now
         */
        /* find encoders for base ID */
        ret = HG_Registered_proc_cb(hgi->hg_class, hgi->id, &flag, &in_cb, &out_cb);
        if(ret != HG_SUCCESS)
            return(ret);
        if(!flag)
            return(HG_NO_MATCH);

        /* register new ID that includes provider id */
816 817 818 819
        ret = margo_register_internal(margo_hg_info_get_instance(hgi), 
            id, in_cb, out_cb, NULL);
        if(ret == 0)
            return(HG_OTHER_ERROR);
Philip Carns's avatar
Philip Carns committed
820 821 822 823 824

        /* should be able to reset now */
        ret = HG_Reset(handle, hgi->addr, id);
        if(ret != HG_SUCCESS)
            return(ret);
825
    }
826 827 828 829 830 831 832

    ret = ABT_eventual_create(sizeof(hret), &eventual);
    if(ret != 0)
    {
        return(HG_NOMEM_ERROR);        
    }

Matthieu Dorier's avatar
Matthieu Dorier committed
833
    *req = eventual;
834

Matthieu Dorier's avatar
Matthieu Dorier committed
835
    return HG_Forward(handle, margo_cb, (void*)eventual, in_struct);
Matthieu Dorier's avatar
Matthieu Dorier committed
836
}
837

Matthieu Dorier's avatar
Matthieu Dorier committed
838 839 840 841
hg_return_t margo_wait(margo_request req)
{
	hg_return_t* waited_hret;
	hg_return_t  hret;
842

Matthieu Dorier's avatar
Matthieu Dorier committed
843 844 845 846
    ABT_eventual_wait(req, (void**)&waited_hret);
	hret = *waited_hret;
    ABT_eventual_free(&req);
	
847
    return(hret);
848 849
}

Matthieu Dorier's avatar
Matthieu Dorier committed
850 851 852 853 854
int margo_test(margo_request req, int* flag)
{
    return ABT_eventual_test(req, NULL, flag);
}

855 856 857 858
typedef struct
{
    hg_handle_t handle;
} margo_forward_timeout_cb_dat;
859

860 861 862 863 864 865 866 867 868 869 870
static void margo_forward_timeout_cb(void *arg)
{
    margo_forward_timeout_cb_dat *timeout_cb_dat =
        (margo_forward_timeout_cb_dat *)arg;

    /* cancel the Mercury op if the forward timed out */
    HG_Cancel(timeout_cb_dat->handle);
    return;
}

hg_return_t margo_forward_timed(
871
    hg_handle_t handle,
872 873
    void *in_struct,
    double timeout_ms)
874 875
{
    int ret;
876
    hg_return_t hret;
877
    margo_instance_id mid;
878
    ABT_eventual eventual;
879
    hg_return_t* waited_hret;
880 881
    margo_timer_t forward_timer;
    margo_forward_timeout_cb_dat timeout_cb_dat;
882 883 884 885 886 887 888

    ret = ABT_eventual_create(sizeof(hret), &eventual);
    if(ret != 0)
    {
        return(HG_NOMEM_ERROR);        
    }

889 890 891
    /* use the handle to get the associated mid */
    mid = margo_hg_handle_get_instance(handle);

892 893 894 895 896
    /* set a timer object to expire when this forward times out */
    timeout_cb_dat.handle = handle;
    margo_timer_init(mid, &forward_timer, margo_forward_timeout_cb,
        &timeout_cb_dat, timeout_ms);

Matthieu Dorier's avatar
Matthieu Dorier committed
897
    hret = HG_Forward(handle, margo_cb, (void*)eventual, in_struct);
898
    if(hret == HG_SUCCESS)
Jonathan Jenkins's avatar
Jonathan Jenkins committed
899 900 901 902 903
    {
        ABT_eventual_wait(eventual, (void**)&waited_hret);
        hret = *waited_hret;
    }

904 905 906 907 908 909 910 911
    /* convert HG_CANCELED to HG_TIMEOUT to indicate op timed out */
    if(hret == HG_CANCELED)
        hret = HG_TIMEOUT;

    /* remove timer if it is still in place (i.e., not timed out) */
    if(hret != HG_TIMEOUT)
        margo_timer_destroy(mid, &forward_timer);

Jonathan Jenkins's avatar
Jonathan Jenkins committed
912 913 914 915 916 917 918 919
    ABT_eventual_free(&eventual);

    return(hret);
}

hg_return_t margo_respond(
    hg_handle_t handle,
    void *out_struct)
Matthieu Dorier's avatar
Matthieu Dorier committed
920 921 922 923 924 925 926 927 928 929 930 931 932
{
    hg_return_t hret;
    margo_request req;
    hret = margo_irespond(handle,out_struct,&req);
    if(hret != HG_SUCCESS)
        return hret;
    return margo_wait(req);
}

hg_return_t margo_irespond(
    hg_handle_t handle,
    void *out_struct,
    margo_request* req)
Jonathan Jenkins's avatar
Jonathan Jenkins committed
933 934 935 936
{
    ABT_eventual eventual;
    int ret;

Matthieu Dorier's avatar
Matthieu Dorier committed
937
    ret = ABT_eventual_create(sizeof(hg_return_t), &eventual);
Jonathan Jenkins's avatar
Jonathan Jenkins committed
938 939 940 941 942
    if(ret != 0)
    {
        return(HG_NOMEM_ERROR);
    }

Matthieu Dorier's avatar
Matthieu Dorier committed
943
    *req = eventual;
944

Matthieu Dorier's avatar
Matthieu Dorier committed
945
    return HG_Respond(handle, margo_cb, (void*)eventual, out_struct);
946 947
}

948 949 950 951 952 953 954
hg_return_t margo_bulk_create(
    margo_instance_id mid,
    hg_uint32_t count,
    void **buf_ptrs,
    const hg_size_t *buf_sizes,
    hg_uint8_t flags,
    hg_bulk_t *handle)
955
{
956 957 958
    return(HG_Bulk_create(mid->hg_class, count,
        buf_ptrs, buf_sizes, flags, handle));
}
959

960 961 962 963
hg_return_t margo_bulk_free(
    hg_bulk_t handle)
{
    return(HG_Bulk_free(handle));
964 965
}

966 967 968 969 970 971 972 973
hg_return_t margo_bulk_deserialize(
    margo_instance_id mid,
    hg_bulk_t *handle,
    const void *buf,
    hg_size_t buf_size)
{
    return(HG_Bulk_deserialize(mid->hg_class, handle, buf, buf_size));
}
974

975
hg_return_t margo_bulk_transfer(
976
    margo_instance_id mid,
977
    hg_bulk_op_t op,
978
    hg_addr_t origin_addr,
979 980 981 982
    hg_bulk_t origin_handle,
    size_t origin_offset,
    hg_bulk_t local_handle,
    size_t local_offset,
983
    size_t size)
Matthieu Dorier's avatar
Matthieu Dorier committed
984 985 986 987
{  
    margo_request req;
    hg_return_t hret = margo_bulk_itransfer(mid,op,origin_addr,
                          origin_handle, origin_offset, local_handle,
Matthieu Dorier's avatar
Matthieu Dorier committed
988
                          local_offset, size, &req);
Matthieu Dorier's avatar
Matthieu Dorier committed
989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003
    if(hret != HG_SUCCESS)
        return hret;
    return margo_wait(req);
}

hg_return_t margo_bulk_itransfer(
    margo_instance_id mid,
    hg_bulk_op_t op,
    hg_addr_t origin_addr,
    hg_bulk_t origin_handle,
    size_t origin_offset,
    hg_bulk_t local_handle,
    size_t local_offset,
    size_t size,
    margo_request* req)
1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014
{
    hg_return_t hret = HG_TIMEOUT;
    ABT_eventual eventual;
    int ret;

    ret = ABT_eventual_create(sizeof(hret), &eventual);
    if(ret != 0)
    {
        return(HG_NOMEM_ERROR);        
    }

Matthieu Dorier's avatar
Matthieu Dorier committed
1015
    *req = eventual;
1016

Matthieu Dorier's avatar
Matthieu Dorier committed
1017 1018
    hret = HG_Bulk_transfer(mid->hg_context, margo_cb,
        (void*)eventual, op, origin_addr, origin_handle, origin_offset, local_handle,
1019
        local_offset, size, HG_OP_ID_IGNORE);
1020 1021 1022 1023

    return(hret);
}

1024 1025 1026 1027
typedef struct
{
    ABT_mutex mutex;
    ABT_cond cond;
Shane Snyder's avatar
Shane Snyder committed
1028
    char is_asleep;
1029 1030 1031 1032 1033 1034 1035 1036 1037
} margo_thread_sleep_cb_dat;

static void margo_thread_sleep_cb(void *arg)
{
    margo_thread_sleep_cb_dat *sleep_cb_dat =
        (margo_thread_sleep_cb_dat *)arg;

    /* wake up the sleeping thread */
    ABT_mutex_lock(sleep_cb_dat->mutex);
1038
    sleep_cb_dat->is_asleep = 0;
1039 1040 1041 1042 1043 1044 1045
    ABT_cond_signal(sleep_cb_dat->cond);
    ABT_mutex_unlock(sleep_cb_dat->mutex);

    return;
}

void margo_thread_sleep(
1046
    margo_instance_id mid,
1047 1048 1049 1050 1051 1052 1053 1054
    double timeout_ms)
{
    margo_timer_t sleep_timer;
    margo_thread_sleep_cb_dat sleep_cb_dat;

    /* set data needed for sleep callback */
    ABT_mutex_create(&(sleep_cb_dat.mutex));
    ABT_cond_create(&(sleep_cb_dat.cond));
1055
    sleep_cb_dat.is_asleep = 1;
1056 1057

    /* initialize the sleep timer */