merge vm random support

12 files changed, 867 insertions, 276 deletions

diff --git a/src/gtest/trex_stateless_gtest.cpp b/src/gtest/trex_stateless_gtest.cpp
index 2718dead..73b7536a 100644
--- a/src/gtest/trex_stateless_gtest.cpp
+++ b/src/gtest/trex_stateless_gtest.cpp
@@ -161,7 +161,7 @@ TEST_F(basic_vm, vm0) {
     StreamVm vm;
 
     vm.add_instruction( new StreamVmInstructionFixChecksumIpv4(20) );
-    vm.add_instruction( new StreamVmInstructionFlowMan( "var1",1,
+    vm.add_instruction( new StreamVmInstructionFlowMan( "var1",8,
                                                         StreamVmInstructionFlowMan::FLOW_VAR_OP_INC,0,1,7 ) 
                         );
     vm.add_instruction( new StreamVmInstructionWriteToPkt( "var1",14, 0,true)
@@ -181,9 +181,7 @@ TEST_F(basic_vm, vm1) {
                         );
     vm.add_instruction( new StreamVmInstructionFixChecksumIpv4(14) );
 
-    vm.set_packet_size(128);
-
-    vm.compile();
+    vm.compile(128);
 
 
     uint32_t program_size=vm.get_dp_instruction_buffer()->get_program_size();
@@ -200,15 +198,13 @@ TEST_F(basic_vm, vm2) {
     StreamVm vm;
 
     vm.add_instruction( new StreamVmInstructionFlowMan( "var1",1,
-                                                        StreamVmInstructionFlowMan::FLOW_VAR_OP_INC,4,1,7 ) 
+                                                        StreamVmInstructionFlowMan::FLOW_VAR_OP_INC,5,1,7 ) 
                         );
     vm.add_instruction( new StreamVmInstructionWriteToPkt( "var1",26, 0,true)
                         );
     //vm.add_instruction( new StreamVmInstructionFixChecksumIpv4(14) );
 
-    vm.set_packet_size(128);
-
-    vm.compile();
+    vm.compile(128);
 
 
     uint32_t program_size=vm.get_dp_instruction_buffer()->get_program_size();
@@ -280,15 +276,13 @@ TEST_F(basic_vm, vm3) {
     StreamVm vm;
 
     vm.add_instruction( new StreamVmInstructionFlowMan( "var1",4 /* size */,
-                                                        StreamVmInstructionFlowMan::FLOW_VAR_OP_INC,4,1,7 ) 
+                                                        StreamVmInstructionFlowMan::FLOW_VAR_OP_INC,5,1,7 ) 
                         );
     vm.add_instruction( new StreamVmInstructionWriteToPkt( "var1",26, 0,true)
                         );
     //vm.add_instruction( new StreamVmInstructionFixChecksumIpv4(14) );
 
-    vm.set_packet_size(128);
-
-    vm.compile();
+    vm.compile(128);
 
 
     uint32_t program_size=vm.get_dp_instruction_buffer()->get_program_size();
@@ -367,16 +361,14 @@ TEST_F(basic_vm, vm4) {
 
     StreamVm vm;
 
-    vm.add_instruction( new StreamVmInstructionFlowMan( "var1",4 /* size */,
-                                                        StreamVmInstructionFlowMan::FLOW_VAR_OP_INC,4,1,7 ) 
+    vm.add_instruction( new StreamVmInstructionFlowMan( "var1", 8 /* size */,
+                                                        StreamVmInstructionFlowMan::FLOW_VAR_OP_INC,5,1,7 ) 
                         );
     vm.add_instruction( new StreamVmInstructionWriteToPkt( "var1",26, 0,false)
                         );
     //vm.add_instruction( new StreamVmInstructionFixChecksumIpv4(14) );
 
-    vm.set_packet_size(128);
-
-    vm.compile();
+    vm.compile(128);
 
 
     uint32_t program_size=vm.get_dp_instruction_buffer()->get_program_size();
@@ -441,8 +433,12 @@ TEST_F(basic_vm, vm4) {
                    test_udp_pkt);
 
         fprintf(stdout," %d \n",i);
-        //utl_DumpBuffer(stdout,test_udp_pkt,PKT_TEST_SIZE,0);
+        utl_DumpBuffer(stdout,test_udp_pkt,PKT_TEST_SIZE,0);
         /* not big */
+        EXPECT_EQ(test_udp_pkt[33],0);
+        EXPECT_EQ(test_udp_pkt[32],0);
+        EXPECT_EQ(test_udp_pkt[31],0);
+        EXPECT_EQ(test_udp_pkt[30],0);
         EXPECT_EQ(test_udp_pkt[29],0);
         EXPECT_EQ(test_udp_pkt[28],0);
         EXPECT_EQ(test_udp_pkt[27],0);
@@ -458,11 +454,11 @@ TEST_F(basic_vm, vm5) {
     StreamVm vm;
 
     vm.add_instruction( new StreamVmInstructionFlowMan( "var1",4 /* size */,
-                                                        StreamVmInstructionFlowMan::FLOW_VAR_OP_INC,4,1,7 ) 
+                                                        StreamVmInstructionFlowMan::FLOW_VAR_OP_INC,5,1,7 ) 
                         );
 
     vm.add_instruction( new StreamVmInstructionFlowMan( "var2",1 /* size */,
-                                                        StreamVmInstructionFlowMan::FLOW_VAR_OP_DEC,25,23,27 ) );
+                                                        StreamVmInstructionFlowMan::FLOW_VAR_OP_DEC,24,23,27 ) );
 
     /* src ip */
     vm.add_instruction( new StreamVmInstructionWriteToPkt( "var1",26, 0,true)
@@ -474,9 +470,7 @@ TEST_F(basic_vm, vm5) {
 
     vm.add_instruction( new StreamVmInstructionFixChecksumIpv4(14) );
 
-    vm.set_packet_size(128);
-
-    vm.compile();
+    vm.compile(128);
 
 
     uint32_t program_size=vm.get_dp_instruction_buffer()->get_program_size();
@@ -607,7 +601,7 @@ TEST_F(basic_vm, vm6) {
                         );
 
     vm.add_instruction( new StreamVmInstructionFlowMan( "var2",1 /* size */,
-                                                        StreamVmInstructionFlowMan::FLOW_VAR_OP_DEC,25,23,27 ) );
+                                                        StreamVmInstructionFlowMan::FLOW_VAR_OP_DEC,24,23,27 ) );
 
     /* src ip */
     vm.add_instruction( new StreamVmInstructionWriteToPkt( "var1",26, 0,true)
@@ -619,9 +613,7 @@ TEST_F(basic_vm, vm6) {
 
     vm.add_instruction( new StreamVmInstructionFixChecksumIpv4(14) );
 
-    vm.set_packet_size(128);
-
-    vm.compile();
+    vm.compile(128);
 
 
     uint32_t program_size=vm.get_dp_instruction_buffer()->get_program_size();
@@ -689,10 +681,7 @@ TEST_F(basic_vm, vm7) {
     vm.add_instruction( new StreamVmInstructionWriteToPkt( "cl1.port",34, 0,true)
                         );
 
-
-    vm.set_packet_size(128);
-
-    vm.compile();
+    vm.compile(128);
 
 
     uint32_t program_size=vm.get_dp_instruction_buffer()->get_program_size();
@@ -759,9 +748,7 @@ TEST_F(basic_vm, vm8) {
                         );
 
 
-    vm.set_packet_size(128);
-
-    vm.compile();
+    vm.compile(128);
 
 
     uint32_t program_size=vm.get_dp_instruction_buffer()->get_program_size();
@@ -825,11 +812,8 @@ static void vm_build_program_seq(StreamVm & vm,
     vm.add_instruction( new StreamVmInstructionWriteToPkt( "tuple_gen.port",34, 0,true)
                         );
 
-
-    vm.set_packet_size(packet_size);
-
     if (should_compile) {
-        vm.compile();
+        vm.compile(packet_size);
     }
 }
 
@@ -896,7 +880,7 @@ TEST_F(basic_vm, vm10) {
 
     EXPECT_EQ(36,vm.get_max_packet_update_offset());
 
-    StreamVmDp * lpDpVm =vm.cloneAsVmDp();
+    StreamVmDp * lpDpVm =vm.generate_dp_object();
 
     EXPECT_EQ(lpDpVm->get_bss_size(),vm.get_bss_size());
 
@@ -991,16 +975,13 @@ TEST_F(basic_vm, vm_syn_attack) {
     vm.add_instruction( new StreamVmInstructionWriteToPkt( "dst_port",34+2, 0,true)
                         );
 
-
-    vm.set_packet_size(128);
-
-    vm.compile();
+    vm.compile(128);
 
     printf(" max packet update %lu \n",(ulong)vm.get_max_packet_update_offset());
 
     EXPECT_EQ(38,vm.get_max_packet_update_offset());
 
-    StreamVmDp * lpDpVm =vm.cloneAsVmDp();
+    StreamVmDp * lpDpVm =vm.generate_dp_object();
 
     EXPECT_EQ(lpDpVm->get_bss_size(),vm.get_bss_size());
 
@@ -2770,76 +2751,193 @@ TEST_F(basic_stl, graph_generator2) {
     delete obj;
 }
 
-/* stress test */
-#if 0 
-TEST_F(basic_stl, graph_generator2) {
-    std::vector<TrexStream *> streams;
-    TrexStreamsGraph graph;
-    TrexStream *stream;
+class VmSplitTest {
 
-     /* add some multi burst streams */
-     stream = new TrexStream(TrexStream::stMULTI_BURST, 0, 1);
-     stream->m_enabled = true;
-     stream->m_self_start = true;
-     stream->m_isg_usec = 100;
+public:
 
-     stream->set_pps(20);
-     stream->set_multi_burst(4918, 321312, 15);
-     stream->m_next_stream_id = -1;
-     stream->m_pkt.len = 64;
+    VmSplitTest(const char *erf_filename) {
+        m_erf_filename = erf_filename;
+        m_stream = NULL;
 
-     streams.push_back(stream);
+        pcap.load_pcap_file("cap2/udp_64B.pcap",0);
+        pcap.update_ip_src(0x10000001);
 
-     stream = new TrexStream(TrexStream::stMULTI_BURST, 0, 2);
-     stream->m_enabled = true;
-     stream->m_self_start = true;
-     stream->m_isg_usec = 59281;
+    }
 
-     stream->set_pps(30);
-     stream->set_multi_burst(4918, 51040, 27);
-     stream->m_next_stream_id = -1;
-     stream->m_pkt.len = 64;
+    ~VmSplitTest() {
+    }
 
-     streams.push_back(stream);
+    void set_stream(TrexStream *stream) {
 
-     stream = new TrexStream(TrexStream::stMULTI_BURST, 0, 3);
-     stream->m_enabled = true;
-     stream->m_self_start = true;
-     stream->m_isg_usec = 59281492;
+        if (m_stream) {
+            delete m_stream;
+            m_stream = NULL;
+        }
 
-     stream->set_pps(40);
-     stream->set_multi_burst(4918, 412312, 2917);
-     stream->m_next_stream_id = -1;
-     stream->m_pkt.len = 64;
+        m_stream = stream;
+        m_stream->m_enabled = true;
+        m_stream->m_self_start = true;
+        
+        pcap.clone_packet_into_stream(stream);
+    }
 
-     streams.push_back(stream);
+    void set_flow_var_as_split(StreamVmInstructionFlowMan::flow_var_op_e op,
+                               uint64_t start,
+                               uint64_t end,
+                               uint64_t init) {
 
+        assert(m_stream);
 
-     /* stream 3 */
-     stream = new TrexStream(TrexStream::stCONTINUOUS, 0, 4);
-     stream->m_enabled = true;
-     stream->m_self_start = true;
+        StreamVmInstructionVar *split_instr = new StreamVmInstructionFlowMan("var1",
+                                                                             8,
+                                                                             op,
+                                                                             init,
+                                                                             start,
+                                                                             end);
 
-     stream->m_isg_usec = 50;
-     stream->set_pps(30);
-     stream->m_next_stream_id = -1;
-     stream->m_pkt.len = 1512;
+        StreamVm &vm = m_stream->m_vm;
 
-     streams.push_back(stream);
+        vm.add_instruction(split_instr);
 
+        vm.add_instruction(new StreamVmInstructionWriteToPkt( "var1", 60 - 8 - 4, 0,true));
 
-     const TrexStreamsGraphObj &obj = graph.generate(streams);
-     printf("event_count is: %lu, max BPS: %f, max PPS: %f\n", obj.get_events().size(), obj.get_max_bps(), obj.get_max_pps());
+        vm.add_instruction(new StreamVmInstructionFixChecksumIpv4(14));
 
-//     for (const TrexStreamsGraphObj::rate_event_st &ev : obj.get_events()) {
-//         printf("time: %f, diff bps: %f, diff pps: %f\n", ev.time, ev.diff_bps, ev.diff_pps);
-//     }
+        vm.set_split_instruction(split_instr);
 
-     for (auto stream : streams) {
-         delete stream;
-     }
-}   
+    }
+
+    void set_client_var_as_split(uint32_t client_min_value,
+                                 uint32_t client_max_value,
+                                 uint16_t port_min,
+                                 uint16_t port_max) {
+
+
+        assert(m_stream);
+
+        StreamVmInstructionVar *split_instr = new StreamVmInstructionFlowClient("var1",
+                                                                                client_min_value,
+                                                                                client_max_value,
+                                                                                port_min,
+                                                                                port_max,
+                                                                                0,
+                                                                                0);
+
+
+        StreamVm &vm = m_stream->m_vm;
+
+        vm.add_instruction(split_instr);
+
+        /* src ip */
+        vm.add_instruction(new StreamVmInstructionWriteToPkt( "var1.ip",26, 0,true));
+        vm.add_instruction(new StreamVmInstructionFixChecksumIpv4(14));
+
+        /* src port */
+        vm.add_instruction(new StreamVmInstructionWriteToPkt("var1.port",34, 0,true));
+
+        vm.set_split_instruction(split_instr);
+    }
+
+    void run(uint8_t dp_core_count, uint8_t dp_core_to_check) {
+        TrexStreamsCompiler compile;
+        std::vector<TrexStreamsCompiledObj *> objs;
+        std::vector<TrexStream *> streams;
+
+        if (m_stream->m_vm.is_vm_empty()) {
+            set_flow_var_as_split(StreamVmInstructionFlowMan::FLOW_VAR_OP_INC,
+                                  0,
+                                  1000,
+                                  0);
+        }
+
+        streams.push_back(m_stream);
+
+        /* compiling for 8 cores */
+        assert(compile.compile(0, streams, objs, dp_core_count));
+
+        /* choose one DP object */
+        TrexStatelessDpStart *lpStartCmd = new TrexStatelessDpStart(0, 0, objs[dp_core_to_check], 1 /*sec */ );
+        objs[dp_core_to_check] = NULL;
+        /* free all the non used DP objects */
+        for (auto obj : objs) {
+            if (obj) {
+                delete obj;
+            }
+        }
+
+
+        CParserOption * po =&CGlobalInfo::m_options;
+        po->preview.setVMode(7);
+        po->preview.setFileWrite(true);
+        po->out_file = m_erf_filename;
+
+        CBasicStl t1;
+        t1.m_msg = lpStartCmd;
+        bool res=t1.init();
+        EXPECT_EQ_UINT32(1, res?1:0);
+
+    }
+
+private:
+    const char *m_erf_filename;
+    TrexStream *m_stream;
+    CPcapLoader pcap;
+};
+
+
+
+TEST_F(basic_stl, vm_split_flow_var_inc) {
+
+    VmSplitTest split("exp/stl_vm_split_flow_var_inc.erf");
+
+    TrexStream stream(TrexStream::stSINGLE_BURST, 0, 0);
+    stream.set_pps(1000);
+
+    split.set_stream(&stream);
+    split.run(8, 4);
+
+}
+
+TEST_F(basic_stl, vm_split_flow_var_small_range) {
+    /* small range */
+    VmSplitTest split("exp/stl_vm_split_flow_var_small_range.erf");
+
+    TrexStream stream(TrexStream::stSINGLE_BURST, 0, 0);
+    stream.set_pps(1000);
+
+    split.set_stream(&stream);
+    split.set_flow_var_as_split(StreamVmInstructionFlowMan::FLOW_VAR_OP_INC, 0, 1, 0);
+    
+    split.run(8, 4);
+
+}
+
+TEST_F(basic_stl, vm_split_flow_var_big_range) {
+    VmSplitTest split("exp/stl_vm_split_flow_var_big_range.erf");
+
+    TrexStream stream(TrexStream::stSINGLE_BURST, 0, 0);
+    stream.set_pps(1000);
+
+    split.set_stream(&stream);
+    split.set_flow_var_as_split(StreamVmInstructionFlowMan::FLOW_VAR_OP_DEC, 1, 1000, 1000);
+
+    split.run(8, 7);
+
+
+}
+
+TEST_F(basic_stl, vm_split_client_var) {
+     VmSplitTest split("exp/stl_vm_split_client_var.erf");
 
-#endif
+    TrexStream stream(TrexStream::stSINGLE_BURST, 0, 0);
+    stream.set_pps(1000);
+
+    split.set_stream(&stream);
+    split.set_client_var_as_split(0x10000001, 0x100000fe, 5000, 5050);
+    
+    split.run(8, 7);
+
+
+}
 
 /********************************************* Itay Tests End *************************************/
diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp
index 8c9eb914..198f8d72 100755
--- a/src/main_dpdk.cpp
+++ b/src/main_dpdk.cpp
@@ -3692,8 +3692,15 @@ void CGlobalTRex::dump_post_test_stats(FILE *fd){
 
     fprintf (fd," summary stats \n");
     fprintf (fd," -------------- \n");
-
-    fprintf (fd," Total-pkt-drop       : %llu pkts \n",(unsigned long long)(pkt_out-pkt_in));
+    
+    if (pkt_in > pkt_out)
+    {
+        fprintf (fd, " Total-pkt-drop       : 0 pkts \n");
+        if (pkt_in > pkt_out * 1.01)
+            fprintf (fd, " Warning : number of rx packets exceeds 101%% of tx packets!\n");
+    }
+    else
+        fprintf (fd, " Total-pkt-drop       : %llu pkts \n", (unsigned long long) (pkt_out - pkt_in));
     fprintf (fd," Total-tx-bytes       : %llu bytes \n", (unsigned long long)pkt_out_bytes);
     fprintf (fd," Total-tx-sw-bytes    : %llu bytes \n", (unsigned long long)sw_pkt_out_bytes);
     fprintf (fd," Total-rx-bytes       : %llu byte \n", (unsigned long long)pkt_in_bytes);
diff --git a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
index d8f7e772..51db0b20 100644
--- a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
+++ b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
@@ -82,7 +82,7 @@ TrexRpcCmdAddStream::_run(const Json::Value &params, Json::Value &result) {
     stream->m_pkt.meta = parse_string(pkt, "meta", result);
 
     /* parse VM */
-    const Json::Value &vm =  parse_array(section ,"vm", result);
+    const Json::Value &vm =  parse_object(section ,"vm", result);
     parse_vm(vm, stream, result);
 
     /* parse RX info */
@@ -230,8 +230,7 @@ TrexRpcCmdAddStream::parse_vm_instr_flow_var(const Json::Value &inst, TrexStream
                                                                 op_type,
                                                                 init_value,
                                                                 min_value,
-                                                                max_value
-                                                                ));
+                                                                max_value));
 }
 
 void 
@@ -249,9 +248,12 @@ TrexRpcCmdAddStream::parse_vm_instr_write_flow_var(const Json::Value &inst, Trex
 
 void 
 TrexRpcCmdAddStream::parse_vm(const Json::Value &vm, TrexStream *stream, Json::Value &result) {
+
+    const Json::Value &instructions =  parse_array(vm ,"instructions", result);
+
     /* array of VM instructions on vm */
-    for (int i = 0; i < vm.size(); i++) {
-        const Json::Value & inst = parse_object(vm, i, result);
+    for (int i = 0; i < instructions.size(); i++) {
+        const Json::Value & inst = parse_object(instructions, i, result);
 
         auto vm_types = {"fix_checksum_ipv4", "flow_var", "write_flow_var","tuple_flow_var"};
         std::string vm_type = parse_choice(inst, "type", vm_types, result);
@@ -273,6 +275,17 @@ TrexRpcCmdAddStream::parse_vm(const Json::Value &vm, TrexStream *stream, Json::V
             throw TrexRpcException("internal error");
         }
     }
+
+    const std::string &var_name = parse_string(vm, "split_by_var", result);
+    if (var_name != "") {
+        StreamVmInstructionVar *instr = stream->m_vm.lookup_var_by_name(var_name);
+        if (!instr) {
+            std::stringstream ss;
+            ss << "VM: request to split by variable '" << var_name << "' but does not exists";
+            generate_parse_err(result, ss.str());
+        }
+        stream->m_vm.set_split_instruction(instr);
+    }
 }
 
 void
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 2984b896..a3f585ad 100644
--- a/src/stateless/cp/trex_stream_vm.cpp
+++ b/src/stateless/cp/trex_stream_vm.cpp
@@ -449,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;
@@ -577,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:
@@ -621,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();
 
+    for (auto instr : m_inst_list) {
+        StreamVmInstruction *new_instr = instr->clone();
+        other.m_inst_list.push_back(new_instr);
 
-void StreamVm::compile() {
+        /* 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() ;
@@ -638,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;
 }
 
 
@@ -648,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 9023c6b7..891e5b51 100644
--- a/src/stateless/cp/trex_stream_vm.h
+++ b/src/stateless/cp/trex_stream_vm.h
@@ -87,18 +87,20 @@ 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;
         }
     }
 
@@ -119,18 +121,20 @@ 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;
         }
     }
 
@@ -152,18 +156,20 @@ 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;
         }
     }
 
@@ -183,18 +189,20 @@ 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;
         }
     }
 
@@ -392,7 +400,7 @@ public:
 class StreamDPVmInstructions {
 public:
     enum INS_TYPE {
-        ditINC8         ,
+        ditINC8     =7 ,
         ditINC16        ,
         ditINC32        ,
         ditINC64        ,
@@ -605,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
  * 
  */
@@ -626,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  */
 };
@@ -641,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
      */
@@ -658,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);
@@ -684,10 +770,6 @@ private:
 
 public:
 
-
-    /* flow var name */
-    std::string   m_var_name;
-
     /* flow var size */
     uint8_t       m_size_bytes;
 
@@ -699,7 +781,6 @@ public:
     uint64_t m_min_value;
     uint64_t m_max_value;
 
-
 };
 
 
@@ -708,7 +789,7 @@ public:
  * 
  * @author hhaim
  */
-class StreamVmInstructionFlowClient : public StreamVmInstruction {
+class StreamVmInstructionFlowClient : public StreamVmInstructionVar {
 
 public:
     enum client_flags_e {
@@ -716,7 +797,7 @@ public:
     };
 
 
-    virtual instruction_type_t get_instruction_type(){
+    virtual instruction_type_t get_instruction_type() const {
         return ( StreamVmInstruction::itFLOW_CLIENT);
     }
 
@@ -728,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;
 
@@ -746,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 
@@ -796,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 */
@@ -831,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) {
@@ -860,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(){
@@ -877,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);
     }
@@ -915,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;
 
 };
 
@@ -940,36 +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);
     }
@@ -1001,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();
 
@@ -1017,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, */
@@ -1046,6 +1197,8 @@ private:
 
 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 */ 
@@ -1056,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 f73824bc..c211b9f5 100644
--- a/src/stateless/dp/trex_stateless_dp_core.cpp
+++ b/src/stateless/dp/trex_stateless_dp_core.cpp
@@ -490,7 +490,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 */
 
@@ -559,7 +562,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;
@@ -589,15 +592,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);
 
@@ -605,17 +608,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;