diff --git a/CMakeLists.txt b/CMakeLists.txt index 07b54f0..036040f 100644 --- a/CMakeLists.txt +++ b/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) diff --git a/cmake/axlConfig.cmake.in b/cmake/axlConfig.cmake.in index 7f6d351..b141ea7 100644 --- a/cmake/axlConfig.cmake.in +++ b/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() diff --git a/cmake/config.h.in b/cmake/config.h.in index cb1da59..33f015a 100644 --- a/cmake/config.h.in +++ b/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 diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 3dcc9d0..bec0ae0 100644 --- a/src/CMakeLists.txt +++ b/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}) diff --git a/src/axl.c b/src/axl.c index f674e74..93a5d25 100644 --- a/src/axl.c +++ b/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; diff --git a/src/axl.h b/src/axl.h index a83ef48..717d93b 100644 --- a/src/axl.h +++ b/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 diff --git a/src/axl_err.c b/src/axl_err.c index 493a67b..f1dfa58 100644 --- a/src/axl_err.c +++ b/src/axl_err.c @@ -10,6 +10,11 @@ /* gethostname */ #include +#include +#include +#include +#include + /* axl version */ #include "axl.h" @@ -17,57 +22,68 @@ * 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); } diff --git a/src/axl_internal.h b/src/axl_internal.h index 10d97c6..acf2080 100644 --- a/src/axl_internal.h +++ b/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__) /* ========================================= diff --git a/src/axl_mpi.c b/src/axl_mpi.c index 07d9719..13f0032 100644 --- a/src/axl_mpi.c +++ b/src/axl_mpi.c @@ -144,59 +144,6 @@ int AXL_Dispatch_comm ( int id, /**< [IN] - transfer hander ID returned from AXL_Create */ MPI_Comm comm) /**< [IN] - communicator used for coordination and flow control */ { -#if 0 - /* lookup transfer info for the given id */ - kvtree* file_list = NULL; - axl_xfer_t xtype = AXL_XFER_NULL; - axl_xfer_state_t xstate = AXL_XFER_STATE_NULL; - if (axl_get_info(id, &file_list, &xtype, &xstate) != AXL_SUCCESS) { - AXL_ERR("Could not find transfer info for UID %d", id); - return AXL_FAILURE; - } - - /* check that handle is in correct state to dispatch */ - if (xstate != AXL_XFER_STATE_CREATED) { - AXL_ERR("Invalid state to dispatch UID %d", id); - return AXL_FAILURE; - } - kvtree_util_set_int(file_list, AXL_KEY_STATE, (int)AXL_XFER_STATE_DISPATCHED); -#endif - -#if 0 - /* create destination directories for each file */ - if (axl_make_directories) { - /* count number of files we have */ - kvtree* file_list = kvtree_get_kv_int(axl_file_lists, AXL_KEY_HANDLE_UID, id); - kvtree* files_hash = kvtree_get(file_list, AXL_KEY_FILES); - int num_files = kvtree_size(files_hash); - - /* allocate pointer for each one */ - const char** files = (const char**) AXL_MALLOC(num_files * sizeof(char*)); - - /* set pointer to each file */ - int i; - char* dest; - kvtree_elem* elem; - while ((elem = axl_get_next_path(id, elem, NULL, &dest))) { - files[i] = dest; - i++; - } - - /* create directories */ - axl_create_dirs(num_files, files, comm); - - /* free list of files */ - axl_free2(&files); - } - - /* TODO: this is hacky */ - /* delegate remaining work to regular dispatch, - * but disable mkdir since we already did that */ - int make_dir = axl_make_directories; - axl_make_directories = 0; - int rc = AXL_Dispatch(id); - axl_make_directories = make_dir; -#endif /* delegate remaining work to regular dispatch */ int rc = AXL_Dispatch(id); diff --git a/src/nnfdm.cpp b/src/nnfdm.cpp new file mode 100644 index 0000000..60180e5 --- /dev/null +++ b/src/nnfdm.cpp @@ -0,0 +1,434 @@ +#include +#include +#include + +#include + +#include "nnfdm/client.h" + +extern "C" { +#include "nnfdm.h" +#include "axl_internal.h" +#include "kvtree.h" +} + +namespace nnfdm = near_node_flash::data_movement; + +namespace { + nnfdm::DataMoverClient* nnfdm_client{nullptr}; + nnfdm::Workflow* nnfdm_workflow{nullptr}; + std::string axl_nnfdm_dcp_arguments; + + std::string print_status(nnfdm::StatusResponse& status_response) + { + std::stringstream ss; + + ss << "Offload Command Status: " << "{" << status_response.state() << "/"; + switch (status_response.state()) { + case nnfdm::StatusResponse::State::STATE_PENDING: ss << "STATE_PENDING"; break; + case nnfdm::StatusResponse::State::STATE_STARTING: ss << "STATE_STARTING"; break; + case nnfdm::StatusResponse::State::STATE_RUNNING: ss << "STATE_RUNNING"; break; + case nnfdm::StatusResponse::State::STATE_COMPLETED: ss <<"STATE_COMPLETED"; break; + case nnfdm::StatusResponse::State::STATE_CANCELLING: ss <<"STATE_CANCELLING"; break; + case nnfdm::StatusResponse::State::STATE_UNKNOWN: ss << "STATE_UNKNOWN"; break; + default: ss << "STATE_???"; break; + } + ss << "}, "; + + ss << "{" << status_response.status() << "/"; + switch (status_response.status()) { + case nnfdm::StatusResponse::Status::STATUS_INVALID: ss << "STATUS_INVALID"; break; + case nnfdm::StatusResponse::Status::STATUS_NOT_FOUND: ss << "STATUS_NOT_FOUND"; break; + case nnfdm::StatusResponse::Status::STATUS_SUCCESS: ss << "STATUS_SUCCESS"; break; + case nnfdm::StatusResponse::Status::STATUS_FAILED: ss << "STATUS_FAILED"; break; + case nnfdm::StatusResponse::Status::STATUS_CANCELLED: ss << "STATUS_CANCELLED"; break; + case nnfdm::StatusResponse::Status::STATUS_UNKNOWN: ss << "STATUS_UNKNOWN"; break; + default: ss << "STATE_???"; break; + } + ss << "}" << std::endl; + + nnfdm::CommandStatus cmd = status_response.commandStatus(); + ss << " Offload Command Status:" << std::endl; + ss << " Command: " << cmd.command << std::endl; + ss << " Progress: " << cmd.progress << "%" << std::endl; + ss << " ElapsedTime: " << cmd.elapsedTime << std::endl; + ss << " LastMessage: " << cmd.lastMessage << std::endl; + ss << " LastMessageTime: " << cmd.lastMessageTime << std::endl; + ss << " Offload StartTime: " << status_response.startTime() << std::endl; + ss << " Offload EndTime: " << status_response.endTime() << std::endl; + ss << " Offload Message: " << status_response.message() << std::endl; + + return ss.str(); + } + + int nnfdm_stat(const char* fname, const char* uid, int64_t max_seconds_to_wait) + { + AXL_DBG(1, "%s", fname); + int rval = 0; + int retry_count = 0; + nnfdm::StatusResponse status_response; + nnfdm::RPCStatus rpc_status{ nnfdm_client->Status(*nnfdm_workflow, nnfdm::StatusRequest{std::string{uid}, max_seconds_to_wait}, &status_response) }; + + if (!rpc_status.ok()) { + AXL_ABORT( -1 + , "NNFDM Status RPC FAILED %d: %s @ %s:%d" + , rpc_status.error_code() + , rpc_status.error_message().c_str() + , __FILE__ + , __LINE__); + /*NOTREACHED*/ + } + + switch (status_response.state()) { + case nnfdm::StatusResponse::State::STATE_PENDING: + case nnfdm::StatusResponse::State::STATE_STARTING: + case nnfdm::StatusResponse::State::STATE_RUNNING: + rval = AXL_STATUS_INPROG; + break; + case nnfdm::StatusResponse::State::STATE_COMPLETED: + if (status_response.status() == nnfdm::StatusResponse::Status::STATUS_SUCCESS) { + AXL_DBG(1, "Offload Complete(%s)", fname ); + rval = AXL_STATUS_DEST; + } + else { + rval = AXL_STATUS_ERROR; + AXL_ERR( "NNFDM Offload Status UNSUCCESSFUL: %d\n%s" + , status_response.status() + , print_status(status_response).c_str()); + } + break; + default: + AXL_ABORT( -1 + , "NNFDM Offload State STATE UNKNOWN: %d\n%s" + , status_response.status() + , print_status(status_response).c_str()); + break; + } + + return rval; + } +} // End of empty namepace + +extern "C" { + +void nnfdm_init() +{ + AXL_DBG(1, "%p", nnfdm_client); + if (nnfdm_client == nullptr) { + const std::string socket_name{"unix:///var/run/nnf-dm.sock"}; + const std::string workflow_name{getenv("DW_WORKFLOW_NAME")}; + const std::string workflow_namespace{getenv("DW_WORKFLOW_NAMESPACE")}; + + nnfdm_client = new nnfdm::DataMoverClient{ socket_name }; + if (nnfdm_client == nullptr) { + AXL_ABORT(-1 + , "NNFDM init: Failed to create data movement client " + "instance for %s@ %s:%d" + , socket_name.c_str() + , __FILE__ + , __LINE__); + } + + nnfdm_workflow = new nnfdm::Workflow{workflow_name, workflow_namespace}; + if (nnfdm_workflow == nullptr) { + AXL_ABORT( -1 + , "nnfdm_init: Failed to create data movement workflow " + "instance for %s/%s@ %s:%d" + , workflow_name.c_str() + , workflow_namespace.c_str() + , __FILE__ + , __LINE__); + } + + char* evar = getenv("AXL_NNFDM_DCP_ARGUMENTS"); + if ( evar != NULL ) { + axl_nnfdm_dcp_arguments = evar; + } + } +} + +void nnfdm_finalize() +{ + AXL_DBG(1, "%p", nnfdm_client); + if (nnfdm_client) { + delete nnfdm_workflow; + delete nnfdm_client; + + nnfdm_workflow = nullptr; + nnfdm_client = nullptr; + } +} + +int nnfdm_start(int id) +{ + AXL_DBG(1, "%d", id); + int rc = AXL_SUCCESS; + + /* Record that we started transfer of this file list */ + kvtree* file_list = axl_kvtrees[id]; + kvtree_util_set_int(file_list, AXL_KEY_STATUS, AXL_STATUS_INPROG); + + kvtree_elem* elem; + kvtree* files = kvtree_get(file_list, AXL_KEY_FILES); + for ( kvtree_elem* elem = kvtree_elem_first(files); + elem != NULL; + elem = kvtree_elem_next(elem)) + { + char* src_filename = kvtree_elem_key(elem); + kvtree* elem_hash = kvtree_elem_hash(elem); + const std::string dm_profile{"scr"}; + + /* get the destination for this file */ + char* dst_filename; + kvtree_util_get_str(elem_hash, AXL_KEY_FILE_DEST, &dst_filename); + + AXL_DBG(1, "nnfdm::CreateRequest(src=%s, dst=%s)" , src_filename, dst_filename); + nnfdm::CreateRequest create_request( + std::string{src_filename} // Source file or directory + , std::string{dst_filename} // Destination file or directory + , false // If True, the data movement command runs `/bin/true` rather than perform actual data movement + , "" // mpirun command line options + , axl_nnfdm_dcp_arguments // Extra options to pass to `dcp` if present in the Data Movement command. + , false // If true, enable server-side logging of stdout when the command is successful. Failures output is always logged. + , true // If true, store stdout in DataMovementStatusResponse.Message when the command is successful. Failure output is always contained in the message. + , -1 // The number of slots specified in the MPI hostfile. A value of 0 disables the use of slots in the hostfile. -1 will defer to the server side configuration. + , -1 // The number of max_slots specified in the MPI hostfile. A value of 0 disables the use of max_slots in the hostfile. -1 will defer to the server side configuration. + , dm_profile // Data movement profile. Empty will default to the default profile. + ); + + AXL_DBG(1, "create_request(%s, %s, %s, ..., %s)", src_filename, dst_filename, axl_nnfdm_dcp_arguments.c_str(), dm_profile.c_str()); + + nnfdm::CreateResponse create_response; + nnfdm::RPCStatus rpc_status = nnfdm_client->Create( *nnfdm_workflow, create_request, &create_response); + if (!rpc_status.ok()) { + AXL_ERR("%s: NNFDM Create(%s, %s) RPC failed with error %d (%s)" + , __PRETTY_FUNCTION__, src_filename, dst_filename + , rpc_status.error_code(), rpc_status.error_message().c_str()); + kvtree_util_set_int(elem_hash, AXL_KEY_FILE_STATUS, AXL_STATUS_ERROR); + rc = AXL_FAILURE; + } + + if (create_response.status() == nnfdm::CreateResponse::Status::STATUS_SUCCESS) { + /* record that the file is in progress */ + kvtree_util_set_int(elem_hash, AXL_KEY_FILE_STATUS, AXL_STATUS_INPROG); + kvtree_util_set_str(elem_hash, AXL_KEY_FILE_SESSION_UID, create_response.uid().c_str()); + } + else { + AXL_ERR("%s: NNFDM Create(%s, %s) Response Status is not SUCCESS: response status: %d - %s" + , __PRETTY_FUNCTION__, src_filename, dst_filename, create_response.status(), create_response.message().c_str() ); + kvtree_util_set_int(elem_hash, AXL_KEY_FILE_STATUS, AXL_STATUS_ERROR); + rc = AXL_FAILURE; + } + } + + return rc; +} + +int nnfdm_test(int id) +{ + AXL_DBG(1, "%d", id); + kvtree* file_list = axl_kvtrees[id]; + kvtree* files = kvtree_get(file_list, AXL_KEY_FILES); + + /* iterate/wait over in-progress files */ + for (kvtree_elem* elem = kvtree_elem_first(files); elem != NULL; elem = kvtree_elem_next(elem)) { + char* src_filename = kvtree_elem_key(elem); + int status; + char* uid; + + kvtree* elem_hash = kvtree_elem_hash(elem); + kvtree_util_get_int(elem_hash, AXL_KEY_FILE_STATUS, &status); + + if (status == AXL_STATUS_DEST) { + continue; /* This one is done */ + } + + kvtree_util_get_str(elem_hash, AXL_KEY_FILE_SESSION_UID, &uid); + status = nnfdm_stat(src_filename, uid, 1); + + if (status != AXL_STATUS_DEST) { + return AXL_FAILURE; /* At least one file is not done */ + } + + kvtree_util_set_int(elem_hash, AXL_KEY_FILE_STATUS, status); + } + + return AXL_SUCCESS; +} + +int nnfdm_cancel(int id) +{ + AXL_DBG(1, "%d", id); + kvtree* file_list = axl_kvtrees[id]; + kvtree* files = kvtree_get(file_list, AXL_KEY_FILES); + + /* iterate/wait over in-progress files */ + for ( kvtree_elem* elem = kvtree_elem_first(files); + elem != NULL; + elem = kvtree_elem_next(elem)) + { + int status; + char* uid; + char* src_filename = kvtree_elem_key(elem); + kvtree* elem_hash = kvtree_elem_hash(elem); + kvtree_util_get_int(elem_hash, AXL_KEY_FILE_STATUS, &status); + + if (status == AXL_STATUS_DEST) { + continue; /* This one is done */ + } + + kvtree_util_get_str(elem_hash, AXL_KEY_FILE_SESSION_UID, &uid); + + nnfdm::CancelRequest cancel_request(uid); + nnfdm::CancelResponse cancel_response; + nnfdm::RPCStatus rpc_status = nnfdm_client->Cancel( *nnfdm_workflow + , cancel_request + , &cancel_response); + if (!rpc_status.ok()) { + AXL_ABORT(-1, + "NNFDM Cancel(uid=%s, file=%s) failed with error %d (%s) @ %s:%d" + , uid + , src_filename + , rpc_status.error_code() + , rpc_status.error_message().c_str() + , __FILE__ + , __LINE__ + ); + /*NOTREACHED*/ + } + + switch (cancel_response.status()) { + case nnfdm::CancelResponse::STATUS_NOT_FOUND: + // + // Assume that unfound uids simply completed previously + // + AXL_DBG(1, "NNFDM Cancel(uid=%s, file=%s) NOTFOUND - IGNORING", uid, src_filename); + continue; + case nnfdm::CancelResponse::STATUS_SUCCESS: + AXL_DBG(1, "NNFDM Cancel(uid=%s, file=%s) Canceled @ %s:%d", uid, src_filename); + break; + default: + AXL_ABORT( -1 + , "NNFDM Cancel(uid=%s, file=%s) " + "Failed with error %d - IGNORING @ %s:%d" + , uid + , src_filename + , cancel_response.status() + , __FILE__ + , __LINE__); + /*NOTEACHED*/ + return 1; + } + + // Now delete the associated uid + nnfdm::DeleteRequest delete_request(std::string{uid}); + nnfdm::DeleteResponse deleteResponse; + rpc_status = nnfdm_client->Delete( *nnfdm_workflow + , delete_request + , &deleteResponse); + + if (!rpc_status.ok()) { + AXL_ABORT( -1 + , "NNFDM Delete(uid=%s, file=%s) RPC FAILED: " + "%d (%s) @ %s:%d" + , uid + , src_filename + , rpc_status.error_code() + , rpc_status.error_message().c_str() + , __FILE__ + , __LINE__); + /*NOTEACHED*/ + } + + /* Delete the request */ + switch (deleteResponse.status()) { + case nnfdm::DeleteResponse::STATUS_SUCCESS: + break; + default: + AXL_ERR( "Warning: NNFDM Offload Delete(%s) UNSUCCESSFUL: %d (%s) @ %s:%d" + , src_filename + , deleteResponse.status() + , deleteResponse.message().c_str() + , __FILE__ + , __LINE__); + break; + } + + kvtree_util_set_int(elem_hash, AXL_KEY_FILE_STATUS, AXL_STATUS_DEST); + } + + return AXL_SUCCESS; +} + +int nnfdm_wait(int id) +{ + AXL_DBG(1, "id=%d", id); + const int64_t max_seconds_to_wait{1}; // wait 1 second + kvtree* file_list = axl_kvtrees[id]; + kvtree* files = kvtree_get(file_list, AXL_KEY_FILES); + + /* iterate/wait over in-progress files */ + for (kvtree_elem* elem = kvtree_elem_first(files); elem != NULL; elem = kvtree_elem_next(elem)) { + char* src_filename = kvtree_elem_key(elem); + kvtree* elem_hash = kvtree_elem_hash(elem); + + char* uid; + kvtree_util_get_str(elem_hash, AXL_KEY_FILE_SESSION_UID, &uid); + + int status; + do { + status = nnfdm_stat(src_filename, uid, max_seconds_to_wait); + } while (status == AXL_STATUS_INPROG); + + nnfdm::DeleteRequest delete_request(std::string{uid}); + nnfdm::DeleteResponse deleteResponse; + + nnfdm::RPCStatus rpc_status = nnfdm_client->Delete(*nnfdm_workflow, delete_request, &deleteResponse); + if (!rpc_status.ok()) { + AXL_ABORT(-1, "NNFDM Delete RPC FAILED: %d (%s)", rpc_status.error_code(), rpc_status.error_message().c_str()); + /*NOTEACHED*/ + } + + /* Delete the request */ + switch (deleteResponse.status()) { + case nnfdm::DeleteResponse::STATUS_SUCCESS: + break; + default: + AXL_ABORT(-1, + "NNFDM Offload Delete(%s) UNSUCCESSFUL: %d (%s)", + src_filename, deleteResponse.status(), deleteResponse.message().c_str()); + return 1; + } + + AXL_DBG(1, "%s Marked Done", src_filename); + kvtree_util_set_int(elem_hash, AXL_KEY_FILE_STATUS, status); + } + + /* iterate over files */ + for (kvtree_elem* elem = kvtree_elem_first(files); elem != NULL; elem = kvtree_elem_next(elem)) { + char* src_filename = kvtree_elem_key(elem); + kvtree* elem_hash = kvtree_elem_hash(elem); + + int status; + kvtree_util_get_int(elem_hash, AXL_KEY_FILE_STATUS, &status); + + switch (status) { + case AXL_STATUS_DEST: + AXL_DBG(1, "%s Is marked as done", src_filename); + break; + case AXL_STATUS_SOURCE: + case AXL_STATUS_ERROR: + default: + AXL_ABORT(-1, + "Wait operation called on file with invalid status @ %s:%d", + __FILE__, __LINE__ + ); + } + } + + /* record transfer complete */ + AXL_DBG(1, "File List is done"); + kvtree_util_set_int(file_list, AXL_KEY_STATUS, AXL_STATUS_DEST); + return AXL_SUCCESS; +} +} diff --git a/src/nnfdm.h b/src/nnfdm.h new file mode 100644 index 0000000..dbb1968 --- /dev/null +++ b/src/nnfdm.h @@ -0,0 +1,19 @@ +#ifndef AXL_ASYNC_NNFDM_H +#define AXL_ASYNC_NNFDM_H + +#ifdef __cplusplus +extern "C" { +#endif + +void nnfdm_init(); +void nnfdm_finalize(); +int nnfdm_start(int id); +int nnfdm_test(int id); +int nnfdm_wait(int id); +int nnfdm_cancel(int id); + +#ifdef __cplusplus +}; +#endif + +#endif //AXL_ASYNC_NNFDM_H diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 79cce0d..3c3bc36 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -22,6 +22,12 @@ ADD_EXECUTABLE(test_config test_config.c) TARGET_LINK_LIBRARIES(axl_cp ${axl_lib}) TARGET_LINK_LIBRARIES(test_config ${axl_lib}) +IF(HAVE_NNFDM) + ADD_EXECUTABLE(axl_test_nnfdm axl_test_nnfdm.cpp) + TARGET_LINK_LIBRARIES(axl_test_nnfdm ${axl_lib}) + ADD_TEST(axl_test_nnfdm axl_test_nnfdm) +ENDIF(HAVE_NNFDM) + ################ # Add tests to ctest ################ diff --git a/test/axl_test_nnfdm.cpp b/test/axl_test_nnfdm.cpp new file mode 100644 index 0000000..36fd231 --- /dev/null +++ b/test/axl_test_nnfdm.cpp @@ -0,0 +1,173 @@ +#include +#include +#include +#include + +#include "nnfdm/client.h" + +using namespace near_node_flash::data_movement; + +int main(int argc, char** argv) +{ + std::string uid; + + DataMoverClient client("unix:///var/run/nnf-dm.sock"); + + { + // Retrieve the version information + + VersionResponse versionResponse; + + RPCStatus status = client.Version(&versionResponse); + if (!status.ok()) { + std::cerr << "Version RPC FAILED (" << status.error_code() << "): " << status.error_message() << std::endl; + return 1; + } + + std::cout << "Data Movement Version: " << versionResponse.version() << std::endl; + std::cout << "Supported API Versions:" << std::endl; + for (auto version : versionResponse.apiversions()) { + std::cout << "\t" << version << std::endl; + } + + return 0; + } + + const char* workflow_name{getenv("DW_WORKFLOW_NAME")}; + const char* workflow_namespace{getenv("DW_WORKFLOW_NAMESPACE")}; + + if (workflow_name == nullptr) { + std::cerr << "DW_WORKFLOW_NAME environment variableis not set, make sure you are running within an allocation" << std::endl; + } + + if (workflow_namespace == nullptr) { + std::cerr << "DW_WORKFLOW_NAMESPACE environment variableis not set, make sure you are running within an allocation" << std::endl; + } + + Workflow workflow(workflow_name, workflow_namespace); + + { + // Create an offload request + CreateRequest create_request( "/mnt/nnf/854e2467-968a-4f40-aa8c-4ac53cb86866-0/foo" + , "/p/lslide/martymcf/bar" + , false // If True, the data movement command runs `/bin/true` rather than perform actual data movement + , "" // mpirun command line options + , "" // Extra options to pass to `dcp` if present in the Data Movement command. + , false // If true, enable server-side logging of stdout when the command is successful. Failures output is always logged. + , true // If true, store stdout in DataMovementStatusResponse.Message when the command is successful. Failure output is always contained in the message. + , -1 // The number of slots specified in the MPI hostfile. A value of 0 disables the use of slots in the hostfile. -1 will defer to the server side configuration. + , -1 // The number of max_slots specified in the MPI hostfile. A value of 0 disables the use of max_slots in the hostfile. -1 will defer to the server side configuration. + , "" // Data movement profile. Empty will default to the default profile. + ); + CreateResponse create_response; + + RPCStatus status = client.Create(workflow, create_request, &create_response); + if (!status.ok()) { + std::cout << "Create RPC FAILED (" << status.error_code() << "): " << status.error_message() << std::endl; + return 1; + } + + switch (create_response.status()) { + case CreateResponse::Status::STATUS_SUCCESS: + uid = create_response.uid(); + std::cout << "Offload Created: UID: " << create_response.uid() << std::endl; + break; + default: + std::cout << "Offload Create FAILED: " << create_response.status() << ": " << create_response.message() << std::endl; + return 1; + } + } + + { + ListRequest listRequest; + ListResponse listResponse; + + RPCStatus status = client.List(workflow, listRequest, &listResponse); + if (!status.ok()) { + std::cout << "List RPC FAILED (" << status.error_code() << "): " << status.error_message() << std::endl; + return 1; + } + + std::cout << "Offload List: " << listResponse.uids().size() << std::endl; + for (auto uid : listResponse.uids()) { + std::cout << "\t" << uid << std::endl; + } + } + + { + StatusRequest statusRequest(uid, 0); + StatusResponse statusResponse; + +RequestStatus: + RPCStatus status = client.Status(workflow, statusRequest, &statusResponse); + if (!status.ok()) { + std::cout << "Status RPC FAILED (" << status.error_code() << "): " << status.error_message() << std::endl; + return 1; + } + + switch (statusResponse.state()) { + case StatusResponse::STATE_PENDING: + case StatusResponse::STATE_STARTING: + case StatusResponse::STATE_RUNNING: + std::cout << "Offload State Pending/Starting/Running" << std::endl; + goto RequestStatus; + case StatusResponse::STATE_COMPLETED: + std::cout << "Offload State Completed" << std::endl; + break; + default: + std::cout << "Offload State STATE UNKNOWN: " << statusResponse.state() << " Status: " << statusResponse.status() << std::endl; + return 1; + } + + switch (statusResponse.status()) { + case StatusResponse::STATUS_SUCCESS: + std::cout << "Offload Status Successful" << std::endl; + break; + default: + std::cout << "Offload Status UNSUCCESSFUL: " << statusResponse.status() << " " << statusResponse.message() << std::endl; + return 1; + } + } + + { + CancelRequest cancelRequest(uid); + CancelResponse cancelResponse; + + RPCStatus status = client.Cancel(workflow, cancelRequest, &cancelResponse); + if (!status.ok()) { + std::cout << "Cancel RPC FAILED (" << status.error_code() << "): " << status.error_message() << std::endl; + return 1; + } + + switch (cancelResponse.status()) { + case CancelResponse::STATUS_SUCCESS: + std::cout << "Offload Cancel Successful" << std::endl; + break; + default: + std::cout << "Offload Cancel UNSUCCESSFUL: " << cancelResponse.status() << " " << cancelResponse.message() << std::endl; + return 1; + } + } + + { + DeleteRequest deleteRequest(uid); + DeleteResponse deleteResponse; + + RPCStatus status = client.Delete(workflow, deleteRequest, &deleteResponse); + if (!status.ok()) { + std::cout << "Delete RPC FAILED (" << status.error_code() << "): " << status.error_message() << std::endl; + return 1; + } + + switch (deleteResponse.status()) { + case DeleteResponse::STATUS_SUCCESS: + std::cout << "Offload Delete Successful" << std::endl; + break; + default: + std::cout << "Offload Delete UNSUCCESSFUL: " << deleteResponse.status() << " " << deleteResponse.message() << std::endl; + return 1; + } + } + + return 0; +} diff --git a/test/nnf/axl.containers/Dockerfile b/test/nnf/axl.containers/Dockerfile new file mode 100644 index 0000000..71f6a02 --- /dev/null +++ b/test/nnf/axl.containers/Dockerfile @@ -0,0 +1,38 @@ +# +# This image contains the NNF MPI File Utils Image that is also used for Data Movement. +# On top of that, we place a simple mpi_hello_world program that will take two arguments +# as input: +# 1) The name of the rabbit-local directory ($DW_JOB_storage) +# 2) The name of the global/shared Lustre directory +# +# This image is loaded +# +# IMPORTANT IMAGE BUILD NOTES: +# +# This image must be built on an x86 LC machine that is configure with SLURM. This image will fail +# to build on flux machines (toss4). +# +# Steps to successful image build: +# ssh oslic +# podman login -u docker.io//axl +# salloc --userns --exclusive +# podman build -t docker.io/& /p/lslide/martymcf/mpirun.log"] + args: ["-c", "mpirun --map-by node -np 2 mpiGraph 1048576 100 1000 > /p/lslide/martymcf/mpiGraph.out"] + volumeMounts: + - mountPath: /p/lslide + name: lslide + volumes: + - name: lslide + persistentVolumeClaim: + claimName: lustre2-default-readwritemany-pvc + Worker: + template: + spec: + imagePullSecrets: + - name: readonly-red-rock-slushy + containers: + - name: axl-container-martymcf + securityContext: + allowPrivilegeEscalation: true + capabilities: + add: + - NET_BIND_SERVICE + - SYS_CHROOT + - AUDIT_WRITE + image: czregistry.llnl.gov:5050/scr/rabbit-containers:mpigraph diff --git a/test/nnf/axl.containers/src/Makefile b/test/nnf/axl.containers/src/Makefile new file mode 100644 index 0000000..d2240ce --- /dev/null +++ b/test/nnf/axl.containers/src/Makefile @@ -0,0 +1,10 @@ +EXECS=mpi_hello_world +MPICC?=mpicc + +all: ${EXECS} + +mpi_hello_world: mpi_hello_world.c + ${MPICC} -o mpi_hello_world mpi_hello_world.c + +clean: + rm -f ${EXECS} \ No newline at end of file diff --git a/test/nnf/axl.containers/src/mpi_hello_world.c b/test/nnf/axl.containers/src/mpi_hello_world.c new file mode 100644 index 0000000..bf95499 --- /dev/null +++ b/test/nnf/axl.containers/src/mpi_hello_world.c @@ -0,0 +1,61 @@ +// Author: Wes Kendall +// Copyright 2011 www.mpitutorial.com +// This code is provided freely with the tutorials on mpitutorial.com. Feel +// free to modify it for your own use. Any distribution of the code must +// either provide a link to www.mpitutorial.com or keep this header intact. +// +// An intro MPI hello world program that uses MPI_Init, MPI_Comm_size, +// MPI_Comm_rank, MPI_Finalize, and MPI_Get_processor_name. +// +#include +#include +#include +#include +#include + +#include + +#include + +int main(int argc, char **argv) +{ + char nnf_storage_path[PATH_MAX]; + char global_storage_path[PATH_MAX]; + char host_name[PATH_MAX]; + + // Initialize the MPI environment. The two arguments to MPI Init are not + // currently used by MPI implementations, but are there in case future + // implementations might need the arguments. + MPI_Init(NULL, NULL); + + // Get the number of processes + int world_size; + MPI_Comm_size(MPI_COMM_WORLD, &world_size); + + // Get the rank of the process + int world_rank; + MPI_Comm_rank(MPI_COMM_WORLD, &world_rank); + + // Get the name of the processor + char processor_name[MPI_MAX_PROCESSOR_NAME]; + int name_len; + MPI_Get_processor_name(processor_name, &name_len); + + if (argc < 3) { + printf("Usage: %s nnf_storage_dir global_storage_dir\n", argv[0]); + return -1; + } + strncpy(nnf_storage_path, argv[1], PATH_MAX); + strncpy(global_storage_path, argv[2], PATH_MAX); + + if (gethostname(host_name, PATH_MAX) < 0) { + printf("%s: gethostname failed\n", argv[0]); + } + + // Print off a hello world message + printf("%s: processor %s, rank %d out of %d processors. NNF Storage path: %s, Global Storage path is: %s\n", + host_name, processor_name, world_rank, world_size, nnf_storage_path, global_storage_path); + + // Finalize the MPI environment. No more MPI calls can be made after this + MPI_Finalize(); +} diff --git a/test/nnf/hpe_demo.container/Dockerfile b/test/nnf/hpe_demo.container/Dockerfile new file mode 100644 index 0000000..f36829a --- /dev/null +++ b/test/nnf/hpe_demo.container/Dockerfile @@ -0,0 +1,20 @@ +# Start from the NearNodeFlash MPI File Utils Image that is used for Data Movement. This is an easy +# way to get MPI File Utils/mpi-operator as a base. Use a stage for building the application. +# Alernatively, use mpi-operator's image directly: +# FROM mpioperator/openmpi:0.3.0 AS build +FROM ghcr.io/nearnodeflash/nnf-mfu:master AS build + +# Install build tools +RUN apt update +RUN apt install -y make libopenmpi-dev + +# Build application +WORKDIR /src +COPY src . +RUN make + +# Final stage - start fresh and don't carry over the build artifacts +FROM ghcr.io/nearnodeflash/nnf-mfu:master + +# Copy application from build stage into final stage +COPY --from=build /src/mpi_hello_world /usr/bin/mpi_hello_world diff --git a/test/nnf/hpe_demo.container/allocation-computes.yaml b/test/nnf/hpe_demo.container/allocation-computes.yaml new file mode 100644 index 0000000..7af123f --- /dev/null +++ b/test/nnf/hpe_demo.container/allocation-computes.yaml @@ -0,0 +1,3 @@ +data: + - name: "hetchy28" + - name: "hetchy29" diff --git a/test/nnf/hpe_demo.container/allocation-servers.yaml b/test/nnf/hpe_demo.container/allocation-servers.yaml new file mode 100644 index 0000000..7118d89 --- /dev/null +++ b/test/nnf/hpe_demo.container/allocation-servers.yaml @@ -0,0 +1,12 @@ +spec: + allocationSets: + - allocationSize: 50000000000 + label: ost + storage: + - allocationCount: 1 + name: hetchy44 + - allocationSize: 50000000000 + label: mdt + storage: + - allocationCount: 1 + name: hetchy44 diff --git a/test/nnf/hpe_demo.container/nnf-container-example.yaml b/test/nnf/hpe_demo.container/nnf-container-example.yaml new file mode 100644 index 0000000..b5e2bf7 --- /dev/null +++ b/test/nnf/hpe_demo.container/nnf-container-example.yaml @@ -0,0 +1,52 @@ +--- +apiVersion: nnf.cray.hpe.com/v1alpha1 +kind: NnfContainerProfile +metadata: + name: demo + namespace: nnf-system +data: + storages: + - name: DW_JOB_my_storage + optional: false + mpiSpec: + mpiReplicaSpecs: + Launcher: + template: + spec: + containers: + - name: nnf-container-example + image: ghcr.io/nearnodeflash/nnf-container-example:latest + command: + - /bin/bash + - -c + - env && df -h + Worker: + template: + spec: + containers: + - name: nnf-container-example + securityContext: + allowPrivilegeEscalation: true + capabilities: + add: + - NET_BIND_SERVICE + - SYS_CHROOT + - AUDIT_WRITE + image: ghcr.io/nearnodeflash/nnf-container-example:latest + +--- +apiVersion: dws.cray.hpe.com/v1alpha1 +kind: Workflow +metadata: + name: demo-container + namespace: default +spec: + desiredState: "Proposal" + dwDirectives: + - "#DW jobdw name=demo-lustre type=lustre capacity=50GB" + - "#DW container name=demo-container profile=demo \ + DW_JOB_my_storage=demo-lustre" + wlmID: "5f239bd8-30db-450b-8c2c-a1a7c8631a1a" + jobID: 900001 + userID: 1050 + groupID: 1051 diff --git a/test/nnf/hpe_demo.container/run_nnf_user_container b/test/nnf/hpe_demo.container/run_nnf_user_container new file mode 100755 index 0000000..37b76ad --- /dev/null +++ b/test/nnf/hpe_demo.container/run_nnf_user_container @@ -0,0 +1,69 @@ +#!/bin/bash + +container_name="demo-container" + +if [ ! -f nnf-container-example.yaml ]; then + echo "Cannot find container example file" + exit 1 +fi + +run_cmd() { + if ! eval $1 > /dev/null 2>&1 ; then + eval $1 + echo "FAILED" + exit 1 + fi +} + +wait_for_good_state() { + seconds=0 + pattern=$1 + + until kubectl get workflows | grep ${container_name} | grep -q "$pattern" + do + sleep 1 + ((seconds++)) + done + echo ${seconds} +} + +get_logs() { + pod=$1 + seconds=0 + + printf "[Getting Logs-------------->]\r" + # until [ $( get_output_lines ${pod} ) -gt 0 ] + until [ $( kubectl logs --tail=100 ${pod} |& wc -l ) -gt 0 ] + do + sleep 1 + ((seconds++)) + done + + printf "[Logs obtained in %02d seconds]\n" $seconds + echo "----------------------" + kubectl logs --tail=100 ${pod} + echo "----------------------" +} + +run_and_wait() { + printf "[Running-->] - %s\r" "$1" + run_cmd "$1" + t=$( wait_for_good_state "$2" ) + printf "[%02d seconds] - %s\n" $t "$1" +} + +run_and_wait "kubectl apply -f nnf-container-example.yaml" "true Completed" + +# +# The following two patch commands finish synchronously and we do not need to wait for them to complete +# +run_cmd "kubectl patch --type merge --patch-file=allocation-computes.yaml computes ${container_name}" +run_cmd "kubectl patch --type merge --patch-file=allocation-servers.yaml servers ${container_name}-0" + +run_and_wait "kubectl patch --type merge workflow ${container_name} --patch '{\"spec\": {\"desiredState\": \"Setup\"}}'" "Setup true Completed" +run_and_wait "kubectl patch --type merge workflow ${container_name} --patch '{\"spec\": {\"desiredState\": \"DataIn\"}}'" "DataIn true Completed" +run_and_wait "kubectl patch --type merge workflow ${container_name} --patch '{\"spec\": {\"desiredState\": \"PreRun\"}}'" "PreRun true Completed" +pod="$( kubectl get pods | grep ${container_name}-launcher | awk '{print $1}' )" +get_logs "$( kubectl get pods | grep ${container_name}-launcher | awk '{print $1}' )" +run_and_wait "kubectl patch --type merge workflow ${container_name} --patch '{\"spec\": {\"desiredState\": \"Teardown\"}}'" "Teardown true Completed" +run_cmd "kubectl delete -f nnf-container-example.yaml" diff --git a/test/nnf/hpe_demo.container/src/Makefile b/test/nnf/hpe_demo.container/src/Makefile new file mode 100644 index 0000000..d2240ce --- /dev/null +++ b/test/nnf/hpe_demo.container/src/Makefile @@ -0,0 +1,10 @@ +EXECS=mpi_hello_world +MPICC?=mpicc + +all: ${EXECS} + +mpi_hello_world: mpi_hello_world.c + ${MPICC} -o mpi_hello_world mpi_hello_world.c + +clean: + rm -f ${EXECS} \ No newline at end of file diff --git a/test/nnf/hpe_demo.container/src/mpi_hello_world.c b/test/nnf/hpe_demo.container/src/mpi_hello_world.c new file mode 100644 index 0000000..6eb0398 --- /dev/null +++ b/test/nnf/hpe_demo.container/src/mpi_hello_world.c @@ -0,0 +1,81 @@ +// Author: Wes Kendall +// Copyright 2011 www.mpitutorial.com +// This code is provided freely with the tutorials on mpitutorial.com. Feel +// free to modify it for your own use. Any distribution of the code must +// either provide a link to www.mpitutorial.com or keep this header intact. +// +// An intro MPI hello world program that uses MPI_Init, MPI_Comm_size, +// MPI_Comm_rank, MPI_Finalize, and MPI_Get_processor_name. +// +#include +#include +#include +#include + +#include + +#include + +int main(int argc, char **argv) +{ + char nnf_storage_path[PATH_MAX]; + + // Initialize the MPI environment. The two arguments to MPI Init are not + // currently used by MPI implementations, but are there in case future + // implementations might need the arguments. + MPI_Init(NULL, NULL); + + // Get the number of processes + int world_size; + MPI_Comm_size(MPI_COMM_WORLD, &world_size); + + // Get the rank of the process + int world_rank; + MPI_Comm_rank(MPI_COMM_WORLD, &world_rank); + + // Get the name of the processor + char processor_name[MPI_MAX_PROCESSOR_NAME]; + int name_len; + MPI_Get_processor_name(processor_name, &name_len); + + if (argc < 2) + { + printf("Storage parameter not supplied\n"); + return -1; + } + strncpy(nnf_storage_path, argv[1], PATH_MAX); + + // Print off a hello world message + printf("Hello world from processor %s, rank %d out of %d processors. NNF Storage path: %s\n", + processor_name, world_rank, world_size, nnf_storage_path); + +#if 0 + // We're using a GFS2 filesystem, which has index mounts for every compute node + // e.g. /mnt/nnf/5d335081-cd0f-4b8a-a1f4-94860a8ae702-0/0/ + // So use the rank to index into the NNF storage path + if (sprintf(nnf_storage_path, "%s/%d/testfile", nnf_storage_path, world_rank) == -1) + { + fprintf(stderr, "rank %d: %s\n", world_rank, strerror(errno)); + return errno; + } + + int fd = open(nnf_storage_path, O_WRONLY | O_CREAT); + if (fd == -1) + { + fprintf(stderr, "rank %d: %s\n", world_rank, strerror(errno)); + return errno; + } + + int res = posix_fallocate(fd, 0, 100); + if (res == -1) + { + fprintf(stderr, "rank %d: %s\n", world_rank, strerror(errno)); + return errno; + } + + printf("rank %d: wrote file to '%s'\n", world_rank, nnf_storage_path); +#endif // 0 + + // Finalize the MPI environment. No more MPI calls can be made after this + MPI_Finalize(); +}