enabled bake

src/CMakeLists.txt

@@ -7,7 +7,8 @@ set(hepnos-src DataStore.cpp
 	       Event.cpp) 
 
 set(hepnos-service-src service/HEPnOSService.cpp
-		       service/ServiceConfig.cpp)
+		       service/ServiceConfig.cpp
+		       service/ConnectionInfoGenerator.cpp)
 
 # load package helper for generating cmake CONFIG packages
 include (CMakePackageConfigHelpers)

src/DataStore.cpp

@@ -163,7 +163,7 @@ DataSet DataStore::createDataSet(const std::string& name) {
 
 void DataStore::shutdown() {
     for(auto addr : m_impl->m_addrs) {
-        margo_shutdown_remote_instance(m_impl->m_mid, addr);
+        margo_shutdown_remote_instance(m_impl->m_mid, addr.second);
     }
 }
 

src/private/ValueTypes.hpp

+/*
+ * (C) 2018 The University of Chicago
+ * 
+ * See COPYRIGHT in top-level directory.
+ */
+#ifndef __PRIVATE_VALUE_TYPES_H
+#define __PRIVATE_VALUE_TYPES_H
+
+#include <cstring>
+#include <cstdlib>
+#include <cstdint>
+#include <memory>
+
+namespace hepnos {
+
+class DataStoreValue {
+
+    size_t           m_object_size;
+    uint64_t         m_server_id;
+    bake_region_id_t m_region_id;
+
+    public:
+
+    DataStoreValue()
+    : m_object_size(0), m_server_id(0) {}
+
+    DataStoreValue(size_t object_size, uint64_t bake_server_id, const bake_region_id_t& region_id)
+    : m_object_size(object_size), m_server_id(bake_server_id), m_region_id(region_id) {}
+
+    size_t getDataSize() const {
+        return m_object_size;
+    }
+
+    const bake_region_id_t& getBakeRegionID() const {
+        return m_region_id;
+    }
+
+    const uint64_t& getBakeServerID() const {
+        return m_server_id;
+    }
+};
+
+}
+
+#endif

src/service/ConnectionInfoGenerator.cpp

+#include <fstream>
+#include <yaml-cpp/yaml.h>
+#include "ConnectionInfoGenerator.hpp"
+
+namespace hepnos {
+
+struct ConnectionInfoGenerator::Impl {
+    std::string m_addr;      // address of this process
+    uint16_t    m_bake_id;   // provider ids for BAKE
+    uint16_t    m_sdskv_id;  // provider ids for SDSKV
+};
+
+ConnectionInfoGenerator::ConnectionInfoGenerator(
+        const std::string& address, 
+        uint16_t sdskv_provider_id,
+        uint16_t bake_provider_id) 
+: m_impl(std::make_unique<Impl>()) {
+    m_impl->m_addr     = address;
+    m_impl->m_bake_id  = bake_provider_id;
+    m_impl->m_sdskv_id = sdskv_provider_id;
+}
+
+ConnectionInfoGenerator::~ConnectionInfoGenerator() {}
+
+void ConnectionInfoGenerator::generateFile(MPI_Comm comm, const std::string& filename) const {
+    int rank, size;
+    const char* addr = m_impl->m_addr.c_str();
+
+    MPI_Comm_rank(comm, &rank);
+    MPI_Comm_size(comm, &size);
+
+    unsigned j=0;
+    while(addr[j] != '\0' && addr[j] != ':') j++;
+    std::string proto(addr, j);
+
+    // Exchange addresses
+    std::vector<char> addresses_buf(128*size);
+    MPI_Gather(addr, 128, MPI_BYTE, addresses_buf.data(), 128, MPI_BYTE, 0, comm);
+
+    // Exchange bake providers info
+    std::vector<uint16_t> bake_pr_ids_buf(size);
+    MPI_Gather(&(m_impl->m_bake_id), 
+            1, MPI_UNSIGNED_SHORT, 
+            bake_pr_ids_buf.data(),
+            1, MPI_UNSIGNED_SHORT, 
+            0, comm);
+
+    // Exchange sdskv providers info
+    std::vector<uint16_t> sdskv_pr_ids_buf(size);
+    MPI_Gather(&(m_impl->m_sdskv_id),
+            1, MPI_UNSIGNED_SHORT, 
+            sdskv_pr_ids_buf.data(),
+            1, MPI_UNSIGNED_SHORT, 
+            0, comm);
+
+    // After this line, the rest is executed only by rank 0
+    if(rank != 0) return;
+
+    std::vector<std::string> addresses;
+    for(unsigned i=0; i < size; i++) {
+        addresses.emplace_back(&addresses_buf[128*i]);
+    }
+
+    YAML::Node config;
+    config["hepnos"]["client"]["protocol"] = proto;
+    YAML::Node providers = config["hepnos"]["providers"];
+    for(unsigned int i=0; i < size; i++) {
+        const auto& provider_addr = addresses[i];
+        if(sdskv_pr_ids_buf[i]) {
+            providers["sdskv"][provider_addr] = sdskv_pr_ids_buf[i];
+        }
+        if(bake_pr_ids_buf[i]) {
+            providers["bake"][provider_addr] = bake_pr_ids_buf[i];
+        }
+    }
+
+    std::ofstream fout(filename);
+    fout << config;
+}
+ 
+}

src/service/ConnectionInfoGenerator.hpp

+#ifndef __HEPNOS_CONNECTION_INFO_GENERATOR_H
+#define __HEPNOS_CONNECTION_INFO_GENERATOR_H
+
+#include <string>
+#include <memory>
+#include <mpi.h>
+
+namespace hepnos {
+
+class ConnectionInfoGenerator {
+
+private:
+
+    class Impl;
+    std::unique_ptr<Impl> m_impl;
+
+public:
+
+    ConnectionInfoGenerator(const std::string& address,
+            uint16_t sdskv_provider_id,
+            uint16_t bake_provider_id);
+    ConnectionInfoGenerator(const ConnectionInfoGenerator&) = delete;
+    ConnectionInfoGenerator(ConnectionInfoGenerator&&) = delete;
+    ConnectionInfoGenerator& operator=(const ConnectionInfoGenerator&) = delete;
+    ConnectionInfoGenerator& operator=(ConnectionInfoGenerator&&) = delete;
+    ~ConnectionInfoGenerator();
+
+    void generateFile(MPI_Comm comm, const std::string& filename) const;
+};
+
+}
+
+#endif

src/service/HEPnOSService.cpp

@@ -12,13 +12,13 @@
 #include <margo.h>
 #include <bake-server.h>
 #include <sdskv-server.h>
-#include <yaml-cpp/yaml.h>
 #include "ServiceConfig.hpp"
+#include "ConnectionInfoGenerator.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_connection_file(MPI_Comm comm, const char* addr, const char* filename);
+//static void generate_connection_file(MPI_Comm comm, const char* addr, const char* filename);
 
 void hepnos_run_service(MPI_Comm comm, const char* config_file, const char* connection_file)
 {
@@ -58,9 +58,10 @@ void hepnos_run_service(MPI_Comm comm, const char* config_file, const char* conn
     hg_size_t self_addr_str_size = 128;
     margo_addr_to_string(mid, self_addr_str, &self_addr_str_size, self_addr);
 
+    uint16_t bake_provider_id = 0;
     if(config->hasStorage()) {
         /* Bake provider initialization */
-        uint16_t bake_mplex_id = 1;
+        bake_provider_id = 1; // XXX we can make that come from the config file
         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 */
@@ -70,18 +71,19 @@ void hepnos_run_service(MPI_Comm comm, const char* config_file, const char* conn
         }
         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);
+        ret = bake_provider_register(mid, bake_provider_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);
     }
 
+    uint8_t sdskv_provider_id = 0;
     if(config->hasDatabase()) {
         /* SDSKV provider initialization */
-        uint8_t sdskv_mplex_id = 1;
+        sdskv_provider_id = 1; // XXX we can make that come from the config file
         sdskv_provider_t sdskv_prov;
-        ret = sdskv_provider_register(mid, sdskv_mplex_id, SDSKV_ABT_POOL_DEFAULT, &sdskv_prov);
+        ret = sdskv_provider_register(mid, sdskv_provider_id, SDSKV_ABT_POOL_DEFAULT, &sdskv_prov);
         ASSERT(ret == 0, "sdskv_provider_register() failed (ret = %d)\n", ret);
 
         /* creating the database */
@@ -98,12 +100,12 @@ void hepnos_run_service(MPI_Comm comm, const char* config_file, const char* conn
 
     margo_addr_free(mid, self_addr);
 
-    // XXX This should be replaced by submitting the connection info to a registry
-    generate_connection_file(MPI_COMM_WORLD, self_addr_str, connection_file);
+    hepnos::ConnectionInfoGenerator fileGen(self_addr_str, sdskv_provider_id, bake_provider_id);
+    fileGen.generateFile(MPI_COMM_WORLD, connection_file);
 
     margo_wait_for_finalize(mid);
 }
-
+/*
 static void generate_connection_file(MPI_Comm comm, const char* addr, const char* filename)
 {
     int rank, size;
@@ -133,3 +135,4 @@ static void generate_connection_file(MPI_Comm comm, const char* addr, const char
     std::ofstream fout(filename);
     fout << config;
 }
+*/

test/config.yaml

@@ -4,3 +4,6 @@ database:
   name: hepnosdb
   path: XXX/$RANK
   type: bdb
+storage:
+  path: /dev/shm/hepnos.$RANK.dat
+  size: 50

test/run-test.sh

@@ -7,7 +7,7 @@ fi
 
 source test-util.sh
 
-TEST_DIR=`$MKTEMP -d tmp/hepnos-XXXXXX`
+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