add support for a program of streams. refactor the dp code

1 files changed, 88 insertions, 13 deletions

diff --git a/src/stateless/dp/trex_stream_node.h b/src/stateless/dp/trex_stream_node.h
index e4cf964d..1e53887b 100644
--- a/src/stateless/dp/trex_stream_node.h
+++ b/src/stateless/dp/trex_stream_node.h
@@ -1,5 +1,6 @@
 /*
  Itay Marom
+ Hanoh Haim
  Cisco Systems, Inc.
 */
 
@@ -27,20 +28,51 @@ limitations under the License.
 class TrexStatelessDpCore;
 #include <trex_stream.h>
 
+class TrexStatelessCpToDpMsgBase;
+
+struct CGenNodeCommand : public CGenNodeBase  {
+
+friend class TrexStatelessDpCore;
+
+public:
+    TrexStatelessCpToDpMsgBase * m_cmd;
+
+    uint8_t             m_pad_end[104];
+
+public:
+    void free_command();
+
+} __rte_cache_aligned;;
+
+
+static_assert(sizeof(CGenNodeCommand) == sizeof(CGenNode), "sizeof(CGenNodeCommand) != sizeof(CGenNode)" );
+
 /* this is a event for stateless */
 struct CGenNodeStateless : public CGenNodeBase  {
 friend class TrexStatelessDpCore;
 
+public:
+    enum {                                          
+            ss_FREE_RESUSE =1, /* should be free by scheduler */
+            ss_INACTIVE    =2, /* will be active by other stream or stopped */
+            ss_ACTIVE      =3  /* the stream is active */ 
+         };
+    typedef uint8_t stream_state_t ;
+
+    static std::string get_stream_state_str(stream_state_t stream_state);
+
 private:
+    /* cache line 0 */
+    /* important stuff here */
     void *              m_cache_mbuf;
 
     double              m_next_time_offset; /* in sec */
     double              m_ibg_sec; /* inter burst time in sec */
 
 
-    uint8_t             m_is_stream_active;
+    stream_state_t      m_state;
     uint8_t             m_port_id;
-    uint8_t             m_stream_type; /* TrexStream::STREAM_TYPE */
+    uint8_t             m_stream_type; /* see TrexStream::STREAM_TYPE ,stream_type_t */
     uint8_t             m_pad;
 
     uint32_t            m_single_burst; /* the number of bursts in case of burst */
@@ -48,14 +80,40 @@ private:
 
     uint32_t            m_multi_bursts; /* in case of multi_burst how many bursts */
 
-
+    /* cache line 1 */
+    TrexStream *        m_ref_stream_info; /* the stream info */
+    CGenNodeStateless  * m_next_stream;
 
     /* pad to match the size of CGenNode */
-    uint8_t             m_pad_end[65];
+    uint8_t             m_pad_end[56];
+
+
 
 
 public:
 
+    uint8_t             get_port_id(){
+        return (m_port_id);
+    }
+
+
+    /* we restart the stream, schedule it using stream isg */
+    inline void update_refresh_time(double cur_time){
+        m_time = cur_time + usec_to_sec(m_ref_stream_info->m_isg_usec);
+    }
+
+    inline bool is_mask_for_free(){
+        return (get_state() == CGenNodeStateless::ss_FREE_RESUSE ?true:false);
+
+    }
+    inline void mark_for_free(){
+        set_state(CGenNodeStateless::ss_FREE_RESUSE);
+        /* only to be safe */
+        m_ref_stream_info= NULL;
+        m_next_stream= NULL;
+
+    }
+
     inline uint8_t  get_stream_type(){
         return (m_stream_type);
     }
@@ -72,11 +130,16 @@ public:
         return (m_multi_bursts);
     }
 
+    inline  void set_state(stream_state_t new_state){
+        m_state=new_state;
+    }
 
-    inline bool is_active() {
-        return m_is_stream_active;
+
+    inline stream_state_t get_state() {
+        return m_state;
     }
 
+    void refresh();
 
     inline void handle_continues(CFlowGenListPerThread *thread) {
         thread->m_node_gen.m_v_if->send_node( (CGenNode *)this);
@@ -100,14 +163,22 @@ public:
         }else{
             m_multi_bursts--;
             if ( m_multi_bursts == 0 ) {
-                /* stop */
-                m_is_stream_active =0;
+                set_state(CGenNodeStateless::ss_INACTIVE);
+                if ( thread->set_stateless_next_node(this,m_next_stream) ){
+                    /* update the next stream time using isg */
+                    m_next_stream->update_refresh_time(m_time);
+
+                    thread->m_node_gen.m_p_queue.push( (CGenNode *)m_next_stream);
+                }else{
+                    // in case of zero we will schedule a command to stop 
+                    // will be called from set_stateless_next_node
+                }
+
             }else{
                 m_time += m_ibg_sec;
                 m_single_burst = m_single_burst_refill;
-
+                thread->m_node_gen.m_p_queue.push( (CGenNode *)this);
             }
-            thread->m_node_gen.m_p_queue.push( (CGenNode *)this);
         }
     }
 
@@ -168,10 +239,14 @@ public:
 
     void free_stl_node();
 
+public:
+    /* debug functions */
 
-    void Dump(FILE *fd){
-        fprintf(fd," %f, %lu, %lu \n",m_time,(ulong)m_port_id,(ulong)get_mbuf_cache_dir());
-    }
+    int get_stream_id();
+
+    static void DumpHeader(FILE *fd);
+
+    void Dump(FILE *fd);
 
 } __rte_cache_aligned;