BIG COMMIT: unix sockets -> ZMQ sockets

CMakeLists.txt

@@ -76,6 +76,10 @@ include_directories(${HIREDIS_INCLUDE_DIRS})
 find_package(libev REQUIRED)
 include_directories(${LIBEV_INCLUDE_DIRS})
 
+# ZeroMQ dependency
+find_package(ZMQ REQUIRED)
+include_directories(${ZMQ_INCLUDE_DIRS})
+
 ################
 # Source files #
 ################
@@ -103,7 +107,8 @@ target_link_libraries(batsim
                       ${OPENSSL_LIBRARIES}
                       ${REDOX_LIBRARY}
                       ${LIBEV_LIBRARY}
-                      ${HIREDIS_LIBRARY})
+                      ${HIREDIS_LIBRARY}
+                      ${ZMQ_LIBRARIES})
 
 ################
 # Installation #

cmake/Modules/FindZMQ.cmake

+# - Try to find ZMQ
+# Once done this will define
+# ZMQ_FOUND - System has ZMQ
+# ZMQ_INCLUDE_DIRS - The ZMQ include directories
+# ZMQ_LIBRARIES - The libraries needed to use ZMQ
+# ZMQ_DEFINITIONS - Compiler switches required for using ZMQ
+
+find_path ( ZMQ_INCLUDE_DIR zmq.h )
+find_library ( ZMQ_LIBRARY NAMES zmq )
+
+set ( ZMQ_LIBRARIES ${ZMQ_LIBRARY} )
+set ( ZMQ_INCLUDE_DIRS ${ZMQ_INCLUDE_DIR} )
+
+include ( FindPackageHandleStandardArgs )
+# handle the QUIETLY and REQUIRED arguments and set ZMQ_FOUND to TRUE
+# if all listed variables are TRUE
+find_package_handle_standard_args ( ZMQ DEFAULT_MSG ZMQ_LIBRARY ZMQ_INCLUDE_DIR )

schedulers/pybatsim/batsim/batsim.py

@@ -6,21 +6,19 @@ import json
 import os
 import re
 import redis
-import socket
 import struct
 import sys
+import zmq
 
 class Batsim(object):
 
-    def __init__(self, scheduler, redis_prefix = None,
+    def __init__(self, scheduler, redis_prefix = 'default',
                  redis_hostname = 'localhost', redis_port = 6379,
                  validatingmachine = None,
-                 server_address = '/tmp/bat_socket', verbose=0):
-        self.server_address = server_address
+                 socket_endpoint = 'tcp://*:5555', verbose=0):
+        self.socket_endpoint = socket_endpoint
         self.verbose = verbose
 
-        if redis_prefix == None:
-            redis_prefix = os.path.abspath(server_address)
         self.redis = DataStorage(redis_prefix, redis_hostname, redis_port)
         self.jobs = dict()
 
@@ -32,15 +30,10 @@ class Batsim(object):
             self.scheduler = validatingmachine(scheduler)
 
         #open connection
-        self._connection = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
-        print("[BATSIM]: connecting to %r" % server_address)
-        try:
-            self._connection.connect(server_address)
-            if self.verbose > 1: print('[BATSIM]: connected')
-        except socket.error:
-            print("[BATSIM]: socket error")
-            raise
-
+        self._context = zmq.Context()
+        self._connection = self._context.socket(zmq.REP)
+        print("[BATSIM]: binding to {addr}".format(addr=self.socket_endpoint))
+        self._connection.bind(self.socket_endpoint)
 
         #initialize some public attributes
         self.last_msg_recv_time = -1
@@ -145,14 +138,8 @@ class Batsim(object):
         return('%.*f' % (6, t))
 
     def _read_bat_msg(self):
-        lg_str = self._connection.recv(4)
-
-        if not lg_str:
-            print("[BATSIM]: connection closed by batsim core")
-            return False
+        msg = self._connection.recv().decode()
 
-        lg = struct.unpack("I",lg_str)[0]
-        msg = self._connection.recv(lg).decode()
         if self.verbose > 0: print('[BATSIM]: from batsim (%r) : %r' % (lg, msg))
         sub_msgs = msg.split('|')
         data = sub_msgs[0].split(":")
@@ -167,13 +154,15 @@ class Batsim(object):
 
         # TODO: handle Z, f and F messages.
 
+        finished_received = False
+
         for i in range(1, len(sub_msgs)):
             data = sub_msgs[i].split(':')
             if data[1] == 'A':
                 self.nb_res = int(self.redis.get('nb_res'))
             elif data[1] == 'Z':
                 print("All jobs have been submitted and completed!")
-                print("TODO: inform schedulers about it...")
+                finished_received = True
             elif data[1] == 'R':
                 self.scheduler.onJobRejection()
             elif data[1] == 'N':
@@ -218,10 +207,8 @@ class Batsim(object):
             msg +=  self._time_to_str(self.time()) +":N"
 
         if self.verbose > 0: print("[BATSIM]:  to  batsim : %r" % msg)
-        lg = struct.pack("i",int(len(msg)))
-        self._connection.sendall(lg)
-        self._connection.sendall(msg.encode())
-        return True
+        self._connection.send(msg.encode())
+        return not finished_received
 
 # High-level access to the Redis data storage system
 class DataStorage(object):

schedulers/pybatsim/launcher.py

@@ -4,15 +4,17 @@
 Run PyBatsim Sschedulers.
 
 Usage:
-    launcher.py <scheduler> [-p] [-v] [-s <socket>] [-r <redis port number>] [-o <options>]
+    launcher.py <scheduler> [-o <options_string>] [options]
 
 Options:
-    -h --help                                      Show this help message and exit.
-    -v --verbose                                   Be verbose.
-    -p --protect                                   Protect the scheduler using a validating machine.
-    -s --socket=<socket>                           Socket to use [default: /tmp/bat_socket]
-    -r --redisport=<port number>                   Redis server port number
-    -o --options=<options>                         A Json string to pass to the scheduler [default: {}]
+    -h --help                           Show this help message and exit.
+    -v --verbose                        Be verbose.
+    -p --protect                        Protect the scheduler using a validating machine.
+    -s --socket-endpoint=<endpoint>     Batsim socket endpoint to use [default: tcp://*:5555]
+    --redis-hostname=<hostname>         Redis server hostname [default: 127.0.0.1]
+    --redis-port=<port>                 Redis server port number [default: 6379]
+    --redis-prefix=<prefix>             Redis prefix [default: default]
+    -o --options=<options_string>       A Json string to pass to the scheduler [default: {}]
 '''
 
 
@@ -59,6 +61,7 @@ def instanciate_scheduler(name, options):
 if __name__ == "__main__":
     #Retrieve arguments
     arguments = docopt(__doc__, version='1.0.0rc2')
+
     if arguments['--verbose']:
         verbose=999
     else:
@@ -72,15 +75,12 @@ if __name__ == "__main__":
     options = json.loads(arguments['--options'])
 
     scheduler_filename = arguments['<scheduler>']
-    socket = arguments['--socket']
-
-    # Redis port
-    if arguments['--redisport']:
-    	redisport = int(arguments['--redisport'])
-    else:
-	redisport = 6379
+    socket_endpoint = arguments['--socket-endpoint']
 
-    # TODO: add Redis arguments (hostname, prefix)
+    # Redis
+    redis_hostname = str(arguments['--redis-hostname'])
+    redis_port = int(arguments['--redis-port'])
+    redis_prefix = str(arguments['--redis-prefix'])
 
     print "Starting simulation..."
     print "Scheduler:", scheduler_filename
@@ -88,7 +88,9 @@ if __name__ == "__main__":
     time_start = time.time()
     scheduler = instanciate_scheduler(scheduler_filename, options=options)
 
-    bs = Batsim(scheduler, validatingmachine=vm, server_address=socket, verbose=verbose, redis_port=redisport)
+    bs = Batsim(scheduler, validatingmachine=vm, socket_endpoint=socket_endpoint,
+                verbose=verbose, redis_hostname=redis_hostname,
+                redis_port=redis_port, redis_prefix=redis_prefix)
 
     bs.start()
     time_ran = str(timedelta(seconds=time.time()-time_start))

src/batsim.cpp

@@ -128,11 +128,12 @@ Most common options:
                                     outputs energy-related files.
 
 Execution context options:
-  -s, --socket <socket_file>        The Unix Domain Socket filename
-                                    [default: /tmp/bat_socket].
+  -s, --socket-endpoint <endpoint>  The Decision process socket endpoint
+                                    Decision process [default: tcp://localhost:5555].
   --redis-hostname <redis_host>     The Redis server hostname
                                     [default: 127.0.0.1]
   --redis-port <redis_port>         The Redis server port [default: 6379].
+  --redis-prefix <prefix>           The Redis prefix [default: default].
 
 Output options:
   -e, --export <prefix>             The export filename prefix used to generate
@@ -295,7 +296,7 @@ Other options:
 
     // Execution context options
     // *************************
-    main_args.socket_filename = args["--socket"].asString();
+    main_args.socket_endpoint = args["--socket-endpoint"].asString();
     main_args.redis_hostname = args["--redis-hostname"].asString();
     try { main_args.redis_port = args["--redis-port"].asLong(); }
     catch(const std::exception &)
@@ -304,6 +305,7 @@ Other options:
                   args["--redis-port"].asString().c_str());
         error = true;
     }
+    main_args.redis_prefix = args["--redis-prefix"].asString();
 
     // Output options
     // **************
@@ -504,13 +506,12 @@ void start_initial_simulation_processes(const MainArguments & main_args,
         XBT_INFO("The process '%s' has been created.", submitter_instance_name.c_str());
     }
 
-
     if (!is_batexec)
     {
         XBT_DEBUG("Creating the 'server' process...");
         ServerProcessArguments * server_args = new ServerProcessArguments;
         server_args->context = context;
-        MSG_process_create("server", uds_server_process, (void*)server_args, master_machine->host);
+        MSG_process_create("server", server_process, (void*)server_args, master_machine->host);
         XBT_INFO("The process 'server' has been created.");
     }
 }
@@ -576,13 +577,12 @@ int main(int argc, char * argv[])
     if (main_args.program_type == ProgramType::BATSIM)
     {
         // Let's prepare Redis's connection
-        context.storage.set_instance_key_prefix(absolute_filename(main_args.socket_filename)); // TODO: main argument?
+        context.storage.set_instance_key_prefix(main_args.redis_prefix);
         context.storage.connect_to_server(main_args.redis_hostname, main_args.redis_port);
 
         // Let's create the socket
-        // TODO: check that the socket is not currently being used
-        context.socket.create_socket(main_args.socket_filename);
-        context.socket.accept_pending_connection();
+        context.zmq_socket = new zmq::socket_t(context.zmq_context, ZMQ_REQ);
+        context.zmq_socket->connect(main_args.socket_endpoint);
 
         // Let's store some metadata about the current instance in the data storage
         context.storage.set("nb_res", std::to_string(context.machines.nb_machines()));
@@ -599,6 +599,9 @@ int main(int argc, char * argv[])
     // Simulation main loop, handled by MSG
     msg_error_t res = MSG_main();
 
+    delete context.zmq_socket;
+    context.zmq_socket = nullptr;
+
     // If SMPI had been used, it should be finalized
     if (context.smpi_used)
         SMPI_finalize();

src/batsim.hpp

@@ -67,7 +67,7 @@ struct MainArguments
     bool energy_used;                                       //!< True if and only if the SimGrid energy plugin should be used.
 
     // Execution context
-    std::string socket_filename;                            //!< The Unix Domain Socket filename
+    std::string socket_endpoint;                            //!< The Decision process socket endpoint
     std::string redis_hostname;                             //!< The Redis (data storage) server host name
     int redis_port;                                         //!< The Redis (data storage) server port
     std::string redis_prefix;                               //!< The Redis (data storage) instance prefix

src/context.hpp

@@ -8,6 +8,8 @@
 #include <chrono>
 #include <vector>
 
+#include <zmq.hpp>
+
 #include "exact_numbers.hpp"
 #include "export.hpp"
 #include "jobs.hpp"
@@ -29,7 +31,9 @@ typedef std::chrono::time_point<std::chrono::high_resolution_clock> my_timestamp
  */
 struct BatsimContext
 {
-    UnixDomainSocket socket;                        //!< The UnixDomainSocket
+    zmq::context_t zmq_context;                     //!< The Zero MQ context
+    zmq::socket_t * zmq_socket = nullptr;           //!< The Zero MQ socket (REQ)
+
     Machines machines;                              //!< The machines
     Workloads workloads;                            //!< The workloads
     Workflows workflows;                            //!< The workflows

src/ipp.hpp

@@ -218,7 +218,7 @@ struct RequestReplyProcessArguments
 };
 
 /**
- * @brief The arguments of the uds_server_process process
+ * @brief The arguments of the server_process process
  */
 struct ServerProcessArguments
 {

src/network.cpp

@@ -168,17 +168,28 @@ int request_reply_scheduler_process(int argc, char *argv[])
                "buffer size is %d. Since information will be lost now or in a near "
                "future, the simulation is aborted.",
                nb_printed_char, date_buf_size);
-    char *sendBuf = (char*) args->send_buffer.c_str();
-    XBT_DEBUG("Buffer received in REQ-REP: '%s'", sendBuf);
+    char *send_buf = (char*) args->send_buffer.c_str();
+    XBT_DEBUG("Buffer received in REQ-REP: '%s'", send_buf);
 
-    context->socket.send(sendBuf);
+    // Let's make sure the message is sent as UTF-8
+    string utf8_message = boost::locale::conv::to_utf<char>(send_buf, "UTF-8");
+
+    // Send the message
+    XBT_INFO("Sending '%s'", utf8_message.c_str());
+    context->zmq_socket->send(utf8_message.data(), utf8_message.size());
 
     auto start = chrono::steady_clock::now();
     string message_received;
 
     try
     {
-         message_received = context->socket.receive();
+        // Get the reply
+        zmq::message_t reply;
+        context->zmq_socket->recv(&reply);
+
+        string raw_message_received((char *)reply.data(), reply.size());
+        message_received = boost::locale::conv::from_utf(raw_message_received, "UTF-8");
+        XBT_INFO("Received '%s'", message_received.c_str());
     }
     catch(const std::runtime_error & error)
     {
@@ -464,23 +475,23 @@ int request_reply_scheduler_process(int argc, char *argv[])
                 xbt_assert(parts2.size() == 3, "Invalid event received ('%s'): messages to ask the waiting time must be composed of 3 parts separated by ':'",
                            event_string.c_str());
 
-		SchedWaitAnswerMessage * message = new SchedWaitAnswerMessage;
+        SchedWaitAnswerMessage * message = new SchedWaitAnswerMessage;
 
                 vector<string> parts3;
                 boost::split(parts3, parts2[2], boost::is_any_of(","), boost::token_compress_on);
-		//                xbt_assert(parts3.size() == 3, "Invalid event received ('%s'): invalid wait answer message content ('%s'): it must be"
-		//                           " formated like R,T,W where R is the number of requested resource, T the requested walltime and W the waiting time",
-		//                           event_string.c_str(), parts2[2].c_str());
+        //                xbt_assert(parts3.size() == 3, "Invalid event received ('%s'): invalid wait answer message content ('%s'): it must be"
+        //                           " formated like R,T,W where R is the number of requested resource, T the requested walltime and W the waiting time",
+        //                           event_string.c_str(), parts2[2].c_str());
 
                 int nb_resources = std::stoi(parts3[1]);
-		double processing_time = std::stod(parts3[2]);
-		double expected_time = std::stod(parts3[3]);
-
-		message->submitter_name = parts3[0];
-		message->nb_resources = nb_resources;
-		message->processing_time = processing_time;
-		message->expected_time = expected_time;
-		
+        double processing_time = std::stod(parts3[2]);
+        double expected_time = std::stod(parts3[3]);
+
+        message->submitter_name = parts3[0];
+        message->nb_resources = nb_resources;
+        message->processing_time = processing_time;
+        message->expected_time = expected_time;
+
                 send_message("server", IPMessageType::SCHED_WAIT_ANSWER, (void*) message);
             } break; // End of case received_stamp == ANSWER_WAIT
 

src/server.cpp

@@ -20,7 +20,7 @@ XBT_LOG_NEW_DEFAULT_CATEGORY(server, "server"); //!< Logging
 
 using namespace std;
 
-int uds_server_process(int argc, char *argv[])
+int server_process(int argc, char *argv[])
 {
     (void) argc;
     (void) argv;
@@ -155,8 +155,8 @@ int uds_server_process(int argc, char *argv[])
             nb_completed_jobs++;
             Job * job = context->workloads.job_at(message->job_id);
 
-            XBT_INFO("Job %s-%d COMPLETED. %d jobs completed so far",
-                     job->workload->name.c_str(),job->number, nb_completed_jobs);
+            XBT_INFO("Job %s!%d COMPLETED. %d jobs completed so far",
+                     job->workload->name.c_str(), job->number, nb_completed_jobs);
 
             send_buffer += '|' + std::to_string(MSG_get_clock()) + ":C:" + message->job_id.to_string();
             XBT_DEBUG( "Message to send to scheduler: %s", send_buffer.c_str());
@@ -435,32 +435,32 @@ int uds_server_process(int argc, char *argv[])
 
         case IPMessageType::SCHED_WAIT_ANSWER:
         {
-	  SchedWaitAnswerMessage * message = new SchedWaitAnswerMessage;
-	  *message = *( (SchedWaitAnswerMessage *) task_data->data);
+      SchedWaitAnswerMessage * message = new SchedWaitAnswerMessage;
+      *message = *( (SchedWaitAnswerMessage *) task_data->data);
 
-	  //	  Submitter * submitter = origin_of_wait_queries.at({message->nb_resources,message->processing_time});
+      //	  Submitter * submitter = origin_of_wait_queries.at({message->nb_resources,message->processing_time});
 
-	  
-	  dsend_message(message->submitter_name, IPMessageType::SCHED_WAIT_ANSWER, (void*) message);
 
-	  //	  origin_of_wait_queries.erase({message->nb_resources,message->processing_time});
+      dsend_message(message->submitter_name, IPMessageType::SCHED_WAIT_ANSWER, (void*) message);
+
+      //	  origin_of_wait_queries.erase({message->nb_resources,message->processing_time});
 
         } break; // end of case SCHED_WAIT_ANSWER
 
         case IPMessageType::WAIT_QUERY:
         {
-	  WaitQueryMessage  * message = (WaitQueryMessage *) task_data->data;
-
-	  //	  XBT_INFO("received : %s , %s\n", to_string(message->nb_resources).c_str(), to_string(message->processing_time).c_str());
-	  send_buffer += "|" + std::to_string(MSG_get_clock()) + ":Q:"
-	    + message->submitter_name.c_str() + ","
-	    + to_string(message->nb_resources).c_str() + ","
-	    + boost::lexical_cast<string>(message->processing_time).c_str();
-	  //	  XBT_INFO("INFO!!! Message to send to scheduler : '%s'", send_buffer.c_str());
-
-	  //Submitter * submitter = submitters.at(message->submitter_name);
-	  //origin_of_wait_queries[{message->nb_resources,message->processing_time}] = submitter;
-	  
+      WaitQueryMessage  * message = (WaitQueryMessage *) task_data->data;
+
+      //	  XBT_INFO("received : %s , %s\n", to_string(message->nb_resources).c_str(), to_string(message->processing_time).c_str());
+      send_buffer += "|" + std::to_string(MSG_get_clock()) + ":Q:"
+        + message->submitter_name.c_str() + ","
+        + to_string(message->nb_resources).c_str() + ","
+        + boost::lexical_cast<string>(message->processing_time).c_str();
+      //	  XBT_INFO("INFO!!! Message to send to scheduler : '%s'", send_buffer.c_str());
+
+      //Submitter * submitter = submitters.at(message->submitter_name);
+      //origin_of_wait_queries[{message->nb_resources,message->processing_time}] = submitter;
+
         } break; // end of case WAIT_QUERY
 
         case IPMessageType::SWITCHED_ON:

src/server.hpp

@@ -11,4 +11,4 @@
  * @param[in] argv The arguments' values
  * @return 0
  */
-int uds_server_process(int argc, char *argv[]);
+int server_process(int argc, char *argv[]);

test/pybatsim_tests.yaml

@@ -21,8 +21,8 @@ implicit_instances:
       timeout: 10
       working_directory: ${base_working_directory}
       output_directory: ${base_output_directory}/results/${instance_id}
-      batsim_command: batsim -p ${platform[filename]} -w ${workload[filename]} -e ${output_directory}/out -s ${output_directory}/socket --mmax-workload ${energy_string}
-      sched_command: python2 ${batsim_dir}/schedulers/pybatsim/launcher.py ${pybatsim_algo[algo_name]} -s ${output_directory}/socket
+      batsim_command: batsim -p ${platform[filename]} -w ${workload[filename]} -e ${output_directory}/out --mmax-workload ${energy_string}
+      sched_command: python2 ${batsim_dir}/schedulers/pybatsim/launcher.py ${pybatsim_algo[algo_name]}
 
 commands_before_instances:
   - ${batsim_dir}/test/is_batsim_dir.py ${base_working_directory}

test/test_batexec.yaml

@@ -19,8 +19,8 @@ implicit_instances:
     generic_instance:
       timeout: 10
       working_directory: ${base_working_directory}
-      output_directory: ${base_output_directory}/results/${pybatsim_algo[name]}_${workload[name]}_${platform[name]}
-      batsim_command: batsim -p ${platform[filename]} -w ${workload[filename]} -e ${output_directory}/out -s ${output_directory}/socket --mmax-workload --batexec
+      output_directory: ${base_output_directory}/results/${workload[name]}_${platform[name]}
+      batsim_command: batsim -p ${platform[filename]} -w ${workload[filename]} -e ${output_directory}/out --mmax-workload --batexec
       sched_command: echo "I do not even exist."
 
 commands_before_instances:

test/test_energy.yaml

@@ -23,8 +23,8 @@ implicit_instances:
       timeout: 10
       working_directory: ${base_working_directory}
       output_directory: ${base_output_directory}/results/${pybatsim_algo[name]}_${workload[name]}_${platform[name]}
-      batsim_command: batsim -p ${platform[filename]} -w ${workload[filename]} -E -e ${output_directory}/out -s ${output_directory}/socket --mmax-workload
-      sched_command: python2 ${batsim_dir}/schedulers/pybatsim/launcher.py ${pybatsim_algo[algo_name]} -s ${output_directory}/socket
+      batsim_command: batsim -p ${platform[filename]} -w ${workload[filename]} -E -e ${output_directory}/out --mmax-workload --redis-prefix ${instance_id}
+      sched_command: python2 ${batsim_dir}/schedulers/pybatsim/launcher.py ${pybatsim_algo[algo_name]} --redis-prefix ${instance_id}
 
 commands_before_instances:
   - ${batsim_dir}/test/is_batsim_dir.py ${base_working_directory}

test/test_long_simulation_time.yaml

@@ -22,8 +22,8 @@ implicit_instances:
       timeout: 10
       working_directory: ${base_working_directory}
       output_directory: ${base_output_directory}/results/${pybatsim_algo[name]}_${workload[name]}_${platform[name]}
-      batsim_command: batsim -p ${platform[filename]} -w ${workload[filename]} -e ${output_directory}/out -s ${output_directory}/socket --mmax-workload
-      sched_command: python2 ${batsim_dir}/schedulers/pybatsim/launcher.py ${pybatsim_algo[algo_name]} -s ${output_directory}/socket
+      batsim_command: batsim -p ${platform[filename]} -w ${workload[filename]} -e ${output_directory}/out --redis-prefix ${instance_id} --mmax-workload
+      sched_command: python2 ${batsim_dir}/schedulers/pybatsim/launcher.py ${pybatsim_algo[algo_name]} --redis-prefix ${instance_id}
 
 commands_before_instances:
   - ${batsim_dir}/test/is_batsim_dir.py ${base_working_directory}

test/test_no_energy.yaml

@@ -20,11 +20,13 @@ implicit_instances:
       pybatsim_algo:
         - {"name":"filler", "algo_name":"fillerSched"}
     generic_instance:
+      variables:
+        socket_port: "$((${instance_number} + 5000))"
       timeout: 10
       working_directory: ${base_working_directory}
       output_directory: ${base_output_directory}/results/${pybatsim_algo[name]}_${workload[name]}_${platform[name]}
-      batsim_command: batsim -p ${platform[filename]} -w ${workload[filename]} -e ${output_directory}/out -s ${output_directory}/socket --mmax-workload
-      sched_command: python2 ${batsim_dir}/schedulers/pybatsim/launcher.py ${pybatsim_algo[algo_name]} -s ${output_directory}/socket
+      batsim_command: batsim -p ${platform[filename]} -w ${workload[filename]} -e ${output_directory}/out --mmax-workload --redis-prefix ${instance_id} --socket-endpoint="tcp://localhost:${socket_port}"
+      sched_command: python2 ${batsim_dir}/schedulers/pybatsim/launcher.py ${pybatsim_algo[algo_name]} --redis-prefix ${instance_id} --socket-endpoint="tcp://*:${socket_port}"
 
 commands_before_instances:
   - ${batsim_dir}/test/is_batsim_dir.py ${base_working_directory}

test/test_same_submit_time.yaml

@@ -22,8 +22,8 @@ implicit_instances:
       timeout: 10
       working_directory: ${base_working_directory}
       output_directory: ${base_output_directory}/results/${instance_id}
-      batsim_command: batsim -p ${platform[filename]} -w ${workload[filename]} -e ${output_directory}/out -s ${output_directory}/socket --mmax-workload -vnetwork-only
-      sched_command: python2 ${batsim_dir}/schedulers/pybatsim/launcher.py ${pybatsim_algo[algo_name]} -s ${output_directory}/socket
+      batsim_command: batsim -p ${platform[filename]} -w ${workload[filename]} -e ${output_directory}/out --redis-prefix ${instance_id} --mmax-workload -vnetwork-only
+      sched_command: python2 ${batsim_dir}/schedulers/pybatsim/launcher.py ${pybatsim_algo[algo_name]} --redis-prefix ${instance_id}
 
       commands_after_execution:
         - cat ${output_directory}/batsim.stderr | grep -o ":S:.*|\|:S:.*'" > ${output_directory}/messages.txt

test/test_smpi.yaml

@@ -25,8 +25,8 @@ implicit_instances:
       timeout: 10
       working_directory: ${base_working_directory}
       output_directory: ${base_output_directory}/results/${pybatsim_algo[name]}_${workload[name]}_${platform[name]}
-      batsim_command: batsim -p ${platform[filename]} -w ${workload[filename]} -e ${output_directory}/out -s ${output_directory}/socket --mmax-workload
-      sched_command: python2 ${batsim_dir}/schedulers/pybatsim/launcher.py ${pybatsim_algo[algo_name]} -s ${output_directory}/socket
+      batsim_command: batsim -p ${platform[filename]} -w ${workload[filename]} -e ${output_directory}/out --redis-prefix ${instance_id} --mmax-workload
+      sched_command: python2 ${batsim_dir}/schedulers/pybatsim/launcher.py ${pybatsim_algo[algo_name]} --redis-prefix ${instance_id}
 
 commands_before_instances:
   - ${batsim_dir}/test/is_batsim_dir.py ${base_working_directory}

test/test_smpi_batexec.yaml

@@ -24,7 +24,7 @@ implicit_instances:
       timeout: 10
       working_directory: ${base_working_directory}
       output_directory: ${base_output_directory}/results/${pybatsim_algo[name]}_${workload[name]}_${platform[name]}
-      batsim_command: batsim -p ${platform[filename]} -w ${workload[filename]} -e ${output_directory}/out -s ${output_directory}/socket --mmax-workload --batexec
+      batsim_command: batsim -p ${platform[filename]} -w ${workload[filename]} -e ${output_directory}/out --mmax-workload --batexec
       sched_command: echo 'I do not exist'
 
 commands_before_instances:

test/test_time_sharing.yaml

@@ -28,8 +28,8 @@ implicit_instances:
       timeout: 20
       working_directory: ${base_working_directory}
       output_directory: ${base_output_directory}/results/${pybatsim_algo[name]}_${workload[name]}_${platform[name]}_${random_seed}
-      batsim_command: batsim -p ${platform[filename]} -w ${workload[filename]} -e ${output_directory}/out -s ${output_directory}/socket --mmax-workload --allow-time-sharing
-      sched_command: python2 ${batsim_dir}/schedulers/pybatsim/launcher.py ${pybatsim_algo[algo_name]} -s ${output_directory}/socket
+      batsim_command: batsim -p ${platform[filename]} -w ${workload[filename]} -e ${output_directory}/out --redis-prefix ${instance_id} --mmax-workload --allow-time-sharing
+      sched_command: python2 ${batsim_dir}/schedulers/pybatsim/launcher.py ${pybatsim_algo[algo_name]} --redis-prefix ${instance_id}
 
 commands_before_instances:
   - ${batsim_dir}/test/is_batsim_dir.py ${base_working_directory}