2 files changed, 170 insertions, 7 deletions
diff --git a/src/stateless/messaging/trex_stateless_messaging.cpp b/src/stateless/messaging/trex_stateless_messaging.cpp
index 95613b41..c2182f3c 100644
--- a/src/stateless/messaging/trex_stateless_messaging.cpp
+++ b/src/stateless/messaging/trex_stateless_messaging.cpp
@@ -241,13 +241,13 @@ TrexDpPortEventMsg::handle() {
 }
 
 /************************* messages from CP to RX **********************/
-bool TrexStatelessRxStartMsg::handle (CRxCoreStateless *rx_core) {
-    rx_core->work();
+bool TrexStatelessRxEnableLatency::handle (CRxCoreStateless *rx_core) {
+    rx_core->enable_latency();
     return true;
 }
 
-bool TrexStatelessRxStopMsg::handle (CRxCoreStateless *rx_core) {
-    rx_core->idle();
+bool TrexStatelessRxDisableLatency::handle (CRxCoreStateless *rx_core) {
+    rx_core->disable_latency();
     return true;
 }
 
@@ -255,3 +255,42 @@ bool TrexStatelessRxQuit::handle (CRxCoreStateless *rx_core) {
     rx_core->quit();
     return true;
 }
+
+
+bool
+TrexStatelessRxStartCapture::handle(CRxCoreStateless *rx_core) {
+    rx_core->start_capture(m_port_id, m_pcap_filename, m_limit, m_shared_counter);
+
+    return true;
+}
+
+bool
+TrexStatelessRxStopCapture::handle(CRxCoreStateless *rx_core) {
+    rx_core->stop_capture(m_port_id);
+
+    return true;
+}
+
+bool
+TrexStatelessRxStartQueue::handle(CRxCoreStateless *rx_core) {
+    rx_core->start_queue(m_port_id, m_size, m_shared_counter);
+
+    return true;
+}
+
+bool
+TrexStatelessRxStopQueue::handle(CRxCoreStateless *rx_core) {
+    rx_core->stop_queue(m_port_id);
+
+    return true;
+}
+
+
+
+bool TrexStatelessRxQueueGetPkts::handle(CRxCoreStateless *rx_core) {
+    RxPacketBuffer *pkt_buffer = rx_core->get_rx_queue_pkts(m_port_id);
+    assert(pkt_buffer);
+    m_reply.set(pkt_buffer);
+
+    return true;
+}
diff --git a/src/stateless/messaging/trex_stateless_messaging.h b/src/stateless/messaging/trex_stateless_messaging.h
index fb2c27ab..52b1662e 100644
--- a/src/stateless/messaging/trex_stateless_messaging.h
+++ b/src/stateless/messaging/trex_stateless_messaging.h
@@ -24,11 +24,15 @@ limitations under the License.
 
 #include "msg_manager.h"
 #include "trex_dp_port_events.h"
+#include "trex_exception.h"
+#include "trex_stateless_rx_defs.h"
+#include "os_time.h"
 
 class TrexStatelessDpCore;
 class CRxCoreStateless;
 class TrexStreamsCompiledObj;
 class CFlowGenListPerThread;
+class RxPacketBuffer;
 
 /**
  * defines the base class for CP to DP messages
@@ -312,7 +316,7 @@ private:
 /************************* messages from DP to CP **********************/
 
 /**
- * defines the base class for CP to DP messages
+ * defines the base class for DP to CP messages
  *
  * @author imarom (27-Oct-15)
  */
@@ -404,11 +408,11 @@ public:
 
 };
 
-class TrexStatelessRxStartMsg : public TrexStatelessCpToRxMsgBase {
+class TrexStatelessRxEnableLatency : public TrexStatelessCpToRxMsgBase {
     bool handle (CRxCoreStateless *rx_core);
 };
 
-class TrexStatelessRxStopMsg : public TrexStatelessCpToRxMsgBase {
+class TrexStatelessRxDisableLatency : public TrexStatelessCpToRxMsgBase {
     bool handle (CRxCoreStateless *rx_core);
 };
 
@@ -416,4 +420,124 @@ class TrexStatelessRxQuit : public TrexStatelessCpToRxMsgBase {
     bool handle (CRxCoreStateless *rx_core);
 };
 
+
+class TrexStatelessRxStartCapture : public TrexStatelessCpToRxMsgBase {
+public:
+    TrexStatelessRxStartCapture(uint8_t port_id, RXCaptureInfo &rx_capture_info) {
+        m_port_id          = port_id;
+        m_limit            = rx_capture_info.m_limit;
+        m_pcap_filename    = rx_capture_info.m_pcap_filename;
+        m_shared_counter   = &rx_capture_info.m_shared_counter;
+    }
+
+    virtual bool handle(CRxCoreStateless *rx_core);
+
+private:
+    uint8_t           m_port_id;
+    std::string       m_pcap_filename;
+    uint64_t          m_limit;
+    uint64_t          *m_shared_counter;
+};
+
+
+class TrexStatelessRxStopCapture : public TrexStatelessCpToRxMsgBase {
+public:
+    TrexStatelessRxStopCapture(uint8_t port_id) {
+        m_port_id = port_id;
+    }
+
+    virtual bool handle(CRxCoreStateless *rx_core);
+
+private:
+    uint8_t           m_port_id;
+};
+
+class TrexStatelessRxStartQueue : public TrexStatelessCpToRxMsgBase {
+public:
+    TrexStatelessRxStartQueue(uint8_t port_id, RXQueueInfo &rx_queue_info) {
+        m_port_id           = port_id;
+        m_size              = rx_queue_info.m_size;
+        m_shared_counter    = &rx_queue_info.m_shared_counter;
+    }
+
+    virtual bool handle(CRxCoreStateless *rx_core);
+
+private:
+    uint8_t           m_port_id;
+    uint64_t          m_size;
+    uint64_t          *m_shared_counter;
+};
+
+class TrexStatelessRxStopQueue : public TrexStatelessCpToRxMsgBase {
+public:
+    TrexStatelessRxStopQueue(uint8_t port_id) {
+        m_port_id = port_id;
+    }
+
+    virtual bool handle(CRxCoreStateless *rx_core);
+
+private:
+    uint8_t           m_port_id;
+};
+
+
+template<typename T> class TrexStatelessMsgReply {
+public:
+    TrexStatelessMsgReply() {
+        m_pending = true;
+    }
+
+    bool is_pending() const {
+        return m_pending;
+    }
+
+    void set(T reply) {
+        m_reply = reply;
+
+        /* before marking as done - memory fence */
+        asm volatile("mfence" ::: "memory");
+        m_pending = false;
+    }
+
+    T wait_for_reply(int timeout_ms = 100, int backoff_ms = 1) {
+        int guard = timeout_ms;
+
+        while (is_pending()) {
+            guard -= backoff_ms;
+            if (guard < 0) {
+                throw TrexException("timeout: RX core has failed to reply");
+            }
+
+            delay(backoff_ms);
+            
+        }
+        return m_reply;
+
+    }
+private:
+    bool  m_pending;
+    T     m_reply;
+};
+
+
+
+class TrexStatelessRxQueueGetPkts : public TrexStatelessCpToRxMsgBase {
+public:
+
+    TrexStatelessRxQueueGetPkts(uint8_t port_id, TrexStatelessMsgReply<RxPacketBuffer *> &reply) : m_reply(reply) {
+        m_port_id = port_id;
+    }
+
+    /**
+     * virtual function to handle a message
+     *
+     */
+    virtual bool handle(CRxCoreStateless *rx_core);
+
+private:
+    uint8_t                                    m_port_id;
+    TrexStatelessMsgReply<RxPacketBuffer*>    &m_reply;
+};
+
+
 #endif /* __TREX_STATELESS_MESSAGING_H__ */