first try at integrating swim-fd directly into ssg

--enable-swim-fd configure flag used to force compile of
swim-fd directory and link into libssg

Makefile.am

@@ -15,7 +15,10 @@ CLEANFILES = $(bin_SCRIPTS)
 MAINTAINERCLEANFILES =
 EXTRA_DIST =
 BUILT_SOURCES =
+src_libssg_la_SOURCES =
 include_HEADERS = include/ssg.h
+noinst_HEADERS += ssg-config.h
+lib_LTLIBRARIES += src/libssg.la
 
 EXTRA_DIST += prepare.sh
 
@@ -25,14 +28,14 @@ AM_CFLAGS =
 
 AM_LIBS =
 
-noinst_HEADERS += ssg-config.h
-
-lib_LTLIBRARIES += src/libssg.la
-src_libssg_la_SOURCES = src/ssg.c
-
 pkgconfigdir = $(libdir)/pkgconfig
 pkgconfig_DATA = maint/ssg.pc
 
 include Make.rules
 
+include $(top_srcdir)/src/Makefile.subdir
+if USE_SWIM_FD
+include $(top_srcdir)/src/swim-fd/Makefile.subdir
+endif
+
 include $(top_srcdir)/tests/Makefile.subdir

configure.ac

@@ -36,7 +36,7 @@ PKG_CONFIG="pkg-config --static"
 
 check_mpi=auto
 AC_ARG_ENABLE([mpi],
-    [--enable-mpi    Enable MPI (default: dynamic check)],
+    [  --enable-mpi            enable MPI (default: dynamic check)],
     [ case "${enableval}" in
      yes) check_mpi=yes ;;
      no)  check_mpi=no ;;
@@ -67,28 +67,15 @@ LIBS="$MARGO_LIBS $LIBS"
 CPPFLAGS="$MARGO_CFLAGS $CPPFLAGS"
 CFLAGS="$MARGO_CFLAGS $CFLAGS"
 
-check_swim_fd=auto
 AC_ARG_ENABLE([swim-fd],
-    [--enable-swim-fd   Enable SWIM failure detection (default: dynamic check)],
-    [case "${enableval}" in
-     yes) check_swim_fd=yes ;;
-     no)  check_swim_fd=no  ;;
-     *)   AC_MSG_ERROR([Invalid value ${enableval} for --enable-swim-fd])
-     esac])
-
-check_swim_fd_status=fail
-if test "x${check_swim_fd}" = xauto -o "x${check_swim_fd}" = xyes ; then
-    PKG_CHECK_MODULES([SWIM_FD],[swim-fd],
-        [AC_DEFINE([HAVE_SWIM_FD], [1], [Defined to 1 if compiled with SWIM failure detection support])
-         LIBS="$SWIM_FD_LIBS $LIBS"
-         CPPFLAGS="$SWIM_FD_CFLAGS $CPPFLAGS"
-         CFLAGS="$SWIM_FD_CFLAGS $CFLAGS"
-         check_swim_fd_status=success], [check_swim_fd_status=fail])
-fi
-
-if test "x${check_swim_fd_status}" = xfail -a "x${check_swim_fd}" = xyes; then
-    AC_MSG_ERROR([SWIM failure detector requested but unable to be used. See config.log])
-fi
+    [  --enable-swim-fd        enable SWIM failure detection (default: disabled)],
+    [if test "x$enable_swim_fd" = "xyes" ; then
+        AC_DEFINE(USE_SWIM_FD, 1, Define to 1 if the SWIM failure detector should be used)
+    elif test "x$enableval" != "xno" ; then
+        AC_MSG_ERROR(bad value ${enable_swim_fd} for --enable-swim-fd)
+    fi],)
+
+AM_CONDITIONAL([USE_SWIM_FD], [test "x${enable_swim_fd}" = xyes])
 
 AC_CONFIG_FILES([Makefile maint/ssg.pc])
 AC_OUTPUT

src/Makefile.subdir

+src_libssg_la_SOURCES += \
+ src/def.h \
+ src/ssg.c

src/def.h

@@ -12,8 +12,8 @@
 #include <abt.h>
 #include <margo.h>
 
-#ifdef HAVE_SWIM_FD
-#include <swim.h>
+#if USE_SWIM_FD
+#include "swim-fd/swim-fd.h"
 #endif
 
 struct ssg
@@ -31,7 +31,7 @@ struct ssg
     ABT_mutex barrier_mutex;
     ABT_cond  barrier_cond;
     ABT_eventual barrier_eventual;
-#if HAVE_SWIM_FD
+#if USE_SWIM_FD
     swim_context_t *swim_ctx;
 #endif
 };

src/ssg.c

@@ -296,7 +296,9 @@ static ssg_t ssg_init_internal(margo_instance_id mid, hg_addr_t self_addr,
     s->barrier_mutex = ABT_MUTEX_NULL;
     s->barrier_cond  = ABT_COND_NULL;
     s->barrier_eventual = ABT_EVENTUAL_NULL;
+#if USE_SWIM_FD
     s->swim_ctx = NULL;
+#endif
 
     // lookup hg addr information for all group members
     hret = ssg_lookup(s);
@@ -306,7 +308,7 @@ static ssg_t ssg_init_internal(margo_instance_id mid, hg_addr_t self_addr,
         goto fini;
     }
 
-#ifdef HAVE_SWIM_FD
+#if USE_SWIM_FD
     // initialize swim failure detector
     if (s->rank != SSG_EXTERNAL_RANK)
     {
@@ -566,7 +568,7 @@ void ssg_finalize(ssg_t s)
 {
     if (s == SSG_NULL) return;
 
-#ifdef HAVE_SWIM_FD
+#if USE_SWIM_FD
     if(s->swim_ctx)
         swim_finalize(s->swim_ctx);
 #endif

src/swim-fd/Makefile.subdir

+src_libssg_la_SOURCES += \
+ src/swim-fd/swim-fd.h \
+ src/swim-fd/swim-fd.c

src/swim-fd/swim-fd.c

+/*
+ * (C) 2016 The University of Chicago
+ * 
+ * See COPYRIGHT in top-level directory.
+ */
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <assert.h>
+#include <time.h>
+
+#include "swim-fd.h"
+
+swim_context_t *swim_init(
+    margo_instance_id mid,
+    ssg_t swim_group,
+    int active)
+{
+    swim_context_t *swim_ctx;
+#if 0
+    int ret;
+
+    /* TODO: this may end up being identical across many processes => hotspots */
+    srand(time(NULL));
+
+    /* allocate structure for storing swim context */
+    swim_ctx = malloc(sizeof(*swim_ctx));
+    assert(swim_ctx);
+    memset(swim_ctx, 0, sizeof(*swim_ctx));
+
+    /* initialize swim state */
+    swim_ctx->mid = mid;
+    swim_ctx->hg_ctx = margo_get_context(mid);
+    swim_ctx->prot_pool = *margo_get_handler_pool(mid);
+    swim_ctx->group = swim_group;
+    swim_ctx->ping_target = SWIM_MEMBER_ID_UNKNOWN;
+
+    /* initialize membership context */
+    swim_init_membership_view(swim_ctx);
+
+    /* set protocol parameters */
+    swim_ctx->prot_period_len = SWIM_DEF_PROTOCOL_PERIOD_LEN;
+    swim_ctx->prot_susp_timeout = SWIM_DEF_SUSPECT_TIMEOUT;
+    swim_ctx->prot_subgroup_sz = SWIM_DEF_SUBGROUP_SIZE;
+
+    swim_register_ping_rpcs(margo_get_class(mid), swim_ctx);
+
+    if(active)
+    {
+        ret = ABT_thread_create(swim_ctx->prot_pool, swim_prot_ult, swim_ctx,
+            ABT_THREAD_ATTR_NULL, &(swim_ctx->prot_thread));
+        if(ret != ABT_SUCCESS)
+        {
+            fprintf(stderr, "Error: unable to create SWIM protocol ULT.\n");
+            return(NULL);
+        }
+    }
+#endif
+
+    return(swim_ctx);
+}
+
+void swim_finalize(swim_context_t *swim_ctx)
+{
+#if 0
+    /* set shutdown flag so ULTs know to start wrapping up */
+    swim_ctx->shutdown_flag = 1;
+
+    if(swim_ctx->prot_thread)
+    {
+        /* wait for the protocol ULT to terminate */
+        DEBUG_LOG("attempting to shutdown ult %p\n", swim_ctx, swim_ctx->prot_thread);
+        ABT_thread_join(swim_ctx->prot_thread);
+        ABT_thread_free(&(swim_ctx->prot_thread));
+        DEBUG_LOG("swim protocol shutdown complete\n", swim_ctx);
+    }
+
+    free(swim_ctx->membership_view);
+    free(swim_ctx);
+#endif
+
+    return;
+}
+

src/swim-fd/swim-fd.h

+/*
+ * (C) 2016 The University of Chicago
+ * 
+ * See COPYRIGHT in top-level directory.
+ */
+
+#ifndef __SWIM_FD_H
+#define __SWIM_FD_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <ssg.h>
+
+/* opaque swim context type */
+typedef struct swim_context swim_context_t;
+
+swim_context_t *swim_init(
+    margo_instance_id m_id,
+    ssg_t swim_group,
+    int active);
+
+void swim_finalize(
+    swim_context_t *swim_ctx);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* __SWIM_FD_H */
+