WIP - NNFDM implementation in AXL

CMakeLists.txt

@@ -39,6 +39,9 @@ MESSAGE(STATUS "ENABLE_IBM_BBAPI: ${ENABLE_IBM_BBAPI}")
 OPTION(ENABLE_CRAY_DW "Whether to enable Cray Datawarp support" OFF)
 MESSAGE(STATUS "ENABLE_CRAY_DW: ${ENABLE_CRAY_DW}")
 
+OPTION(ENABLE_HPE_NNFDM "Whether to enable HPE Data Mover support" ON)
+MESSAGE(STATUS "ENABLE_HPE_NNFDM: ${ENABLE_HPE_NNFDM}")
+
 OPTION(ENABLE_TESTS "Whether to build tests" ON)
 MESSAGE(STATUS "ENABLE_TESTS: ${ENABLE_TESTS}")
 
@@ -75,6 +78,19 @@ IF(ENABLE_CRAY_DW)
     LIST(APPEND AXL_EXTERNAL_STATIC_LIBS ${DATAWARP_LIBRARIES})
 ENDIF(ENABLE_CRAY_DW)
 
+## DataMover
+IF(ENABLE_HPE_NNFDM)
+  FIND_PACKAGE(NNFDM REQUIRED)
+  IF(NNFDM_FOUND)
+    SET(HAVE_NNFDM TRUE)
+    INCLUDE_DIRECTORIES(${NNFDM_INCLUDE_DIR})
+    MESSAGE(STATUS "NNFDM_INCLUDE_DIR: ${NNFDM_INCLUDE_DIR}")
+    MESSAGE(STATUS "NNFDM_LIBRARIES: ${NNFDM_LIBRARIES}")
+    LIST(APPEND AXL_EXTERNAL_LIBS ${NNFDM_LIBRARIES})
+    LIST(APPEND AXL_EXTERNAL_STATIC_LIBS ${NNFDM_LIBRARIES})
+  ENDIF(NNFDM_FOUND)
+ENDIF(ENABLE_HPE_NNFDM)
+
 ## IBM Burst Buffer API
 IF(ENABLE_IBM_BBAPI)
     FIND_PACKAGE(BBAPI)

cmake/axlConfig.cmake.in

@@ -19,6 +19,10 @@ if (@HAVE_BBAPI@)
   find_dependency(BBAPI REQUIRED)
 endif()
 
+if (@HAVE_NNFDM@)
+  find_dependency(nnfdm REQUIRED)
+endif()
+
 if (@MPI@)
   find_dependency(MPI REQUIRED)
 endif()

cmake/config.h.in

@@ -1,5 +1,6 @@
 // Machine Specific Libs
 #cmakedefine HAVE_PTHREADS
 #cmakedefine HAVE_DATAWARP
+#cmakedefine HAVE_NNFDM
 #cmakedefine HAVE_BBAPI
 #cmakedefine HAVE_BBAPI_FALLBACK

src/CMakeLists.txt

@@ -34,6 +34,10 @@ IF(HAVE_DATAWARP)
     LIST(APPEND libaxl_srcs axl_async_datawarp.c)
 ENDIF(HAVE_DATAWARP)
 
+IF(HAVE_NNFDM)
+    LIST(APPEND libaxl_srcs nnfdm.cpp)
+ENDIF(HAVE_NNFDM)
+
 IF(BUILD_SHARED_LIBS)
     # Default AXL library is withOUT MPI
     ADD_LIBRARY(axl_o OBJECT ${libaxl_srcs})

src/axl.c

@@ -38,6 +38,10 @@
 #include "axl_async_bbapi.h"
 #endif /* HAVE_BBAPI */
 
+#ifdef HAVE_NNFDM
+#include "nnfdm.h"
+#endif /* HAVE_NNFDM */
+
 #ifdef HAVE_DATAWARP
 #include "axl_async_datawarp.h"
 #endif /* HAVE_DATAWARP */
@@ -91,6 +95,7 @@ static unsigned int axl_kvtrees_count = 0;
 static int bbapi_is_loaded = 0;
 #endif
 
+
 /* Allocate a new kvtree and return the AXL ID for it.  If state_file is
  * specified, then populate the kvtree with it's data. */
 static int axl_alloc_id(const char* state_file)
@@ -205,10 +210,12 @@ static axl_xfer_t axl_detect_native_xfer(void)
      * DataWarp libraries.  In the real world, our supercomputer is only going
      * to have one of those libraries, so just use whatever we find at
      * build time. */
-#ifdef HAVE_BBAPI
+#if defined(HAVE_BBAPI)
     xtype = AXL_XFER_ASYNC_BBAPI;
-#elif HAVE_DATAWARP
+#elif defined(HAVE_DATAWARP)
     xtype = AXL_XFER_ASYNC_DW;
+#elif defined(HAVE_NNFDM)
+    xtype = AXL_XFER_ASYNC_NNFDM;
 #else
     xtype = AXL_XFER_SYNC;
 #endif
@@ -293,6 +300,11 @@ int AXL_Finalize (void)
     }
 #endif
 
+#ifdef HAVE_NNFDM
+    nnfdm_finalize();
+#endif
+
+
     /* decrement reference count and free data structures on last call */
     axl_init_count--;
     if (axl_init_count == 0) {
@@ -688,6 +700,15 @@ int AXL_Create(axl_xfer_t xtype, const char* name, const char* state_file)
 #endif /* HAVE_DATAWARP */
         break;
 
+    case AXL_XFER_ASYNC_NNFDM:
+#ifndef HAVE_NNFDM
+        AXL_ERR("NNFDM requested but not enabled during build");
+        rc = AXL_FAILURE;
+#else
+        nnfdm_init();
+#endif /* HAVE_NNFDM */
+        break;
+
     default:
         AXL_ERR("Unknown transfer type (%d)", (int) xtype);
         rc = AXL_FAILURE;
@@ -805,6 +826,16 @@ static int __AXL_Add (int id, const char* src, const char* dest)
     case AXL_XFER_SYNC:
         break;
 
+#ifdef HAVE_DATAWARP
+    case AXL_XFER_ASYNC_DW:
+        break;
+#endif /* HAVE_DATAWARP */
+
+#ifdef HAVE_NNFDM
+    case AXL_XFER_ASYNC_NNFDM:
+        break;
+#endif /* HAVE_NNFDM */
+
 #ifdef HAVE_PTHREADS
     case AXL_XFER_PTHREAD:
         break;
@@ -821,11 +852,6 @@ static int __AXL_Add (int id, const char* src, const char* dest)
         break;
 #endif /* HAVE_BBAPI */
 
-#ifdef HAVE_DATAWARP
-    case AXL_XFER_ASYNC_DW:
-        break;
-#endif /* HAVE_DATAWARP */
-
     default:
         AXL_ERR("Unknown transfer type (%d)", (int) xtype);
         rc = AXL_FAILURE;
@@ -1146,6 +1172,17 @@ int __AXL_Dispatch (int id, int resume)
         break;
 #endif /* HAVE_DATAWARP */
 
+#ifdef HAVE_NNFDM
+    case AXL_XFER_ASYNC_NNFDM:
+        if (resume) {
+            AXL_ERR("AXL_Resume() isn't supported yet for NNFDM");
+            rc = AXL_FAILURE;
+            break;
+        }
+        rc = nnfdm_start(id);
+        break;
+#endif /* HAVE_NNFDM */
+
     default:
         AXL_ERR("Unknown transfer type (%d)", (int) xtype);
         rc = AXL_FAILURE;
@@ -1297,6 +1334,12 @@ int AXL_Test (int id)
         break;
 #endif /* HAVE_DATAWARP */
 
+#ifdef HAVE_NNFDM
+    case AXL_XFER_ASYNC_NNFDM:
+        rc = nnfdm_test(id);
+        break;
+#endif /* HAVE_NNFDM */
+
     default:
         AXL_ERR("Unknown transfer type (%d)", (int) xtype);
         rc = AXL_FAILURE;
@@ -1367,6 +1410,12 @@ int AXL_Wait (int id)
         break;
 #endif /* HAVE_DATAWARP */
 
+#ifdef HAVE_NNFDM
+    case AXL_XFER_ASYNC_NNFDM:
+        rc = nnfdm_wait(id);
+        break;
+#endif /* HAVE_NNDFM */
+
     default:
         AXL_ERR("Unknown transfer type (%d)", (int) xtype);
         rc = AXL_FAILURE;
@@ -1465,6 +1514,12 @@ int AXL_Cancel (int id)
         break;
 #endif
 
+#ifdef HAVE_NNFDM
+    case AXL_XFER_ASYNC_NNFDM:
+        rc = nnfdm_cancel(id);
+        break;
+#endif
+
     default:
         AXL_ERR("Unknown transfer type (%d)", (int) xtype);
         rc = AXL_FAILURE;

src/axl.h

@@ -42,6 +42,7 @@ typedef enum {
     AXL_XFER_SYNC,          /* synchronous copy */
     AXL_XFER_ASYNC_DAEMON,  /* async daemon process (not used, but kept to maintain enum values) */
     AXL_XFER_ASYNC_DW,      /* Cray Datawarp */
+    AXL_XFER_ASYNC_NNFDM,   /* HPE DataMover */
     AXL_XFER_ASYNC_BBAPI,   /* IBM Burst Buffer API */
     AXL_XFER_NATIVE,        /* Autodetect and use the native API (BBAPI, DW,
                              * etc) for this node type.  It may or may not

src/axl_err.c

@@ -10,64 +10,80 @@
 /* gethostname */
 #include <unistd.h>
 
+#include <string.h>
+#include <errno.h>
+#include <sys/time.h>
+#include <time.h>
+
 /* axl version */
 #include "axl.h"
 
 /* current debug level for AXL library,
  * set in AXL_Init used in axl_dbg */
 int axl_debug;
 
-/* print message to stdout if axl_debug is set and it is >= level */
-void axl_dbg(int level, const char* fmt, ...)
+static void axl_print_date_and_hostinfo(const char* title) 
 {
-    /* get my hostname */
     char hostname[256];
     if (gethostname(hostname, sizeof(hostname)) != 0) {
         /* TODO: error! */
     }
-
-    va_list argp;
+
+    struct timeval curTime;
+    gettimeofday(&curTime, NULL);
+    int milli = curTime.tv_usec / 1000;
+
+    time_t rawtime;
+    struct tm * timeinfo;
+    char buffer[80];
+
+    time(&rawtime);
+    timeinfo = localtime(&rawtime);
+
+    strftime(buffer, 80, "%Y-%m-%d %H:%M:%S", timeinfo);
+
+    char currentTime[84] = "";
+    sprintf(currentTime, "%s:%d", buffer, milli);
+
+    fprintf(stderr, "%s %s:%d %s:", currentTime, title, getpid(), hostname);
+}
+
+/* print message to stderr if axl_debug is set and it is >= level */
+void axl_dbg(int level, const char* fmt, ...)
+{
     if (level == 0 || (axl_debug > 0 && axl_debug >= level)) {
-        fprintf(stdout, "AXL %s: %s: ", AXL_VERSION, hostname);
+        axl_print_date_and_hostinfo("AXL");
+
+        va_list argp;
         va_start(argp, fmt);
-        vfprintf(stdout, fmt, argp);
+        vfprintf(stderr, fmt, argp);
         va_end(argp);
-        fprintf(stdout, "\n");
+        fprintf(stderr, "\n");
     }
 }
 
-/* print error message to stdout */
+/* print error message to stderr */
 void axl_err(const char* fmt, ...)
 {
-    /* get my hostname */
-    char hostname[256];
-    if (gethostname(hostname, sizeof(hostname)) != 0) {
-        /* TODO: error! */
-    }
+    axl_print_date_and_hostinfo("AXL ERROR");
 
     va_list argp;
-    fprintf(stdout, "AXL %s ERROR: %s: ", AXL_VERSION, hostname);
     va_start(argp, fmt);
-    vfprintf(stdout, fmt, argp);
+    vfprintf(stderr, fmt, argp);
     va_end(argp);
-    fprintf(stdout, "\n");
+    fprintf(stderr, "\n");
 }
 
 /* print abort message and kill run */
 void axl_abort(int rc, const char* fmt, ...)
 {
-    /* get my hostname */
-    char hostname[256];
-    if (gethostname(hostname, sizeof(hostname)) != 0) {
-        /* TODO: error! */
-    }
-
+    axl_print_date_and_hostinfo("AXL ABORT");
+
     va_list argp;
-    fprintf(stderr, "AXL %s ABORT: %s: ", AXL_VERSION, hostname);
     va_start(argp, fmt);
     vfprintf(stderr, fmt, argp);
     va_end(argp);
     fprintf(stderr, "\n");
-  
+
     exit(rc);
 }

src/axl_internal.h

@@ -54,6 +54,8 @@ extern int axl_rank;
 #define AXL_KEY_FILE_STATUS   ("STATUS")
 #define AXL_KEY_FILE_CRC      ("CRC")
 #define AXL_KEY_STATE_FILE    ("STATE_FILE")
+#define AXL_KEY_FILE_SESSION_UID ("NNFDM_Session_ID")
+
 
 /* TRANSFER STATUS */
 #define AXL_STATUS_SOURCE (1)
@@ -63,8 +65,9 @@ extern int axl_rank;
 
 /* attaches function name, file name, and line number to error messages
  * https://gcc.gnu.org/onlinedocs/cpp/Variadic-Macros.html */
-#define AXL_ERR(format, ...) axl_err(format " @ %s %s:%d", ##__VA_ARGS__, __func__, __FILE__, __LINE__)
-#define AXL_DBG(level, format, ...) axl_dbg(level, format " @ %s %s:%d", ##__VA_ARGS__, __func__, __FILE__, __LINE__)
+#define AXL_ERR(format, ...)        axl_err(       " @ %s:%d " format, __func__, __LINE__, ##__VA_ARGS__)
+#define AXL_DBG(level, format, ...) axl_dbg(level, " @ %s:%d " format, __func__, __LINE__, ##__VA_ARGS__)
+#define AXL_ABORT(rc, format, ...)  axl_abort(rc,  " @ %s:%d " format, __func__, __LINE__, ##__VA_ARGS__)
 
 /*
 =========================================