added support for YAML-based server configuration

bin/hepnos-daemon.cpp

@@ -20,14 +20,14 @@
 void usage(void)
 {
     fprintf(stderr, "Usage: hepnos-daemon <addr> <config>\n");
-    fprintf(stderr, "  <addr>    the Mercury address to listen on (e.g. tcp://)\n");
-    fprintf(stderr, "  <config>  path to the YAML file to generate for clients\n");
+    fprintf(stderr, "  <config> path to the YAML file containing the service configuration\n");
+    fprintf(stderr, "  <connection>  path to the YAML file to generate for clients\n");
     exit(-1);
 }
 
 int main(int argc, char *argv[])
 {
-    char* listen_addr;
+    char* connection_file;
     char* config_file;
     int rank;
 
@@ -42,10 +42,10 @@ int main(int argc, char *argv[])
         exit(0);
     }
 
-    listen_addr = argv[1];
-    config_file = argv[2];
+    config_file = argv[1];
+    connection_file = argv[2];
 
-    hepnos_run_service(MPI_COMM_WORLD, listen_addr, config_file);
+    hepnos_run_service(MPI_COMM_WORLD, config_file, connection_file);
 
     MPI_Finalize();
 }

include/hepnos-service.h

@@ -7,7 +7,7 @@
 extern "C" {
 #endif
 
-void hepnos_run_service(MPI_Comm comm, const char* listen_addr, const char* config_file);
+void hepnos_run_service(MPI_Comm comm, const char* config_file, const char* connection_file);
 
 #ifdef __cplusplus
 }

src/CMakeLists.txt

@@ -6,7 +6,8 @@ set(hepnos-src DataStore.cpp
 	       SubRun.cpp
 	       Event.cpp) 
 
-set(hepnos-service-src service/HEPnOSService.cpp)
+set(hepnos-service-src service/HEPnOSService.cpp
+		       service/ServiceConfig.cpp)
 
 # load package helper for generating cmake CONFIG packages
 include (CMakePackageConfigHelpers)
@@ -19,7 +20,7 @@ set (hepnos-pkg "share/cmake/hepnos")
 #
 set (HEPNOS_VERSION_MAJOR 0)
 set (HEPNOS_VERSION_MINOR 1)
-set (HEPNOS_VERSION_PATCH 0)
+set (HEPNOS_VERSION_PATCH 2)
 set (hepnos-vers "${HEPNOS_VERSION_MAJOR}.${HEPNOS_VERSION_MINOR}")
 set (HEPNOS_VERSION "${hepnos-vers}.${HEPNOS_VERSION_PATCH}")
 
@@ -44,6 +45,8 @@ target_include_directories (hepnos-service PUBLIC $<INSTALL_INTERFACE:include>)
 # local include's BEFORE, in case old incompatable .h files in prefix/include
 target_include_directories (hepnos-service BEFORE PUBLIC
     $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/../include>)
+target_include_directories (hepnos-service BEFORE PUBLIC
+    $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/service>)
 
 # for shared libs, establish the lib version
 set_target_properties (hepnos-service

src/service/HEPnOSService.cpp

@@ -13,13 +13,14 @@
 #include <bake-server.h>
 #include <sdskv-server.h>
 #include <yaml-cpp/yaml.h>
+#include "ServiceConfig.hpp"
 #include "hepnos-service.h"
 
 #define ASSERT(__cond, __msg, ...) { if(!(__cond)) { fprintf(stderr, "[%s:%d] " __msg, __FILE__, __LINE__, __VA_ARGS__); exit(-1); } }
 
-static void generate_config_file(MPI_Comm comm, const char* addr, const char* config_file);
+static void generate_connection_file(MPI_Comm comm, const char* addr, const char* filename);
 
-void hepnos_run_service(MPI_Comm comm, const char* listen_addr, const char* config_file)
+void hepnos_run_service(MPI_Comm comm, const char* config_file, const char* connection_file)
 {
     margo_instance_id mid;
     int ret;
@@ -27,11 +28,24 @@ void hepnos_run_service(MPI_Comm comm, const char* listen_addr, const char* conf
 
     MPI_Comm_rank(comm, &rank);
 
+    /* load configuration */
+    std::unique_ptr<hepnos::ServiceConfig> config;
+    try {
+        config = std::make_unique<hepnos::ServiceConfig>(config_file, rank);
+    } catch(const std::exception& e) {
+        std::cerr << "Error: when reading configuration:" << std::endl;
+        std::cerr << "  " << e.what() << std::endl;
+        std::cerr << "Aborting." << std::endl;
+        MPI_Abort(MPI_COMM_WORLD, -1);
+        return;
+    }
+
     /* Margo initialization */
-    mid = margo_init(listen_addr, MARGO_SERVER_MODE, 0, -1);
+    mid = margo_init(config->getAddress().c_str(), MARGO_SERVER_MODE, 0, -1);
     if (mid == MARGO_INSTANCE_NULL)
     {
-        fprintf(stderr, "Error: Unable to initialize margo\n");
+        std::cerr << "Error: unable to initialize margo" << std::endl;
+        std::cerr << "Aborting." << std::endl;
         MPI_Abort(MPI_COMM_WORLD, -1);
         return;
     }
@@ -44,43 +58,53 @@ void hepnos_run_service(MPI_Comm comm, const char* listen_addr, const char* conf
     hg_size_t self_addr_str_size = 128;
     margo_addr_to_string(mid, self_addr_str, &self_addr_str_size, self_addr);
 
-    /* Bake provider initialization */
-    uint16_t bake_mplex_id = 1;
-    char bake_target_name[128];
-    sprintf(bake_target_name, "/dev/shm/hepnos.%d.dat", rank);
-    /* create the bake target if it does not exist */
-    if(-1 == access(bake_target_name, F_OK)) {
-        // XXX creating a pool of 10MB - this should come from a config file
-        ret = bake_makepool(bake_target_name, 10*1024*1024, 0664);
-        ASSERT(ret == 0, "bake_makepool() failed (ret = %d)\n", ret);
+    if(config->hasStorage()) {
+        /* Bake provider initialization */
+        uint16_t bake_mplex_id = 1;
+        const char* bake_target_name = config->getStoragePath().c_str();
+        size_t bake_target_size = config->getStorageSize()*(1024*1024);
+        /* create the bake target if it does not exist */
+        if(-1 == access(bake_target_name, F_OK)) {
+            ret = bake_makepool(bake_target_name, bake_target_size, 0664);
+            ASSERT(ret == 0, "bake_makepool() failed (ret = %d)\n", ret);
+        }
+        bake_provider_t bake_prov;
+        bake_target_id_t bake_tid;
+        ret = bake_provider_register(mid, bake_mplex_id, BAKE_ABT_POOL_DEFAULT, &bake_prov);
+        ASSERT(ret == 0, "bake_provider_register() failed (ret = %d)\n", ret);
+        ret = bake_provider_add_storage_target(bake_prov, bake_target_name, &bake_tid);
+        ASSERT(ret == 0, "bake_provider_add_storage_target() failed to add target %s (ret = %d)\n",
+                bake_target_name, ret);
+    }
+
+    if(config->hasDatabase()) {
+        /* SDSKV provider initialization */
+        uint8_t sdskv_mplex_id = 1;
+        sdskv_provider_t sdskv_prov;
+        ret = sdskv_provider_register(mid, sdskv_mplex_id, SDSKV_ABT_POOL_DEFAULT, &sdskv_prov);
+        ASSERT(ret == 0, "sdskv_provider_register() failed (ret = %d)\n", ret);
+
+        /* creating the database */
+        const char* db_path = config->getDatabasePath().c_str();
+        const char* db_name = config->getDatabaseName().c_str();
+        sdskv_db_type_t db_type;
+        if(config->getDatabaseType() == "map") db_type = KVDB_MAP;
+        if(config->getDatabaseType() == "ldb") db_type = KVDB_LEVELDB;
+        if(config->getDatabaseType() == "bdb") db_type = KVDB_BERKELEYDB;
+        sdskv_database_id_t db_id;
+        ret = sdskv_provider_add_database(sdskv_prov, db_name, db_path, db_type, SDSKV_COMPARE_DEFAULT,  &db_id);
+        ASSERT(ret == 0, "sdskv_provider_add_database() failed (ret = %d)\n", ret);
     }
-    bake_provider_t bake_prov;
-    bake_target_id_t bake_tid;
-    ret = bake_provider_register(mid, bake_mplex_id, BAKE_ABT_POOL_DEFAULT, &bake_prov);
-    ASSERT(ret == 0, "bake_provider_register() failed (ret = %d)\n", ret);
-    ret = bake_provider_add_storage_target(bake_prov, bake_target_name, &bake_tid);
-    ASSERT(ret == 0, "bake_provider_add_storage_target() failed to add target %s (ret = %d)\n",
-            bake_target_name, ret);
-
-    /* SDSKV provider initialization */
-    uint8_t sdskv_mplex_id = 1;
-    sdskv_provider_t sdskv_prov;
-    ret = sdskv_provider_register(mid, sdskv_mplex_id, SDSKV_ABT_POOL_DEFAULT, &sdskv_prov);
-    ASSERT(ret == 0, "sdskv_provider_register() failed (ret = %d)\n", ret);
-
-    // XXX creating the database - this should come from a config file
-    sdskv_database_id_t db_id;
-    ret = sdskv_provider_add_database(sdskv_prov, "hepnosdb", "", KVDB_MAP, SDSKV_COMPARE_DEFAULT,  &db_id);
-    ASSERT(ret == 0, "sdskv_provider_add_database() failed (ret = %d)\n", ret);
 
     margo_addr_free(mid, self_addr);
 
-    generate_config_file(MPI_COMM_WORLD, self_addr_str, config_file);
+    // XXX This should be replaced by submitting the connection info to a registry
+    generate_connection_file(MPI_COMM_WORLD, self_addr_str, connection_file);
 
     margo_wait_for_finalize(mid);
 }
 
-static void generate_config_file(MPI_Comm comm, const char* addr, const char* config_file)
+static void generate_connection_file(MPI_Comm comm, const char* addr, const char* filename)
 {
     int rank, size;
     MPI_Comm_rank(comm, &rank);
@@ -106,6 +130,6 @@ static void generate_config_file(MPI_Comm comm, const char* addr, const char* co
     for(auto& s :  addresses)
         config["hepnos"]["providers"]["sdskv"][s] = 1;
 
-    std::ofstream fout(config_file);
+    std::ofstream fout(filename);
     fout << config;
 }

src/service/ServiceConfig.cpp

+#include "ServiceConfig.hpp"
+#include "hepnos/Exception.hpp"
+#include <yaml-cpp/yaml.h>
+
+namespace hepnos {
+
+struct ServiceConfig::Impl {
+
+    std::string m_address;
+    bool        m_hasDatabase;
+    std::string m_databasePath;
+    std::string m_databaseName;
+    std::string m_databaseType;
+    bool        m_hasStorage;
+    std::string m_storagePath;
+    size_t      m_storageSize;
+
+};
+
+static YAML::Node loadAndValidate(const std::string& filename);
+static std::string insertRankIn(const std::string& str, int rank);
+
+ServiceConfig::ServiceConfig(const std::string& filename, int rank)
+: m_impl(std::make_unique<Impl>()) {
+    
+    YAML::Node config       = loadAndValidate(filename);
+    YAML::Node address      = config["address"];
+    YAML::Node db_node      = config["database"];
+    YAML::Node storage_node = config["storage"];
+    m_impl->m_address = address.as<std::string>();
+    if(!db_node) {
+        m_impl->m_hasDatabase  = false;
+    } else {
+        m_impl->m_hasDatabase  = true;
+        m_impl->m_databasePath = insertRankIn(db_node["path"].as<std::string>(), rank);
+        m_impl->m_databaseName = db_node["name"].as<std::string>();
+        m_impl->m_databaseType = db_node["type"].as<std::string>();
+    }
+    if(!storage_node) {
+        m_impl->m_hasStorage   = false;
+    } else {
+        m_impl->m_hasStorage   = true;
+        m_impl->m_storagePath  = insertRankIn(storage_node["path"].as<std::string>(), rank);
+        m_impl->m_storageSize  = storage_node["size"].as<size_t>();
+    }
+}
+
+ServiceConfig::~ServiceConfig() {}
+
+const std::string& ServiceConfig::getAddress() const {
+    return m_impl->m_address;
+}
+
+bool ServiceConfig::hasDatabase() const {
+    return m_impl->m_hasDatabase;
+}
+
+const std::string& ServiceConfig::getDatabasePath() const {
+    return m_impl->m_databasePath;
+}
+
+const std::string& ServiceConfig::getDatabaseName() const {
+    return m_impl->m_databaseName;
+}
+
+const std::string& ServiceConfig::getDatabaseType() const {
+    return m_impl->m_databaseType;
+}
+
+bool ServiceConfig::hasStorage() const {
+    return m_impl->m_hasStorage;
+}
+
+const std::string& ServiceConfig::getStoragePath() const {
+    return m_impl->m_storagePath;
+}
+
+size_t ServiceConfig::getStorageSize() const {
+    return m_impl->m_storageSize;
+}
+
+static YAML::Node loadAndValidate(const std::string& filename) {
+    YAML::Node config = YAML::LoadFile(filename);
+    if(!config["address"]) {
+        throw Exception("\"address\" field not found in configuration file.");
+    }
+    if(!config["database"]) {
+        throw Exception("\"database\" field not found in configuration file.");
+    }
+    if(!config["database"]["path"]) {
+        throw Exception("\"database.path\" field not found in configuration file.");
+    }
+    if(!config["database"]["name"]) {
+        throw Exception("\"database.name\" field not found in configuration file.");
+    }
+    if(!config["database"]["type"]) {
+        throw Exception("\"database.type\" field not found in configuration file.");
+    }
+    std::string db_type = config["database"]["type"].as<std::string>();
+    if(db_type != "map"
+    && db_type != "ldb"
+    && db_type != "bdb") {
+        throw Exception("\"database.type\" field should be \"map\", \"ldb\", or \"bdb\".");
+    }
+    if(config["storage"]) {
+        if(!config["storage"]["path"]) {
+            throw Exception("\"storage.path\" field not found in configuration file.");
+        }
+        if(!config["storage"]["size"]) {
+            throw Exception("\"storage.size\" field not found in configuration file.");
+        }
+    }
+    return config;
+}
+
+static std::string insertRankIn(const std::string& str, int rank) {
+    size_t index = 0;
+    std::string result = str;
+    std::stringstream ssrank;
+    ssrank << rank;
+    std::string srank = ssrank.str();
+    while (true) {
+        index = result.find("$RANK", index);
+        if (index == std::string::npos) break;
+        if(rank >= 0) {
+            result.replace(index, 5, srank.c_str());
+        } else {
+            result.replace(index, 5, "");
+        }
+        index += 5;
+    }
+    return result;
+}
+
+}
+

src/service/ServiceConfig.hpp

+#ifndef __HEPNOS_SERVICE_CONFIG_H
+#define __HEPNOS_SERVICE_CONFIG_H
+
+#include <string>
+#include <memory>
+#include <yaml-cpp/yaml.h>
+
+namespace hepnos {
+
+class ServiceConfig {
+
+private:
+    class Impl;
+    std::unique_ptr<Impl> m_impl;
+
+public:
+
+    ServiceConfig(const std::string& filename, int rank=-1);
+    
+    ServiceConfig(const ServiceConfig&) = delete;
+    ServiceConfig(ServiceConfig&&) = delete;
+    ServiceConfig& operator=(const ServiceConfig&) = delete;
+    ServiceConfig& operator=(ServiceConfig&&) = delete;
+    ~ServiceConfig();
+
+    const std::string& getAddress() const;
+    bool hasDatabase() const;
+    const std::string& getDatabasePath() const;
+    const std::string& getDatabaseName() const;
+    const std::string& getDatabaseType() const;
+    bool hasStorage() const;
+    const std::string& getStoragePath() const;
+    size_t getStorageSize() const;
+};
+
+}
+
+#endif

test/CMakeLists.txt

 # Move the configuration file used for the configuration test
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/run-test.sh ${CMAKE_CURRENT_BINARY_DIR}/run-test.sh COPYONLY)
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/test-util.sh ${CMAKE_CURRENT_BINARY_DIR}/test-util.sh COPYONLY)
+configure_file(${CMAKE_CURRENT_SOURCE_DIR}/config.yaml ${CMAKE_CURRENT_BINARY_DIR}/config.yaml COPYONLY)
 
 add_executable(example example.cpp)
 target_link_libraries(example hepnos ${Boost_LIBRARIES})

test/config.yaml

+---
+address: tcp://
+database:
+  name: hepnosdb
+  path: XXX/$RANK
+  type: bdb

test/run-test.sh

@@ -7,15 +7,18 @@ fi
 
 source test-util.sh
 
-TEST_DIR=`$MKTEMP -d /tmp/hepnos-XXXXXX`
-CFG_FILE=$TEST_DIR/config.yml
+TEST_DIR=`$MKTEMP -d tmp/hepnos-XXXXXX`
+CON_FILE=$TEST_DIR/connection.yaml
+cp config.yaml $TEST_DIR/config.yaml
+CFG_FILE=$TEST_DIR/config.yaml
+sed -i -e "s|XXX|${TEST_DIR}/database|g" $CFG_FILE
 
-hepnos_test_start_servers 2 1 20 $CFG_FILE
+hepnos_test_start_servers 2 1 20 $CFG_FILE $CON_FILE
 
-export HEPNOS_CONFIG_FILE=$CFG_FILE
+export HEPNOS_CONFIG_FILE=$CON_FILE
 
 # run a connect test client
-run_to 10 $1 $CFG_FILE
+run_to 10 $1 $CON_FILE
 if [ $? -ne 0 ]; then
     wait
     exit 1

test/test-util.sh

@@ -19,11 +19,11 @@ function hepnos_test_start_servers()
     nservers=${1:-4}
     startwait=${2:-15}
     maxtime=${3:-120}
-    cfile=${4:-testconfig.yml}
-
+    config=${4:-config.yaml}
+    cfile=${5:-connection.yaml}
     rm -rf ${cfile}
 
-    run_to $maxtime mpirun -np $nservers ../bin/hepnos-daemon tcp:// $cfile &
+    run_to $maxtime mpirun -np $nservers ../bin/hepnos-daemon $config $cfile &
     if [ $? -ne 0 ]; then
         # TODO: this doesn't actually work; can't check return code of
         # something executing in background.  We have to rely on the