Updates connection of components

Author: Mariana Miranda

files/core_config_file

@@ -1,6 +1,6 @@
 controller: core
-core_address: 0.0.0.0:50053
+core_address: 0.0.0.0:50051
 control_type: 1
 system_limit: 1000000
 housekeeping_rules_file: ../files/posix_layer_housekeeping_rules_static_op
-policies_rules_file: ../files/static_rules_with_time_job
+policies_rules_file: ../files/static_rules_with_time_file_job

files/local_config_file

@@ -1,3 +1,3 @@
 controller: local
-core_address: 0.0.0.0:50053
-local_address: 0.0.0.0:50052
+core_address: 0.0.0.0:50051
+local_address: 0.0.0.0:50053

files/local_config_file_2

@@ -0,0 +1,3 @@
+controller: local
+core_address: 0.0.0.0:50051
+local_address: 0.0.0.0:50052

files/posix_layer_housekeeping_rules_static_op

@@ -2,7 +2,7 @@
 2 create_channel 2000 posix 2000 write no_op
 3 create_channel 3000 posix 3000 open no_op
 4 create_channel 4000 posix 4000 close no_op
-5 create_object 1000 1 posix read no_op drl 10000 1073741824
-6 create_object 2000 1 posix write no_op drl 10000 1073741824
-7 create_object 3000 1 posix open no_op drl 10000 1073741824
-8 create_object 4000 1 posix close no_op drl 10000 1073741824
+5 create_object 1000 1 posix no_op no_op drl 10000 1073741824
+6 create_object 2000 1 posix no_op no_op drl 10000 1073741824
+7 create_object 3000 1 posix no_op no_op drl 10000 1073741824
+8 create_object 4000 1 posix no_op no_op drl 10000 1073741824

include/cheferd/controller/control_application.hpp

@@ -18,6 +18,14 @@ using namespace std::chrono;
 
 namespace cheferd {
 
+//  Detailed information related to a data plane stage.
+struct StageInfo {
+    std::string m_stage_name;
+    std::string m_stage_env;
+    std::string m_stage_user;
+    std::string m_local_address;
+};
+
 class ControlApplication {
 
 protected:

include/cheferd/controller/core_control_application.hpp

@@ -15,13 +15,6 @@ namespace cheferd {
 // Threshold to decide if a new should be enforced.
 #define IOPS_THRESHOLD 10
 
-//  Detailed information related to a data plane stage.
-struct StageInfoCore {
-    std::string m_stage_name;
-    std::string m_stage_env;
-    std::string m_stage_user;
-    std::string m_local_address;
-};
 
 /**
  * CoreControlApplication class.
@@ -42,7 +35,8 @@ class CoreControlApplication : public ControlApplication {
     std::queue<std::string> local_queue;
 
     // Related to the registration of new data plane stages.
-    std::queue<std::tuple<std::string, std::pair<std::string, std::string>, std::string>>
+    std::mutex pending_register_stage_lock_;
+    std::queue<std::unique_ptr<StageInfo>>
         local_to_data_queue_;
 
     // Type of control (e.g., Static, Dynamic, ...).
@@ -56,9 +50,9 @@ class CoreControlApplication : public ControlApplication {
     // controller. (e.g., <"0.0.0.0:50052", ["job1+1", "job1+2"]>)
     std::unordered_map<std::string, std::vector<std::string>> local_to_stages;
 
-    // StageID -> StageInfoCore. Details each stage information.
-    // (e.g., <"job1+1", StageInfoCore>)
-    std::unordered_map<std::string, StageInfoCore> stage_info_detailed;
+    // StageID -> StageInfo. Details each stage information.
+    // (e.g., <"job1+1", StageInfo>)
+    std::unordered_map<std::string, std::unique_ptr<StageInfo>> stage_info_detailed;
 
     // AppID -> < LocalControllerID -> [Envs] >. Holds info about the location and  of each job
     // instance. (e.g., <"job1", <"0.0.0.0:50052", [1,2]>>
@@ -157,7 +151,7 @@ class CoreControlApplication : public ControlApplication {
 
     void operator() () override;
 
-    LocalControllerSession* register_local_controller_session (
+    void register_local_controller_session (
         const std::string& local_controller_address);
 
     void register_stage_session (const std::string& local_controller_address,

include/cheferd/controller/local_control_application.hpp

@@ -23,7 +23,7 @@
 using controllers_grpc_interface::ACK;
 using controllers_grpc_interface::ConnectReply;
 using controllers_grpc_interface::ConnectRequest;
-using controllers_grpc_interface::StageInfo;
+using controllers_grpc_interface::StageInfoConnect;
 using controllers_grpc_interface::StageReadyRaw;
 using controllers_grpc_interface::StatsGlobalMap;
 using grpc::Server;
@@ -55,7 +55,10 @@ class LocalControlApplication : public GlobalToLocal::Service, public ControlApp
     std::unordered_map<std::string, std::unique_ptr<DataPlaneSession>> data_sessions_;
     std::unordered_map<std::string, std::unique_ptr<DataPlaneSession>> preparing_data_sessions_;
 
-    std::unordered_map<int, std::unique_ptr<HandshakeSession>> pending_data_sessions_;
+    // Related to the registration of new local controller sessions.
+    std::mutex pending_data_plane_sessions_lock_;
+    std::queue<std::unique_ptr<HandshakeSession>> pending_data_sessions_;
+
 
     std::unordered_map<std::string, std::vector<std::pair<int, int>>> operation_to_channel_object;
 
@@ -81,7 +84,7 @@ class LocalControlApplication : public GlobalToLocal::Service, public ControlApp
      * @param index
      * @return
      */
-    PStatus stage_handshake (int index);
+    void handle_data_plane_sessions ();
 
     /**
      * Sleep:
@@ -92,8 +95,8 @@ class LocalControlApplication : public GlobalToLocal::Service, public ControlApp
      * CallStageHandshake:
      * @return
      */
-    std::tuple<const std::string, const std::string, const std::string> call_stage_handshake (
-        const int& index);
+    std::unique_ptr<StageInfo> call_stage_handshake (
+        HandshakeSession* handshake_session);
 
     /*
      * SubmitHousekeepingRules:
@@ -150,7 +153,7 @@ class LocalControlApplication : public GlobalToLocal::Service, public ControlApp
     void operator() () override;
 
     // TODO: FIX this in control application parent class
-    HandshakeSession* register_stage_session (int index, int i);
+    void register_stage_session (int socket_t);
 
     void stop_feedback_loop () override;
 

include/cheferd/networking/core_connection_manager.hpp

@@ -28,7 +28,7 @@
 using controllers_grpc_interface::ACK;
 using controllers_grpc_interface::ConnectReply;
 using controllers_grpc_interface::ConnectRequest;
-using controllers_grpc_interface::StageInfo;
+using controllers_grpc_interface::StageInfoConnect;
 using grpc::Server;
 using grpc::ServerBuilder;
 using grpc::ServerContext;
@@ -68,7 +68,7 @@ class CoreConnectionManager : public LocalToGlobal::Service, public ConnectionMa
         ConnectReply* reply) override;
 
     Status ConnectStageToGlobal (ServerContext* context,
-        const StageInfo* request,
+        const StageInfoConnect* request,
         ConnectReply* reply) override;
 
 public:

include/cheferd/networking/local_interface.hpp

@@ -31,7 +31,7 @@ using controllers_grpc_interface::ConnectReply;
 using controllers_grpc_interface::ConnectRequest;
 using controllers_grpc_interface::EnforcementOpRules;
 using controllers_grpc_interface::EnforcementRules;
-using controllers_grpc_interface::StageInfo;
+using controllers_grpc_interface::StageInfoConnect;
 using controllers_grpc_interface::StageReadyRaw;
 using controllers_grpc_interface::StatsGlobalMap;
 using grpc::Server;

include/cheferd/session/handshake_session.hpp

@@ -25,7 +25,7 @@ namespace cheferd {
 class HandshakeSession {
 
 private:
-    long session_id_;
+    long socket_id_;
 
     std::queue<std::string> submission_queue_; // queue that contains the request to submit to the
     // data plane
@@ -106,7 +106,7 @@ class HandshakeSession {
 
     /**
      * HandshakeSession parameterized constructor.
-     * @param id Data plane stage session id.
+     * @param id Data plane stage socket id.
      */
     explicit HandshakeSession (long id);
 
@@ -118,10 +118,9 @@ class HandshakeSession {
     /**
      * StartSession: begin the enforcement session between the controller and
      * the data plane stage.
-     * @param socket Socket identifier of the communication channel between the
      * controller and the data plane stage.
      */
-    void StartSession (int socket);
+    void StartSession ();
 
     void RemoveSession ();
 

protos/controllers_grpc_interface.proto

@@ -91,16 +91,16 @@ message StatsGlobal {
 
 service LocalToGlobal{
   rpc ConnectLocalToGlobal (ConnectRequest) returns (ConnectReply) {}
-  rpc ConnectStageToGlobal (StageInfo) returns (ConnectReply) {}
-  rpc DisconnectStageToGlobal (StageInfo) returns (ConnectReply) {}
+  rpc ConnectStageToGlobal (StageInfoConnect) returns (ConnectReply) {}
+  rpc DisconnectStageToGlobal (StageInfoConnect) returns (ConnectReply) {}
 }
 
 // The request message containing the user's name.
 message ConnectRequest {
   string user_address = 1;
 }
 
-message StageInfo {
+message StageInfoConnect {
   string local_address = 1;
   string stage_name = 2;
   string stage_env = 3;