Redis added into Batsim: cmake+cli args+connect

CMakeLists.txt

@@ -57,6 +57,17 @@ include_directories(${Boost_INCLUDE_DIR})
 find_package(rapidjson REQUIRED)
 include_directories(${RAPIDJSON_INCLUDE_DIRS})
 
+## Redox dependency
+find_package(redox REQUIRED)
+include_directories(${REDOX_INCLUDE_DIR})
+
+## Redox sub dependencies
+find_package(hiredis REQUIRED)
+include_directories(${HIREDIS_INCLUDE_DIRS})
+
+find_package(libev REQUIRED)
+include_directories(${LIBEV_INCLUDE_DIRS})
+
 ################
 # Source files #
 ################
@@ -68,8 +79,12 @@ file(GLOB batsim_SRC
 add_executable(batsim ${batsim_SRC})
 
 # Libraries to link
-target_link_libraries(batsim ${SIMGRID_LIBRARIES} ${Boost_SYSTEM_LIBRARY_DEBUG}
-                      ${Boost_FILESYSTEM_LIBRARY_DEBUG})
+target_link_libraries(batsim ${SIMGRID_LIBRARIES}
+                      ${Boost_SYSTEM_LIBRARY_DEBUG}
+                      ${Boost_FILESYSTEM_LIBRARY_DEBUG}
+                      ${REDOX_LIBRARY}
+                      ${LIBEV_LIBRARY}
+                      ${HIREDIS_LIBRARY})
 
 ################
 # Installation #

cmake/Modules/Findhiredis.cmake

+# File got from redox repo, https://github.com/hmartiro/redox.
+# Try to find hiredis
+# Once done, this will define
+#
+# HIREDIS_FOUND        - system has hiredis
+# HIREDIS_INCLUDE_DIRS - hiredis include directories
+# HIREDIS_LIBRARIES    - libraries need to use hiredis
+
+if(HIREDIS_INCLUDE_DIRS AND HIREDIS_LIBRARIES)
+  set(HIREDIS_FIND_QUIETLY TRUE)
+else()
+  find_path(
+    HIREDIS_INCLUDE_DIR
+    NAMES hiredis/hiredis.h
+    HINTS ${HIREDIS_ROOT_DIR}
+    PATH_SUFFIXES include)
+
+  find_library(
+    HIREDIS_LIBRARY
+    NAMES hiredis
+    HINTS ${HIREDIS_ROOT_DIR}
+    PATH_SUFFIXES ${CMAKE_INSTALL_LIBDIR})
+
+  set(HIREDIS_INCLUDE_DIRS ${HIREDIS_INCLUDE_DIR})
+  set(HIREDIS_LIBRARIES ${HIREDIS_LIBRARY})
+
+  include (FindPackageHandleStandardArgs)
+  find_package_handle_standard_args(
+    hiredis DEFAULT_MSG HIREDIS_LIBRARY HIREDIS_INCLUDE_DIR)
+
+  mark_as_advanced(HIREDIS_LIBRARY HIREDIS_INCLUDE_DIR)
+endif()

cmake/Modules/Findlibev.cmake

+# File got from redox repo, https://github.com/hmartiro/redox.
+# Try to find libev
+# Once done, this will define
+#
+# LIBEV_FOUND        - system has libev
+# LIBEV_INCLUDE_DIRS - libev include directories
+# LIBEV_LIBRARIES    - libraries needed to use libev
+
+if(LIBEV_INCLUDE_DIRS AND LIBEV_LIBRARIES)
+  set(LIBEV_FIND_QUIETLY TRUE)
+else()
+  find_path(
+    LIBEV_INCLUDE_DIR
+    NAMES ev.h
+    HINTS ${LIBEV_ROOT_DIR}
+    PATH_SUFFIXES include)
+
+  find_library(
+    LIBEV_LIBRARY
+    NAME ev
+    HINTS ${LIBEV_ROOT_DIR}
+    PATH_SUFFIXES ${CMAKE_INSTALL_LIBDIR})
+
+  set(LIBEV_INCLUDE_DIRS ${LIBEV_INCLUDE_DIR})
+  set(LIBEV_LIBRARIES ${LIBEV_LIBRARY})
+
+  include(FindPackageHandleStandardArgs)
+  find_package_handle_standard_args(
+    libev DEFAULT_MSG LIBEV_LIBRARY LIBEV_INCLUDE_DIR)
+
+  mark_as_advanced(LIBEV_LIBRARY LIBEV_INCLUDE_DIR)
+endif()

cmake/Modules/Findredox.cmake

+# Try fo find Redox.
+# Once done, this will define:
+#   REDOX_FOUND
+#   REDOX_INCLUDE_DIRS
+#   REDOX_LIBRARIES
+
+find_path(
+    REDOX_INCLUDE_DIR
+    NAMES redox.hpp
+    HINTS ${REDOX_ROOT_DIR})
+
+find_library(
+    REDOX_LIBRARY
+    NAMES redox
+    HINTS ${REDOX_ROOT_DIR}
+    PATH_SUFFIXES ${CMAKE_INSTALL_LIBDIR})
+
+set(REDOX_INCLUDE_DIRS ${REDOX_INCLUDE_DIR})
+set(REDOX_LIBRARIES ${REDOX_LIBRARY})
+
+include (FindPackageHandleStandardArgs)
+    find_package_handle_standard_args(
+    redox DEFAULT_MSG REDOX_LIBRARY REDOX_INCLUDE_DIR)
+
+mark_as_advanced(REDOX_LIBRARY REDOX_INCLUDE_DIR)

src/batsim.cpp

@@ -55,6 +55,9 @@ struct MainArguments
     std::string masterHostName = "master_host";             //!< The name of the SimGrid host which runs scheduler processes and not user tasks
     std::string exportPrefix = "out";                       //!< The filename prefix used to export simulation information
 
+    std::string redis_hostname = "127.0.0.1";               //!< The Redis (data storage) server host name
+    int redis_port = 6379;                                  //!< The Redis (data storage) server port
+
     int limit_machines_count = -1;                          //!< The number of machines to use to compute jobs. -1 : no limit. >= 1 : the number of computation machines
     bool limit_machines_count_by_workload = false;          //!< If set to true, the number of computing machiens to use should be limited by the workload description
 
@@ -84,6 +87,9 @@ int parse_opt (int key, char *arg, struct argp_state *state)
     case 'h':
         mainArgs->allow_space_sharing = true;
         break;
+    case 'H':
+        mainArgs->redis_hostname = arg;
+        break;
     case 'e':
         mainArgs->exportPrefix = arg;
         break;
@@ -111,6 +117,20 @@ int parse_opt (int key, char *arg, struct argp_state *state)
     case 'p':
         mainArgs->energy_used = true;
         break;
+    case 'P':
+    {
+        int ivalue = stoi(arg);
+
+        if ((ivalue <= 0) || (ivalue > 65535))
+        {
+            mainArgs->abort = true;
+            mainArgs->abortReason += "\n  invalid PORT positional argument (" + to_string(ivalue) +
+                                     "): it should be a valid port: integer in range [1,65535].";
+        }
+
+        mainArgs->redis_port = ivalue;
+        break;
+    }
     case 'v':
     {
         string sArg = arg;
@@ -189,10 +209,12 @@ int main(int argc, char * argv[])
     {
         {"export", 'e', "FILENAME_PREFIX", 0, "The export filename prefix used to generate simulation output. Default value: 'out'", 0},
         {"allow-space-sharing", 'h', 0, 0, "Allows space sharing: the same resource can compute several jobs at the same time", 0},
+        {"redis-hostname", 'H', "HOSTNAME", 0, "Sets the host name of the remote Redis (data storage) server.", 0},
         {"limit-machine-count", 'l', "M", 0, "Allows to limit the number of computing machines to use. If M == -1 (default), all the machines described in PLATFORM_FILE are used (but the master_host). If M >= 1, only the first M machines will be used to comupte jobs.", 0},
         {"limit-machine-count-by-worload", 'L', 0, 0, "If set, allows to limit the number of computing machines to use. This number is read from the workload file. If both limit-machine-count and limit-machine-count-by-worload are set, the minimum of the two will be used.", 0},
         {"master-host", 'm', "NAME", 0, "The name of the host in PLATFORM_FILE which will run SimGrid scheduling processes and won't be used to compute tasks. Default value: 'master_host'", 0},
         {"energy-plugin", 'p', 0, 0, "Enables energy-aware experiments", 0},
+        {"redis-port", 'P', "PORT", 0, "Sets the port of the remote Redis (data storage) server.", 0},
         {"quiet", 'q', 0, 0, "Shortcut for --verbosity=quiet", 0},
         {"socket", 's', "FILENAME", 0, "Unix Domain Socket filename. Default value: '/tmp/bat_socket'", 0},
         {"process-tracing", 't', 0, 0, "Enables SimGrid process tracing (shortcut for SimGrid options ----cfg=tracing:1 --cfg=tracing/msg/process:1)", 0},
@@ -350,6 +372,10 @@ int main(int argc, char * argv[])
     context.socket.create_socket(mainArgs.socketFilename);
     context.socket.accept_pending_connection();
 
+    // Redis
+    context.storage.set_instance_key_prefix(absolute_filename(mainArgs.socketFilename));
+    context.storage.connect_to_server(mainArgs.redis_hostname, mainArgs.redis_port);
+
     // Main processes running
     XBT_INFO("Creating jobs_submitter process...");
     JobSubmitterProcessArguments * submitterArgs = new JobSubmitterProcessArguments;

src/context.hpp

@@ -11,6 +11,7 @@
 #include "profiles.hpp"
 #include "export.hpp"
 #include "pstate.hpp"
+#include "storage.hpp"
 
 /**
  * @brief The Batsim context
@@ -25,6 +26,7 @@ struct BatsimContext
     PStateChangeTracer pstate_tracer;               //!< The PStateChangeTracer
     EnergyConsumptionTracer energy_tracer;          //!< The EnergyConsumptionTracer
     CurrentSwitches current_switches;               //!< The current switches
+    RedisStorage storage;                           //!< The RedisStorage
 
     long double energy_first_job_submission = -1;   //!< The amount of consumed energy (J) when the first job is submitted
     long double energy_last_job_completion;         //!< The amount of consumed energy (J) when the last job is completed

src/network.cpp

@@ -435,3 +435,18 @@ int request_reply_scheduler_process(int argc, char *argv[])
     delete args;
     return 0;
 }
+
+std::string absolute_filename(const std::string & filename)
+{
+    xbt_assert(filename.length() > 0);
+
+    // Let's assume filenames starting by "/" are absolute.
+    if (filename[0] == '/')
+        return filename;
+
+    char cwd_buf[PATH_MAX];
+    char * getcwd_ret = getcwd(cwd_buf, PATH_MAX);
+    xbt_assert(getcwd_ret == cwd_buf, "getcwd failed");
+
+    return string(getcwd_ret) + '/' + filename;
+}

src/network.hpp

@@ -89,3 +89,10 @@ private:
  * @return 0
  */
 int request_reply_scheduler_process(int argc, char *argv[]);
+
+/**
+ * @brief Computes the absolute filename of a given file
+ * @param[in] filename The name of the file (not necessarily existing).
+ * @return The absolute filename corresponding to the given filename
+ */
+std::string absolute_filename(const std::string & filename);

src/storage.cpp

+#include "storage.hpp"
+
+#include <simgrid.h>
+
+RedisStorage::RedisStorage()
+{
+    //TODO: wrap redox logging into simgrid?
+}
+
+RedisStorage::~RedisStorage()
+{
+    if (_is_connected)
+    {
+        disconnect();
+    }
+}
+
+void RedisStorage::set_instance_key_prefix(const std::string & key_prefix)
+{
+    _instance_key_prefix = key_prefix;
+}
+
+void RedisStorage::connect_to_server(const std::string & host,
+                                     int port,
+                                     std::function<void (int)> connection_callback)
+{
+    xbt_assert(!_is_connected, "Bad RedisStorage::connect_to_server call: "
+               "Already connected");
+
+    _is_connected = _redox.connect(host, port, connection_callback);
+    xbt_assert(_is_connected, "Error: could not connect to Redis server "
+               "(host='%s', port=%d)", host.c_str(), port);
+
+}
+
+void RedisStorage::disconnect()
+{
+    xbt_assert(_is_connected, "Bad RedisStorage::connect_to_server call: "
+               "Not connected");
+
+    _redox.disconnect();
+}
+
+std::string RedisStorage::get(const std::string & key)
+{
+    xbt_assert(_is_connected, "Bad RedisStorage::get call: Not connected");
+    return _redox.get(build_key(key));
+}
+
+bool RedisStorage::set(const std::string &key, const std::string &value)
+{
+    xbt_assert(_is_connected, "Bad RedisStorage::get call: Not connected");
+    return _redox.set(build_key(key), value);
+}
+
+bool RedisStorage::del(const std::string &key)
+{
+    xbt_assert(_is_connected, "Bad RedisStorage::get call: Not connected");
+    return _redox.del(build_key(key));
+}
+
+std::string RedisStorage::instance_key_prefix() const
+{
+    return _instance_key_prefix;
+}
+
+std::string RedisStorage::key_subparts_separator() const
+{
+    return _key_subparts_separator;
+}
+
+std::string RedisStorage::build_key(const std::string & user_given_key) const
+{
+    return _instance_key_prefix + _key_subparts_separator + user_given_key;
+}

src/storage.hpp

+#pragma once
+
+#include <string>
+
+#include <redox.hpp>
+
+/**
+ * @brief Wrapper class around Redox, a Redis client library.
+ * @details This class provides blocking methods that communicate
+ * with a Redis server. Furthermore, this class defines an instance
+ * key prefix and adds this prefix to every user-given key, in order
+ * to make the concurrent executions of Batsim easier.
+ */
+class RedisStorage
+{
+public:
+    /**
+     * @brief Builds a RedisStorage
+     */
+    RedisStorage();
+    /**
+     * @brief Destroys a RedisStorage
+     */
+    ~RedisStorage();
+
+    /**
+     * @brief Sets the instance key prefix
+     * @param[in] key_prefix The new key prefix
+     */
+    void set_instance_key_prefix(const std::string & key_prefix);
+
+    /**
+     * @brief Connects to a Redis server
+     * @param[in] host The server hostname
+     * @param[in] port The server port
+     * @param connection_callback The callback function to call on connection
+     */
+    void connect_to_server(const std::string & host = redox::REDIS_DEFAULT_HOST,
+                           int port = redox::REDIS_DEFAULT_PORT,
+                           std::function< void(int)> connection_callback = nullptr);
+    /**
+     * @brief Disconnects from the server
+     */
+    void disconnect();
+
+    /**
+     * @brief Gets the value associated with the given key
+     * @param[in] key The key
+     * @return The value associated with the given key
+     */
+    std::string get(const std::string & key);
+
+    /**
+     * @brief Sets a key-value in the Redis server
+     * @param[in] key The key
+     * @param[in] value The value to associate with the key
+     * @return true if succeeded, false otherwise.
+     */
+    bool set(const std::string & key,
+             const std::string & value);
+
+    /**
+     * @brief Deletes a key-value association from the Redis server
+     * @param[in] key The key which should be deleted from the Redis server
+     * @return true if succeeded, false otherwise.
+     */
+    bool del(const std::string & key);
+
+    /**
+     * @brief Returns the instance key prefix.
+     * @return The instance key prefix.
+     */
+    std::string instance_key_prefix() const;
+
+    /**
+     * @brief Returns the key subparts separator.
+     * @return The key subparts separator.
+     */
+    std::string key_subparts_separator() const;
+
+private:
+    /**
+     * @brief Build a final key from a user-given key.
+     * @param[in] user_given_key The user-given key
+     * @return The real key corresponding to the user-given key
+     */
+    std::string build_key(const std::string & user_given_key) const;
+
+private:
+    bool _is_connected = false; //!< True if and only if the instance is connected to a Redis server
+    redox::Redox _redox; //!< The Redox instance
+    std::string _instance_key_prefix = ""; //!< The instance key prefix, which is added before to every user-given key.
+    std::string _key_subparts_separator = ":"; //!< The key subparts separator, which is put between the instance key prefix and the user-given key.
+};