added async publisher to the RPC server

4 files changed, 452 insertions, 241 deletions

diff --git a/src/stateless/trex_stateless.cpp b/src/stateless/trex_stateless.cpp
index 3ec419ea..0b7947a0 100644
--- a/src/stateless/trex_stateless.cpp
+++ b/src/stateless/trex_stateless.cpp
@@ -19,6 +19,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 #include <trex_stateless_api.h>
+#include <trex_stateless_port.h>
 
 using namespace std;
 
@@ -73,107 +74,64 @@ uint8_t TrexStateless::get_port_count() {
     return m_port_count;
 }
 
+void 
+TrexStateless::update_stats() {
 
-/***************************
- * trex stateless port stats
- * 
- **************************/
-TrexPortStats::TrexPortStats() {
-    m_stats = {0};
-}
-
-/***************************
- * trex stateless port
- * 
- **************************/
-TrexStatelessPort::TrexStatelessPort(uint8_t port_id) : m_port_id(port_id) {
-    m_port_state = PORT_STATE_UP_IDLE;
-    clear_owner();
-}
-
+    /* update CPU util. */
+    #ifdef TREX_RPC_MOCK_SERVER
+        m_stats.m_stats.m_cpu_util = 0;
+    #else
+        m_stats.m_stats.m_cpu_util = 0;
+    #endif
 
-/**
- * starts the traffic on the port
- * 
- */
-TrexStatelessPort::rc_e
-TrexStatelessPort::start_traffic(void) {
+    /* for every port update and accumulate */
+    for (uint8_t i = 0; i < m_port_count; i++) {
+        m_ports[i]->update_stats();
 
-    if (m_port_state != PORT_STATE_UP_IDLE) {
-        return (RC_ERR_BAD_STATE_FOR_OP);
-    }
+        const TrexPortStats & port_stats = m_ports[i]->get_stats();
 
-    if (get_stream_table()->size() == 0) {
-        return (RC_ERR_NO_STREAMS);
-    }
+        m_stats.m_stats.m_tx_bps += port_stats.m_stats.m_tx_bps;
+        m_stats.m_stats.m_rx_bps += port_stats.m_stats.m_rx_bps;
 
-    m_port_state = PORT_STATE_TRANSMITTING;
+        m_stats.m_stats.m_tx_pps += port_stats.m_stats.m_tx_pps;
+        m_stats.m_stats.m_rx_pps += port_stats.m_stats.m_rx_pps;
 
-    /* real code goes here */
-    return (RC_OK);
-}
+        m_stats.m_stats.m_total_tx_pkts += port_stats.m_stats.m_total_tx_pkts;
+        m_stats.m_stats.m_total_rx_pkts += port_stats.m_stats.m_total_rx_pkts;
 
-void 
-TrexStatelessPort::stop_traffic(void) {
+        m_stats.m_stats.m_total_tx_bytes += port_stats.m_stats.m_total_tx_bytes;
+        m_stats.m_stats.m_total_rx_bytes += port_stats.m_stats.m_total_rx_bytes;
 
-    /* real code goes here */
-    if (m_port_state == PORT_STATE_TRANSMITTING) {
-        m_port_state = PORT_STATE_UP_IDLE;
+        m_stats.m_stats.m_tx_rx_errors += port_stats.m_stats.m_tx_rx_errors;
     }
 }
 
-/**
-* access the stream table
-* 
-*/
-TrexStreamTable * TrexStatelessPort::get_stream_table() {
-    return &m_stream_table;
-}
-
+void
+TrexStateless::encode_stats(Json::Value &global) {
 
-std::string 
-TrexStatelessPort::get_state_as_string() {
+    global["cpu_util"] = m_stats.m_stats.m_cpu_util;
 
-    switch (get_state()) {
-    case PORT_STATE_DOWN:
-        return "down";
+    global["tx_bps"]   = m_stats.m_stats.m_tx_bps;
+    global["rx_bps"]   = m_stats.m_stats.m_rx_bps;
 
-    case PORT_STATE_UP_IDLE:
-        return  "idle";
+    global["tx_pps"]   = m_stats.m_stats.m_tx_pps;
+    global["rx_pps"]   = m_stats.m_stats.m_rx_pps;
 
-    case PORT_STATE_TRANSMITTING:
-        return "transmitting";
-    }
+    global["total_tx_pkts"] = Json::Value::UInt64(m_stats.m_stats.m_total_tx_pkts);
+    global["total_rx_pkts"] = Json::Value::UInt64(m_stats.m_stats.m_total_rx_pkts);
 
-    return "unknown";
-}
+    global["total_tx_bytes"] = Json::Value::UInt64(m_stats.m_stats.m_total_tx_bytes);
+    global["total_rx_bytes"] = Json::Value::UInt64(m_stats.m_stats.m_total_rx_bytes);
 
-void
-TrexStatelessPort::get_properties(string &driver, string &speed) {
+    global["tx_rx_errors"]    = Json::Value::UInt64(m_stats.m_stats.m_tx_rx_errors);
 
-    /* take this from DPDK */
-    driver = "e1000";
-    speed  = "1 Gbps";
-}
+    for (uint8_t i = 0; i < m_port_count; i++) {
+        std::stringstream ss;
 
+        ss << "port " << i;
+        Json::Value &port_section = global[ss.str()];
 
-/**
- * generate a random connection handler
- * 
- */
-std::string 
-TrexStatelessPort::generate_handler() {
-    std::stringstream ss;
-
-    static const char alphanum[] =
-        "0123456789"
-        "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
-        "abcdefghijklmnopqrstuvwxyz";
-
-    /* generate 8 bytes of random handler */
-    for (int i = 0; i < 8; ++i) {
-        ss << alphanum[rand() % (sizeof(alphanum) - 1)];
+        m_ports[i]->encode_stats(port_section);
     }
-
-    return (ss.str());
 }
+
diff --git a/src/stateless/trex_stateless_api.h b/src/stateless/trex_stateless_api.h
index fd88baf7..e2bf4e1c 100644
--- a/src/stateless/trex_stateless_api.h
+++ b/src/stateless/trex_stateless_api.h
@@ -41,167 +41,36 @@ public:
     }
 };
 
-/**
- * TRex stateless port stats
- * 
- * @author imarom (24-Sep-15)
- */
-class TrexPortStats {
-
-public:
-    TrexPortStats();
-
-public:
-    struct {
-        uint64_t tx_pps;
-        uint64_t tx_bps;
-        uint64_t total_tx_pkts;
-        uint64_t total_tx_bytes;
-
-        uint64_t rx_pps;
-        uint64_t rx_bps;
-        uint64_t total_rx_pkts;
-        uint64_t total_rx_bytes;
-
-        uint64_t tx_rx_errors;
-    } m_stats;
-};
+class TrexStatelessPort;
 
 /**
- * describes a stateless port
+ * unified stats
  * 
- * @author imarom (31-Aug-15)
+ * @author imarom (06-Oct-15)
  */
-class TrexStatelessPort {
+class TrexStatelessStats {
 public:
-
-    /**
-     * port state
-     */
-    enum port_state_e {
-        PORT_STATE_DOWN,
-        PORT_STATE_UP_IDLE,
-        PORT_STATE_TRANSMITTING
-    };
-
-    /**
-     * describess different error codes for port operations
-     */
-    enum rc_e {
-        RC_OK,
-        RC_ERR_BAD_STATE_FOR_OP,
-        RC_ERR_NO_STREAMS,
-        RC_ERR_FAILED_TO_COMPILE_STREAMS
-    };
-
-    TrexStatelessPort(uint8_t port_id);
-
-    /**
-     * start traffic
-     * 
-     */
-    rc_e start_traffic(void);
-
-    /**
-     * stop traffic
-     * 
-     */
-    void stop_traffic(void);
-
-    /**
-     * access the stream table
-     * 
-     */
-    TrexStreamTable *get_stream_table();
-
-    /**
-     * get the port state
-     * 
-     */
-    port_state_e get_state() {
-        return m_port_state;
+    TrexStatelessStats() {
+        m_stats = {0};
     }
 
-    /**
-     * port state as string
-     * 
-     */
-    std::string get_state_as_string();
-
-    /**
-     * fill up properties of the port
-     * 
-     * @author imarom (16-Sep-15)
-     * 
-     * @param driver 
-     * @param speed 
-     */
-    void get_properties(std::string &driver, std::string &speed);
-
-    /**
-    * query for ownership
-    * 
-    */
-    const std::string &get_owner() {
-        return m_owner;
-    }
-
-    /**
-     * owner handler 
-     * for the connection 
-     * 
-     */
-    const std::string &get_owner_handler() {
-        return m_owner_handler;
-    }
-
-    bool is_free_to_aquire() {
-        return (m_owner == "none");
-    }
-
-    /**
-    * take ownership of the server array 
-    * this is static 
-    * ownership is total 
-    * 
-    */
-    void set_owner(const std::string &owner) {
-        m_owner = owner;
-        m_owner_handler = generate_handler();
-    }
-
-    void clear_owner() {
-        m_owner = "none";
-        m_owner_handler = "";
-    }
-
-    bool verify_owner_handler(const std::string &handler) {
-
-        return ( (m_owner != "none") && (m_owner_handler == handler) );
-
-    }
-
-    /**
-     * update the values of the stats
-     * 
-     * @author imarom (24-Sep-15)
-     */
-    void update_stats();
-
-    const TrexPortStats & get_stats() {
-        return m_stats;
-    }
-
-private:
-
-    std::string generate_handler();
-
-    TrexStreamTable  m_stream_table;
-    uint8_t          m_port_id;
-    port_state_e     m_port_state;
-    std::string      m_owner;
-    std::string      m_owner_handler;
-    TrexPortStats    m_stats;
+    struct {
+        double m_cpu_util;
+
+        double m_tx_bps;
+        double m_rx_bps;
+       
+        double m_tx_pps;
+        double m_rx_pps;
+       
+        uint64_t m_total_tx_pkts;
+        uint64_t m_total_rx_pkts;
+       
+        uint64_t m_total_tx_bytes;
+        uint64_t m_total_rx_bytes;
+       
+        uint64_t m_tx_rx_errors;
+    } m_stats;
 };
 
 /**
@@ -232,8 +101,22 @@ public:
         return instance;
     }
 
-    TrexStatelessPort *get_port_by_id(uint8_t port_id);
-    uint8_t            get_port_count();
+    TrexStatelessPort * get_port_by_id(uint8_t port_id);
+    uint8_t             get_port_count();
+
+    /**
+     * update all the stats (deep update)
+     * (include all the ports and global stats)
+     * 
+     */
+    void               update_stats();
+
+    /**
+     * fetch all the stats
+     * 
+     */
+    void                encode_stats(Json::Value &global);
+
 
 protected:
     TrexStateless();
@@ -248,9 +131,11 @@ protected:
     TrexStateless(TrexStateless const&)      = delete;  
     void operator=(TrexStateless const&)     = delete;
 
-    bool                m_is_configured;
-    TrexStatelessPort  **m_ports;
-    uint8_t             m_port_count;
+    bool                 m_is_configured;
+    TrexStatelessPort   **m_ports;
+    uint8_t              m_port_count;
+
+    TrexStatelessStats   m_stats;
 };
 
 #endif /* __TREX_STATELESS_API_H__ */
diff --git a/src/stateless/trex_stateless_port.cpp b/src/stateless/trex_stateless_port.cpp
new file mode 100644
index 00000000..7322ef8a
--- /dev/null
+++ b/src/stateless/trex_stateless_port.cpp
@@ -0,0 +1,172 @@
+/*
+ Itay Marom
+ Cisco Systems, Inc.
+*/
+
+/*
+Copyright (c) 2015-2015 Cisco Systems, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+#include <trex_stateless_api.h>
+#include <trex_stateless_port.h>
+#include <string>
+
+using namespace std;
+
+/***************************
+ * trex stateless port
+ * 
+ **************************/
+TrexStatelessPort::TrexStatelessPort(uint8_t port_id) : m_port_id(port_id) {
+    m_port_state = PORT_STATE_UP_IDLE;
+    clear_owner();
+}
+
+
+/**
+ * starts the traffic on the port
+ * 
+ */
+TrexStatelessPort::rc_e
+TrexStatelessPort::start_traffic(void) {
+
+    if (m_port_state != PORT_STATE_UP_IDLE) {
+        return (RC_ERR_BAD_STATE_FOR_OP);
+    }
+
+    if (get_stream_table()->size() == 0) {
+        return (RC_ERR_NO_STREAMS);
+    }
+
+    m_port_state = PORT_STATE_TRANSMITTING;
+
+    /* real code goes here */
+    return (RC_OK);
+}
+
+void 
+TrexStatelessPort::stop_traffic(void) {
+
+    /* real code goes here */
+    if (m_port_state == PORT_STATE_TRANSMITTING) {
+        m_port_state = PORT_STATE_UP_IDLE;
+    }
+}
+
+/**
+* access the stream table
+* 
+*/
+TrexStreamTable * TrexStatelessPort::get_stream_table() {
+    return &m_stream_table;
+}
+
+
+std::string 
+TrexStatelessPort::get_state_as_string() {
+
+    switch (get_state()) {
+    case PORT_STATE_DOWN:
+        return "down";
+
+    case PORT_STATE_UP_IDLE:
+        return  "idle";
+
+    case PORT_STATE_TRANSMITTING:
+        return "transmitting";
+    }
+
+    return "unknown";
+}
+
+void
+TrexStatelessPort::get_properties(string &driver, string &speed) {
+
+    /* take this from DPDK */
+    driver = "e1000";
+    speed  = "1 Gbps";
+}
+
+
+/**
+ * generate a random connection handler
+ * 
+ */
+std::string 
+TrexStatelessPort::generate_handler() {
+    std::stringstream ss;
+
+    static const char alphanum[] =
+        "0123456789"
+        "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
+        "abcdefghijklmnopqrstuvwxyz";
+
+    /* generate 8 bytes of random handler */
+    for (int i = 0; i < 8; ++i) {
+        ss << alphanum[rand() % (sizeof(alphanum) - 1)];
+    }
+
+    return (ss.str());
+}
+
+/**
+ * update stats for the port
+ * 
+ */
+void 
+TrexStatelessPort::update_stats() {
+    #ifdef TREX_RPC_MOCK_SERVER
+    /* do lies - its a mock */
+    m_stats.m_stats.m_tx_bps = rand() % 10000;
+    m_stats.m_stats.m_rx_bps = rand() % 10000;
+
+    m_stats.m_stats.m_tx_pps = m_stats.m_stats.m_tx_bps / (64 + rand() % 1000);
+    m_stats.m_stats.m_rx_pps = m_stats.m_stats.m_rx_bps / (64 + rand() % 1000);
+
+
+    m_stats.m_stats.m_total_tx_bytes += m_stats.m_stats.m_tx_bps;
+    m_stats.m_stats.m_total_rx_bytes += m_stats.m_stats.m_rx_bps;
+
+    m_stats.m_stats.m_total_tx_pkts += m_stats.m_stats.m_tx_pps;
+    m_stats.m_stats.m_total_rx_pkts += m_stats.m_stats.m_rx_pps;
+
+    #else
+    /* real update work */
+    #endif
+}
+
+const TrexPortStats &
+TrexStatelessPort::get_stats() {
+    return m_stats;
+}
+
+void
+TrexStatelessPort::encode_stats(Json::Value &port) {
+
+    port["tx_bps"]          = m_stats.m_stats.m_tx_bps;
+    port["rx_bps"]          = m_stats.m_stats.m_rx_bps;
+
+    port["tx_pps"]          = m_stats.m_stats.m_tx_pps;
+    port["rx_pps"]          = m_stats.m_stats.m_rx_pps;
+
+    port["total_tx_pkts"]   = Json::Value::UInt64(m_stats.m_stats.m_total_tx_pkts);
+    port["total_rx_pkts"]   = Json::Value::UInt64(m_stats.m_stats.m_total_rx_pkts);
+
+    port["total_tx_bytes"]  = Json::Value::UInt64(m_stats.m_stats.m_total_tx_bytes);
+    port["total_rx_bytes"]  = Json::Value::UInt64(m_stats.m_stats.m_total_rx_bytes);
+    
+    port["tx_rx_errors"]    = Json::Value::UInt64(m_stats.m_stats.m_tx_rx_errors);
+}
+
+
diff --git a/src/stateless/trex_stateless_port.h b/src/stateless/trex_stateless_port.h
new file mode 100644
index 00000000..ea98ddae
--- /dev/null
+++ b/src/stateless/trex_stateless_port.h
@@ -0,0 +1,196 @@
+/*
+ Itay Marom
+ Cisco Systems, Inc.
+*/
+
+/*
+Copyright (c) 2015-2015 Cisco Systems, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+#ifndef __TREX_STATELESS_PORT_H__
+#define __TREX_STATELESS_PORT_H__
+
+#include <trex_stream_api.h>
+
+/**
+ * TRex stateless port stats
+ * 
+ * @author imarom (24-Sep-15)
+ */
+class TrexPortStats {
+
+public:
+    TrexPortStats() {
+        m_stats = {0};
+    }
+
+public:
+    struct {
+
+        double m_tx_bps;
+        double m_rx_bps;
+       
+        double m_tx_pps;
+        double m_rx_pps;
+       
+        uint64_t m_total_tx_pkts;
+        uint64_t m_total_rx_pkts;
+       
+        uint64_t m_total_tx_bytes;
+        uint64_t m_total_rx_bytes;
+       
+        uint64_t m_tx_rx_errors;
+    } m_stats;
+};
+
+/**
+ * describes a stateless port
+ * 
+ * @author imarom (31-Aug-15)
+ */
+class TrexStatelessPort {
+public:
+
+    /**
+     * port state
+     */
+    enum port_state_e {
+        PORT_STATE_DOWN,
+        PORT_STATE_UP_IDLE,
+        PORT_STATE_TRANSMITTING
+    };
+
+    /**
+     * describess different error codes for port operations
+     */
+    enum rc_e {
+        RC_OK,
+        RC_ERR_BAD_STATE_FOR_OP,
+        RC_ERR_NO_STREAMS,
+        RC_ERR_FAILED_TO_COMPILE_STREAMS
+    };
+
+    TrexStatelessPort(uint8_t port_id);
+
+    /**
+     * start traffic
+     * 
+     */
+    rc_e start_traffic(void);
+
+    /**
+     * stop traffic
+     * 
+     */
+    void stop_traffic(void);
+
+    /**
+     * access the stream table
+     * 
+     */
+    TrexStreamTable *get_stream_table();
+
+    /**
+     * get the port state
+     * 
+     */
+    port_state_e get_state() {
+        return m_port_state;
+    }
+
+    /**
+     * port state as string
+     * 
+     */
+    std::string get_state_as_string();
+
+    /**
+     * fill up properties of the port
+     * 
+     * @author imarom (16-Sep-15)
+     * 
+     * @param driver 
+     * @param speed 
+     */
+    void get_properties(std::string &driver, std::string &speed);
+
+    /**
+    * query for ownership
+    * 
+    */
+    const std::string &get_owner() {
+        return m_owner;
+    }
+
+    /**
+     * owner handler 
+     * for the connection 
+     * 
+     */
+    const std::string &get_owner_handler() {
+        return m_owner_handler;
+    }
+
+    bool is_free_to_aquire() {
+        return (m_owner == "none");
+    }
+
+    /**
+    * take ownership of the server array 
+    * this is static 
+    * ownership is total 
+    * 
+    */
+    void set_owner(const std::string &owner) {
+        m_owner = owner;
+        m_owner_handler = generate_handler();
+    }
+
+    void clear_owner() {
+        m_owner = "none";
+        m_owner_handler = "";
+    }
+
+    bool verify_owner_handler(const std::string &handler) {
+
+        return ( (m_owner != "none") && (m_owner_handler == handler) );
+
+    }
+
+    /**
+     * update the values of the stats
+     * 
+     */
+    void update_stats();
+
+    const TrexPortStats & get_stats();
+
+    /**
+     * encode stats as JSON
+     */
+    void encode_stats(Json::Value &port);
+
+private:
+
+    std::string generate_handler();
+
+    TrexStreamTable  m_stream_table;
+    uint8_t          m_port_id;
+    port_state_e     m_port_state;
+    std::string      m_owner;
+    std::string      m_owner_handler;
+    TrexPortStats    m_stats;
+};
+
+#endif /* __TREX_STATELESS_PORT_H__ */