Merge from origin

10 files changed, 127 insertions, 42 deletions

diff --git a/src/internal_api/trex_platform_api.h b/src/internal_api/trex_platform_api.h
index 3ae49da8..f8bc10d5 100644
--- a/src/internal_api/trex_platform_api.h
+++ b/src/internal_api/trex_platform_api.h
@@ -110,6 +110,7 @@ public:
     virtual void get_global_stats(TrexPlatformGlobalStats &stats) const = 0;
     virtual void get_interface_stats(uint8_t interface_id, TrexPlatformInterfaceStats &stats) const = 0;
     virtual void get_interface_info(uint8_t interface_id, std::string &driver_name, driver_speed_e &speed) const = 0;
+    virtual void publish_async_data_now(uint32_t key) const = 0;
     virtual uint8_t get_dp_core_count() const = 0;
     
     virtual ~TrexPlatformApi() {}
@@ -127,6 +128,7 @@ public:
     void get_global_stats(TrexPlatformGlobalStats &stats) const;
     void get_interface_stats(uint8_t interface_id, TrexPlatformInterfaceStats &stats) const;
     void get_interface_info(uint8_t interface_id, std::string &driver_name, driver_speed_e &speed) const;
+    void publish_async_data_now(uint32_t key) const;
     uint8_t get_dp_core_count() const;
     
 };
@@ -146,6 +148,7 @@ public:
         speed = SPEED_INVALID;
     }
 
+    void publish_async_data_now(uint32_t key) const {}
     uint8_t get_dp_core_count() const;
 };
 
diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp
index fe6c6bdb..d40c4c8b 100755
--- a/src/main_dpdk.cpp
+++ b/src/main_dpdk.cpp
@@ -2609,8 +2609,12 @@ private:
     bool is_all_cores_finished();
 
 public:
+
+    void publish_async_data();
+    void publish_async_barrier(uint32_t key);
+
     void dump_stats(FILE *fd,
-                    std::string & json,CGlobalStats::DumpFormat format);
+                    CGlobalStats::DumpFormat format);
     void dump_template_info(std::string & json);
     bool sanity_check();
     void update_stats(void);
@@ -2649,6 +2653,7 @@ private:
     CLatencyVmPort      m_latency_vm_vports[BP_MAX_PORTS]; /* vm driver */
     CLatencyPktInfo     m_latency_pkt;
     TrexPublisher       m_zmq_publisher;
+    CGlobalStats        m_stats;
 
 public:
     TrexStateless       *m_trex_stateless;
@@ -3448,11 +3453,11 @@ void CGlobalTRex::dump_template_info(std::string & json){
     json+="]}" ;
 }
 
-void CGlobalTRex::dump_stats(FILE *fd,std::string & json,
-                                CGlobalStats::DumpFormat format){
-    CGlobalStats  stats;
+void CGlobalTRex::dump_stats(FILE *fd, CGlobalStats::DumpFormat format){
+
     update_stats();
-    get_stats(stats);
+    get_stats(m_stats);
+
     if (format==CGlobalStats::dmpTABLE) {
         if ( m_io_modes.m_g_mode == CTrexGlobalIoMode::gNORMAL ){
             switch (m_io_modes.m_pp_mode ){
@@ -3461,11 +3466,11 @@ void CGlobalTRex::dump_stats(FILE *fd,std::string & json,
                     break;
                 case CTrexGlobalIoMode::ppTABLE:
                     fprintf(fd,"\n-Per port stats table \n");
-                    stats.Dump(fd,CGlobalStats::dmpTABLE);
+                    m_stats.Dump(fd,CGlobalStats::dmpTABLE);
                     break;
                 case CTrexGlobalIoMode::ppSTANDARD:
                     fprintf(fd,"\n-Per port stats - standard\n");
-                    stats.Dump(fd,CGlobalStats::dmpSTANDARD);
+                    m_stats.Dump(fd,CGlobalStats::dmpSTANDARD);
                     break;
             };
 
@@ -3475,22 +3480,62 @@ void CGlobalTRex::dump_stats(FILE *fd,std::string & json,
                     break;
                 case   CTrexGlobalIoMode::apENABLE:
                     fprintf(fd,"\n-Global stats enabled \n");
-                    stats.DumpAllPorts(fd);
+                    m_stats.DumpAllPorts(fd);
                     break;
             };
         }
     }else{
         /* at exit , always need to dump it in standartd mode for scripts*/
-        stats.Dump(fd,format);
-        stats.DumpAllPorts(fd);
+        m_stats.Dump(fd,format);
+        m_stats.DumpAllPorts(fd);
     }
-    stats.dump_json(json);
+    
+}
+
+
+void 
+CGlobalTRex::publish_async_data() {
+     std::string json;
+
+     m_stats.dump_json(json);
+     m_zmq_publisher.publish_json(json);
+
+     /* generator json , all cores are the same just sample the first one */
+     m_fl.m_threads_info[0]->m_node_gen.dump_json(json);
+     m_zmq_publisher.publish_json(json);
+
+        
+     if ( !get_is_stateless() ){
+         dump_template_info(json);
+         m_zmq_publisher.publish_json(json);
+     }
+
+     if ( get_is_rx_check_mode() ) {
+         m_mg.rx_check_dump_json(json );
+         m_zmq_publisher.publish_json(json);
+     }
+
+     /* backward compatible */
+     m_mg.dump_json(json );
+     m_zmq_publisher.publish_json(json);
+
+     /* more info */
+     m_mg.dump_json_v2(json );
+     m_zmq_publisher.publish_json(json);
+
+     /* stateless info - nothing for now */
+     //m_trex_stateless->generate_publish_snapshot(json);
+     //m_zmq_publisher.publish_json(json);
 }
 
+void 
+CGlobalTRex::publish_async_barrier(uint32_t key) {
+    m_zmq_publisher.publish_barrier(key);
+}
 
-int CGlobalTRex::run_in_master(){
+int CGlobalTRex::run_in_master() {
 
-    std::string json;
+   
     bool was_stopped=false;
 
     if ( get_is_stateless() ) {
@@ -3530,7 +3575,7 @@ int CGlobalTRex::run_in_master(){
             m_io_modes.DumpHelp(stdout);
         }
 
-        dump_stats(stdout,json,CGlobalStats::dmpTABLE);
+        dump_stats(stdout,CGlobalStats::dmpTABLE);
 
         if (m_io_modes.m_g_mode == CTrexGlobalIoMode::gNORMAL ) {
     		fprintf (stdout," current time    : %.1f sec  \n",now_sec());
@@ -3542,16 +3587,6 @@ int CGlobalTRex::run_in_master(){
     		fprintf (stdout," test duration   : %.1f sec  \n",d);
         }
 
-        m_zmq_publisher.publish_json(json);
-
-        /* generator json , all cores are the same just sample the first one */
-        m_fl.m_threads_info[0]->m_node_gen.dump_json(json);
-        m_zmq_publisher.publish_json(json);
-
-        if ( !get_is_stateless() ){
-            dump_template_info(json);
-            m_zmq_publisher.publish_json(json);
-        }
 
         if ( !CGlobalInfo::m_options.is_latency_disabled() ){
             m_mg.update();
@@ -3591,24 +3626,12 @@ int CGlobalTRex::run_in_master(){
 
             }
 
-            if ( get_is_rx_check_mode() ) {
-                m_mg.rx_check_dump_json(json );
-                m_zmq_publisher.publish_json(json);
-            }
-
-            /* backward compatible */
-            m_mg.dump_json(json );
-            m_zmq_publisher.publish_json(json);
-
-            /* more info */
-            m_mg.dump_json_v2(json );
-            m_zmq_publisher.publish_json(json);
+          
 
         }
 
-        /* stateless info */
-        m_trex_stateless->generate_publish_snapshot(json);
-        m_zmq_publisher.publish_json(json);
+        /* publish data */
+        publish_async_data();
 
         /* check from messages from DP */
         check_for_dp_messages();
@@ -3679,11 +3702,10 @@ int CGlobalTRex::run_in_core(virtual_thread_id_t virt_core_id){
 int CGlobalTRex::stop_master(){
 
     delay(1000);
-    std::string json;
     fprintf(stdout," ==================\n");
     fprintf(stdout," interface sum \n");
     fprintf(stdout," ==================\n");
-    dump_stats(stdout,json,CGlobalStats::dmpSTANDARD);
+    dump_stats(stdout,CGlobalStats::dmpSTANDARD);
     fprintf(stdout," ==================\n");
     fprintf(stdout," \n\n");
 
@@ -3724,7 +3746,7 @@ int CGlobalTRex::stop_master(){
         m_mg.DumpRxCheckVerification(stdout,total_tx_rx_check);
     }
     
-    dump_stats(stdout,json,CGlobalStats::dmpSTANDARD);
+    dump_stats(stdout,CGlobalStats::dmpSTANDARD);
     dump_post_test_stats(stdout);
     m_fl.Delete();
 
@@ -4903,3 +4925,10 @@ TrexDpdkPlatformApi::get_interface_info(uint8_t interface_id,
     driver_name = CTRexExtendedDriverDb::Ins()->get_driver_name();
     speed = CTRexExtendedDriverDb::Ins()->get_drv()->get_driver_speed();
 }
+
+void
+TrexDpdkPlatformApi::publish_async_data_now(uint32_t key) const {
+    g_trex.publish_async_data();
+    g_trex.publish_async_barrier(key);
+}
+
diff --git a/src/mock/trex_platform_api_mock.cpp b/src/mock/trex_platform_api_mock.cpp
index 416c4b69..7cacd96c 100644
--- a/src/mock/trex_platform_api_mock.cpp
+++ b/src/mock/trex_platform_api_mock.cpp
@@ -51,3 +51,4 @@ void
 TrexMockPlatformApi::port_id_to_cores(uint8_t port_id, std::vector<std::pair<uint8_t, uint8_t>> &cores_id_list) const {
     cores_id_list.push_back(std::make_pair(0, 0));
 }
+
diff --git a/src/publisher/trex_publisher.cpp b/src/publisher/trex_publisher.cpp
index 35653069..f56d56df 100644
--- a/src/publisher/trex_publisher.cpp
+++ b/src/publisher/trex_publisher.cpp
@@ -94,6 +94,21 @@ TrexPublisher::publish_event(event_type_e type, const Json::Value &data) {
     publish_json(s);
 }
 
+void
+TrexPublisher::publish_barrier(uint32_t key) {
+    Json::FastWriter writer;
+    Json::Value value;
+    std::string s;
+    
+    value["name"] = "trex-barrier";
+    value["type"] = key;
+    value["data"] = Json::objectValue;
+
+    s = writer.write(value);
+    publish_json(s);
+}
+
+
 /**
  * error handling
  * 
diff --git a/src/publisher/trex_publisher.h b/src/publisher/trex_publisher.h
index 52978476..f086babb 100644
--- a/src/publisher/trex_publisher.h
+++ b/src/publisher/trex_publisher.h
@@ -53,8 +53,20 @@ public:
         
     };
 
+    /**
+     * publishes an async event
+     * 
+     */
     virtual void publish_event(event_type_e type, const Json::Value &data = Json::nullValue);
 
+    /**
+     * publishes a barrier requested by the user
+     * 
+     * @author imarom (17-Jan-16)
+     * 
+     */
+    virtual void publish_barrier(uint32_t key);
+
 private:
     void show_zmq_last_error(const std::string &err);
 private:
diff --git a/src/rpc-server/commands/trex_rpc_cmd_general.cpp b/src/rpc-server/commands/trex_rpc_cmd_general.cpp
index a701f6db..66999144 100644
--- a/src/rpc-server/commands/trex_rpc_cmd_general.cpp
+++ b/src/rpc-server/commands/trex_rpc_cmd_general.cpp
@@ -317,3 +317,19 @@ TrexRpcCmdGetPortStatus::_run(const Json::Value &params, Json::Value &result) {
     return (TREX_RPC_CMD_OK);
 }
 
+/**
+ * publish async data now (fast flush)
+ * 
+ */
+trex_rpc_cmd_rc_e
+TrexRpcPublishNow::_run(const Json::Value &params, Json::Value &result) {
+    TrexStateless *main = get_stateless_obj();
+
+    uint32_t key = parse_uint32(params, "key", result);
+
+    main->get_platform_api()->publish_async_data_now(key);
+
+    result["result"] = Json::objectValue;
+    return (TREX_RPC_CMD_OK);
+
+}
diff --git a/src/rpc-server/commands/trex_rpc_cmds.h b/src/rpc-server/commands/trex_rpc_cmds.h
index b1750053..081398d1 100644
--- a/src/rpc-server/commands/trex_rpc_cmds.h
+++ b/src/rpc-server/commands/trex_rpc_cmds.h
@@ -56,6 +56,7 @@ TREX_RPC_CMD_DEFINE(TrexRpcCmdTestSub,    "test_sub", 2, false);
  * general cmds
  */
 TREX_RPC_CMD_DEFINE(TrexRpcCmdPing,       "ping",                 0, false);
+TREX_RPC_CMD_DEFINE(TrexRpcPublishNow,    "publish_now",          1, false);
 TREX_RPC_CMD_DEFINE(TrexRpcCmdGetCmds,    "get_supported_cmds",   0, false);
 TREX_RPC_CMD_DEFINE(TrexRpcCmdGetVersion, "get_version",          0, false);
 
diff --git a/src/rpc-server/trex_rpc_async_server.cpp b/src/rpc-server/trex_rpc_async_server.cpp
index 6e5fbfc6..aee92539 100644
--- a/src/rpc-server/trex_rpc_async_server.cpp
+++ b/src/rpc-server/trex_rpc_async_server.cpp
@@ -51,6 +51,8 @@ TrexRpcServerAsync::_prepare() {
  */
 void 
 TrexRpcServerAsync::_rpc_thread_cb() {
+/* disabled, using the main publisher */
+#if 0
     std::stringstream ss;
 
     /* create a socket based on the configuration */
@@ -105,6 +107,7 @@ TrexRpcServerAsync::_rpc_thread_cb() {
 
     /* must be closed from the same thread */
     zmq_close(m_socket);
+#endif
 }
 
 void 
diff --git a/src/rpc-server/trex_rpc_cmds_table.cpp b/src/rpc-server/trex_rpc_cmds_table.cpp
index 82c723b7..5218cd0a 100644
--- a/src/rpc-server/trex_rpc_cmds_table.cpp
+++ b/src/rpc-server/trex_rpc_cmds_table.cpp
@@ -34,6 +34,7 @@ TrexRpcCommandsTable::TrexRpcCommandsTable() {
 
     /* general */
     register_command(new TrexRpcCmdPing());
+    register_command(new TrexRpcPublishNow());
     register_command(new TrexRpcCmdGetCmds());
     register_command(new TrexRpcCmdGetVersion());
     register_command(new TrexRpcCmdGetSysInfo());
diff --git a/src/sim/trex_sim_stateless.cpp b/src/sim/trex_sim_stateless.cpp
index 215315e0..13f264cf 100644
--- a/src/sim/trex_sim_stateless.cpp
+++ b/src/sim/trex_sim_stateless.cpp
@@ -97,6 +97,10 @@ public:
         }
     }
 
+    virtual void publish_async_data_now(uint32_t key) const {
+
+    }
+
 private:
     int m_dp_core_count;
 };