margo-p2p-bw.c 17.6 KB
Newer Older
Philip Carns's avatar
Philip Carns committed
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27
/*
 * Copyright (c) 2017 UChicago Argonne, LLC
 *
 * See COPYRIGHT in top-level directory.
 */

#include "ssg-config.h"

#include <unistd.h>
#include <stdio.h>
#include <string.h>
#include <assert.h>

#include <mpi.h>

#include <margo.h>
#ifdef HAVE_ABT_SNOOZER
#include <abt-snoozer.h>
#endif
#include <mercury.h>
#include <abt.h>
#include <ssg.h>
#include <ssg-mpi.h>

struct options
{
    int xfer_size;
28
    int duration_seconds;
Philip Carns's avatar
Philip Carns committed
29 30 31 32 33 34 35 36 37 38
    int concurrency;
    int threads;
    int snoozer_flag_client;
    int snoozer_flag_server;
    unsigned int mercury_timeout_client;
    unsigned int mercury_timeout_server;
    char* diag_file_name;
    char* na_transport;
};

39 40
#define BW_TOTAL_MEM_SIZE 2097152UL

Philip Carns's avatar
Philip Carns committed
41 42 43
static void parse_args(int argc, char **argv, struct options *opts);
static void usage(void);

44 45
MERCURY_GEN_PROC(bw_rpc_in_t,
        ((hg_bulk_t)(bulk_handle)))
Philip Carns's avatar
Philip Carns committed
46 47 48 49 50
DECLARE_MARGO_RPC_HANDLER(bw_ult);

static int run_benchmark(hg_id_t id, ssg_member_id_t target, 
    ssg_group_id_t gid, margo_instance_id mid);

51 52 53 54
struct bw_worker_arg
{
    double start_tm;
    margo_instance_id mid;
Philip Carns's avatar
Philip Carns committed
55 56 57 58
    ABT_mutex *cur_off_mutex;
    size_t *cur_off;
    hg_bulk_t *client_bulk_handle;
    const hg_addr_t *target_addr;
59 60
    hg_size_t bytes_moved;
    double end_ts;
61
    hg_bulk_op_t op;
62 63
};

Philip Carns's avatar
Philip Carns committed
64
static void bw_worker(void *_arg);
Philip Carns's avatar
Philip Carns committed
65

66 67 68
static hg_id_t g_bw_id;
static ABT_pool g_transfer_pool;
static ABT_eventual g_bw_done_eventual;
Philip Carns's avatar
Philip Carns committed
69
static struct options g_opts;
70 71 72
static char *g_buffer = NULL;
static hg_size_t g_buffer_size = BW_TOTAL_MEM_SIZE;
static hg_bulk_t g_bulk_handle = HG_BULK_NULL;
Philip Carns's avatar
Philip Carns committed
73

Philip Carns's avatar
Philip Carns committed
74 75 76 77 78 79 80 81 82 83 84 85 86 87
int main(int argc, char **argv) 
{
    margo_instance_id mid;
    int nranks;
    hg_context_t *hg_context;
    hg_class_t *hg_class;
    ABT_xstream xstream;
    ABT_pool pool;
    int ret;
    ssg_group_id_t gid;
    ssg_member_id_t self;
    int rank;
    int namelen;
    char processor_name[MPI_MAX_PROCESSOR_NAME];
88
    ABT_xstream *bw_worker_xstreams = NULL;
Philip Carns's avatar
Philip Carns committed
89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106

    ABT_init(argc, argv);
    MPI_Init(&argc, &argv);

    /* 2 process one-way bandwidth measurement only */
    MPI_Comm_size(MPI_COMM_WORLD, &nranks);
    if(nranks != 2)
    {
        usage();
        exit(EXIT_FAILURE);
    }
    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
    MPI_Get_processor_name(processor_name,&namelen);
    printf("Process %d of %d is on %s\n",
	rank, nranks, processor_name);

    parse_args(argc, argv, &g_opts);

107 108 109 110 111 112 113 114 115
    /* allocate one big buffer for rdma transfers */
    g_buffer = calloc(g_buffer_size, 1);
    if(!g_buffer)
    {
        perror("calloc");
        fprintf(stderr, "Error: unable to allocate %lu byte buffer.\n", g_buffer_size);
        return(-1);
    }

Philip Carns's avatar
Philip Carns committed
116 117 118 119 120 121 122 123 124 125 126 127 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 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194
    /* boilerplate HG initialization steps */
    /***************************************/

    if((rank == 0 && g_opts.mercury_timeout_client == 0) ||
       (rank == 1 && g_opts.mercury_timeout_server == 0))
    {
        struct hg_init_info hii;
        
        /* If mercury timeout of zero is requested, then set
         * init option to NO_BLOCK.  This allows some transports to go
         * faster because they do not have to set up or maintain the data
         * structures necessary for signaling completion on blocked
         * operations.
         */
        memset(&hii, 0, sizeof(hii));
        hii.na_init_info.progress_mode = NA_NO_BLOCK;
        hg_class = HG_Init_opt(g_opts.na_transport, HG_TRUE, &hii);
    }
    else
    {
        hg_class = HG_Init(g_opts.na_transport, HG_TRUE);
    }
    if(!hg_class)
    {
        fprintf(stderr, "Error: HG_Init()\n");
        return(-1);
    }
    hg_context = HG_Context_create(hg_class);
    if(!hg_context)
    {
        fprintf(stderr, "Error: HG_Context_create()\n");
        HG_Finalize(hg_class);
        return(-1);
    }

    if((rank == 0 && g_opts.snoozer_flag_client) || 
        (rank == 1 && g_opts.snoozer_flag_server))
    {
#ifdef HAVE_ABT_SNOOZER
        /* set primary ES to idle without polling in scheduler */
        ret = ABT_snoozer_xstream_self_set();
        if(ret != 0)
        {
            fprintf(stderr, "Error: ABT_snoozer_xstream_self_set()\n");
            return(-1);
        }
#else
        fprintf(stderr, "Error: abt-snoozer scheduler is not supported\n");
        return(-1);
#endif
    }

    /* get main pool for running mercury progress and RPC handlers */
    ret = ABT_xstream_self(&xstream);
    if(ret != 0)
    {
        fprintf(stderr, "Error: ABT_xstream_self()\n");
        return(-1);
    }   
    ret = ABT_xstream_get_main_pools(xstream, 1, &pool);
    if(ret != 0)
    {
        fprintf(stderr, "Error: ABT_xstream_get_main_pools()\n");
        return(-1);
    }

    /* actually start margo */
    mid = margo_init_pool(pool, pool, hg_context);
    assert(mid);

    if(g_opts.diag_file_name)
        margo_diag_start(mid);

    /* adjust mercury timeout in Margo if requested */
    if(rank == 0 && g_opts.mercury_timeout_client != UINT_MAX)
        margo_set_param(mid, MARGO_PARAM_PROGRESS_TIMEOUT_UB, &g_opts.mercury_timeout_client);
    if(rank == 1 && g_opts.mercury_timeout_server != UINT_MAX)
        margo_set_param(mid, MARGO_PARAM_PROGRESS_TIMEOUT_UB, &g_opts.mercury_timeout_server);

195
    g_bw_id = MARGO_REGISTER_MPLEX(
Philip Carns's avatar
Philip Carns committed
196 197
        mid, 
        "bw_rpc", 
198
        bw_rpc_in_t,
Philip Carns's avatar
Philip Carns committed
199 200 201 202 203 204 205 206 207 208 209 210 211 212 213
        void,
        bw_ult,
        MARGO_DEFAULT_MPLEX_ID,
        NULL);

    /* set up group */
    ret = ssg_init(mid);
    assert(ret == 0);
    gid = ssg_group_create_mpi("margo-p2p-latency", MPI_COMM_WORLD, NULL, NULL);
    assert(gid != SSG_GROUP_ID_NULL);

    assert(ssg_get_group_size(gid) == 2);

    self = ssg_get_group_self_id(gid);

214
    if(self == 1)
Philip Carns's avatar
Philip Carns committed
215
    {
216 217
        /* server side: prep everything before letting the client initiate
         * benchmark
218
         */
219 220 221 222
        void* buffer = g_buffer;

        /* register memory for xfer */
        ret = margo_bulk_create(mid, 1, &buffer, &g_buffer_size, HG_BULK_READWRITE, &g_bulk_handle);
Philip Carns's avatar
Philip Carns committed
223
        assert(ret == 0);
224

225
        /* set up abt pool */
Philip Carns's avatar
Philip Carns committed
226 227 228
        if(g_opts.threads == 0)
        {
            /* run bulk transfers from primary pool on server */
229
            g_transfer_pool = pool;
Philip Carns's avatar
Philip Carns committed
230 231 232
        }
        else
        {
233 234 235 236
            /* run bulk transfers from a dedicated pool */
            bw_worker_xstreams = malloc(
                    g_opts.threads * sizeof(*bw_worker_xstreams));
            assert(bw_worker_xstreams);
237
            ret = ABT_snoozer_xstream_create(g_opts.threads, &g_transfer_pool,
238 239
                    bw_worker_xstreams);
            assert(ret == ABT_SUCCESS);
Philip Carns's avatar
Philip Carns committed
240 241
        }

242 243
        /* signaling mechanism for server to exit at conclusion of test */
        ret = ABT_eventual_create(0, &g_bw_done_eventual);
Philip Carns's avatar
Philip Carns committed
244
        assert(ret == 0);
245 246 247 248 249 250 251 252 253 254 255 256 257 258
    }

    MPI_Barrier(MPI_COMM_WORLD);

    if(self == 0)
    {
        /* ssg id 0 (client) initiates benchmark */
        ret = run_benchmark(g_bw_id, 1, gid, mid);
        assert(ret == 0);
    }
    else
    {
        /* ssg id 1 (server) waits for test RPC to complete */
        int i;
Philip Carns's avatar
Philip Carns committed
259

260
        ABT_eventual_wait(g_bw_done_eventual, NULL);
261 262 263 264 265 266 267 268

        /* cleanup dedicated pool if needed */
        for (i = 0; i < g_opts.threads; i++) {
            ABT_xstream_join(bw_worker_xstreams[i]);
            ABT_xstream_free(&bw_worker_xstreams[i]);
        }
        if(bw_worker_xstreams)
            free(bw_worker_xstreams);
269 270
    
        margo_bulk_free(g_bulk_handle);
Philip Carns's avatar
Philip Carns committed
271 272 273 274 275 276 277 278
    }

    ssg_group_destroy(gid);
    ssg_finalize();

    if(g_opts.diag_file_name)
        margo_diag_dump(mid, g_opts.diag_file_name, 1);

279 280
    free(g_buffer);

Philip Carns's avatar
Philip Carns committed
281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311
    margo_finalize(mid);
    HG_Context_destroy(hg_context);
    HG_Finalize(hg_class);
    MPI_Finalize();
    ABT_finalize();

    return 0;
}

static void parse_args(int argc, char **argv, struct options *opts)
{
    int opt;
    int ret;
    char clientflag, serverflag;

    memset(opts, 0, sizeof(*opts));

    opts->concurrency = 1;

#ifdef HAVE_ABT_SNOOZER
    /* default to enabling snoozer scheduler on both client and server */
    opts->snoozer_flag_client = 1;
    opts->snoozer_flag_server = 1;
#else
    opts->snoozer_flag_client = 0;
    opts->snoozer_flag_server = 0;
#endif
    /* default to using whatever the standard timeout is in margo */
    opts->mercury_timeout_client = UINT_MAX;
    opts->mercury_timeout_server = UINT_MAX; 

312
    while((opt = getopt(argc, argv, "n:x:c:T:d:s:t:D:")) != -1)
Philip Carns's avatar
Philip Carns committed
313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339
    {
        switch(opt)
        {
            case 'd':
                opts->diag_file_name = strdup(optarg);
                if(!opts->diag_file_name)
                {
                    perror("strdup");
                    exit(EXIT_FAILURE);
                }
                break;
            case 'x':
                ret = sscanf(optarg, "%d", &opts->xfer_size);
                if(ret != 1)
                {
                    usage();
                    exit(EXIT_FAILURE);
                }
                break;
            case 'c':
                ret = sscanf(optarg, "%d", &opts->concurrency);
                if(ret != 1)
                {
                    usage();
                    exit(EXIT_FAILURE);
                }
                break;
340
            case 'T':
Philip Carns's avatar
Philip Carns committed
341 342 343 344 345 346 347
                ret = sscanf(optarg, "%d", &opts->threads);
                if(ret != 1)
                {
                    usage();
                    exit(EXIT_FAILURE);
                }
                break;
348 349 350 351 352 353 354 355
            case 'D':
                ret = sscanf(optarg, "%d", &opts->duration_seconds);
                if(ret != 1)
                {
                    usage();
                    exit(EXIT_FAILURE);
                }
                break;
Philip Carns's avatar
Philip Carns committed
356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389
            case 's':
                ret = sscanf(optarg, "%c,%c", &clientflag, &serverflag);
                if(ret != 2)
                {
                    usage();
                    exit(EXIT_FAILURE);
                }
                if(clientflag == '0') opts->snoozer_flag_client = 0;
                else if(clientflag == '1') opts->snoozer_flag_client = 1;
                if(serverflag == '0') opts->snoozer_flag_server = 0;
                else if(serverflag == '1') opts->snoozer_flag_server = 1;
                break;
            case 't':
                ret = sscanf(optarg, "%u,%u", &opts->mercury_timeout_client, &opts->mercury_timeout_server);
                if(ret != 2)
                {
                    usage();
                    exit(EXIT_FAILURE);
                }
                break;
            case 'n':
                opts->na_transport = strdup(optarg);
                if(!opts->na_transport)
                {
                    perror("strdup");
                    exit(EXIT_FAILURE);
                }
                break;
            default:
                usage();
                exit(EXIT_FAILURE);
        }
    }

390
    if(opts->xfer_size < 1 || opts->concurrency < 1 || opts->duration_seconds < 1 || !opts->na_transport)
Philip Carns's avatar
Philip Carns committed
391 392 393 394 395 396 397 398 399 400 401 402
    {
        usage();
        exit(EXIT_FAILURE);
    }

    return;
}

static void usage(void)
{
    fprintf(stderr,
        "Usage: "
403 404 405
        "margo-p2p-bw -x <xfer_size> -D <duration> -n <na>\n"
        "\t-x <xfer_size> - size of each bulk tranfer in bytes\n"
        "\t-D <duration> - duration of test in seconds\n"
Philip Carns's avatar
Philip Carns committed
406 407
        "\t-n <na> - na transport\n"
        "\t[-c concurrency] - number of concurrent operations to issue with ULTs\n"
408
        "\t[-T <os threads] - number of dedicated operating system threads to run ULTs on\n"
Philip Carns's avatar
Philip Carns committed
409 410 411
        "\t[-d filename] - enable diagnostics output \n"
        "\t[-s <bool,bool>] - specify if snoozer scheduler is used on client and server\n"
        "\t\t(e.g., -s 0,1 means snoozer disabled on client and enabled on server)\n"
412
        "\t\texample: mpiexec -n 2 ./margo-p2p-bw -x 4096 -D 30 -n verbs://\n"
Philip Carns's avatar
Philip Carns committed
413 414 415 416 417 418 419 420
        "\t\t(must be run with exactly 2 processes\n");
    
    return;
}

/* service an RPC that runs the bandwidth test */
static void bw_ult(hg_handle_t handle)
{
Philip Carns's avatar
Philip Carns committed
421
    int i;
Philip Carns's avatar
Philip Carns committed
422
    bw_rpc_in_t in;
Philip Carns's avatar
Philip Carns committed
423
    ABT_thread *tid_array;
424
    struct bw_worker_arg *arg_array;
Philip Carns's avatar
Philip Carns committed
425
    int ret;
426 427 428
    double start_time;
    margo_instance_id mid;
    const struct hg_info *hgi;
Philip Carns's avatar
Philip Carns committed
429 430
    size_t cur_off = 0;
    ABT_mutex cur_off_mutex;
431 432 433 434
    unsigned long bytes_moved = 0;
    double end_ts = 0.0;
    unsigned long bytes_to_check = 0;
    hg_size_t x;
435

Philip Carns's avatar
Philip Carns committed
436 437
    ABT_mutex_create(&cur_off_mutex);
    
438 439 440 441 442
    /* get handle info and margo instance */
    hgi = margo_get_info(handle);
    assert(hgi);
    mid = margo_hg_info_get_instance(hgi);
    assert(mid != MARGO_INSTANCE_NULL);
Philip Carns's avatar
Philip Carns committed
443

Philip Carns's avatar
Philip Carns committed
444 445 446
    ret = margo_get_input(handle, &in);
        assert(ret == HG_SUCCESS);

Philip Carns's avatar
Philip Carns committed
447 448
    tid_array = malloc(g_opts.concurrency * sizeof(*tid_array));
    assert(tid_array);
449
    arg_array = calloc(g_opts.concurrency, sizeof(*arg_array));
450
    assert(arg_array);
Philip Carns's avatar
Philip Carns committed
451

452
    start_time = ABT_get_wtime();
Philip Carns's avatar
Philip Carns committed
453 454 455
    /* create requested number of workers to run transfer */
    for(i=0; i<g_opts.concurrency; i++)
    {
456 457
        arg_array[i].start_tm = start_time;
        arg_array[i].mid = mid;
Philip Carns's avatar
Philip Carns committed
458 459 460 461
        arg_array[i].cur_off = &cur_off;
        arg_array[i].cur_off_mutex = &cur_off_mutex;
        arg_array[i].client_bulk_handle = &in.bulk_handle;
        arg_array[i].target_addr = &hgi->addr;
462
        arg_array[i].op = HG_BULK_PULL;
Philip Carns's avatar
Philip Carns committed
463

464
        ret = ABT_thread_create(g_transfer_pool, bw_worker, &arg_array[i], ABT_THREAD_ATTR_NULL, &tid_array[i]);
Philip Carns's avatar
Philip Carns committed
465 466 467 468 469 470 471
        assert(ret == 0);
    }

    for(i=0; i<g_opts.concurrency; i++)
    {
        ABT_thread_join(tid_array[i]);
        ABT_thread_free(&tid_array[i]);
472 473 474 475
        
        bytes_moved += arg_array[i].bytes_moved;
        if(arg_array[i].end_ts > end_ts)
            end_ts = arg_array[i].end_ts;
Philip Carns's avatar
Philip Carns committed
476 477
    }

478 479 480 481 482 483 484 485 486
    /* calculate how many bytes of the buffer have been transferred */
    bytes_to_check = (g_buffer_size / g_opts.xfer_size) * g_opts.xfer_size;
    if(bytes_moved < bytes_to_check)
        bytes_to_check = bytes_moved;

    /* check integrity of fill pattern.  Note that this isn't as strong as
     * checking every RDMA transfer separately since we are looping around
     * and overwriting in a ring-buffer style.  We could corrupt early and
     * then get it right later and miss the problem.
487
     */
488
    for(x=0; x<(bytes_to_check/sizeof(x)); x++)
489
    {
490
        assert(((hg_size_t*)g_buffer)[x] == x);
491 492 493 494 495
        /* while we are at it, modify pattern so that we can check integrity
         * on client side for reverse direction transfer
         */
        ((hg_size_t*)g_buffer)[x] = x+1;
    }
496 497 498 499 500 501 502 503

    printf("<op>\t<concurrency>\t<threads>\t<bytes>\t<seconds>\t<MiB/s>\n");
    printf("PULL\t%d\t%d\t%lu\t%f\t%f\n",
        g_opts.concurrency,
        g_opts.threads,
        bytes_moved,
        (end_ts-start_time),
        ((double)bytes_moved/(end_ts-start_time))/(1024.0*1024.0));
504

505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543
    /* pause a moment */
    margo_thread_sleep(mid, 100);

    cur_off = 0;
    memset(arg_array, 0, sizeof(*arg_array) * g_opts.concurrency);
    start_time = ABT_get_wtime();

    /* create requested number of workers to run transfer */
    for(i=0; i<g_opts.concurrency; i++)
    {
        arg_array[i].start_tm = start_time;
        arg_array[i].mid = mid;
        arg_array[i].cur_off = &cur_off;
        arg_array[i].cur_off_mutex = &cur_off_mutex;
        arg_array[i].client_bulk_handle = &in.bulk_handle;
        arg_array[i].target_addr = &hgi->addr;
        arg_array[i].op = HG_BULK_PUSH;

        ret = ABT_thread_create(g_transfer_pool, bw_worker, &arg_array[i], ABT_THREAD_ATTR_NULL, &tid_array[i]);
        assert(ret == 0);
    }

    for(i=0; i<g_opts.concurrency; i++)
    {
        ABT_thread_join(tid_array[i]);
        ABT_thread_free(&tid_array[i]);
        
        bytes_moved += arg_array[i].bytes_moved;
        if(arg_array[i].end_ts > end_ts)
            end_ts = arg_array[i].end_ts;
    }

    printf("PUSH\t%d\t%d\t%lu\t%f\t%f\n",
        g_opts.concurrency,
        g_opts.threads,
        bytes_moved,
        (end_ts-start_time),
        ((double)bytes_moved/(end_ts-start_time))/(1024.0*1024.0));

Philip Carns's avatar
Philip Carns committed
544
    margo_respond(handle, NULL);
Philip Carns's avatar
Philip Carns committed
545
    margo_free_input(handle, &in);
Philip Carns's avatar
Philip Carns committed
546 547
    margo_destroy(handle);

Philip Carns's avatar
Philip Carns committed
548
    free(tid_array);
549
    free(arg_array);
Philip Carns's avatar
Philip Carns committed
550

551
    ABT_eventual_set(g_bw_done_eventual, NULL, 0);
Philip Carns's avatar
Philip Carns committed
552

Philip Carns's avatar
Philip Carns committed
553 554
    ABT_mutex_free(&cur_off_mutex);

Philip Carns's avatar
Philip Carns committed
555 556 557 558 559 560 561 562 563 564
    return;
}
DEFINE_MARGO_RPC_HANDLER(bw_ult)

static int run_benchmark(hg_id_t id, ssg_member_id_t target, 
    ssg_group_id_t gid, margo_instance_id mid)
{
    hg_handle_t handle;
    hg_addr_t target_addr;
    int ret;
565 566
    bw_rpc_in_t in;
    void* buffer = g_buffer;
567 568 569 570 571
    hg_size_t i;

    /* fill pattern in origin buffer */
    for(i=0; i<(g_buffer_size/sizeof(i)); i++)
        ((hg_size_t*)buffer)[i] = i;
Philip Carns's avatar
Philip Carns committed
572 573 574 575 576 577 578

    target_addr = ssg_get_addr(gid, target);
    assert(target_addr != HG_ADDR_NULL);

    ret = margo_create(mid, target_addr, id, &handle);
    assert(ret == 0);

579 580 581
    ret = margo_bulk_create(mid, 1, &buffer, &g_buffer_size, HG_BULK_READWRITE, &in.bulk_handle);
    assert(ret == 0);

Philip Carns's avatar
Philip Carns committed
582
    ret = margo_forward(handle, &in);
Philip Carns's avatar
Philip Carns committed
583 584
    assert(ret == 0);

585
    margo_bulk_free(in.bulk_handle);
Philip Carns's avatar
Philip Carns committed
586 587 588 589 590
    margo_destroy(handle);

    return(0);
}

Philip Carns's avatar
Philip Carns committed
591 592 593
/* function that assists in transferring data until end condition is met */
static void bw_worker(void *_arg)
{
594 595
    struct bw_worker_arg *arg = _arg;
    double now;
Philip Carns's avatar
Philip Carns committed
596 597
    size_t my_off;
    int ret;
598

599
    // printf("# DBG: worker started.\n");
600 601 602 603 604

    now = ABT_get_wtime();

    while((now - arg->start_tm) < g_opts.duration_seconds)
    {
Philip Carns's avatar
Philip Carns committed
605 606 607 608 609 610 611 612 613 614
        /* find the offset for this transfer and then increment for next
         * one
         */
        ABT_mutex_spinlock(*arg->cur_off_mutex);
        my_off = *arg->cur_off;
        (*arg->cur_off) += g_opts.xfer_size;
        if(*arg->cur_off > g_buffer_size)
            *arg->cur_off = 0;
        ABT_mutex_unlock(*arg->cur_off_mutex);

615
        ret = margo_bulk_transfer(arg->mid, arg->op,
Philip Carns's avatar
Philip Carns committed
616 617 618
                *arg->target_addr, *arg->client_bulk_handle, my_off, g_bulk_handle, my_off, g_opts.xfer_size);
        assert(ret == 0);

619
        arg->bytes_moved += g_opts.xfer_size;
620 621
        now = ABT_get_wtime();
    }
622
    arg->end_ts = now;
623

624
    // printf("# DBG: worker stopped.\n");
Philip Carns's avatar
Philip Carns committed
625 626
    return;
}