margo.c 40.9 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

#include <margo-config.h>
15
#include <time.h>
Philip Carns's avatar
Philip Carns committed
16
#include <math.h>
17 18

#include "margo.h"
19
#include "margo-timer.h"
Philip Carns's avatar
Philip Carns committed
20
#include "utlist.h"
21
#include "uthash.h"
22

23
#define DEFAULT_MERCURY_PROGRESS_TIMEOUT_UB 100 /* 100 milliseconds */
Shane Snyder's avatar
Shane Snyder committed
24
#define DEFAULT_MERCURY_HANDLE_CACHE_SIZE 32
25

Philip Carns's avatar
Philip Carns committed
26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41
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
42 43 44 45 46 47 48
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 */
};

49 50 51 52 53 54 55
struct margo_finalize_cb
{
    void(*callback)(void*);
    void* uargs;
    struct margo_finalize_cb* next;
};

56 57
struct margo_timer_list; /* defined in margo-timer.c */

58 59
struct margo_instance
{
Shane Snyder's avatar
Shane Snyder committed
60
    /* mercury/argobots state */
61 62
    hg_context_t *hg_context;
    hg_class_t *hg_class;
63 64 65
    ABT_pool handler_pool;
    ABT_pool progress_pool;

66
    /* internal to margo for this particular instance */
67
    int margo_init;
68
    int abt_init;
69 70
    ABT_thread hg_progress_tid;
    int hg_progress_shutdown_flag;
Jonathan Jenkins's avatar
Jonathan Jenkins committed
71
    ABT_xstream progress_xstream;
Jonathan Jenkins's avatar
Jonathan Jenkins committed
72 73 74
    int owns_progress_pool;
    ABT_xstream *rpc_xstreams;
    int num_handler_pool_threads;
75
    unsigned int hg_progress_timeout_ub;
76 77 78

    /* control logic for callers waiting on margo to be finalized */
    int finalize_flag;
Jonathan Jenkins's avatar
Jonathan Jenkins committed
79
    int refcount;
80 81
    ABT_mutex finalize_mutex;
    ABT_cond finalize_cond;
82
    struct margo_finalize_cb* finalize_cb;
83

84 85 86 87 88 89
    /* control logic to prevent margo_finalize from destroying
       the instance when some operations are pending */
    unsigned pending_operations;
    ABT_mutex pending_operations_mtx;
    int finalize_requested;

Matthieu Dorier's avatar
Matthieu Dorier committed
90 91 92 93
    /* control logic for shutting down */
    hg_id_t shutdown_rpc_id;
    int enable_remote_shutdown;

94 95
    /* timer data */
    struct margo_timer_list* timer_list;
Shane Snyder's avatar
Shane Snyder committed
96 97 98
    /* 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;
99
    ABT_mutex handle_cache_mtx; /* mutex protecting access to above caches */
Shane Snyder's avatar
Shane Snyder committed
100

Philip Carns's avatar
Philip Carns committed
101 102 103 104 105 106 107 108 109 110 111
    /* 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;
112 113
};

114 115 116
struct margo_rpc_data
{
	margo_instance_id mid;
117
    ABT_pool pool;
118 119 120
	void* user_data;
	void (*user_free_callback)(void *);
};
121

Matthieu Dorier's avatar
Matthieu Dorier committed
122 123
MERCURY_GEN_PROC(margo_shutdown_out_t, ((int32_t)(ret)))

124
static void hg_progress_fn(void* foo);
125
static void margo_rpc_data_free(void* ptr);
Matthieu Dorier's avatar
Matthieu Dorier committed
126 127
static void remote_shutdown_ult(hg_handle_t handle);
DECLARE_MARGO_RPC_HANDLER(remote_shutdown_ult);
128

129 130 131 132 133 134 135 136 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
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
165 166 167 168 169 170
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);
171
static hg_id_t margo_register_internal(margo_instance_id mid, hg_id_t id,
172
    hg_proc_cb_t in_proc_cb, hg_proc_cb_t out_proc_cb, hg_rpc_cb_t rpc_cb, ABT_pool pool);
Shane Snyder's avatar
Shane Snyder committed
173

174
margo_instance_id margo_init(const char *addr_str, int mode,
175
    int use_progress_thread, int rpc_thread_count)
176
{
Jonathan Jenkins's avatar
Jonathan Jenkins committed
177 178
    ABT_xstream progress_xstream = ABT_XSTREAM_NULL;
    ABT_pool progress_pool = ABT_POOL_NULL;
179 180 181
    ABT_sched progress_sched;
    ABT_sched self_sched;
    ABT_xstream self_xstream;
Jonathan Jenkins's avatar
Jonathan Jenkins committed
182
    ABT_xstream *rpc_xstreams = NULL;
183
    ABT_sched *rpc_scheds = NULL;
Jonathan Jenkins's avatar
Jonathan Jenkins committed
184 185
    ABT_xstream rpc_xstream = ABT_XSTREAM_NULL;
    ABT_pool rpc_pool = ABT_POOL_NULL;
186 187
    hg_class_t *hg_class = NULL;
    hg_context_t *hg_context = NULL;
188
    int listen_flag = (mode == MARGO_CLIENT_MODE) ? HG_FALSE : HG_TRUE;
189
    int abt_init = 0;
Jonathan Jenkins's avatar
Jonathan Jenkins committed
190
    int i;
191 192 193
    int ret;
    struct margo_instance *mid = MARGO_INSTANCE_NULL;

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

Philip Carns's avatar
Philip Carns committed
196 197 198 199 200 201 202 203 204 205 206 207
    /* 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");

208 209 210 211 212 213
    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;
    }
214

215 216 217 218 219 220 221 222
    /* set caller (self) ES to sleep when idle by using sched_wait */
    ret = ABT_sched_create_basic(ABT_SCHED_BASIC_WAIT, 0, NULL, 
        ABT_SCHED_CONFIG_NULL, &self_sched);
    if(ret != ABT_SUCCESS) goto err;
    ret = ABT_xstream_self(&self_xstream);
    if(ret != ABT_SUCCESS) goto err;
    ret = ABT_xstream_set_main_sched(self_xstream, self_sched);
    if(ret != ABT_SUCCESS) goto err;
Jonathan Jenkins's avatar
Jonathan Jenkins committed
223 224 225

    if (use_progress_thread)
    {
226 227 228 229 230 231
        /* create an xstream to run progress engine */
        ret = ABT_sched_create_basic(ABT_SCHED_BASIC_WAIT, 0, NULL,
           ABT_SCHED_CONFIG_NULL, &progress_sched);
        if (ret != ABT_SUCCESS) goto err;
        ret = ABT_xstream_create(progress_sched, &progress_xstream);
        if (ret != ABT_SUCCESS) goto err;
Jonathan Jenkins's avatar
Jonathan Jenkins committed
232 233 234 235 236 237
    }
    else
    {
        ret = ABT_xstream_self(&progress_xstream);
        if (ret != ABT_SUCCESS) goto err;
    }
238 239
    ret = ABT_xstream_get_main_pools(progress_xstream, 1, &progress_pool);
    if (ret != ABT_SUCCESS) goto err;
Jonathan Jenkins's avatar
Jonathan Jenkins committed
240

241
    if (rpc_thread_count > 0)
Jonathan Jenkins's avatar
Jonathan Jenkins committed
242
    {
243
        /* create a collection of xstreams to run RPCs */
244 245
        rpc_xstreams = calloc(rpc_thread_count, sizeof(*rpc_xstreams));
        if (rpc_xstreams == NULL) goto err;
246 247 248
        rpc_scheds = calloc(rpc_thread_count, sizeof(*rpc_scheds));
        if (rpc_scheds == NULL) goto err;
        ret = ABT_pool_create_basic(ABT_POOL_FIFO_WAIT, ABT_POOL_ACCESS_MPMC, ABT_TRUE, &rpc_pool);
249
        if (ret != ABT_SUCCESS) goto err;
250 251
        for(i=0; i<rpc_thread_count; i++) 
        {
Philip Carns's avatar
Philip Carns committed
252
            ret = ABT_sched_create_basic(ABT_SCHED_BASIC_WAIT, 1, &rpc_pool,
253 254 255
               ABT_SCHED_CONFIG_NULL, &rpc_scheds[i]);
            if (ret != ABT_SUCCESS) goto err;
            ret = ABT_xstream_create(rpc_scheds[i], rpc_xstreams+i);
256 257
            if (ret != ABT_SUCCESS) goto err;
        }
258 259 260 261 262 263 264 265 266 267 268
    }
    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
269 270
    }

271 272 273 274 275 276
    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
277 278 279
    mid = margo_init_pool(progress_pool, rpc_pool, hg_context);
    if (mid == MARGO_INSTANCE_NULL) goto err;

280
    mid->margo_init = 1;
281
    mid->abt_init = abt_init;
Jonathan Jenkins's avatar
Jonathan Jenkins committed
282 283 284
    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
285
    mid->rpc_xstreams = rpc_xstreams;
286

Jonathan Jenkins's avatar
Jonathan Jenkins committed
287 288 289
    return mid;

err:
290 291
    if(mid)
    {
292
        margo_timer_list_free(mid->timer_list);
293 294 295 296
        ABT_mutex_free(&mid->finalize_mutex);
        ABT_cond_free(&mid->finalize_cond);
        free(mid);
    }
Jonathan Jenkins's avatar
Jonathan Jenkins committed
297 298 299 300 301 302 303 304 305 306 307 308 309
    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);
310
        free(rpc_scheds);
Jonathan Jenkins's avatar
Jonathan Jenkins committed
311
    }
312 313 314 315
    if(hg_context)
        HG_Context_destroy(hg_context);
    if(hg_class)
        HG_Finalize(hg_class);
316 317
    if(abt_init)
        ABT_finalize();
Jonathan Jenkins's avatar
Jonathan Jenkins committed
318 319 320 321
    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
322
    hg_context_t *hg_context)
323 324
{
    int ret;
Shane Snyder's avatar
Shane Snyder committed
325
    hg_return_t hret;
326 327
    struct margo_instance *mid;

Matthieu Dorier's avatar
Matthieu Dorier committed
328
    mid = calloc(1,sizeof(*mid));
329
    if(!mid) goto err;
330
    memset(mid, 0, sizeof(*mid));
331

332 333 334
    ABT_mutex_create(&mid->finalize_mutex);
    ABT_cond_create(&mid->finalize_cond);

335 336
    mid->progress_pool = progress_pool;
    mid->handler_pool = handler_pool;
Jonathan Jenkins's avatar
Jonathan Jenkins committed
337
    mid->hg_class = HG_Context_get_class(hg_context);
338
    mid->hg_context = hg_context;
339
    mid->hg_progress_timeout_ub = DEFAULT_MERCURY_PROGRESS_TIMEOUT_UB;
340

Jonathan Jenkins's avatar
Jonathan Jenkins committed
341
    mid->refcount = 1;
342
    mid->finalize_cb = NULL;
Matthieu Dorier's avatar
Matthieu Dorier committed
343
    mid->enable_remote_shutdown = 0;
344

345 346 347 348
    mid->pending_operations = 0;
    ABT_mutex_create(&mid->pending_operations_mtx);
    mid->finalize_requested = 0;

349 350
    mid->timer_list = margo_timer_list_create();
    if(mid->timer_list == NULL) goto err;
351

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

356
    ret = ABT_thread_create(mid->progress_pool, hg_progress_fn, mid, 
357
        ABT_THREAD_ATTR_NULL, &mid->hg_progress_tid);
358 359
    if(ret != 0) goto err;

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

363 364
    return mid;

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

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

382 383 384 385 386 387 388 389 390
    /* 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);
    }

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

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

    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
413 414
    margo_handle_cache_destroy(mid);

415 416 417 418 419 420
    if (mid->margo_init)
    {
        if (mid->hg_context)
            HG_Context_destroy(mid->hg_context);
        if (mid->hg_class)
            HG_Finalize(mid->hg_class);
421 422
        if (mid->abt_init)
            ABT_finalize();
423 424
    }

Jonathan Jenkins's avatar
Jonathan Jenkins committed
425 426 427
    free(mid);
}

428
void margo_finalize(margo_instance_id mid)
429
{
Jonathan Jenkins's avatar
Jonathan Jenkins committed
430
    int do_cleanup;
431

432 433 434 435 436 437 438 439 440 441
    /* check if there are pending operations */
    int pending;
    ABT_mutex_lock(mid->pending_operations_mtx);
    pending = mid->pending_operations;
    ABT_mutex_unlock(mid->pending_operations_mtx);
    if(pending) {
        mid->finalize_requested = 1;
        return;
    }

442
    /* tell progress thread to wrap things up */
443
    mid->hg_progress_shutdown_flag = 1;
444 445

    /* wait for it to shutdown cleanly */
446 447
    ABT_thread_join(mid->hg_progress_tid);
    ABT_thread_free(&mid->hg_progress_tid);
448

449 450 451 452
    ABT_mutex_lock(mid->finalize_mutex);
    mid->finalize_flag = 1;
    ABT_cond_broadcast(mid->finalize_cond);

Jonathan Jenkins's avatar
Jonathan Jenkins committed
453 454
    mid->refcount--;
    do_cleanup = mid->refcount == 0;
455

Jonathan Jenkins's avatar
Jonathan Jenkins committed
456 457 458 459 460 461 462
    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);
463 464 465 466 467 468

    return;
}

void margo_wait_for_finalize(margo_instance_id mid)
{
Jonathan Jenkins's avatar
Jonathan Jenkins committed
469
    int do_cleanup;
470 471 472

    ABT_mutex_lock(mid->finalize_mutex);

Jonathan Jenkins's avatar
Jonathan Jenkins committed
473
        mid->refcount++;
474 475 476 477
            
        while(!mid->finalize_flag)
            ABT_cond_wait(mid->finalize_cond, mid->finalize_mutex);

Jonathan Jenkins's avatar
Jonathan Jenkins committed
478 479 480
        mid->refcount--;
        do_cleanup = mid->refcount == 0;

481
    ABT_mutex_unlock(mid->finalize_mutex);
Jonathan Jenkins's avatar
Jonathan Jenkins committed
482 483 484 485

    if (do_cleanup)
        margo_cleanup(mid);

486 487 488
    return;
}

489 490 491 492 493 494 495
hg_bool_t margo_is_listening(
            margo_instance_id mid)
{
    if(!mid) return HG_FALSE;
    return HG_Class_is_listening(mid->hg_class);
}

496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512
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
513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528
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;

529
    hret = margo_forward(handle, NULL);
Matthieu Dorier's avatar
Matthieu Dorier committed
530 531 532 533 534 535 536 537 538 539 540 541 542 543
    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;
    }

544
    margo_free_output(handle, &out);
Matthieu Dorier's avatar
Matthieu Dorier committed
545 546 547 548 549
    margo_destroy(handle);

    return out.ret;
}

550
hg_id_t margo_provider_register_name(margo_instance_id mid, const char *func_name,
551
    hg_proc_cb_t in_proc_cb, hg_proc_cb_t out_proc_cb, hg_rpc_cb_t rpc_cb,
552
    uint16_t provider_id, ABT_pool pool)
553
{
554
    hg_id_t id;
555 556 557 558 559
    int ret;

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

561
    ret = margo_register_internal(mid, id, in_proc_cb, out_proc_cb, rpc_cb, pool);
562
    if(ret == 0)
563 564 565
        return(0);

    return(id);
566 567
}

Matthieu Dorier's avatar
Matthieu Dorier committed
568 569 570 571 572 573 574
hg_return_t margo_deregister(
        margo_instance_id mid,
        hg_id_t rpc_id)
{
    return HG_Deregister(mid->hg_class, rpc_id);
}

575 576
hg_return_t margo_registered_name(margo_instance_id mid, const char *func_name,
    hg_id_t *id, hg_bool_t *flag)
577
{
578 579
    *id = gen_id(func_name, 0);
    return(HG_Registered(mid->hg_class, *id, flag));
580 581
}

Matthieu Dorier's avatar
Matthieu Dorier committed
582
hg_return_t margo_provider_registered_name(margo_instance_id mid, const char *func_name,
583
    uint16_t provider_id, hg_id_t *id, hg_bool_t *flag)
584
{
585 586
    *id = gen_id(func_name, provider_id);

587
    return HG_Registered(mid->hg_class, *id, flag);
588 589
}

590 591 592 593 594 595 596
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 
597
		= (struct margo_rpc_data*) HG_Registered_data(mid->hg_class, id);
598
	if(!margo_data) return HG_OTHER_ERROR;
599 600 601
    if(margo_data->user_data && margo_data->user_free_callback) {
        (margo_data->user_free_callback)(margo_data->user_data);
    }
602 603 604 605 606 607 608 609 610 611 612 613 614
	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;
}

615 616 617 618
hg_return_t margo_registered_disable_response(
    margo_instance_id mid,
    hg_id_t id,
    int disable_flag)
619
{
620
    return(HG_Registered_disable_response(mid->hg_class, id, disable_flag));
621
}
622

623 624 625 626 627 628 629 630 631 632 633 634
hg_return_t margo_registered_disabled_response(
    margo_instance_id mid,
    hg_id_t id,
    int* disabled_flag)
{
    hg_bool_t b;
    hg_return_t ret = HG_Registered_disabled_response(mid->hg_class, id, &b);
    if(ret != HG_SUCCESS) return ret;
    *disabled_flag = b;
    return HG_SUCCESS;
}

635
struct lookup_cb_evt
636
{
637
    hg_return_t hret;
638 639 640 641 642 643
    hg_addr_t addr;
};

static hg_return_t margo_addr_lookup_cb(const struct hg_cb_info *info)
{
    struct lookup_cb_evt evt;
644
    evt.hret = info->ret;
645
    evt.addr = info->info.lookup.addr;
Matthieu Dorier's avatar
Matthieu Dorier committed
646
    ABT_eventual eventual = (ABT_eventual)(info->arg);
647 648

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

651 652 653
    return(HG_SUCCESS);
}

654 655 656 657
hg_return_t margo_addr_lookup(
    margo_instance_id mid,
    const char   *name,
    hg_addr_t    *addr)
658
{
659
    hg_return_t hret;
660 661 662
    struct lookup_cb_evt *evt;
    ABT_eventual eventual;
    int ret;
663

664 665 666 667 668 669
    ret = ABT_eventual_create(sizeof(*evt), &eventual);
    if(ret != 0)
    {
        return(HG_NOMEM_ERROR);        
    }

670
    hret = HG_Addr_lookup(mid->hg_context, margo_addr_lookup_cb,
Matthieu Dorier's avatar
Matthieu Dorier committed
671
        (void*)eventual, name, HG_OP_ID_IGNORE);
672
    if(hret == HG_SUCCESS)
673 674 675
    {
        ABT_eventual_wait(eventual, (void**)&evt);
        *addr = evt->addr;
676
        hret = evt->hret;
677 678 679 680
    }

    ABT_eventual_free(&eventual);

681
    return(hret);
682 683 684 685 686
}

hg_return_t margo_addr_free(
    margo_instance_id mid,
    hg_addr_t addr)
687
{
688 689
    return(HG_Addr_free(mid->hg_class, addr));
}
690

691 692 693 694 695
hg_return_t margo_addr_self(
    margo_instance_id mid,
    hg_addr_t *addr)
{
    return(HG_Addr_self(mid->hg_class, addr));
696 697
}

698 699 700 701 702 703 704 705 706
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(
707
    margo_instance_id mid,
708 709 710 711 712 713 714 715 716 717
    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)
{
718
    hg_return_t hret = HG_OTHER_ERROR;
Shane Snyder's avatar
Shane Snyder committed
719 720 721 722 723 724 725 726

    /* 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);
    }
727

Shane Snyder's avatar
Shane Snyder committed
728
    return hret;
729 730
}

731
hg_return_t margo_destroy(hg_handle_t handle)
732
{
733 734 735
    if(handle == HG_HANDLE_NULL)
        return HG_SUCCESS;

736
    margo_instance_id mid;
737
    hg_return_t hret = HG_OTHER_ERROR;
Shane Snyder's avatar
Shane Snyder committed
738

739 740 741
    /* use the handle to get the associated mid */
    mid = margo_hg_handle_get_instance(handle);

Shane Snyder's avatar
Shane Snyder committed
742 743 744 745 746 747 748
    /* 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);
    }
749

Shane Snyder's avatar
Shane Snyder committed
750
    return hret;
751 752 753 754 755
}

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
756
    ABT_eventual eventual = (ABT_eventual)(info->arg);
757 758

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

764
hg_return_t margo_provider_forward(
765
    uint16_t provider_id,
766 767
    hg_handle_t handle,
    void *in_struct)
768 769 770
{
	hg_return_t hret;
	margo_request req;
771
	hret = margo_provider_iforward(provider_id, handle, in_struct, &req);
772 773 774 775 776
	if(hret != HG_SUCCESS) 
		return hret;
	return margo_wait(req);
}

777
hg_return_t margo_provider_iforward(
778
    uint16_t provider_id,
779 780 781
    hg_handle_t handle,
    void *in_struct,
    margo_request* req)
782 783
{
    hg_return_t hret = HG_TIMEOUT;
784
    ABT_eventual eventual;
785
    int ret;
786 787 788 789 790 791 792 793
    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
794
    id = mux_id(hgi->id, provider_id);
795

796 797 798 799 800
    hg_bool_t is_registered;
    ret = HG_Registered(hgi->hg_class, id, &is_registered);
    if(ret != HG_SUCCESS)
        return(ret);
    if(!is_registered)
801 802 803 804 805 806 807 808 809 810 811 812
    {
        /* 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 */
813
        ret = margo_register_internal(margo_hg_info_get_instance(hgi), 
814
            id, in_cb, out_cb, NULL, ABT_POOL_NULL);
815 816
        if(ret == 0)
            return(HG_OTHER_ERROR);
817
    }
818 819 820
    ret = HG_Reset(handle, hgi->addr, id);
    if(ret != HG_SUCCESS)
        return(ret);
821 822 823 824 825 826 827

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

828
    *req = eventual;
829

Matthieu Dorier's avatar
Matthieu Dorier committed
830
    return HG_Forward(handle, margo_cb, (void*)eventual, in_struct);
831
}
832

833 834 835 836
hg_return_t margo_wait(margo_request req)
{
	hg_return_t* waited_hret;
	hg_return_t  hret;
837

838 839 840 841
    ABT_eventual_wait(req, (void**)&waited_hret);
	hret = *waited_hret;
    ABT_eventual_free(&req);
	
842
    return(hret);
843 844
}

Matthieu Dorier's avatar
Matthieu Dorier committed
845 846 847 848 849
int margo_test(margo_request req, int* flag)
{
    return ABT_eventual_test(req, NULL, flag);
}

850 851 852 853
typedef struct
{
    hg_handle_t handle;
} margo_forward_timeout_cb_dat;
854

855 856 857 858 859 860 861 862 863 864 865
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(
866
    hg_handle_t handle,
867 868
    void *in_struct,
    double timeout_ms)
869 870
{
    int ret;
871
    hg_return_t hret;
872
    margo_instance_id mid;
873
    ABT_eventual eventual;
874
    hg_return_t* waited_hret;
875 876
    margo_timer_t forward_timer;
    margo_forward_timeout_cb_dat timeout_cb_dat;
877 878 879 880 881 882 883

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

884 885 886
    /* use the handle to get the associated mid */
    mid = margo_hg_handle_get_instance(handle);

887 888 889 890 891
    /* 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
892
    hret = HG_Forward(handle, margo_cb, (void*)eventual, in_struct);
893
    if(hret == HG_SUCCESS)
Jonathan Jenkins's avatar
Jonathan Jenkins committed
894 895 896 897 898
    {
        ABT_eventual_wait(eventual, (void**)&waited_hret);
        hret = *waited_hret;
    }

899 900 901 902 903 904 905 906
    /* 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
907 908 909 910 911 912 913 914
    ABT_eventual_free(&eventual);

    return(hret);
}

hg_return_t margo_respond(
    hg_handle_t handle,
    void *out_struct)
915 916 917 918 919 920 921 922 923 924 925 926 927
{
    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
928 929 930 931
{
    ABT_eventual eventual;
    int ret;

Matthieu Dorier's avatar
Matthieu Dorier committed
932
    ret = ABT_eventual_create(sizeof(hg_return_t), &eventual);
Jonathan Jenkins's avatar
Jonathan Jenkins committed
933 934 935 936 937
    if(ret != 0)
    {
        return(HG_NOMEM_ERROR);
    }

938
    *req = eventual;
939

Matthieu Dorier's avatar
Matthieu Dorier committed
940
    return HG_Respond(handle, margo_cb, (void*)eventual, out_struct);
941 942
}

943 944 945 946 947 948 949
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)
950
{
951 952 953
    return(HG_Bulk_create(mid->hg_class, count,
        buf_ptrs, buf_sizes, flags, handle));
}
954

955 956 957 958
hg_return_t margo_bulk_free(
    hg_bulk_t handle)
{
    return(HG_Bulk_free(handle));
959 960
}

961 962 963 964 965 966 967 968
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));
}
969

970
hg_return_t margo_bulk_transfer(
971
    margo_instance_id mid,
972
    hg_bulk_op_t op,
973
    hg_addr_t origin_addr,
974 975 976 977
    hg_bulk_t origin_handle,
    size_t origin_offset,
    hg_bulk_t local_handle,
    size_t local_offset,
978
    size_t size)
979 980 981 982
{  
    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
983
                          local_offset, size, &req);
984 985 986 987 988 989 990 991 992 993 994 995 996 997 998
    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)
999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009
{
    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);        
    }

1010
    *req = eventual;
1011

Matthieu Dorier's avatar
Matthieu Dorier committed
1012 1013
    hret = HG_Bulk_transfer(mid->hg_context, margo_cb,
        (void*)eventual, op, origin_addr, origin_handle, origin_offset, local_handle,
1014
        local_offset, size, HG_OP_ID_IGNORE);
1015 1016 1017 1018

    return(hret);
}

1019 1020 1021 1022
typedef struct
{
    ABT_mutex mutex;
    ABT_cond cond;
Shane Snyder's avatar
Shane Snyder committed
1023
    char is_asleep;
1024 1025 1026 1027 1028 1029 1030 1031 1032
} 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);
1033
    sleep_cb_dat->is_asleep = 0;
1034 1035 1036 1037 1038 1039 1040
    ABT_cond_signal(sleep_cb_dat->cond);
    ABT_mutex_unlock(sleep_cb_dat->mutex);

    return;
}

void margo_thread_sleep(
1041
    margo_instance_id mid,
1042 1043 1044 1045 1046 1047 1048 1049
    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));
1050
    sleep_cb_dat.is_asleep = 1;
1051 1052

    /* initialize the sleep timer */
1053
    margo_timer_init(mid, &sleep_timer, margo_thread_sleep_cb,
1054 1055 1056 1057
        &sleep_cb_dat, timeout_ms);

    /* yield thread for specified timeout */
    ABT_mutex_lock(sleep_cb_dat.mutex);
1058 1059
    while(sleep_cb_dat.is_asleep)
        ABT_cond_wait(sleep_cb_dat.cond, sleep_cb_dat.mutex);
1060 1061
    ABT_mutex_unlock(sleep_cb_dat.mutex);

1062 1063 1064 1065
    /* clean up */
    ABT_mutex_free(&sleep_cb_dat.mutex);
    ABT_cond_free(&sleep_cb_dat.cond);

1066 1067 1068
    return;
}

1069
int margo_get_handler_pool(margo_instance_id mid, ABT_pool* pool)
1070
{
1071 1072 1073 1074 1075 1076
    if(mid) {
        *pool = mid->handler_pool;
        return 0;
    } else {
        return -1;
    }
1077
}
1078

1079 1080 1081 1082
hg_context_t* margo_get_context(margo_instance_id mid)
{
    return(mid->hg_context);
}
1083

1084 1085 1086
hg_class_t* margo_get_class(margo_instance_id mid)
{
    return(mid->hg_class);
1087
}
Philip Carns's avatar
Philip Carns committed
1088

1089
ABT_pool margo_hg_handle_get_handler_pool(hg_handle_t h)
1090
{
1091 1092 1093 1094 1095 1096
    struct margo_rpc_data* data;
    const struct hg_info* info;
    ABT_pool pool;
    
    info = HG_Get_info(h);
    if(!info) return ABT_POOL_NULL;
1097

1098 1099
    data = (struct margo_rpc_data*) HG_Registered_data(info->hg_class, info->id);
    if(!data) return ABT_POOL_NULL;
1100

1101
    pool = data->pool;
1102 1103 1104
    if(pool == ABT_POOL_NULL)
        margo_get_handler_pool(data->mid, &pool);

1105 1106
    return pool;
}
1107

1108 1109 1110 1111 1112 1113 1114 1115
margo_instance_id margo_hg_info_get_instance(const struct hg_info *info)
{
    struct margo_rpc_data* data = 
        (struct margo_rpc_data*) HG_Registered_data(info->hg_class, info->id);
    if(!data) return MARGO_INSTANCE_NULL;
    return data->mid;
}

1116 1117 1118 1119 1120 1121 1122
margo_instance_id margo_hg_handle_get_instance(hg_handle_t h)
{
    struct margo_rpc_data* data;
    const struct hg_info* info;
    
    info = HG_Get_info(h);
    if(!info) return MARGO_INSTANCE_NULL;
Philip Carns's avatar
Philip Carns committed
1123

1124 1125
    data = (struct margo_rpc_data*) HG_Registered_data(info->hg_class, info->id);
    if(!data) return MARGO_INSTANCE_NULL;
1126

1127
    return data->mid;
1128 1129
}

1130
static void margo_rpc_data_free(void* ptr)
Philip Carns's avatar
Philip Carns committed
1131
{
1132 1133 1134 1135 1136 1137
	struct margo_rpc_data* data = (struct margo_rpc_data*) ptr;
	if(data->user_data && data->user_free_callback) {
		data->user_free_callback(data->user_data);
	}
	free(ptr);
}
1138

1139 1140 1141 1142 1143 1144 1145
/* dedicated thread function to drive Mercury progress */
static void hg_progress_fn(void* foo)
{
    int ret;
    unsigned int actual_count;
    struct margo_instance *mid = (struct margo_instance *)foo;
    size_t size;
1146
    unsigned int hg_progress_timeout = mid->hg_progress_timeout_ub;
1147 1148
    double next_timer_exp;
    int trigger_happened;
1149 1150
    double tm1, tm2;
    int diag_enabled = 0;
1151

1152 1153 1154 1155
    while(!mid->hg_progress_shutdown_flag)
    {
        trigger_happened = 0;
        do {
1156 1157 1158 1159
            /* save value of instance diag variable, in case it is modified
             * while we are in loop 
             */
            diag_enabled = mid->diag_enabled;
1160

1161
            if(diag_enabled) tm1 = ABT_get_wtime();
1162
            ret = HG_Trigger(mid->hg_context, 0, 1, &actual_count);
1163 1164 1165 1166 1167
            if(diag_enabled)
            {
                tm2 = ABT_get_wtime();
                __DIAG_UPDATE(mid->diag_trigger_elapsed, (tm2-tm1));
            }
1168

1169 1170 1171
            if(ret == HG_SUCCESS && actual_count > 0)
                trigger_happened = 1;
        } while((ret == HG_SUCCESS) && actual_count && !mid->hg_progress_shutdown_flag);
1172

1173 1174
        if(trigger_happened)
            ABT_thread_yield();
1175

1176
        ABT_pool_get_size(mid->progress_pool, &size);
1177
        /* Are there any other threads executing in this pool that are *not*
1178 1179 1180 1181
         * blocked ?  If so then, we can't sleep here or else those threads 
         * will not get a chance to execute.
         * TODO: check is ABT_pool_get_size returns the number of ULT/tasks
         * that can be executed including this one, or not including this one.
1182
         */
1183
        if(size > 0)
1184 1185 1186 1187 1188 1189 1190 1191
        {
            /* TODO: this is being executed more than is necessary (i.e.
             * in cases where there are other legitimate ULTs eligible
             * for execution that are not blocking on any events, Margo
             * or otherwise). Maybe we need an abt scheduling tweak here
             * to make sure that this ULT is the lowest priority in that
             * scenario.
             */
1192
            if(diag_enabled) tm1 = ABT_get_wtime();
1193
            ret = HG_Progress(mid->hg_context, 0);
1194 1195 1196 1197 1198 1199
            if(diag_enabled)
            {
                tm2 = ABT_get_wtime();
                __DIAG_UPDATE(mid->diag_progress_elapsed_zero_timeout, (tm2-tm1));
                __DIAG_UPDATE(mid->diag_progress_timeout_value, 0);
            }
1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218
            if(ret == HG_SUCCESS)
            {
                /* Mercury completed something; loop around to trigger
                 * callbacks 
                 */
            }
            else if(ret == HG_TIMEOUT)
            {
                /* No completion; yield here to allow other ULTs to run */
                ABT_thread_yield();
            }
            else
            {
                /* TODO: error handling */
                fprintf(stderr, "WARNING: unexpected return code (%d) from HG_Progress()\n", ret);
            }
        }
        else
        {
1219
            hg_progress_timeout = mid->hg_progress_timeout_ub;
1220 1221 1222 1223 1224 1225 1226 1227 1228
            ret = margo_timer_get_next_expiration(mid, &next_timer_exp);
            if(ret == 0)
            {
                /* there is a queued timer, don't block long enough
                 * to keep this timer waiting
                 */
                if(next_timer_exp >= 0.0)
                {
                    next_timer_exp *= 1000; /* convert to milliseconds */
1229
                    if(next_timer_exp < mid->hg_progress_timeout_ub)
1230 1231 1232 1233 1234 1235 1236
                        hg_progress_timeout = (unsigned int)next_timer_exp;
                }
                else
                {
                    hg_progress_timeout = 0;
                }
            }
1237
            if(diag_enabled) tm1 = ABT_get_wtime();
1238
            ret = HG_Progress(mid->hg_context, hg_progress_timeout);
1239 1240 1241 1242 1243 1244 1245 1246 1247 1248
            if(diag_enabled)
            {
                tm2 = ABT_get_wtime();
                if(hg_progress_timeout == 0)
                    __DIAG_UPDATE(mid->diag_progress_elapsed_zero_timeout, (tm2-tm1));
                else
                    __DIAG_UPDATE(mid->diag_progress_elapsed_nonzero_timeout, (tm2-tm1));
                    
                __DIAG_UPDATE(mid->diag_progress_timeout_value, hg_progress_timeout);
            }
1249 1250 1251 1252 1253 1254
            if(ret != HG_SUCCESS && ret != HG_TIMEOUT)
            {
                /* TODO: error handling */
                fprintf(stderr, "WARNING: unexpected return code (%d) from HG_Progress()\n", ret);
            }
        }
1255

1256 1257 1258
        /* check for any expired timers */
        margo_check_timers(mid);
    }
1259

1260
    return;
Philip Carns's avatar
Philip Carns committed
1261
}
Philip Carns's avatar
Philip Carns committed
1262 1263 1264 1265 1266 1267 1268


void margo_diag_start(margo_instance_id mid)
{
    mid->diag_enabled = 1;
}

1269
static void print_diag_data(FILE *file, const char* name, const char *description, struct diag_data *data)
Philip Carns's avatar
Philip Carns committed
1270
{
1271 1272 1273 1274 1275 1276 1277 1278
    double avg;

    fprintf(file, "# %s\n", description);
    if(data->count != 0)
        avg = data->cumulative/data->count;
    else
        avg = 0;
    fprintf(file, "%s\t%.9f\t%.9f\t%.9f\t%d\n", name, avg, data->min, data->max, data->count);
Philip Carns's avatar
Philip Carns committed
1279 1280 1281
    return;
}

1282
void margo_diag_dump(margo_instance_id mid, const char* file, int uniquify)
Philip Carns's avatar
Philip Carns committed
1283 1284 1285
{
    FILE *outfile;
    time_t ltime;
1286
    char revised_file_name[256] = {0};
Philip Carns's avatar
Philip Carns committed
1287 1288 1289

    assert(mid->diag_enabled);

1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304
    if(uniquify)
    {
        char hostname[128] = {0};
        int pid;

        gethostname(hostname, 128);
        pid = getpid();

        sprintf(revised_file_name, "%s-%s-%d", file, hostname, pid);
    }
    else
    {
        sprintf(revised_file_name, "%s", file);
    }

Philip Carns's avatar
Philip Carns committed
1305 1306 1307 1308 1309 1310
    if(strcmp("-", file) == 0)
    {