1. fixed ZMQ message limitation

2. added some scale yamls for IMIX (300 streams, 1000 streams)
3. return objects are always complex objects (not strings) - for backward compatability
4. some minor adjustments to ZMQ socket timeouts

6 files changed, 57 insertions, 35 deletions

diff --git a/src/platform_cfg.cpp b/src/platform_cfg.cpp
index 547cc3ad..ca42aa31 100755
--- a/src/platform_cfg.cpp
+++ b/src/platform_cfg.cpp
@@ -35,8 +35,8 @@ void CPlatformMemoryYamlInfo::reset(){
        m_mbuf[MBUF_64]           = m_mbuf[MBUF_64]*2;
        m_mbuf[MBUF_2048]         = CONST_NB_MBUF_2_10G/2;
 
-       m_mbuf[TRAFFIC_MBUF_64]           = m_mbuf[MBUF_64]*2;
-       m_mbuf[TRAFFIC_MBUF_2048]         = CONST_NB_MBUF_2_10G*4;
+       m_mbuf[TRAFFIC_MBUF_64]           = m_mbuf[MBUF_64] * 4;
+       m_mbuf[TRAFFIC_MBUF_2048]         = CONST_NB_MBUF_2_10G * 8;
 
        m_mbuf[MBUF_DP_FLOWS]     = (1024*1024/2);
        m_mbuf[MBUF_GLOBAL_FLOWS] =(10*1024/2);
diff --git a/src/rpc-server/commands/trex_rpc_cmd_general.cpp b/src/rpc-server/commands/trex_rpc_cmd_general.cpp
index 9570aae7..5cea055c 100644
--- a/src/rpc-server/commands/trex_rpc_cmd_general.cpp
+++ b/src/rpc-server/commands/trex_rpc_cmd_general.cpp
@@ -43,7 +43,7 @@ using namespace std;
 trex_rpc_cmd_rc_e 
 TrexRpcCmdPing::_run(const Json::Value &params, Json::Value &result) {
 
-    result["result"] = "ACK";
+    result["result"] = Json::objectValue;
     return (TREX_RPC_CMD_OK);
 }
 
@@ -272,7 +272,7 @@ TrexRpcCmdRelease::_run(const Json::Value &params, Json::Value &result) {
         generate_execute_err(result, ex.what());
     }
 
-    result["result"] = "ACK";
+    result["result"] = Json::objectValue;
 
     return (TREX_RPC_CMD_OK);
 }
diff --git a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
index dea4c171..51df3159 100644
--- a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
+++ b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
@@ -122,7 +122,7 @@ TrexRpcCmdAddStream::_run(const Json::Value &params, Json::Value &result) {
         generate_execute_err(result, ex.what());
     }
 
-    result["result"] = "ACK";
+    result["result"] = Json::objectValue;
 
     return (TREX_RPC_CMD_OK);
 }
@@ -350,7 +350,7 @@ TrexRpcCmdRemoveStream::_run(const Json::Value &params, Json::Value &result) {
     
     delete stream;
 
-    result["result"] = "ACK";
+    result["result"] = Json::objectValue;
 
     return (TREX_RPC_CMD_OK);
 }
@@ -379,7 +379,7 @@ TrexRpcCmdRemoveAllStreams::_run(const Json::Value &params, Json::Value &result)
     }
 
 
-    result["result"] = "ACK";
+    result["result"] = Json::objectValue;
 
     return (TREX_RPC_CMD_OK);
 }
@@ -493,7 +493,7 @@ TrexRpcCmdStartTraffic::_run(const Json::Value &params, Json::Value &result) {
         generate_execute_err(result, ex.what());
     }
 
-    result["result"] = "ACK";
+    result["result"]["multiplier"] = port->get_multiplier();
 
     return (TREX_RPC_CMD_OK);
 }
@@ -520,7 +520,7 @@ TrexRpcCmdStopTraffic::_run(const Json::Value &params, Json::Value &result) {
         generate_execute_err(result, ex.what());
     }
 
-    result["result"] = "ACK";
+    result["result"] = Json::objectValue;
 
     return (TREX_RPC_CMD_OK);
 }
@@ -589,7 +589,7 @@ TrexRpcCmdPauseTraffic::_run(const Json::Value &params, Json::Value &result) {
         generate_execute_err(result, ex.what());
     }
 
-    result["result"] = "ACK";
+    result["result"] = Json::objectValue;
 
     return (TREX_RPC_CMD_OK);
 }
@@ -617,7 +617,7 @@ TrexRpcCmdResumeTraffic::_run(const Json::Value &params, Json::Value &result) {
         generate_execute_err(result, ex.what());
     }
 
-    result["result"] = "ACK";
+    result["result"] = Json::objectValue;
 
     return (TREX_RPC_CMD_OK);
 }
@@ -656,7 +656,7 @@ TrexRpcCmdUpdateTraffic::_run(const Json::Value &params, Json::Value &result) {
         generate_execute_err(result, ex.what());
     }
 
-    result["result"] = "ACK";
+    result["result"]["multiplier"] = port->get_multiplier();
 
     return (TREX_RPC_CMD_OK);
 }
diff --git a/src/rpc-server/trex_rpc_req_resp_server.cpp b/src/rpc-server/trex_rpc_req_resp_server.cpp
index 9147f75d..eb7825ac 100644
--- a/src/rpc-server/trex_rpc_req_resp_server.cpp
+++ b/src/rpc-server/trex_rpc_req_resp_server.cpp
@@ -26,6 +26,7 @@ limitations under the License.
 #include <unistd.h>
 #include <sstream>
 #include <iostream>
+#include <assert.h>
 
 #include <zmq.h>
 #include <json/json.h>
@@ -70,28 +71,14 @@ void TrexRpcServerReqRes::_rpc_thread_cb() {
 
     /* server main loop */
     while (m_is_running) {
-        int msg_size = zmq_recv (m_socket, m_msg_buffer, sizeof(m_msg_buffer), 0);
-
-        /* msg_size of -1 is an error - decode it */
-        if (msg_size == -1) {
-            /* normal shutdown and zmq_term was called */
-            if (errno == ETERM) {
-                break;
-            } else {
-                throw TrexRpcException("Unhandled error of zmq_recv");
-            }
-        }
+        std::string request;
 
-        if (msg_size >= sizeof(m_msg_buffer)) {
-            std::stringstream ss;
-            ss << "RPC request of '" << msg_size << "' exceeds maximum message size which is '" << sizeof(m_msg_buffer) << "'";
-            handle_server_error(ss.str());
-            continue;
+        /* get the next request */
+        bool rc = fetch_one_request(request);
+        if (!rc) {
+            break;
         }
 
-        /* transform it to a string */
-        std::string request((const char *)m_msg_buffer, msg_size);
-
         verbose_json("Server Received: ", TrexJsonRpcV2Parser::pretty_json_str(request));
 
         handle_request(request);
@@ -101,6 +88,35 @@ void TrexRpcServerReqRes::_rpc_thread_cb() {
     zmq_close(m_socket);
 }
 
+bool
+TrexRpcServerReqRes::fetch_one_request(std::string &msg) {
+
+    zmq_msg_t zmq_msg;
+    int rc;
+
+    rc = zmq_msg_init(&zmq_msg);
+    assert(rc == 0);
+
+    rc = zmq_msg_recv (&zmq_msg, m_socket, 0);
+
+    if (rc == -1) {
+        zmq_msg_close(&zmq_msg);
+        /* normal shutdown and zmq_term was called */
+        if (errno == ETERM) {
+            return false;
+        } else {
+            throw TrexRpcException("Unhandled error of zmq_recv");
+        }
+    }
+
+    const char *data =  (const char *)zmq_msg_data(&zmq_msg);
+    size_t len = zmq_msg_size(&zmq_msg);
+    msg.append(data, len);
+
+    zmq_msg_close(&zmq_msg);
+    return true;
+}
+
 /**
  * stops the ZMQ based RPC server
  * 
diff --git a/src/rpc-server/trex_rpc_req_resp_server.h b/src/rpc-server/trex_rpc_req_resp_server.h
index bc38c0ef..2876206c 100644
--- a/src/rpc-server/trex_rpc_req_resp_server.h
+++ b/src/rpc-server/trex_rpc_req_resp_server.h
@@ -39,14 +39,12 @@ protected:
     void _stop_rpc_thread();
 
 private:
-
+    bool fetch_one_request(std::string &msg);
     void handle_request(const std::string &request);
     void handle_server_error(const std::string &specific_err);
 
-    static const int    RPC_MAX_MSG_SIZE = (200 * 1024);
     void               *m_context;
     void               *m_socket;
-    uint8_t             m_msg_buffer[RPC_MAX_MSG_SIZE];
 };
 
 
diff --git a/src/stateless/cp/trex_stateless_port.h b/src/stateless/cp/trex_stateless_port.h
index 45eb16e8..2d15a1cc 100644
--- a/src/stateless/cp/trex_stateless_port.h
+++ b/src/stateless/cp/trex_stateless_port.h
@@ -79,7 +79,7 @@ public:
      * start traffic
      * throws TrexException in case of an error
      */
-    void start_traffic(const TrexPortMultiplier &mul, double duration = -1);
+    void start_traffic(const TrexPortMultiplier &mul, double duration);
 
     /**
      * stop traffic
@@ -211,6 +211,14 @@ public:
     }
 
 
+    /**
+     * returns the traffic multiplier currently being used by the DP
+     * 
+     */
+    double get_multiplier() {
+        return (m_current_per_core_m * m_cores_id_list.size()); 
+    }
+
 private: