From 73574943ae438985f37aae3ea52e9713c55ef62e Mon Sep 17 00:00:00 2001 From: imarom Date: Tue, 29 Sep 2015 19:47:45 +0300 Subject: just a checkpoint on async server --- src/rpc-server/trex_rpc_async_server.cpp | 77 ++++++++++++++++++++++++++++++++ 1 file changed, 77 insertions(+) create mode 100644 src/rpc-server/trex_rpc_async_server.cpp (limited to 'src/rpc-server/trex_rpc_async_server.cpp') diff --git a/src/rpc-server/trex_rpc_async_server.cpp b/src/rpc-server/trex_rpc_async_server.cpp new file mode 100644 index 00000000..76549cbd --- /dev/null +++ b/src/rpc-server/trex_rpc_async_server.cpp @@ -0,0 +1,77 @@ +/* + 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 +#include +#include +#include + +/** + * ZMQ based publisher server + * + */ +TrexRpcServerAsync::TrexRpcServerAsync(const TrexRpcServerConfig &cfg) : TrexRpcServerInterface(cfg, "publisher") { + /* ZMQ is not thread safe - this should be outside */ + m_context = zmq_ctx_new(); +} + +void +TrexRpcServerAsync::_rpc_thread_cb() { + std::stringstream ss; + + /* create a socket based on the configuration */ + m_socket = zmq_socket (m_context, ZMQ_PUB); + + switch (m_cfg.get_protocol()) { + case TrexRpcServerConfig::RPC_PROT_TCP: + ss << "tcp://*:"; + break; + default: + throw TrexRpcException("unknown protocol for RPC"); + } + + ss << m_cfg.get_port(); + + /* bind the scoket */ + int rc = zmq_bind (m_socket, ss.str().c_str()); + if (rc != 0) { + throw TrexRpcException("Unable to start ZMQ server at: " + ss.str()); + } + + /* while the server is running - publish results */ + while (m_is_running) { + /* update all ports for their stats */ + uint8_t port_count = TrexStateless::get_instance().get_port_count(); + for (uint8_t i = 0; i < port_count; i++) { + TrexStateless::get_instance().get_port_by_id(i).update_stats(); + const TrexPortStats &stats = TrexStateless::get_instance().get_port_by_id(i).get_stats(); + + + + } + } +} + +void +TrexRpcServerAsync::_stop_rpc_thread() { + m_is_running = false; + this->m_thread.join(); + zmq_term(m_context); +} -- cgit From 4d53d6e2633caed782067965b1b4422b45dab4a2 Mon Sep 17 00:00:00 2001 From: imarom Date: Wed, 7 Oct 2015 14:57:48 +0200 Subject: added async publisher to the RPC server --- src/rpc-server/trex_rpc_async_server.cpp | 36 +++++++++++++++++++++++++------- 1 file changed, 29 insertions(+), 7 deletions(-) (limited to 'src/rpc-server/trex_rpc_async_server.cpp') diff --git a/src/rpc-server/trex_rpc_async_server.cpp b/src/rpc-server/trex_rpc_async_server.cpp index 76549cbd..40d16dfe 100644 --- a/src/rpc-server/trex_rpc_async_server.cpp +++ b/src/rpc-server/trex_rpc_async_server.cpp @@ -18,10 +18,19 @@ 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. */ + +/* required for sleep_for c++ 2011 + https://bugs.launchpad.net/ubuntu/+source/gcc-4.4/+bug/608145 +*/ +#define _GLIBCXX_USE_NANOSLEEP + #include +#include #include #include #include +#include +#include /** * ZMQ based publisher server @@ -32,6 +41,10 @@ TrexRpcServerAsync::TrexRpcServerAsync(const TrexRpcServerConfig &cfg) : TrexRpc m_context = zmq_ctx_new(); } +/** + * publisher thread + * + */ void TrexRpcServerAsync::_rpc_thread_cb() { std::stringstream ss; @@ -57,21 +70,30 @@ TrexRpcServerAsync::_rpc_thread_cb() { /* while the server is running - publish results */ while (m_is_running) { - /* update all ports for their stats */ - uint8_t port_count = TrexStateless::get_instance().get_port_count(); - for (uint8_t i = 0; i < port_count; i++) { - TrexStateless::get_instance().get_port_by_id(i).update_stats(); - const TrexPortStats &stats = TrexStateless::get_instance().get_port_by_id(i).get_stats(); + Json::Value snapshot; + Json::FastWriter writer; + + /* trigger a full update for stats */ + TrexStateless::get_instance().update_stats(); + + /* encode them to JSON */ + TrexStateless::get_instance().encode_stats(snapshot); + + /* write to string and publish */ + std::string snapshot_str = writer.write(snapshot); + zmq_send(m_socket, snapshot_str.c_str(), snapshot_str.size(), 0); + //std::cout << "sending " << snapshot_str << "\n"; + /* relax for some time */ + std::this_thread::sleep_for (std::chrono::milliseconds(1000)); - } } } void TrexRpcServerAsync::_stop_rpc_thread() { m_is_running = false; - this->m_thread.join(); + this->m_thread->join(); zmq_term(m_context); } -- cgit From 74b648a86c16933680b03a736afe3f0305b4f6d2 Mon Sep 17 00:00:00 2001 From: imarom Date: Thu, 8 Oct 2015 08:34:43 +0200 Subject: some file renaming --- src/rpc-server/trex_rpc_async_server.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'src/rpc-server/trex_rpc_async_server.cpp') diff --git a/src/rpc-server/trex_rpc_async_server.cpp b/src/rpc-server/trex_rpc_async_server.cpp index 40d16dfe..01f03af3 100644 --- a/src/rpc-server/trex_rpc_async_server.cpp +++ b/src/rpc-server/trex_rpc_async_server.cpp @@ -24,7 +24,7 @@ limitations under the License. */ #define _GLIBCXX_USE_NANOSLEEP -#include +#include #include #include #include -- cgit From 09c9d77dc2f5a89924bd27226727220801a5df13 Mon Sep 17 00:00:00 2001 From: imarom Date: Thu, 8 Oct 2015 10:23:33 +0200 Subject: fixed some bugs in the async server also added affinity to the stateless main object --- src/rpc-server/trex_rpc_async_server.cpp | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) (limited to 'src/rpc-server/trex_rpc_async_server.cpp') diff --git a/src/rpc-server/trex_rpc_async_server.cpp b/src/rpc-server/trex_rpc_async_server.cpp index 01f03af3..3313e42e 100644 --- a/src/rpc-server/trex_rpc_async_server.cpp +++ b/src/rpc-server/trex_rpc_async_server.cpp @@ -87,13 +87,13 @@ TrexRpcServerAsync::_rpc_thread_cb() { /* relax for some time */ std::this_thread::sleep_for (std::chrono::milliseconds(1000)); - } + + /* must be closed from the same thread */ + zmq_close(m_socket); } void TrexRpcServerAsync::_stop_rpc_thread() { - m_is_running = false; - this->m_thread->join(); zmq_term(m_context); } -- cgit From 6c7880b9881ed6690954f0c29259dd0b584b3970 Mon Sep 17 00:00:00 2001 From: imarom Date: Sun, 11 Oct 2015 14:42:24 +0200 Subject: DP cores now inject a single packet as a dummy to see stats --- src/rpc-server/trex_rpc_async_server.cpp | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) (limited to 'src/rpc-server/trex_rpc_async_server.cpp') diff --git a/src/rpc-server/trex_rpc_async_server.cpp b/src/rpc-server/trex_rpc_async_server.cpp index 3313e42e..f4d21f2f 100644 --- a/src/rpc-server/trex_rpc_async_server.cpp +++ b/src/rpc-server/trex_rpc_async_server.cpp @@ -36,7 +36,7 @@ limitations under the License. * ZMQ based publisher server * */ -TrexRpcServerAsync::TrexRpcServerAsync(const TrexRpcServerConfig &cfg) : TrexRpcServerInterface(cfg, "publisher") { +TrexRpcServerAsync::TrexRpcServerAsync(const TrexRpcServerConfig &cfg, std::mutex *lock) : TrexRpcServerInterface(cfg, "publisher", lock) { /* ZMQ is not thread safe - this should be outside */ m_context = zmq_ctx_new(); } @@ -73,9 +73,19 @@ TrexRpcServerAsync::_rpc_thread_cb() { Json::Value snapshot; Json::FastWriter writer; + /* if lock was provided - take it */ + if (m_lock) { + m_lock->lock(); + } + /* trigger a full update for stats */ TrexStateless::get_instance().update_stats(); + /* done with the lock */ + if (m_lock) { + m_lock->unlock(); + } + /* encode them to JSON */ TrexStateless::get_instance().encode_stats(snapshot); -- cgit