margo-p2p-bw.c 16.2 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 62
};

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

65 66 67
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
68
static struct options g_opts;
69 70 71
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
72

Philip Carns's avatar
Philip Carns committed
73 74 75 76 77 78 79 80 81 82 83 84 85 86
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];
87
    ABT_xstream *bw_worker_xstreams = NULL;
Philip Carns's avatar
Philip Carns committed
88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105

    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);

106 107 108 109 110 111 112 113 114
    /* 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
115 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
    /* 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);

194
    g_bw_id = MARGO_REGISTER_MPLEX(
Philip Carns's avatar
Philip Carns committed
195 196
        mid, 
        "bw_rpc", 
197
        bw_rpc_in_t,
Philip Carns's avatar
Philip Carns committed
198 199 200 201 202 203 204 205 206 207 208 209 210 211 212
        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);

213
    if(self == 1)
Philip Carns's avatar
Philip Carns committed
214
    {
215 216
        /* server side: prep everything before letting the client initiate
         * benchmark
217
         */
218 219 220 221
        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
222
        assert(ret == 0);
223

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

241 242
        /* 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
243
        assert(ret == 0);
244 245 246 247 248 249 250 251 252 253 254 255 256 257
    }

    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
258

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

        /* 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);
268 269
    
        margo_bulk_free(g_bulk_handle);
Philip Carns's avatar
Philip Carns committed
270 271 272 273 274 275 276 277
    }

    ssg_group_destroy(gid);
    ssg_finalize();

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

278 279
    free(g_buffer);

Philip Carns's avatar
Philip Carns committed
280 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
    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; 

311
    while((opt = getopt(argc, argv, "n:x:c:T:d:s:t:D:")) != -1)
Philip Carns's avatar
Philip Carns committed
312 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
    {
        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;
339
            case 'T':
Philip Carns's avatar
Philip Carns committed
340 341 342 343 344 345 346
                ret = sscanf(optarg, "%d", &opts->threads);
                if(ret != 1)
                {
                    usage();
                    exit(EXIT_FAILURE);
                }
                break;
347 348 349 350 351 352 353 354
            case 'D':
                ret = sscanf(optarg, "%d", &opts->duration_seconds);
                if(ret != 1)
                {
                    usage();
                    exit(EXIT_FAILURE);
                }
                break;
Philip Carns's avatar
Philip Carns committed
355 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
            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);
        }
    }

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

    return;
}

static void usage(void)
{
    fprintf(stderr,
        "Usage: "
402 403 404
        "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
405 406
        "\t-n <na> - na transport\n"
        "\t[-c concurrency] - number of concurrent operations to issue with ULTs\n"
407
        "\t[-T <os threads] - number of dedicated operating system threads to run ULTs on\n"
Philip Carns's avatar
Philip Carns committed
408 409 410
        "\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"
411
        "\t\texample: mpiexec -n 2 ./margo-p2p-bw -x 4096 -D 30 -n verbs://\n"
Philip Carns's avatar
Philip Carns committed
412 413 414 415 416 417 418 419
        "\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
420
    int i;
Philip Carns's avatar
Philip Carns committed
421
    bw_rpc_in_t in;
Philip Carns's avatar
Philip Carns committed
422
    ABT_thread *tid_array;
423
    struct bw_worker_arg *arg_array;
Philip Carns's avatar
Philip Carns committed
424
    int ret;
425 426 427
    double start_time;
    margo_instance_id mid;
    const struct hg_info *hgi;
Philip Carns's avatar
Philip Carns committed
428 429
    size_t cur_off = 0;
    ABT_mutex cur_off_mutex;
430 431 432 433
    unsigned long bytes_moved = 0;
    double end_ts = 0.0;
    unsigned long bytes_to_check = 0;
    hg_size_t x;
434

Philip Carns's avatar
Philip Carns committed
435 436
    ABT_mutex_create(&cur_off_mutex);
    
437 438 439 440 441
    /* 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
442

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

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

451
    start_time = ABT_get_wtime();
Philip Carns's avatar
Philip Carns committed
452 453 454
    /* create requested number of workers to run transfer */
    for(i=0; i<g_opts.concurrency; i++)
    {
455 456
        arg_array[i].start_tm = start_time;
        arg_array[i].mid = mid;
Philip Carns's avatar
Philip Carns committed
457 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
        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
463 464 465 466 467 468 469
        assert(ret == 0);
    }

    for(i=0; i<g_opts.concurrency; i++)
    {
        ABT_thread_join(tid_array[i]);
        ABT_thread_free(&tid_array[i]);
470 471 472 473
        
        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
474 475
    }

476 477 478 479 480 481 482 483 484
    /* 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.
485
     */
486 487 488 489 490 491 492 493 494 495
    for(x=0; x<(bytes_to_check/sizeof(x)); x++)
        assert(((hg_size_t*)g_buffer)[x] == x);

    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));
496

Philip Carns's avatar
Philip Carns committed
497
    margo_respond(handle, NULL);
Philip Carns's avatar
Philip Carns committed
498
    margo_free_input(handle, &in);
Philip Carns's avatar
Philip Carns committed
499 500
    margo_destroy(handle);

Philip Carns's avatar
Philip Carns committed
501
    free(tid_array);
502
    free(arg_array);
Philip Carns's avatar
Philip Carns committed
503

504
    ABT_eventual_set(g_bw_done_eventual, NULL, 0);
Philip Carns's avatar
Philip Carns committed
505

Philip Carns's avatar
Philip Carns committed
506 507
    ABT_mutex_free(&cur_off_mutex);

Philip Carns's avatar
Philip Carns committed
508 509 510 511 512 513 514 515 516 517
    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;
518 519
    bw_rpc_in_t in;
    void* buffer = g_buffer;
520 521 522 523 524
    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
525 526 527 528 529 530 531

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

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

532 533 534
    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
535
    ret = margo_forward(handle, &in);
Philip Carns's avatar
Philip Carns committed
536 537
    assert(ret == 0);

538
    margo_bulk_free(in.bulk_handle);
Philip Carns's avatar
Philip Carns committed
539 540 541 542 543
    margo_destroy(handle);

    return(0);
}

Philip Carns's avatar
Philip Carns committed
544 545 546
/* function that assists in transferring data until end condition is met */
static void bw_worker(void *_arg)
{
547 548
    struct bw_worker_arg *arg = _arg;
    double now;
Philip Carns's avatar
Philip Carns committed
549 550
    size_t my_off;
    int ret;
551

552
    // printf("# DBG: worker started.\n");
553 554 555

    now = ABT_get_wtime();

Philip Carns's avatar
Philip Carns committed
556
    /* TODO: need two runs, one in each direction? */
557 558
    while((now - arg->start_tm) < g_opts.duration_seconds)
    {
Philip Carns's avatar
Philip Carns committed
559 560 561 562 563 564 565 566 567 568 569 570 571 572
        /* 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);

        ret = margo_bulk_transfer(arg->mid, HG_BULK_PULL,
                *arg->target_addr, *arg->client_bulk_handle, my_off, g_bulk_handle, my_off, g_opts.xfer_size);
        assert(ret == 0);

573
        arg->bytes_moved += g_opts.xfer_size;
574 575
        now = ABT_get_wtime();
    }
576
    arg->end_ts = now;
577

578
    // printf("# DBG: worker stopped.\n");
Philip Carns's avatar
Philip Carns committed
579 580
    return;
}