Merge branch 'master' into dan_stateless

13 files changed, 777 insertions, 194 deletions

diff --git a/src/stateless/cp/trex_stateless.h b/src/stateless/cp/trex_stateless.h
index 5c11be1e..59be9241 100644
--- a/src/stateless/cp/trex_stateless.h
+++ b/src/stateless/cp/trex_stateless.h
@@ -161,6 +161,10 @@ public:
         return m_ports;
     }
 
+    TrexRpcServer * get_rpc_server() {
+        return m_rpc_server;
+    }
+
 protected:
 
     /* no copy or assignment */
@@ -168,7 +172,7 @@ protected:
     void operator=(TrexStateless const&)     = delete;
 
     /* RPC server array */
-    TrexRpcServer        *m_rpc_server;
+    TrexRpcServer                        *m_rpc_server;
 
     /* ports */
     std::vector <TrexStatelessPort *>    m_ports;
diff --git a/src/stateless/cp/trex_stateless_port.cpp b/src/stateless/cp/trex_stateless_port.cpp
index 3a64f8c5..aa34e87b 100644
--- a/src/stateless/cp/trex_stateless_port.cpp
+++ b/src/stateless/cp/trex_stateless_port.cpp
@@ -72,6 +72,8 @@ TrexStatelessPort::TrexStatelessPort(uint8_t port_id, const TrexPlatformApi *api
 
     /* init the events DP DB */
     m_dp_events.create(this);
+
+    m_graph_obj = NULL;
 }
 
 
diff --git a/src/stateless/cp/trex_stream.cpp b/src/stateless/cp/trex_stream.cpp
index 02f43a3a..976cfa07 100644
--- a/src/stateless/cp/trex_stream.cpp
+++ b/src/stateless/cp/trex_stream.cpp
@@ -54,32 +54,19 @@ std::string TrexStream::get_stream_type_str(stream_type_t stream_type){
 
 
 void
-TrexStream::compile() {
+TrexStream::vm_compile() {
 
     /* in case there are no instructions - nothing to do */
     if (m_vm.is_vm_empty()) {
-        m_has_vm = false;
         return;
     }
 
-    m_has_vm = true;
+    /* compile */
+    m_vm.compile(m_pkt.len);
 
-    m_vm.set_packet_size(m_pkt.len);
+    /* create DP object */
+    m_vm_dp = m_vm.generate_dp_object();
 
-    m_vm.compile();
-
-    #if 0
-    m_vm.Dump(stdout);
-    #endif
-
-    m_vm_dp = m_vm.cloneAsVmDp();
-
-    /* calc m_vm_prefix_size which is the size of the writable packet */
-    uint16_t max_pkt_offset = m_vm_dp->get_max_packet_update_offset();
-    uint16_t pkt_size = m_pkt.len;
-
-    /* calculate the mbuf size that we should allocate */
-    m_vm_prefix_size = calc_writable_mbuf_size(max_pkt_offset, pkt_size);
 }
 
 
@@ -133,8 +120,6 @@ TrexStream::TrexStream(uint8_t type,
 
     m_pkt.binary = NULL;
     m_pkt.len    = 0;
-    m_has_vm = false;
-    m_vm_prefix_size = 0;
 
     m_rx_check.m_enable = false;
 
@@ -150,9 +135,9 @@ TrexStream::~TrexStream() {
     if (m_pkt.binary) {
         delete [] m_pkt.binary;
     }
-    if ( m_vm_dp ){
-        delete  m_vm_dp;
-        m_vm_dp=NULL;
+    if (m_vm_dp){
+        delete m_vm_dp;
+        m_vm_dp = NULL;
     }
 }
 
diff --git a/src/stateless/cp/trex_stream.h b/src/stateless/cp/trex_stream.h
index 720246f6..b4f19111 100644
--- a/src/stateless/cp/trex_stream.h
+++ b/src/stateless/cp/trex_stream.h
@@ -65,7 +65,7 @@ static inline uint16_t get_log2_size(uint16_t size){
  * 
  */
 static inline uint16_t calc_writable_mbuf_size(uint16_t max_offset_writable,
-                                 uint16_t pkt_size){
+                                               uint16_t pkt_size){
 
     if ( pkt_size<=64 ){
         return (pkt_size);
@@ -177,36 +177,42 @@ public:
     }
 
     /* create new stream */
-    TrexStream * clone_as_dp() const {
-
-            TrexStream *dp = new TrexStream(m_type,m_port_id,m_stream_id);
-            dp->m_has_vm = m_has_vm;
-            if (m_vm_dp) {
-                /* should have vm */
-                assert(m_has_vm);
-                dp->m_vm_dp = m_vm_dp->clone();
-            }else{
-                dp->m_vm_dp = NULL;
-            }
-            dp->m_vm_prefix_size = m_vm_prefix_size;
-
-            dp->m_isg_usec      = m_isg_usec;
-            dp->m_next_stream_id = m_next_stream_id;
-
-            dp->m_enabled    = m_enabled;
-            dp->m_self_start = m_self_start;
-
-            /* deep copy */
-            dp->m_pkt.clone(m_pkt.binary,m_pkt.len);
-
-            dp->m_rx_check              =   m_rx_check;
-            dp->m_pps                   =   m_pps;
-            dp->m_burst_total_pkts      =   m_burst_total_pkts;
-            dp->m_num_bursts            =   m_num_bursts;
-            dp->m_ibg_usec              =   m_ibg_usec ;
-            return (dp);
+    TrexStream * clone() const {
+
+        /* not all fields will be cloned */
+
+        TrexStream *dp = new TrexStream(m_type,m_port_id,m_stream_id);
+        if (m_vm_dp) {
+            dp->m_vm_dp = m_vm_dp->clone();
+        } else {
+            dp->m_vm_dp = NULL;
+        }
+
+        dp->m_isg_usec      = m_isg_usec;
+        dp->m_next_stream_id = m_next_stream_id;
+
+        dp->m_enabled    = m_enabled;
+        dp->m_self_start = m_self_start;
+
+        /* deep copy */
+        dp->m_pkt.clone(m_pkt.binary,m_pkt.len);
+
+        dp->m_rx_check              =   m_rx_check;
+        dp->m_pps                   =   m_pps;
+        dp->m_burst_total_pkts      =   m_burst_total_pkts;
+        dp->m_num_bursts            =   m_num_bursts;
+        dp->m_ibg_usec              =   m_ibg_usec;
+
+        return(dp);
     }
 
+    /* release the DP object */
+    void release_dp_object() {
+        if (m_vm_dp) {
+            delete m_vm_dp;
+            m_vm_dp = NULL;
+        }
+    }
 
     double get_burst_length_usec() const {
         return ( (m_burst_total_pkts / m_pps) * 1000 * 1000);
@@ -219,27 +225,20 @@ public:
 
     void Dump(FILE *fd);
 
-    bool is_vm(){
-        return (  m_has_vm );
+    StreamVmDp * getDpVm(){
+        return (m_vm_dp);
     }
 
-    StreamVmDp  * getDpVm(){
-        return ( m_vm_dp);
-    }
-
-    void post_vm_compile();
-
     /**
      * internal compilation of stream (for DP)
      * 
      */
-    void compile();
+    void vm_compile();
 
 public:
     /* basic */
     uint8_t       m_type;
     uint8_t       m_port_id;
-    uint16_t      m_vm_prefix_size;          /* writeable mbuf size */
     uint32_t      m_stream_id;              /* id from RPC can be anything */
     
 
@@ -250,16 +249,15 @@ public:
     /* indicators */
     bool          m_enabled;
     bool          m_self_start;
-    bool          m_has_vm; /* do we have instructions to run */
 
 
-    StreamVmDp  * m_vm_dp; /* compile VM  */
+    /* VM CP and DP */
+    StreamVm      m_vm;
+    StreamVmDp   *m_vm_dp;
 
     CStreamPktData   m_pkt;
     /* pkt */
 
-    /* VM */
-    StreamVm m_vm;
 
     /* RX check */
     struct {
diff --git a/src/stateless/cp/trex_stream_vm.cpp b/src/stateless/cp/trex_stream_vm.cpp
index e10e1a81..a3f585ad 100644
--- a/src/stateless/cp/trex_stream_vm.cpp
+++ b/src/stateless/cp/trex_stream_vm.cpp
@@ -191,7 +191,29 @@ void StreamVm::build_flow_var_table() {
     var_clear_table();
     m_cur_var_offset=0;
     uint32_t ins_id=0;
+    m_is_random_var=false;
         /* scan all flow var instruction and build */
+
+    for (auto inst : m_inst_list) {
+        if ( inst->get_instruction_type() == StreamVmInstruction::itFLOW_MAN ){
+            StreamVmInstructionFlowMan * ins_man=(StreamVmInstructionFlowMan *)inst;
+            if (ins_man->m_op ==StreamVmInstructionFlowMan::FLOW_VAR_OP_RANDOM){
+                m_is_random_var =true;
+            }
+        }
+    }
+
+    /* if we found allocate BSS +4 bytes */
+    if ( m_is_random_var ){
+        VmFlowVarRec var;
+
+        var.m_offset = m_cur_var_offset;
+        var.m_ins.m_ins_flowv = NULL;
+        var.m_size_bytes = sizeof(uint32_t);
+        var_add("___random___",var);
+        m_cur_var_offset += sizeof(uint32_t);
+    }
+
     for (auto inst : m_inst_list) {
         if ( inst->get_instruction_type() == StreamVmInstruction::itFLOW_MAN ){
 
@@ -207,6 +229,7 @@ void StreamVm::build_flow_var_table() {
                 ss << "instruction id '" << ins_id << "' flow variable name " << ins_man->m_var_name << " already exists";
                 err(ss.str());
             }else{
+
                 var.m_offset=m_cur_var_offset;
                 var.m_ins.m_ins_flowv = ins_man;
                 var.m_size_bytes = ins_man->m_size_bytes;
@@ -426,7 +449,7 @@ void StreamVm::build_program(){
                     op = StreamDPVmInstructions::ditRANDOM64 ;
                 }
 
-                StreamDPOpFlowVar32 fv64;
+                StreamDPOpFlowVar64 fv64;
                 fv64.m_op = op;
                 fv64.m_flow_offset = get_var_offset(lpMan->m_var_name);
                 fv64.m_min_val     = (uint64_t)lpMan->m_min_value;
@@ -541,6 +564,11 @@ void StreamVm::build_bss() {
     alloc_bss();
     uint8_t * p=(uint8_t *)m_bss;
 
+    if ( m_is_random_var ){
+        *((uint32_t*)p)=rand();
+        p+=sizeof(uint32_t);
+    }
+
     for (auto inst : m_inst_list) {
 
         if ( inst->get_instruction_type() == StreamVmInstruction::itFLOW_MAN ){
@@ -549,19 +577,19 @@ void StreamVm::build_bss() {
 
             switch (ins_man->m_size_bytes) {
             case 1:
-                *p=(uint8_t)ins_man->m_init_value;
+                *p=(uint8_t)ins_man->get_bss_init_value();
                 p+=1;
                 break;
             case 2:
-                *((uint16_t*)p)=(uint16_t)ins_man->m_init_value;
+                *((uint16_t*)p)=(uint16_t)ins_man->get_bss_init_value();
                 p+=2;
                 break;
             case 4:
-                *((uint32_t*)p)=(uint32_t)ins_man->m_init_value;
+                *((uint32_t*)p)=(uint32_t)ins_man->get_bss_init_value();
                 p+=4;
                 break;
             case 8:
-                *((uint64_t*)p)=(uint64_t)ins_man->m_init_value;
+                *((uint64_t*)p)=(uint64_t)ins_man->get_bss_init_value();
                 p+=8;
                 break;
             default:
@@ -593,9 +621,58 @@ void StreamVm::build_bss() {
     }
 }
 
+/**
+ * set the VM split instruction 
+ * instr is a pointer to an instruction inside 
+ * the VM program 
+ * 
+ */
+void
+StreamVm::set_split_instruction(StreamVmInstructionVar *instr) {
+    m_split_instr = instr;
+}
+
+/**
+ * copy instructions from this VM to 'other'
+ * 
+ * @author imarom (22-Dec-15)
+ * 
+ * @param other 
+ */
+void 
+StreamVm::copy_instructions(StreamVm &other) const {
+    /* clear previous if any exists */
+    for (auto instr : other.m_inst_list) {
+        delete instr;
+    }
 
+    other.m_inst_list.clear();
 
-void StreamVm::compile() {
+    for (auto instr : m_inst_list) {
+        StreamVmInstruction *new_instr = instr->clone();
+        other.m_inst_list.push_back(new_instr);
+
+        /* for the split instruction - find the right one */
+        if (instr == m_split_instr) {
+            /* dynamic cast must succeed here */
+            other.m_split_instr = dynamic_cast<StreamVmInstructionVar *>(new_instr);
+            assert(other.m_split_instr);
+        }
+    }
+
+}
+
+/**
+ * actual work - compile the VM
+ * 
+ */
+void StreamVm::compile(uint16_t pkt_len) {
+
+    if (is_vm_empty()) {
+        return;
+    }
+
+    m_pkt_size = pkt_len;
 
     /* build flow var offset table */
     build_flow_var_table() ;
@@ -610,6 +687,11 @@ void StreamVm::compile() {
         ss << "maximum offset is" << get_max_packet_update_offset() << " bigger than maximum " <<svMAX_PACKET_OFFSET_CHANGE;
         err(ss.str());
     }
+
+    /* calculate the mbuf size that we should allocate */
+    m_prefix_size = calc_writable_mbuf_size(get_max_packet_update_offset(), m_pkt_size);
+
+    m_is_compiled = true;
 }
 
 
@@ -620,6 +702,30 @@ StreamVm::~StreamVm() {
     free_bss();
 }
 
+/** 
+* return a pointer to a flow var / client var 
+* by name if exists, otherwise NULL 
+* 
+*/
+StreamVmInstructionVar *
+StreamVm::lookup_var_by_name(const std::string &var_name) {
+    for (StreamVmInstruction *inst : m_inst_list) {
+
+        /* try to cast up to a variable */
+        StreamVmInstructionVar *var = dynamic_cast<StreamVmInstructionVar *>(inst);
+        if (!var) {
+            continue;
+        }
+
+        if (var->get_var_name() == var_name) {
+            return var;
+        }
+
+    }
+
+    return NULL;
+}
+
 
 void StreamVm::Dump(FILE *fd){
     fprintf(fd," instructions \n");
diff --git a/src/stateless/cp/trex_stream_vm.h b/src/stateless/cp/trex_stream_vm.h
index e65a87e3..891e5b51 100644
--- a/src/stateless/cp/trex_stream_vm.h
+++ b/src/stateless/cp/trex_stream_vm.h
@@ -32,6 +32,46 @@ limitations under the License.
 
 
 
+//https://software.intel.com/en-us/articles/fast-random-number-generator-on-the-intel-pentiumr-4-processor/
+
+//Used to seed the generator.
+inline void fast_srand(uint32_t &g_seed, int seed ){
+  g_seed = seed;
+}
+
+
+//fastrand routine returns one integer, similar output value range as C lib.
+
+inline int fastrand(uint32_t &g_seed)
+{
+  g_seed = (214013*g_seed+2531011);
+  return (g_seed>>16)&0x7FFF;
+}         
+
+static inline void vm_srand(uint32_t * per_thread_seed,uint64_t seedval)
+{
+    fast_srand( *per_thread_seed,seedval );
+}
+
+static inline uint32_t vm_rand16(uint32_t * per_thread_seed)
+{
+	return ( fastrand(*per_thread_seed));
+}
+
+static inline uint32_t vm_rand32(uint32_t * per_thread_seed)
+{
+	return ( (vm_rand16(per_thread_seed)<<16)+vm_rand16(per_thread_seed));
+}
+
+static inline uint64_t vm_rand64(uint32_t * per_thread_seed)
+{
+    uint64_t res;
+
+    res=((uint64_t)vm_rand32(per_thread_seed)<<32)+vm_rand32(per_thread_seed);
+
+    return (res);
+}
+                          
 
 class StreamVm;
 
@@ -47,24 +87,26 @@ public:
     void dump(FILE *fd,std::string opt);
 
     inline void run_inc(uint8_t * flow_var) {
-        uint8_t * p=(flow_var+m_flow_offset);
-        *p=*p+1;
-        if (*p>m_max_val) {
-            *p=m_min_val;
+        uint8_t *p = (flow_var + m_flow_offset);
+        if (*p == m_max_val) {
+            *p = m_min_val;
+        } else {
+            *p = *p + 1;
         }
     }
 
     inline void run_dec(uint8_t * flow_var) {
-        uint8_t * p=(flow_var+m_flow_offset);
-        *p=*p-1;
-        if (*p<m_min_val) {
-            *p=m_max_val;
+        uint8_t *p = (flow_var + m_flow_offset);
+        if (*p == m_min_val) {
+            *p = m_max_val;
+        } else {
+            *p = *p - 1;
         }
     }
 
-    inline void run_rand(uint8_t * flow_var) {
+    inline void run_rand(uint8_t * flow_var,uint32_t *per_thread_random) {
         uint8_t * p=(flow_var+m_flow_offset);
-        *p= m_min_val + (rand() % (int)(m_max_val - m_min_val + 1));
+        *p= m_min_val + (vm_rand16(per_thread_random)  % (int)(m_max_val - m_min_val + 1));
     }
 
 
@@ -79,24 +121,26 @@ public:
     void dump(FILE *fd,std::string opt);
 
     inline void run_inc(uint8_t * flow_var) {
-        uint16_t * p=(uint16_t *)(flow_var+m_flow_offset);
-        *p=*p+1;
-        if (*p>m_max_val) {
-            *p=m_min_val;
+        uint16_t *p = (uint16_t *)(flow_var + m_flow_offset);
+        if (*p == m_max_val) {
+            *p = m_min_val;
+        } else {
+            *p = *p + 1;
         }
     }
 
     inline void run_dec(uint8_t * flow_var) {
-        uint16_t * p=(uint16_t *)(flow_var+m_flow_offset);
-        *p=*p-1;
-        if (*p<m_min_val) {
-            *p=m_max_val;
+        uint16_t *p = (uint16_t *)(flow_var + m_flow_offset);
+        if (*p == m_min_val) {
+            *p = m_max_val;
+        } else {
+            *p = *p - 1;
         }
     }
 
-    inline void run_rand(uint8_t * flow_var) {
+    inline void run_rand(uint8_t * flow_var,uint32_t *per_thread_random) {
         uint16_t * p=(uint16_t *)(flow_var+m_flow_offset);
-        *p= m_min_val + (rand() % (int)(m_max_val - m_min_val + 1));
+        *p= m_min_val + (vm_rand16(per_thread_random) % (int)(m_max_val - m_min_val + 1));
     }
 
 
@@ -112,24 +156,26 @@ public:
     void dump(FILE *fd,std::string opt);
 
     inline void run_inc(uint8_t * flow_var) {
-        uint32_t * p=(uint32_t *)(flow_var+m_flow_offset);
-        *p=*p+1;
-        if (*p>m_max_val) {
-            *p=m_min_val;
+        uint32_t *p = (uint32_t *)(flow_var + m_flow_offset);
+        if (*p == m_max_val) {
+            *p = m_min_val;
+        } else {
+            *p = *p + 1;
         }
     }
 
     inline void run_dec(uint8_t * flow_var) {
-        uint32_t * p=(uint32_t *)(flow_var+m_flow_offset);
-        *p=*p-1;
-        if (*p<m_min_val) {
-            *p=m_max_val;
+        uint32_t *p = (uint32_t *)(flow_var + m_flow_offset);
+        if (*p == m_min_val) {
+            *p = m_max_val;
+        } else {
+            *p = *p - 1;
         }
     }
 
-    inline void run_rand(uint8_t * flow_var) {
+    inline void run_rand(uint8_t * flow_var,uint32_t *per_thread_random) {
         uint32_t * p=(uint32_t *)(flow_var+m_flow_offset);
-        *p= m_min_val + (rand() % (int)(m_max_val - m_min_val + 1));
+        *p= m_min_val + (vm_rand32(per_thread_random) % (int)(m_max_val - m_min_val + 1));
     }
 
 } __attribute__((packed)) ;
@@ -143,24 +189,26 @@ public:
     void dump(FILE *fd,std::string opt);
 
     inline void run_inc(uint8_t * flow_var) {
-        uint64_t * p=(uint64_t *)(flow_var+m_flow_offset);
-        *p=*p+1;
-        if (*p>m_max_val) {
-            *p=m_min_val;
+        uint64_t *p = (uint64_t *)(flow_var + m_flow_offset);
+        if (*p == m_max_val) {
+            *p = m_min_val;
+        } else {
+            *p = *p + 1;
         }
     }
 
     inline void run_dec(uint8_t * flow_var) {
-        uint64_t * p=(uint64_t *)(flow_var+m_flow_offset);
-        *p=*p-1;
-        if (*p<m_min_val) {
-            *p=m_max_val;
+        uint64_t *p = (uint64_t *)(flow_var + m_flow_offset);
+        if (*p == m_min_val) {
+            *p = m_max_val;
+        } else {
+            *p = *p - 1;
         }
     }
 
-    inline void run_rand(uint8_t * flow_var) {
+    inline void run_rand(uint8_t * flow_var,uint32_t *per_thread_random) {
         uint64_t * p=(uint64_t *)(flow_var+m_flow_offset);
-        *p= m_min_val + (rand() % (int)(m_max_val - m_min_val + 1));
+        *p= m_min_val + ( vm_rand64(per_thread_random)  % (int)(m_max_val - m_min_val + 1));
     }
 
 
@@ -352,7 +400,7 @@ public:
 class StreamDPVmInstructions {
 public:
     enum INS_TYPE {
-        ditINC8         ,
+        ditINC8     =7 ,
         ditINC16        ,
         ditINC32        ,
         ditINC64        ,
@@ -395,7 +443,8 @@ private:
 
 class StreamDPVmInstructionsRunner {
 public:
-    inline void run(uint32_t program_size,
+    inline void run(uint32_t * per_thread_random,
+                    uint32_t program_size,
                     uint8_t * program,  /* program */
                     uint8_t * flow_var, /* flow var */
                     uint8_t * pkt);      /* pkt */
@@ -403,7 +452,8 @@ public:
 };
 
 
-inline void StreamDPVmInstructionsRunner::run(uint32_t program_size,
+inline void StreamDPVmInstructionsRunner::run(uint32_t * per_thread_random,
+                                              uint32_t program_size,
                                               uint8_t * program,  /* program */
                                               uint8_t * flow_var, /* flow var */
                                               uint8_t * pkt){
@@ -488,22 +538,22 @@ inline void StreamDPVmInstructionsRunner::run(uint32_t program_size,
 
         case  StreamDPVmInstructions::ditRANDOM8 :
             ua.lpv8 =(StreamDPOpFlowVar8 *)p;
-            ua.lpv8->run_rand(flow_var);
+            ua.lpv8->run_rand(flow_var,per_thread_random);
             p+=sizeof(StreamDPOpFlowVar8);
             break;
         case  StreamDPVmInstructions::ditRANDOM16 :
             ua.lpv16 =(StreamDPOpFlowVar16 *)p;
-            ua.lpv16->run_rand(flow_var);
+            ua.lpv16->run_rand(flow_var,per_thread_random);
             p+=sizeof(StreamDPOpFlowVar16);
             break;
         case  StreamDPVmInstructions::ditRANDOM32 :
             ua.lpv32 =(StreamDPOpFlowVar32 *)p;
-            ua.lpv32->run_rand(flow_var);
+            ua.lpv32->run_rand(flow_var,per_thread_random);
             p+=sizeof(StreamDPOpFlowVar32);
             break;
         case  StreamDPVmInstructions::ditRANDOM64 :
             ua.lpv64 =(StreamDPOpFlowVar64 *)p;
-            ua.lpv64->run_rand(flow_var);
+            ua.lpv64->run_rand(flow_var,per_thread_random);
             p+=sizeof(StreamDPOpFlowVar64);
             break;
 
@@ -563,18 +613,57 @@ public:
 
     typedef uint8_t instruction_type_t ;
 
-    virtual instruction_type_t get_instruction_type()=0;
+    virtual instruction_type_t get_instruction_type() const = 0;
 
     virtual ~StreamVmInstruction();
 
     virtual void Dump(FILE *fd)=0;
     
+    virtual StreamVmInstruction * clone() = 0;
+
+    /* by default an instruction is not splitable */
+    virtual bool is_splitable() const {
+        return false;
+    }
 
 private:
     static const std::string m_name;
 };
 
 /**
+ * abstract class that defines a flow var
+ * 
+ * @author imarom (23-Dec-15)
+ */
+class StreamVmInstructionVar : public StreamVmInstruction {
+
+public:
+
+    StreamVmInstructionVar(const std::string &var_name) : m_var_name(var_name) {
+
+    }
+
+    const std::string & get_var_name() {
+        return m_var_name;
+    }
+
+    virtual bool is_splitable() const {
+        return true;
+    }
+
+    /**
+     * what is the split range for this var
+     * 
+     */
+    virtual uint64_t get_splitable_range() const = 0;
+
+public:
+    
+    /* flow var name */
+    const std::string m_var_name;
+};
+
+/**
  * fix checksum for ipv4
  * 
  */
@@ -584,12 +673,16 @@ public:
 
     }
 
-    virtual instruction_type_t get_instruction_type(){
+    virtual instruction_type_t get_instruction_type() const {
         return ( StreamVmInstruction::itFIX_IPV4_CS);
     }
 
     virtual void Dump(FILE *fd);
 
+    virtual StreamVmInstruction * clone() {
+        return new StreamVmInstructionFixChecksumIpv4(m_pkt_offset);
+    }
+
 public:
     uint16_t m_pkt_offset; /* the offset of IPv4 header from the start of the packet  */
 };
@@ -599,14 +692,18 @@ public:
  * 
  * @author imarom (07-Sep-15)
  */
-class StreamVmInstructionFlowMan : public StreamVmInstruction {
+class StreamVmInstructionFlowMan : public StreamVmInstructionVar {
 
 public:
 
-    virtual instruction_type_t get_instruction_type(){
+    virtual instruction_type_t get_instruction_type() const {
         return ( StreamVmInstruction::itFLOW_MAN);
     }
 
+    virtual uint64_t get_splitable_range() const {
+        return (m_max_value - m_min_value + 1);
+    }
+
     /**
      * different types of operations on the object
      */
@@ -616,25 +713,56 @@ public:
         FLOW_VAR_OP_RANDOM
     };
 
+
+    /**
+     * for BSS we take one previous value 
+     * because the VM will be executed before writing to pkt 
+     * so the init value is one step's advanced 
+     * 
+     */
+    uint64_t get_bss_init_value() const {
+        uint64_t init = m_init_value;
+
+        switch (m_op) {
+        case FLOW_VAR_OP_INC:
+            return (init == m_min_value ? m_max_value : (init - 1));
+
+        case FLOW_VAR_OP_DEC:
+            return (init == m_max_value ? m_min_value : (init + 1));
+
+        default:
+            return init;
+        }
+
+    }
+
     StreamVmInstructionFlowMan(const std::string &var_name,
                                uint8_t size,
                                flow_var_op_e op,
                                uint64_t init_value,
                                uint64_t min_value,
-                               uint64_t max_value) : 
-                                                     m_var_name(var_name),
-                                                     m_size_bytes(size),
-                                                     m_op(op),
-                                                     m_init_value(init_value),
-                                                     m_min_value(min_value),
-                                                     m_max_value(max_value) {
+                               uint64_t max_value) : StreamVmInstructionVar(var_name) {
 
+        m_op = op;
+        m_size_bytes = size;
+        m_init_value = init_value;
+        m_min_value  = min_value;
+        m_max_value  = max_value;
     }
 
     virtual void Dump(FILE *fd);
 
     void sanity_check(uint32_t ins_id,StreamVm *lp);
 
+    virtual StreamVmInstruction * clone() {
+        return new StreamVmInstructionFlowMan(m_var_name,
+                                              m_size_bytes,
+                                              m_op,
+                                              m_init_value,
+                                              m_min_value,
+                                              m_max_value);
+    }
+
 private:
     void sanity_check_valid_range(uint32_t ins_id,StreamVm *lp);
     void sanity_check_valid_size(uint32_t ins_id,StreamVm *lp);
@@ -642,10 +770,6 @@ private:
 
 public:
 
-
-    /* flow var name */
-    std::string   m_var_name;
-
     /* flow var size */
     uint8_t       m_size_bytes;
 
@@ -657,7 +781,6 @@ public:
     uint64_t m_min_value;
     uint64_t m_max_value;
 
-
 };
 
 
@@ -666,7 +789,7 @@ public:
  * 
  * @author hhaim
  */
-class StreamVmInstructionFlowClient : public StreamVmInstruction {
+class StreamVmInstructionFlowClient : public StreamVmInstructionVar {
 
 public:
     enum client_flags_e {
@@ -674,7 +797,7 @@ public:
     };
 
 
-    virtual instruction_type_t get_instruction_type(){
+    virtual instruction_type_t get_instruction_type() const {
         return ( StreamVmInstruction::itFLOW_CLIENT);
     }
 
@@ -686,8 +809,8 @@ public:
                                uint16_t port_max,
                                uint32_t limit_num_flows, /* zero means don't limit */
                                uint16_t flags
-                               ) { 
-        m_var_name   = var_name;
+                               ) : StreamVmInstructionVar(var_name) { 
+
         m_client_min = client_min_value;
         m_client_max = client_max_value;
 
@@ -704,15 +827,34 @@ public:
         return  (4+2+4);
     }
 
+    uint32_t get_ip_range() const {
+        return (m_client_max - m_client_min + 1);
+    }
+
+    uint16_t get_port_range() const {
+        return (m_port_max - m_port_min + 1);
+    }
+
+    virtual uint64_t get_splitable_range() const {
+        return get_ip_range();
+    }
+
     bool is_unlimited_flows(){
         return ( (m_flags &   StreamVmInstructionFlowClient::CLIENT_F_UNLIMITED_FLOWS ) == 
                   StreamVmInstructionFlowClient::CLIENT_F_UNLIMITED_FLOWS );
     }
-public:
 
+    virtual StreamVmInstruction * clone() {
+        return new StreamVmInstructionFlowClient(m_var_name,
+                                                 m_client_min,
+                                                 m_client_max,
+                                                 m_port_min,
+                                                 m_port_max,
+                                                 m_limit_num_flows,
+                                                 m_flags);
+    }
 
-    /* flow var name */
-    std::string   m_var_name;
+public:
 
     uint32_t m_client_min;  // min ip 
     uint32_t m_client_max;  // max ip 
@@ -754,12 +896,19 @@ public:
                                                         m_add_value(add_value),
                                                         m_is_big_endian(is_big_endian) {}
 
-    virtual instruction_type_t get_instruction_type(){
+    virtual instruction_type_t get_instruction_type() const {
         return ( StreamVmInstruction::itPKT_WR);
     }
 
     virtual void Dump(FILE *fd);
 
+    virtual StreamVmInstruction * clone() {
+        return new StreamVmInstructionWriteToPkt(m_flow_var_name,
+                                                 m_pkt_offset,
+                                                 m_add_value,
+                                                 m_is_big_endian);
+    }
+
 public:
 
     /* flow var name to write */
@@ -789,13 +938,15 @@ public:
         m_bss_size=0;
         m_program_size=0;
         m_max_pkt_offset_change=0;
+        m_prefix_size = 0;
     }
 
     StreamVmDp( uint8_t * bss,
                 uint16_t bss_size,
                 uint8_t * prog,
                 uint16_t prog_size,
-                uint16_t max_pkt_offset
+                uint16_t max_pkt_offset,
+                uint16_t prefix_size
                 ){
 
         if (bss) {
@@ -818,7 +969,9 @@ public:
             m_program_ptr = NULL;
             m_program_size=0;
         }
-        m_max_pkt_offset_change =max_pkt_offset;
+
+        m_max_pkt_offset_change = max_pkt_offset;
+        m_prefix_size = prefix_size;
     }
 
     ~StreamVmDp(){
@@ -835,12 +988,13 @@ public:
     }
 
     StreamVmDp * clone() const {
-        StreamVmDp * lp= new StreamVmDp(m_bss_ptr,
-                                        m_bss_size,
-                                        m_program_ptr,
-                                        m_program_size,
-                                        m_max_pkt_offset_change
-                                        );
+        StreamVmDp * lp = new StreamVmDp(m_bss_ptr,
+                                         m_bss_size,
+                                         m_program_ptr,
+                                         m_program_size,
+                                         m_max_pkt_offset_change,
+                                         m_prefix_size
+                                         );
         assert(lp);
         return (lp);
     }
@@ -873,12 +1027,21 @@ public:
         return (m_max_pkt_offset_change);
     }
 
+    uint16_t get_prefix_size() {
+        return m_prefix_size;
+    }
+
+    void set_prefix_size(uint16_t prefix_size) {
+        m_prefix_size = prefix_size;
+    }
+
 private:
     uint8_t *  m_bss_ptr; /* pointer to the data section */
     uint8_t *  m_program_ptr; /* pointer to the program */
     uint16_t   m_bss_size;
     uint16_t   m_program_size; /* program size*/
     uint16_t   m_max_pkt_offset_change;
+    uint16_t   m_prefix_size;
 
 };
 
@@ -898,35 +1061,52 @@ public:
 
 
     StreamVm(){
+        m_prefix_size=0;
         m_bss=0;
         m_pkt_size=0;
         m_cur_var_offset=0;
+        m_is_random_var=false;
+        m_split_instr=NULL;
+        m_is_compiled = false;
     }
 
 
-    /* set packet size */
-    void set_packet_size(uint16_t pkt_size){
-        m_pkt_size = pkt_size;
+    uint16_t get_packet_size() const {
+        return ( m_pkt_size);
     }
 
-    uint16_t get_packet_size(){
-        return ( m_pkt_size);
+
+    void set_split_instruction(StreamVmInstructionVar *instr);
+
+    StreamVmInstructionVar * get_split_instruction() {
+        return m_split_instr;
     }
 
+    StreamVmDp * generate_dp_object(){
 
-    StreamVmDp * cloneAsVmDp(){
+        if (!m_is_compiled) {
+            return NULL;
+        }
 
         StreamVmDp * lp= new StreamVmDp(get_bss_ptr(),
                                         get_bss_size(),
                                         get_dp_instruction_buffer()->get_program(),
                                         get_dp_instruction_buffer()->get_program_size(),
-                                        get_max_packet_update_offset()
+                                        get_max_packet_update_offset(),
+                                        get_prefix_size()
                                         );
         assert(lp);
         return (lp);
         
     }
 
+    /**
+     * clone VM instructions
+     * 
+     */
+    void copy_instructions(StreamVm &other) const;
+
+    
     bool is_vm_empty() {
         return (m_inst_list.size() == 0);
     }
@@ -958,14 +1138,20 @@ public:
         return ( m_max_field_update );
     }
 
+    uint16_t get_prefix_size() {
+        return m_prefix_size;
+    }
 
+    bool is_compiled() {
+        return m_is_compiled;
+    }
 
     /**
      * compile the VM 
      * return true of success, o.w false 
      * 
      */
-    void compile();
+    void compile(uint16_t pkt_len);
 
     ~StreamVm();
 
@@ -974,6 +1160,14 @@ public:
     /* raise exception */
     void  err(const std::string &err);
 
+
+    /**
+     * return a pointer to a flow var / client var 
+     * by name if exists, otherwise NULL 
+     * 
+     */
+    StreamVmInstructionVar * lookup_var_by_name(const std::string &var_name);
+
 private:
 
     /* lookup for varible offset, */
@@ -1002,6 +1196,9 @@ private:
     void add_field_cnt(uint16_t new_cnt);
 
 private:
+    bool                               m_is_random_var;
+    bool                               m_is_compiled;
+    uint16_t                           m_prefix_size;
     uint16_t                           m_pkt_size;
     uint16_t                           m_cur_var_offset;
     uint16_t                           m_max_field_update; /* the location of the last byte that is going to be changed in the packet */ 
@@ -1012,6 +1209,8 @@ private:
 
     StreamDPVmInstructions             m_instructions;
     
+    StreamVmInstructionVar             *m_split_instr;
+    
 };
 
 #endif /* __TREX_STREAM_VM_API_H__ */
diff --git a/src/stateless/cp/trex_streams_compiler.cpp b/src/stateless/cp/trex_streams_compiler.cpp
index c4900e66..6bcddc1d 100644
--- a/src/stateless/cp/trex_streams_compiler.cpp
+++ b/src/stateless/cp/trex_streams_compiler.cpp
@@ -21,12 +21,16 @@ limitations under the License.
 
 #include <string>
 #include <sstream>
-#include <trex_streams_compiler.h>
-#include <trex_stream.h>
 #include <assert.h>
-#include <trex_stateless.h>
 #include <iostream>
 
+#include <trex_streams_compiler.h>
+#include <trex_stateless.h>
+#include <trex_vm_splitter.h>
+#include <trex_stream.h>
+
+
+
 /**
  * describes a graph node in the pre compile check
  * 
@@ -175,12 +179,11 @@ TrexStreamsCompiledObj::clone() {
      * clone each element
      */
     for (auto obj : m_objs) {
-        TrexStream *new_stream = obj.m_stream->clone_as_dp();
+        TrexStream *new_stream = obj.m_stream->clone();
         new_compiled_obj->add_compiled_stream(new_stream);
     }
 
     return new_compiled_obj;
-
 }
 
 void TrexStreamsCompiledObj::Dump(FILE *fd){
@@ -463,19 +466,16 @@ TrexStreamsCompiler::compile_stream(const TrexStream *stream,
         new_next_id = nodes.get(stream->m_next_stream_id)->m_compressed_stream_id;
     }
 
+    std::vector<TrexStream *> core_streams(dp_core_count);
+
     /* calculate rate */
     double per_core_rate          = (stream->m_pps * (factor / dp_core_count));
     int per_core_burst_total_pkts = (stream->m_burst_total_pkts / dp_core_count);
 
-    /* compile VM */
-    /* fix this const away problem */
-    ((TrexStream *)stream)->compile();
-
-    std::vector<TrexStream *> per_core_streams(dp_core_count);
 
     /* for each core - creates its own version of the stream */
     for (uint8_t i = 0; i < dp_core_count; i++) {
-        TrexStream *dp_stream = stream->clone_as_dp();
+        TrexStream *dp_stream = stream->clone();
 
         /* fix stream ID */
         dp_stream->fix_dp_stream_id(new_id, new_next_id);
@@ -485,16 +485,20 @@ TrexStreamsCompiler::compile_stream(const TrexStream *stream,
         dp_stream->m_pps               = per_core_rate;
         dp_stream->m_burst_total_pkts  = per_core_burst_total_pkts;
 
-        per_core_streams[i] = dp_stream;
+        core_streams[i] = dp_stream;
     }
 
     /* take care of remainder from a burst */
     int burst_remainder = stream->m_burst_total_pkts - (per_core_burst_total_pkts * dp_core_count);
-    per_core_streams[0]->m_burst_total_pkts += burst_remainder;
+    core_streams[0]->m_burst_total_pkts += burst_remainder;
+
+    /* handle VM (split if needed) */
+    TrexVmSplitter vm_splitter;
+    vm_splitter.split( (TrexStream *)stream, core_streams);
 
     /* attach the compiled stream of every core to its object */
     for (uint8_t i = 0; i < dp_core_count; i++) {
-        objs[i]->add_compiled_stream(per_core_streams[i]);
+        objs[i]->add_compiled_stream(core_streams[i]);
     }
 
 
diff --git a/src/stateless/cp/trex_streams_compiler.h b/src/stateless/cp/trex_streams_compiler.h
index d2b0cd1d..4b61dcfa 100644
--- a/src/stateless/cp/trex_streams_compiler.h
+++ b/src/stateless/cp/trex_streams_compiler.h
@@ -125,8 +125,6 @@ private:
                         std::vector<TrexStreamsCompiledObj *> &objs,
                         GraphNodeMap &nodes);
 
-    void compile_stream_vm(TrexStream *stream);
-
     std::vector<std::string> m_warnings;
 };
 
diff --git a/src/stateless/cp/trex_vm_splitter.cpp b/src/stateless/cp/trex_vm_splitter.cpp
new file mode 100644
index 00000000..9465718f
--- /dev/null
+++ b/src/stateless/cp/trex_vm_splitter.cpp
@@ -0,0 +1,191 @@
+/*
+ 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 <trex_vm_splitter.h>
+#include <trex_stateless.h>
+
+/**
+ * split a specific stream's VM to multiple cores
+ * number of cores is implied by the size of the vector
+ * 
+ */
+void
+TrexVmSplitter::split(TrexStream *stream, std::vector<TrexStream *> core_streams) {
+
+    /* nothing to do if no VM */
+    if (stream->m_vm.is_vm_empty()) {
+        return;
+    }
+
+    /* prepare some vars */
+    m_dp_core_count = core_streams.size();
+    m_core_streams  = &core_streams;
+    m_stream        = stream;
+
+    /* if we cannot split - compile the main and duplicate */
+    bool rc = split_internal();
+    if (!rc) {
+
+        /* compile the stream and simply clone it to all streams */
+        m_stream->vm_compile();
+
+        /* for every core - simply clone the DP object */
+        for (TrexStream *core_stream : *m_core_streams) {
+            core_stream->m_vm_dp = m_stream->m_vm_dp->clone();
+        }
+
+        /* no need for the reference stream DP object */
+        delete m_stream->m_vm_dp;
+        m_stream->m_vm_dp = NULL;
+    }
+}
+
+bool
+TrexVmSplitter::split_internal() {
+
+    const StreamVmInstructionVar *split_instr = m_stream->m_vm.get_split_instruction();
+
+    /* if no split instruction was specified - fall back*/
+    if (split_instr == NULL) {
+        return false;
+    }
+
+    if (split_instr->get_instruction_type() == StreamVmInstruction::itFLOW_MAN) {
+        return split_by_flow_var( (const StreamVmInstructionFlowMan *)split_instr );
+
+    } else if (split_instr->get_instruction_type() == StreamVmInstruction::itFLOW_CLIENT) {
+        return split_by_flow_client_var( (const StreamVmInstructionFlowClient *)split_instr );
+
+    } else {
+        throw TrexException("VM splitter : cannot split by instruction which is not flow var or flow client var");
+    }
+
+}
+
+/**
+ * split VM by flow var 
+ * 
+ * @author imarom (20-Dec-15)
+ * 
+ * @param instr 
+ * 
+ * @return bool 
+ */
+bool
+TrexVmSplitter::split_by_flow_var(const StreamVmInstructionFlowMan *instr) {
+    /* no point in splitting random */
+    if (instr->m_op == StreamVmInstructionFlowMan::FLOW_VAR_OP_RANDOM) {
+        return false;
+    }
+
+    /* if the range is too small - it is unsplitable  */
+    if (instr->get_splitable_range() < m_dp_core_count) {
+        return false;
+    }
+
+    /* we need to split - duplicate VM now */
+    duplicate_vm();
+
+    /* calculate range splitting */
+    uint64_t range = instr->get_splitable_range();
+
+    uint64_t range_part = range / m_dp_core_count;
+    uint64_t leftover   = range % m_dp_core_count;
+
+    /* first core handles a bit more */
+    uint64_t start   = instr->m_min_value;
+    uint64_t end     = start + range_part + leftover - 1;
+
+
+    /* do work */
+    for (TrexStream *core_stream : *m_core_streams) {
+
+        /* get the per-core instruction to split */
+        StreamVmInstructionFlowMan *per_core_instr = (StreamVmInstructionFlowMan *)core_stream->m_vm.get_split_instruction();
+
+        per_core_instr->m_min_value  = start;
+        per_core_instr->m_max_value  = end; 
+
+        /* after split this has no meaning - choose it as we see fit */
+        per_core_instr->m_init_value = (per_core_instr->m_op == StreamVmInstructionFlowMan::FLOW_VAR_OP_DEC ? end : start);
+
+        core_stream->vm_compile();
+
+        start = end + 1;
+        end   = start + range_part - 1;
+    }
+
+    return true;
+}
+
+
+bool
+TrexVmSplitter::split_by_flow_client_var(const StreamVmInstructionFlowClient *instr) {
+
+    /* if the range is too small - it is unsplitable  */
+    if (instr->get_ip_range() < m_dp_core_count) {
+        return false;
+    }
+
+    /* we need to split - duplicate VM now */
+    duplicate_vm();
+
+    /* calculate range splitting */
+    uint64_t range = instr->get_ip_range();
+
+    uint64_t range_part = range / m_dp_core_count;
+    uint64_t leftover   = range % m_dp_core_count;
+
+    /* first core handles a bit more */
+    uint64_t start   = instr->m_client_min;
+    uint64_t end     = start + range_part + leftover - 1;
+
+
+    /* do work */
+    for (TrexStream *core_stream : *m_core_streams) {
+
+        /* get the per-core instruction to split */
+        StreamVmInstructionFlowClient *per_core_instr = (StreamVmInstructionFlowClient *)core_stream->m_vm.get_split_instruction();
+
+        per_core_instr->m_client_min  = start;
+        per_core_instr->m_client_max  = end; 
+
+        core_stream->vm_compile();
+
+        start = end + 1;
+        end   = start + range_part - 1;
+    }
+
+    return true;
+}
+
+/**
+ * duplicate the VM instructions
+ * to all the cores
+ */
+void
+TrexVmSplitter::duplicate_vm() {
+    /* for each core - duplicate the instructions */
+    for (TrexStream *core_stream : *m_core_streams) {
+        m_stream->m_vm.copy_instructions(core_stream->m_vm);
+    }
+}
+
diff --git a/src/stateless/cp/trex_vm_splitter.h b/src/stateless/cp/trex_vm_splitter.h
new file mode 100644
index 00000000..dac71c21
--- /dev/null
+++ b/src/stateless/cp/trex_vm_splitter.h
@@ -0,0 +1,60 @@
+/*
+ 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.
+*/
+#ifndef __TREX_VM_SPLITTER_H__
+#define __TREX_VM_SPLITTER_H__
+
+#include <trex_stream.h>
+
+/**
+ * TRex VM splitter is used to split 
+ * VM instructions around cores 
+ *  
+ * 
+ * @author imarom (23-Dec-15)
+ */
+class TrexVmSplitter {
+
+public:
+
+    TrexVmSplitter() {
+        m_dp_core_count = 0;
+    }
+
+    /**
+     * split a stream's VM to per core streams
+     */
+    void split(TrexStream *stream, std::vector<TrexStream *> core_streams);
+
+
+private:
+    bool split_internal();
+    bool split_by_flow_var(const StreamVmInstructionFlowMan *instr);
+    bool split_by_flow_client_var(const StreamVmInstructionFlowClient *instr);
+
+    void duplicate_vm();
+
+    TrexStream                 *m_stream;
+    std::vector<TrexStream *>  *m_core_streams;
+    uint8_t                     m_dp_core_count;
+};
+
+
+#endif /* __TREX_VM_SPLITTER_H__ */
diff --git a/src/stateless/dp/trex_stateless_dp_core.cpp b/src/stateless/dp/trex_stateless_dp_core.cpp
index 585ff2c7..0a9a88ab 100644
--- a/src/stateless/dp/trex_stateless_dp_core.cpp
+++ b/src/stateless/dp/trex_stateless_dp_core.cpp
@@ -69,12 +69,31 @@ void CGenNodeStateless::Dump(FILE *fd){
 }
 
 
+
+void CGenNodeStateless::refresh_vm_bss(){
+    if ( m_vm_flow_var ) {
+        StreamVmDp  * vm_s=m_ref_stream_info->m_vm_dp;
+        assert(vm_s);
+        memcpy(m_vm_flow_var,vm_s->get_bss(),vm_s->get_bss_size());
+    }
+}
+
+
+/**
+ * this function called when stream restart after it was inactive
+ */
 void CGenNodeStateless::refresh(){
 
     /* refill the stream info */
     m_single_burst    = m_single_burst_refill;
     m_multi_bursts    = m_ref_stream_info->m_num_bursts;
     m_state           = CGenNodeStateless::ss_ACTIVE;
+
+    /* refresh init value */
+#if 0 
+    /* TBD should add a JSON varible for that */
+    refresh_vm_bss();
+#endif
 }
 
 
@@ -124,7 +143,8 @@ rte_mbuf_t   * CGenNodeStateless::alloc_node_with_vm(){
     /* run the VM program */
     StreamDPVmInstructionsRunner runner;
 
-    runner.run( m_vm_program_size, 
+    runner.run( (uint32_t*)m_vm_flow_var,
+                m_vm_program_size, 
                 m_vm_program,
                 m_vm_flow_var,
                 (uint8_t*)p);
@@ -302,7 +322,7 @@ bool TrexStatelessDpCore::set_stateless_next_node(CGenNodeStateless * cur_node,
 
         /* can't be FREE_RESUSE */
         assert(state != CGenNodeStateless::ss_FREE_RESUSE);
-        if (next_node->get_state() == CGenNodeStateless::ss_INACTIVE ) {
+        if (state == CGenNodeStateless::ss_INACTIVE ) {
 
             /* refill start info and scedule, no update in active streams  */
             next_node->refresh();
@@ -333,8 +353,11 @@ void
 TrexStatelessDpCore::idle_state_loop() {
 
     while (m_state == STATE_IDLE) {
-        periodic_check_for_cp_messages();
-        delay(200);
+        bool had_msg = periodic_check_for_cp_messages();
+        /* if no message - backoff for some time */
+        if (!had_msg) {
+            delay(200);
+        }
     }
 }
 
@@ -470,7 +493,10 @@ TrexStatelessDpCore::add_stream(TrexStatelessDpPerPort * lp_port,
     node->m_cache_mbuf=0;
     node->m_type = CGenNode::STATELESS_PKT;
 
-    node->m_ref_stream_info  =   stream->clone_as_dp();
+    /* clone the stream from control plane memory to DP memory */
+    node->m_ref_stream_info = stream->clone();
+    /* no need for this memory anymore on the control plane memory */
+    stream->release_dp_object();
 
     node->m_next_stream=0; /* will be fixed later */
 
@@ -539,7 +565,7 @@ TrexStatelessDpCore::add_stream(TrexStatelessDpPerPort * lp_port,
     node->set_mbuf_cache_dir(dir);
 
 
-    if (stream->is_vm()  == false ) {
+    if (node->m_ref_stream_info->getDpVm() == NULL) {
         /* no VM */
 
         node->m_vm_flow_var =  NULL;
@@ -569,15 +595,15 @@ TrexStatelessDpCore::add_stream(TrexStatelessDpPerPort * lp_port,
 
         StreamVmDp  * lpDpVm = local_mem_stream->getDpVm();
 
-        node->m_vm_flow_var =  lpDpVm->clone_bss(); /* clone the flow var */
-        node->m_vm_program  =  lpDpVm->get_program(); /* same ref to the program */
-        node->m_vm_program_size =lpDpVm->get_program_size();
+        node->m_vm_flow_var      = lpDpVm->clone_bss(); /* clone the flow var */
+        node->m_vm_program       = lpDpVm->get_program(); /* same ref to the program */
+        node->m_vm_program_size  = lpDpVm->get_program_size();
 
 
         /* we need to copy the object */
-        if ( pkt_size > stream->m_vm_prefix_size  ) {
+        if ( pkt_size > lpDpVm->get_prefix_size() ) {
             /* we need const packet */
-            uint16_t const_pkt_size  = pkt_size - stream->m_vm_prefix_size ;
+            uint16_t const_pkt_size  = pkt_size - lpDpVm->get_prefix_size() ;
             rte_mbuf_t *m = CGlobalInfo::pktmbuf_alloc(node->get_socket_id(), const_pkt_size );
             assert(m);
 
@@ -585,17 +611,18 @@ TrexStatelessDpCore::add_stream(TrexStatelessDpPerPort * lp_port,
             assert(p);
 
             /* copy packet data */
-            memcpy(p,(stream_pkt+ stream->m_vm_prefix_size),const_pkt_size);
+            memcpy(p,(stream_pkt + lpDpVm->get_prefix_size()),const_pkt_size);
 
             node->set_const_mbuf(m);
         }
 
 
-        if (stream->m_vm_prefix_size > pkt_size ) {
-            stream->m_vm_prefix_size = pkt_size;
+        if (lpDpVm->get_prefix_size() > pkt_size ) {
+            lpDpVm->set_prefix_size(pkt_size);
         }
+
         /* copy the headr */
-        uint16_t header_size = stream->m_vm_prefix_size;
+        uint16_t header_size = lpDpVm->get_prefix_size();
         assert(header_size);
         node->alloc_prefix_header(header_size);
         uint8_t *p=node->m_original_packet_data_prefix;
diff --git a/src/stateless/dp/trex_stateless_dp_core.h b/src/stateless/dp/trex_stateless_dp_core.h
index 7dc4a2b2..efdb364c 100644
--- a/src/stateless/dp/trex_stateless_dp_core.h
+++ b/src/stateless/dp/trex_stateless_dp_core.h
@@ -185,12 +185,12 @@ public:
      * 
      * @author imarom (27-Oct-15)
      */
-    void periodic_check_for_cp_messages() {
+    bool periodic_check_for_cp_messages() {
         // doing this inline for performance reasons
 
         /* fast path */
         if ( likely ( m_ring_from_cp->isEmpty() ) ) {
-            return;
+            return false;
         }
 
         while ( true ) {
@@ -204,6 +204,8 @@ public:
             handle_cp_msg(msg);
         }
 
+        return true;
+
     }
 
     /* quit the main loop, work in both stateless in stateful, don't free memory trigger from  master  */
diff --git a/src/stateless/dp/trex_stream_node.h b/src/stateless/dp/trex_stream_node.h
index d33785fe..70a66e6a 100644
--- a/src/stateless/dp/trex_stream_node.h
+++ b/src/stateless/dp/trex_stream_node.h
@@ -102,11 +102,13 @@ private:
     uint8_t *            m_vm_flow_var; /* pointer to the vm flow var */
     uint8_t *            m_vm_program;  /* pointer to the program */
     uint16_t             m_vm_program_size; /* up to 64K op codes */
+    uint16_t             m_pad2;
+    uint32_t             m_pad3;
 
     /* End Fast Field VM Section */
 
     /* pad to match the size of CGenNode */
-    uint8_t             m_pad_end[30];
+    uint8_t             m_pad_end[20];
 
 
 public:
@@ -329,6 +331,11 @@ public:
 
     void Dump(FILE *fd);
 
+private:
+
+    void refresh_vm_bss();
+
+
 } __rte_cache_aligned;
 
 static_assert(sizeof(CGenNodeStateless) == sizeof(CGenNode), "sizeof(CGenNodeStateless) != sizeof(CGenNode)" );