Commit f1e3831a authored by Matthieu Dorier's avatar Matthieu Dorier

Merge branch 'dev-config' into 'master'

Dev config

See merge request !1
parents 08bad35b 537d3ff4
......@@ -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();
}
......
......@@ -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
}
......
......@@ -6,7 +6,9 @@ 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
service/ConnectionInfoGenerator.cpp)
# load package helper for generating cmake CONFIG packages
include (CMakePackageConfigHelpers)
......@@ -19,7 +21,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 +46,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
......
......@@ -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);
}
}
......
This diff is collapsed.
/*
* (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
#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;
}
}
#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
......@@ -12,14 +12,15 @@
#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_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,55 @@ 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);
uint16_t bake_provider_id = 0;
if(config->hasStorage()) {
/* Bake provider initialization */
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 */
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_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 */
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_provider_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);
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_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 +132,7 @@ 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;
}
*/
#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;
}
}
#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
# 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})
......
---
address: tcp://
database:
name: hepnosdb
path: XXX/$RANK
type: bdb
storage:
path: /dev/shm/hepnos.$RANK.dat
size: 50
......@@ -8,14 +8,17 @@ fi
source test-util.sh
TEST_DIR=`$MKTEMP -d /tmp/hepnos-XXXXXX`
CFG_FILE=$TEST_DIR/config.yml
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
......@@ -27,5 +30,5 @@ wait
# cleanup
rm -rf $TEST_DIR
rm -rf /dev/shm/hepnos.dat
exit 0
......@@ -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
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment