4 files changed, 119 insertions, 13 deletions

diff --git a/src/stateless/cp/trex_dp_port_events.cpp b/src/stateless/cp/trex_dp_port_events.cpp
index 1321a362..fc96e00a 100644
--- a/src/stateless/cp/trex_dp_port_events.cpp
+++ b/src/stateless/cp/trex_dp_port_events.cpp
@@ -78,6 +78,9 @@ protected:
     virtual void on_event() {
         /* do nothing */
     }
+    virtual void on_error(int thread_id) {
+        /* do nothing */
+    }
 };
 
 void
@@ -105,14 +108,14 @@ TrexDpPortEvents::barrier() {
  * 
  */
 void 
-TrexDpPortEvents::on_core_reporting_in(int event_id, int thread_id) {
+TrexDpPortEvents::on_core_reporting_in(int event_id, int thread_id, bool status) {
     TrexDpPortEvent *event = lookup(event_id);
     /* event might have been deleted */
     if (!event) {
         return;
     }
 
-    bool done = event->on_core_reporting_in(thread_id);
+    bool done = event->on_core_reporting_in(thread_id, status);
 
     if (done) {
         destroy_event(event_id);
@@ -150,7 +153,7 @@ TrexDpPortEvent::init(TrexStatelessPort *port, int event_id, int timeout_ms) {
 }
 
 bool
-TrexDpPortEvent::on_core_reporting_in(int thread_id) {
+TrexDpPortEvent::on_core_reporting_in(int thread_id, bool status) {
     /* mark sure no double signal */
     if (m_signal.at(thread_id)) {
         std::stringstream err;
@@ -163,6 +166,11 @@ TrexDpPortEvent::on_core_reporting_in(int thread_id) {
     m_signal.at(thread_id) = true;
     m_pending_cnt--;
 
+    /* if any core reported an error - mark as a failure */
+    if (!status) {
+        on_error(thread_id);
+    }
+
     /* event occured */
     if (m_pending_cnt == 0) {
         on_event();
diff --git a/src/stateless/cp/trex_dp_port_events.h b/src/stateless/cp/trex_dp_port_events.h
index 3b8c8633..681e47ab 100644
--- a/src/stateless/cp/trex_dp_port_events.h
+++ b/src/stateless/cp/trex_dp_port_events.h
@@ -48,13 +48,22 @@ protected:
      */
     virtual void on_event() = 0;
 
+    /**
+     * when a thread ID encounter an error
+     * 
+     * @author imarom (20-Apr-16)
+     * 
+     * @param thread_id 
+     */
+    virtual void on_error(int thread_id) = 0;
+
     TrexStatelessPort *get_port() {
         return m_port;
     }
 
 private:
     void init(TrexStatelessPort *port, int event_id, int timeout_ms);
-    bool on_core_reporting_in(int thread_id);
+    bool on_core_reporting_in(int thread_id, bool status = true);
 
     std::unordered_map<int, bool>  m_signal;
     int                            m_pending_cnt;
@@ -98,7 +107,7 @@ public:
     /**
      * a core has reached the event 
      */
-    void on_core_reporting_in(int event_id, int thread_id);
+    void on_core_reporting_in(int event_id, int thread_id, bool status = true);
 
 private:
     TrexDpPortEvent *lookup(int event_id);
diff --git a/src/stateless/cp/trex_stateless_port.cpp b/src/stateless/cp/trex_stateless_port.cpp
index 90142d9b..360cc7d6 100644
--- a/src/stateless/cp/trex_stateless_port.cpp
+++ b/src/stateless/cp/trex_stateless_port.cpp
@@ -24,6 +24,7 @@ limitations under the License.
 #include <trex_stateless_messaging.h>
 #include <trex_streams_compiler.h>
 #include <common/basic_utils.h>
+#include <common/captureFile.h>
 
 #include <string>
 
@@ -70,6 +71,20 @@ protected:
         assert(get_port()->m_pending_async_stop_event != TrexDpPortEvents::INVALID_ID);
         get_port()->m_pending_async_stop_event = TrexDpPortEvents::INVALID_ID;
     }
+
+    /**
+     * when a DP core encountered an error
+     * 
+     * @author imarom (20-Apr-16)
+     */
+    virtual void on_error(int thread_id) {
+        Json::Value data;
+
+        data["port_id"]   = get_port()->get_port_id();
+        data["thread_id"] = thread_id;
+
+        get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_ERROR, data);
+    }
 };
 
 /***************************
@@ -240,6 +255,13 @@ TrexStatelessPort::start_traffic(const TrexPortMultiplier &mul, double duration,
 }
 
 
+bool TrexStatelessPort::is_active() const {
+    return   (  (m_port_state == PORT_STATE_TX) 
+             || (m_port_state == PORT_STATE_PAUSE)
+             || (m_port_state == PORT_STATE_PCAP_TX)
+             );
+}
+
 /**
  * stop traffic on port
  * 
@@ -249,9 +271,7 @@ TrexStatelessPort::start_traffic(const TrexPortMultiplier &mul, double duration,
  */
 void
 TrexStatelessPort::stop_traffic(void) {
-
-    if (!( (m_port_state == PORT_STATE_TX) 
-        || (m_port_state == PORT_STATE_PAUSE) )) {
+    if (!is_active()) {
         return;
     }
 
@@ -395,6 +415,55 @@ TrexStatelessPort::update_traffic(const TrexPortMultiplier &mul, bool force) {
 
 }
 
+void
+TrexStatelessPort::push_remote(const std::string &pcap_filename,
+                               double ipg_usec,
+                               double speedup,
+                               uint32_t count,
+                               double duration) {
+
+    /* command allowed only on state stream */
+    verify_state(PORT_STATE_IDLE | PORT_STATE_STREAMS);
+
+    /* check that file exists */
+    CCapReaderBase *reader;
+    std::stringstream ss;
+    reader = CCapReaderFactory::CreateReader((char *)pcap_filename.c_str(), 0, ss);
+    if (!reader) {
+        throw TrexException(ss.str());
+    }
+    delete reader;
+
+    /* only one core gets to play */
+    int tx_core = m_cores_id_list[0];
+
+    /* create async event */
+    assert(m_pending_async_stop_event == TrexDpPortEvents::INVALID_ID);
+    m_pending_async_stop_event = m_dp_events.create_event(new AsyncStopEvent());
+
+    /* mark all other cores as done */
+    for (int index = 1; index < m_cores_id_list.size(); index++) {
+        /* mimic an end event */
+        m_dp_events.on_core_reporting_in(m_pending_async_stop_event, m_cores_id_list[index]);
+    }
+
+    /* send a message to core */
+    change_state(PORT_STATE_PCAP_TX);
+    TrexStatelessCpToDpMsgBase *push_msg = new TrexStatelessDpPushPCAP(m_port_id,
+                                                                       m_pending_async_stop_event,
+                                                                       pcap_filename,
+                                                                       ipg_usec,
+                                                                       speedup,
+                                                                       count,
+                                                                       duration);
+    send_message_to_dp(tx_core, push_msg);
+
+    /* update subscribers */    
+    Json::Value data;
+    data["port_id"] = m_port_id;
+    get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_STARTED, data);
+}
+
 std::string 
 TrexStatelessPort::get_state_as_string() const {
 
@@ -413,6 +482,9 @@ TrexStatelessPort::get_state_as_string() const {
 
     case PORT_STATE_PAUSE:
         return "PAUSE";
+
+    case PORT_STATE_PCAP_TX:
+        return "PCAP_TX";
     }
 
     return "UNKNOWN";
diff --git a/src/stateless/cp/trex_stateless_port.h b/src/stateless/cp/trex_stateless_port.h
index 520940d8..8856b429 100644
--- a/src/stateless/cp/trex_stateless_port.h
+++ b/src/stateless/cp/trex_stateless_port.h
@@ -129,11 +129,12 @@ public:
      * port state
      */
     enum port_state_e {
-        PORT_STATE_DOWN = 0x1,
-        PORT_STATE_IDLE = 0x2,
-        PORT_STATE_STREAMS = 0x4,
-        PORT_STATE_TX = 0x8,
-        PORT_STATE_PAUSE = 0x10,
+        PORT_STATE_DOWN     = 0x1,
+        PORT_STATE_IDLE     = 0x2,
+        PORT_STATE_STREAMS  = 0x4,
+        PORT_STATE_TX       = 0x8,
+        PORT_STATE_PAUSE    = 0x10,
+        PORT_STATE_PCAP_TX  = 0x20,
     };
 
     /**
@@ -212,6 +213,16 @@ public:
     void update_traffic(const TrexPortMultiplier &mul, bool force);
 
     /**
+     * push a PCAP file onto the port
+     * 
+     */
+    void push_remote(const std::string &pcap_filename,
+                     double ipg_usec,
+                     double speedup,
+                     uint32_t count,
+                     double duration);
+
+    /**
      * get the port state
      *
      */
@@ -220,6 +231,12 @@ public:
     }
 
     /**
+     * return true if the port is active
+     * (paused is considered active)
+     */
+    bool is_active() const;
+
+    /**
      * port state as string
      *
      */