initial ssg impl

COPYRIGHT

+Copyright (c) 2016, UChicago Argonne, LLC
+All Rights Reserved
+SDS TOOLS (ANL-SF-16-009)
+
+OPEN SOURCE LICENSE
+
+Under the terms of Contract No. DE-AC02-06CH11357 with UChicago Argonne,
+LLC, the U.S. Government retains certain rights in this software.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions
+are met:
+
+1. Redistributions of source code must retain the above copyright notice,
+this list of conditions and the following disclaimer.
+
+2. Redistributions in binary form must reproduce the above copyright
+notice, this list of conditions and the following disclaimer in the
+documentation and/or other materials provided with the distribution.
+
+3. Neither the names of UChicago Argonne, LLC or the Department of
+Energy nor the names of its contributors may be used to endorse or
+promote products derived from this software without specific prior
+written permission.
+
+******************************************************************************
+DISCLAIMER
+
+THE SOFTWARE IS SUPPLIED "AS IS" WITHOUT WARRANTY OF ANY KIND.
+
+NEITHER THE UNTED STATES GOVERNMENT, NOR THE UNITED STATES DEPARTMENT
+OF ENERGY, NOR UCHICAGO ARGONNE, LLC, NOR ANY OF THEIR EMPLOYEES,
+MAKES ANY WARRANTY, EXPRESS OR IMPLIED, OR ASSUMES ANY LEGAL LIABILITY
+OR RESPONSIBILITY FOR THE ACCURACY, COMPLETENESS, OR USEFULNESS OF ANY
+INFORMATION, DATA, APPARATUS, PRODUCT, OR PROCESS DISCLOSED, OR REPRESENTS
+THAT ITS USE WOULD NOT INFRINGE PRIVATELY OWNED RIGHTS.
+
+******************************************************************************

README.md

+SSG is a Simple, Stupid Grouping mechanism for Mercury. It provides
+mechanisms for bootstrapping a set of pre-existing mercury processes. So
+far, we have the following:
+
+- MPI bootstrap (this works well with CCI, where the addresses you pass in
+  aren't the addresses used)
+- config-file bootstrap (where each process is assumed to exist in the
+  membership list - CCI can't currently be used with this method)
+
+Serializers for the ssg data structure are also provided.
+
+# Building
+
+(if configuring for the first time)
+./prepare.sh
+
+./configure [standard options] PKG\_CONFIG\_PATH=/path/to/mercury/pkgconfig
+make
+make install
+
+MPI support is by default optionally included. If you wish to compile with MPI support, set CC=mpicc (or equivalent) in configure. If you wish to disable MPI entirely, use --disable-mpi (you can also force MPI inclusion through --enable-mpi).
+
+# Documentation
+
+The example is the best documentation so far. Check out examples/ssg-example.c
+(and the other files) for usage. The example program initializes ssg, pings
+peer servers as defined by their ssg rank, and shuts down via a chain
+communication.
+
+# Running examples
+
+cd to your build directory and run:
+$top\_level\_dir/examples/run-example-\*.sh
+
+See the script contents for how these are run.

configure.ac

+#                                               -*- Autoconf -*-
+# Process this file with autoconf to produce a configure script.
+
+AC_PREREQ([2.67])
+AC_INIT([ssg], [0.1], [],[],[]) 
+
+AC_CANONICAL_TARGET
+AC_CANONICAL_SYSTEM
+AC_CANONICAL_HOST
+
+AM_INIT_AUTOMAKE([foreign subdir-objects -Wall])
+
+# we should remove this soon, only needed for automake 1.10 and older
+m4_ifdef([AM_SILENT_RULES], [AM_SILENT_RULES([yes])])
+
+AC_CONFIG_SRCDIR([README.md])
+AC_CONFIG_HEADERS([ssg-config.h])
+AC_CONFIG_MACRO_DIR([m4])
+
+AC_LANG([C])
+
+# Checks for programs.
+AC_PROG_CC
+AC_PROG_CC_C99
+AC_PROG_RANLIB
+
+dnl
+dnl Add warning flags by default
+dnl
+CFLAGS="-Wall -Wextra $CFLAGS"
+CXXFLAGS="-Wall -Wextra $CXXFLAGS"
+
+dnl
+dnl Verify pkg-config
+dnl
+PKG_PROG_PKG_CONFIG
+PKG_CONFIG="pkg-config --static"
+
+PKG_CHECK_MODULES_STATIC([MERCURY],[mercury],[],
+   [AC_MSG_ERROR([Could not find working mercury installation!])])
+LIBS="$MERCURY_LIBS $LIBS"
+CPPFLAGS="$MERCURY_CFLAGS $CPPFLAGS"
+CFLAGS="$MERCURY_CFLAGS $CFLAGS"
+
+check_mpi=auto
+AC_ARG_ENABLE([mpi],
+    [--enable-mpi    Enable MPI (default: dynamic check)],
+    [ case "${enableval}" in
+     yes) check_mpi=yes ;;
+     no)  check_mpi=no ;;
+     *) AC_MSG_ERROR([bad value ${enableval} for --enable-mpi]) ;;
+     esac],
+    [])
+
+
+check_mpi_status=fail
+if test "x${check_mpi}" = xauto -o "x${check_mpi}" = xyes ; then
+    AC_MSG_CHECKING([If MPI programs can be compiled])
+    AC_LINK_IFELSE(
+        [AC_LANG_PROGRAM([[#include<mpi.h>]], [[MPI_Init(0,0);]])],
+        [AC_DEFINE([HAVE_MPI], [1], [Define to 1 if compiled with MPI support])
+         AC_MSG_RESULT([yes])
+         check_mpi_status=success],
+        [AC_MSG_RESULT([no])])
+fi
+
+if test "x${check_mpi_status}" = xfail -a "x${check_mpi}" = xyes; then
+    AC_MSG_ERROR([MPI requested but unable to be used. Did you specify an MPI compiler?])
+fi
+
+AM_CONDITIONAL([HAVE_MPI], [test "x${check_mpi_status}" = xsuccess])
+
+AC_CONFIG_FILES([Makefile maint/ssg.pc])
+AC_OUTPUT

examples/example.conf

+bmi+tcp://localhost:3344
+bmi+tcp://localhost:3345
+bmi+tcp://localhost:3346

examples/rpc.c

+/*
+ * Copyright (c) 2016 UChicago Argonne, LLC
+ *
+ * See COPYRIGHT in top-level directory.
+ */
+
+#include <assert.h>
+#include <stdio.h>
+
+#include <mercury.h>
+#include <ssg.h>
+#include "rpc.h"
+
+hg_return_t ping_rpc_handler(hg_handle_t h)
+{
+    hg_return_t hret;
+    ping_t out;
+    ping_t in;
+    struct hg_info *info;
+    rpc_context_t *c;
+
+    hret = HG_Get_input(h, &in);
+    assert(hret == HG_SUCCESS);
+
+    info = HG_Get_info(h);
+    assert(info != NULL);
+
+    // get ssg data
+    c = HG_Registered_data(info->hg_class, info->id);
+    assert(c != NULL && c->s != SSG_NULL);
+    out.rank = ssg_get_rank(c->s);
+    assert(out.rank != SSG_RANK_UNKNOWN && out.rank != SSG_EXTERNAL_RANK);
+
+    printf("%d: got ping from rank %d\n", out.rank, in.rank);
+
+    HG_Respond(h, NULL, NULL, &out);
+
+    hret = HG_Free_input(h, &in);
+    assert(hret == HG_SUCCESS);
+    hret = HG_Destroy(h);
+    assert(hret == HG_SUCCESS);
+    return HG_SUCCESS;
+}
+
+static hg_return_t shutdown_post_respond(const struct hg_cb_info *cb_info)
+{
+    hg_handle_t h;
+    struct hg_info *info;
+    rpc_context_t *c;
+
+    h = cb_info->info.respond.handle;
+    info = HG_Get_info(h);
+    assert(info != NULL);
+
+    c = HG_Registered_data(info->hg_class, info->id);
+    printf("%d: post-respond, setting shutdown flag\n", ssg_get_rank(c->s));
+
+    c->shutdown_flag = 1;
+    HG_Destroy(h);
+    return HG_SUCCESS;
+}
+
+static hg_return_t shutdown_post_forward(const struct hg_cb_info *cb_info)
+{
+    hg_handle_t fwd_handle, resp_handle;
+    rpc_context_t *c;
+    int rank;
+    hg_return_t hret;
+    struct hg_info *info;
+
+    // RPC has completed, respond to previous rank
+    fwd_handle = cb_info->info.forward.handle;
+    resp_handle = cb_info->arg;
+    info = HG_Get_info(fwd_handle);
+    c = HG_Registered_data(info->hg_class, info->id);
+    assert(c != NULL && c->s != SSG_NULL);
+    rank = ssg_get_rank(c->s);
+    assert(rank != SSG_RANK_UNKNOWN && rank != SSG_EXTERNAL_RANK);
+    if (rank > 0) {
+        printf("%d: sending shutdown response\n", rank);
+        hret = HG_Respond(resp_handle, &shutdown_post_respond, NULL, NULL);
+        HG_Destroy(resp_handle);
+        assert(hret == HG_SUCCESS);
+        return HG_SUCCESS;
+    }
+    else {
+        c->shutdown_flag = 1;
+        printf("%d: noone to respond to, setting shutdown flag\n", rank);
+    }
+
+    HG_Destroy(fwd_handle);
+    return HG_SUCCESS;
+}
+
+// shutdown - do a ring communication for simplicity, really would want some
+// multicast or something
+hg_return_t shutdown_rpc_handler(hg_handle_t h)
+{
+    hg_return_t hret;
+    struct hg_info *info;
+    int rank;
+    rpc_context_t *c;
+
+    info = HG_Get_info(h);
+    assert(info != NULL);
+
+    // get ssg data
+    c = HG_Registered_data(info->hg_class, info->id);
+    assert(c != NULL && c->s != SSG_NULL);
+    rank = ssg_get_rank(c->s);
+    assert(rank != SSG_RANK_UNKNOWN && rank != SSG_EXTERNAL_RANK);
+
+    printf("%d: received shutdown request\n", rank);
+
+    // forward shutdown to neighbor
+    rank++;
+    // end-of the line, respond and shut down
+    if (rank == ssg_get_count(c->s)) {
+        printf("%d: sending response and setting shutdown flag\n", rank-1);
+        hret = HG_Respond(h, &shutdown_post_respond, NULL, NULL);
+        assert(hret == HG_SUCCESS);
+        hret = HG_Destroy(h);
+        assert(hret == HG_SUCCESS);
+        c->shutdown_flag = 1;
+    }
+    else {
+        hg_handle_t next_handle;
+        na_addr_t next_addr;
+
+        next_addr = ssg_get_addr(c->s, rank);
+        assert(next_addr != NULL);
+
+        hret = HG_Create(info->context, next_addr, info->id, &next_handle);
+        assert(hret == HG_SUCCESS);
+
+        printf("%d: forwarding shutdown to next\n", rank-1);
+        hret = HG_Forward(next_handle, &shutdown_post_forward, h, NULL);
+        assert(hret == HG_SUCCESS);
+
+        hret = HG_Destroy(next_handle);
+        assert(hret == HG_SUCCESS);
+    }
+
+    return HG_SUCCESS;
+}

examples/rpc.h

+/*
+ * Copyright (c) 2016 UChicago Argonne, LLC
+ *
+ * See COPYRIGHT in top-level directory.
+ */
+
+#pragma once
+
+#include <mercury.h>
+#include <mercury_macros.h>
+#include <ssg.h>
+
+/* visible API for example RPC operation */
+
+typedef struct rpc_context
+{
+    ssg_t s;
+    int shutdown_flag;
+} rpc_context_t;
+
+MERCURY_GEN_PROC(ping_t, ((int32_t)(rank)))
+
+hg_return_t ping_rpc_handler(hg_handle_t h);
+hg_return_t shutdown_rpc_handler(hg_handle_t h);

examples/run-example-conf.sh

+#!/bin/bash
+
+# run me from the top-level build dir
+examples/ssg-example -s 2 bmi+tcp://localhost:3344 conf ../examples/example.conf &
+examples/ssg-example -s 2 bmi+tcp://localhost:3345 conf ../examples/example.conf &
+examples/ssg-example -s 2 bmi+tcp://localhost:3346 conf ../examples/example.conf

examples/run-example-mpi.sh

+#!/bin/bash
+
+# run me from the top-level build dir
+mpirun -np 3 examples/ssg-example -s 0 cci+sm://localhost:3344 conf ../examples/example.conf

examples/ssg-example.c

+/*
+ * Copyright (c) 2016 UChicago Argonne, LLC
+ *
+ * See COPYRIGHT in top-level directory.
+ */
+
+#include <unistd.h>
+#include <stdio.h>
+#include <string.h>
+
+#include <mercury.h>
+#include <na.h>
+#include <mercury_util/mercury_request.h>
+#include <ssg.h>
+#include <ssg-config.h>
+#include "rpc.h"
+
+#ifdef HAVE_MPI
+#include <ssg-mpi.h>
+#endif
+
+#define DIE_IF(cond_expr, err_fmt, ...) \
+    do { \
+        if (cond_expr) { \
+            fprintf(stderr, "ERROR at %s:%d (" #cond_expr "): " \
+                    err_fmt "\n", __FILE__, __LINE__, ##__VA_ARGS__); \
+            exit(1); \
+        } \
+    } while(0)
+
+static void usage()
+{
+    fputs("Usage: "
+          "./ssg-example [-s <time>] <addr> <config mode> [config file]\n"
+          "  -s <time> - time to sleep before doing lookup\n"
+          "  <config mode> - \"mpi\" (if supported) or \"conf\"\n"
+          "  if conf is the mode, then [config file] is required\n",
+          stderr);
+}
+
+// hg_request_class_t progress/trigger
+static int progress(unsigned int timeout, void *arg)
+{
+    if (HG_Progress((hg_context_t*)arg, timeout) == HG_SUCCESS)
+        return HG_UTIL_SUCCESS;
+    else
+        return HG_UTIL_FAIL;
+}
+static int trigger(unsigned int timeout, unsigned int *flag, void *arg)
+{
+    (void)arg;
+    if (HG_Trigger((hg_context_t*)arg, timeout, 1, flag) != HG_SUCCESS) {
+        return HG_UTIL_FAIL;
+    }
+    else {
+        *flag = (*flag) ? HG_UTIL_TRUE : HG_UTIL_FALSE;
+        return HG_UTIL_SUCCESS;
+    }
+}
+
+int main(int argc, char *argv[])
+{
+    // mercury
+    na_class_t *nacl;
+    na_context_t *nactx;
+    hg_class_t *hgcl;
+    hg_context_t *hgctx;
+    hg_request_class_t *reqcl;
+    hg_request_t *req;
+    hg_id_t ping_id, shutdown_id;
+    hg_handle_t ping_handle = HG_HANDLE_NULL;
+
+    // args
+    const char * addr_str;
+    const char * mode;
+    int sleep_time = 0;
+
+    // process state
+    rpc_context_t c = { SSG_NULL, 0 };
+    int rank; // not mpi
+
+    // comm vars
+    int peer_rank;
+    na_addr_t peer_addr;
+    ping_t ping_in;
+    unsigned int req_complete_flag = 0;
+
+    // return codes
+    na_return_t nret;
+    hg_return_t hret;
+    int ret;
+
+#ifdef HAVE_MPI
+    MPI_Init(&argc, &argv);
+#endif
+
+    argc--; argv++;
+
+    if (!argc) { usage(); return 1; }
+
+    if (strcmp(argv[0], "-s") == 0) {
+        if (argc < 2) { usage(); return 1; }
+        sleep_time = atoi(argv[1]);
+        argc -= 2; argv += 2;
+    }
+
+    if (!argc) { usage(); return 1; }
+    addr_str = argv[0];
+    argc--; argv++;
+
+    // init NA, HG
+    nacl = NA_Initialize(addr_str, NA_TRUE);
+    DIE_IF(nacl == NULL, "NA_Initialize");
+    nactx = NA_Context_create(nacl);
+    DIE_IF(nactx == NULL, "NA_Context_create");
+    hgcl = HG_Init_na(nacl, nactx);
+    DIE_IF(hgcl == NULL, "HG_Init");
+    hgctx = HG_Context_create(hgcl);
+    DIE_IF(hgctx == NULL, "HG_Context_create");
+
+    ping_id =
+        MERCURY_REGISTER(hgcl, "ping", ping_t, ping_t, &ping_rpc_handler);
+    shutdown_id =
+        MERCURY_REGISTER(hgcl, "shutdown", void, void, &shutdown_rpc_handler);
+
+    hret = HG_Register_data(hgcl, ping_id, &c, NULL);
+    DIE_IF(hret != HG_SUCCESS, "HG_Register_data");
+    hret = HG_Register_data(hgcl, shutdown_id, &c, NULL);
+    DIE_IF(hret != HG_SUCCESS, "HG_Register_data");
+
+    // parse mode and attempt to initialize ssg
+    if (!argc) { usage(); return 1; }
+    mode = argv[0];
+    argc--; argv++;
+    if (strcmp(mode, "mpi") == 0) {
+#ifdef HAVE_MPI
+        c.s = ssg_init_mpi(nacl, MPI_COMM_WORLD);
+        sleep_time = 0; // ignore sleeping
+#else
+        fprintf(stderr, "Error: MPI support not built in\n");
+        return 1;
+#endif
+    }
+    else if (strcmp(mode, "conf") == 0) {
+        const char * conf;
+        if (!argc) { usage(); return 1; }
+        conf = argv[0];
+        argc--; argv++;
+        c.s = ssg_init_config(conf);
+    }
+    else {
+        fprintf(stderr, "Error: bad mode passed in %s\n", mode);
+        return 1;
+    }
+
+    DIE_IF(c.s == SSG_NULL, "ssg_init (mode %s)", mode);
+
+    if (sleep_time >= 0) sleep(sleep_time);
+
+    // resolve group addresses
+    nret = ssg_lookup(nacl, nactx, c.s);
+    DIE_IF(nret != NA_SUCCESS, "ssg_lookup");
+
+    // get my (non-mpi) rank
+    rank = ssg_get_rank(c.s);
+    DIE_IF(rank == SSG_RANK_UNKNOWN || rank == SSG_EXTERNAL_RANK,
+            "ssg_get_rank - bad rank %d", rank);
+
+    // initialize request shim
+    reqcl = hg_request_init(&progress, &trigger, hgctx);
+    DIE_IF(reqcl == NULL, "hg_request_init");
+    req = hg_request_create(reqcl);
+    DIE_IF(req == NULL, "hg_request_create");
+
+    // sanity check count - if we're on our own, don't bother sending RPCs
+    if (ssg_get_count(c.s) == 1)
+        goto cleanup;
+
+    // all ready to go - ping my neighbor rank
+    peer_rank = (rank+1) % ssg_get_count(c.s);
+    peer_addr = ssg_get_addr(c.s, peer_rank);
+    DIE_IF(peer_addr == NA_ADDR_NULL, "ssg_get_addr(%d)", peer_rank);
+
+    printf("%d: pinging %d\n", rank, peer_rank);
+    hret = HG_Create(hgctx, peer_addr, ping_id, &ping_handle);
+    DIE_IF(hret != HG_SUCCESS, "HG_Create");
+    ping_in.rank = rank;
+    hret = HG_Forward(ping_handle, &hg_request_complete_cb, req, &ping_in);
+    DIE_IF(hret != HG_SUCCESS, "HG_Forward");
+    ret = hg_request_wait(req, HG_MAX_IDLE_TIME, &req_complete_flag);
+    DIE_IF(ret == HG_UTIL_FAIL, "ping failed");
+    DIE_IF(req_complete_flag == 0, "ping timed out");
+
+    // rank 0 - initialize the shutdown process. All others - enter progress
+    if (rank != 0) {
+        unsigned int num_trigger;
+        do {
+            do {
+                num_trigger = 0;
+                hret = HG_Trigger(hgctx, 0, 1, &num_trigger);
+            } while (hret == HG_SUCCESS && num_trigger == 1);
+
+            hret = HG_Progress(hgctx, c.shutdown_flag ? 100 : HG_MAX_IDLE_TIME);
+        } while ((hret == HG_SUCCESS || hret == HG_TIMEOUT) && !c.shutdown_flag);
+        DIE_IF(hret != HG_SUCCESS && hret != HG_TIMEOUT, "HG_Progress");
+        printf("%d: shutting down\n", rank);
+
+        // trigger remaining
+        do {
+            num_trigger = 0;
+            hret = HG_Trigger(hgctx, 0, 1, &num_trigger);
+        } while (hret == HG_SUCCESS && num_trigger == 1);
+    }
+    else {
+        printf("%d: initiating shutdown\n", rank);
+        hg_handle_t shutdown_handle = HG_HANDLE_NULL;
+        hret = HG_Create(hgctx, peer_addr, shutdown_id, &shutdown_handle);
+        DIE_IF(hret != HG_SUCCESS, "HG_Create");
+        hret = HG_Forward(shutdown_handle, &hg_request_complete_cb, req, NULL);
+        DIE_IF(hret != HG_SUCCESS, "HG_Forward");
+        req_complete_flag = 0;
+        ret = hg_request_wait(req, HG_MAX_IDLE_TIME, &req_complete_flag);
+        DIE_IF(ret != HG_UTIL_SUCCESS, "hg_request_wait");
+        DIE_IF(req_complete_flag == 0, "hg_request_wait timeout");
+        HG_Destroy(shutdown_handle);
+    }
+
+cleanup:
+    printf("%d: cleaning up\n", rank);
+    // cleanup
+    HG_Destroy(ping_handle);
+    ssg_finalize(c.s);
+    hg_request_destroy(req);
+    hg_request_finalize(reqcl);
+    HG_Context_destroy(hgctx);
+    HG_Finalize(hgcl);
+    NA_Context_destroy(nacl, nactx);
+    NA_Finalize(nacl);
+
+#ifdef HAVE_MPI
+    MPI_Finalize();
+#endif
+    return 0;
+}

include/ssg-mpi.h

+/*
+ * Copyright (c) 2016 UChicago Argonne, LLC
+ *
+ * See COPYRIGHT in top-level directory.
+ */
+
+#pragma once
+
+// mpi based initialization for ssg
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <mpi.h>
+#include <na.h>
+
+#include "ssg.h"
+
+// mpi based (no config file) - all participants (defined by the input
+// communicator) do a global address exchange
+// in this case, the caller has already initialized NA with its address
+ssg_t ssg_init_mpi(na_class_t *nacl, MPI_Comm comm);
+
+/**
+ * vim: ft=c sw=4 ts=4 sts=4 tw=80 expandtab
+ */

include/ssg.h

+/*
+ * Copyright (c) 2016 UChicago Argonne, LLC
+ *
+ * See COPYRIGHT in top-level directory.
+ */
+
+#pragma once
+
+// "simple stupid group" interface
+//
+// Contains trivial wireup and connection management functionality, with a
+// model of a static (at startup) member list.
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <mercury.h>
+#include <na.h>