add client command

7 files changed, 514 insertions, 15 deletions

diff --git a/.gitignore b/.gitignore
index 0e580c8e..de45a1ef 100644
--- a/.gitignore
+++ b/.gitignore
@@ -35,6 +35,11 @@ scripts/exp/stl_vm_enable0-0.erf
 scripts/exp/stl_vm_enable1-0.erf
 scripts/exp/stl_vm_enable2-0.erf
 scripts/exp/udp_64B_vm6.pcap
+scripts/exp/udp_64B_vm7.pcap
+scripts/exp/udp_64B_vm8.pcap
+scripts/exp/udp_64B_vm9.pcap
+
+
 
 
 
diff --git a/scripts/exp/udp_64B_vm7-ex.pcap b/scripts/exp/udp_64B_vm7-ex.pcap
new file mode 100644
index 00000000..e45a3459
--- /dev/null
+++ b/scripts/exp/udp_64B_vm7-ex.pcap
diff --git a/scripts/exp/udp_64B_vm8-ex.pcap b/scripts/exp/udp_64B_vm8-ex.pcap
new file mode 100644
index 00000000..b19bd0d0
--- /dev/null
+++ b/scripts/exp/udp_64B_vm8-ex.pcap
diff --git a/scripts/exp/udp_64B_vm9-ex.pcap b/scripts/exp/udp_64B_vm9-ex.pcap
new file mode 100644
index 00000000..7428e5d1
--- /dev/null
+++ b/scripts/exp/udp_64B_vm9-ex.pcap
diff --git a/src/gtest/trex_stateless_gtest.cpp b/src/gtest/trex_stateless_gtest.cpp
index 38fb1633..bc89ec54 100644
--- a/src/gtest/trex_stateless_gtest.cpp
+++ b/src/gtest/trex_stateless_gtest.cpp
@@ -470,7 +470,7 @@ TEST_F(basic_vm, vm5) {
         0x01,0x01,0x01,0x01, /*26 */
         0x3d,0xad,0x72,0x1b, /*30 */
 
-        0x11,0x11,
+        0x11,0x11,           /*34 */
         0x11,0x11,
 
         0x00,0x6d,
@@ -630,6 +630,205 @@ TEST_F(basic_vm, vm6) {
     EXPECT_EQ(1, res1?1:0);
 }
 
+/* test client command */
+TEST_F(basic_vm, vm7) {
+
+
+
+    StreamVm vm;
+
+    vm.add_instruction( new StreamVmInstructionFlowClient( "cl1",
+                                                           0x10000001,
+                                                           0x10000004,
+                                                           1025,
+                                                           1027,
+                                                           100,
+                                                           0) );
+
+    /* src ip */
+    vm.add_instruction( new StreamVmInstructionWriteToPkt( "cl1.ip",26, 0,true)
+                        );
+
+    vm.add_instruction( new StreamVmInstructionFixChecksumIpv4(14) );
+
+    /* src port */
+    vm.add_instruction( new StreamVmInstructionWriteToPkt( "cl1.port",34, 0,true)
+                        );
+
+
+    vm.set_packet_size(128);
+
+    vm.compile_next();
+
+
+    uint32_t program_size=vm.get_dp_instruction_buffer()->get_program_size();
+
+    printf (" program size : %lu \n",(ulong)program_size);
+
+
+    vm.Dump(stdout);
+
+    CPcapLoader pcap;
+    pcap.load_pcap_file("cap2/udp_64B.pcap",0);
+
+    
+
+    CFileWriterBase * lpWriter=CCapWriterFactory::CreateWriter(LIBPCAP,(char *)"exp/udp_64B_vm7.pcap");
+    assert(lpWriter);
+
+
+    StreamDPVmInstructionsRunner runner;
+
+    int i;
+    for (i=0; i<20; i++) {
+        runner.run(program_size, 
+                   vm.get_dp_instruction_buffer()->get_program(),
+                   vm.get_bss_ptr(),
+                   (uint8_t*)pcap.m_raw.raw);
+
+        assert(lpWriter->write_packet(&pcap.m_raw));
+    }
+
+    delete lpWriter;
+
+    CErfCmp cmp;
+
+    bool res1=cmp.compare("exp/udp_64B_vm7.pcap","exp/udp_64B_vm7-ex.pcap");
+    EXPECT_EQ(1, res1?1:0);
+}
+
+TEST_F(basic_vm, vm8) {
+
+
+
+    StreamVm vm;
+
+    vm.add_instruction( new StreamVmInstructionFlowClient( "cl1",
+                                                           0x10000001,
+                                                           0x10000006,
+                                                           1025,
+                                                           1027,
+                                                           4,
+                                                           0) );
+
+    /* src ip */
+    vm.add_instruction( new StreamVmInstructionWriteToPkt( "cl1.ip",26, 0,true)
+                        );
+
+    vm.add_instruction( new StreamVmInstructionFixChecksumIpv4(14) );
+
+    /* src port */
+    vm.add_instruction( new StreamVmInstructionWriteToPkt( "cl1.port",34, 0,true)
+                        );
+
+
+    vm.set_packet_size(128);
+
+    vm.compile_next();
+
+
+    uint32_t program_size=vm.get_dp_instruction_buffer()->get_program_size();
+
+    printf (" program size : %lu \n",(ulong)program_size);
+
+
+    vm.Dump(stdout);
+
+    CPcapLoader pcap;
+    pcap.load_pcap_file("cap2/udp_64B.pcap",0);
+
+    
+
+    CFileWriterBase * lpWriter=CCapWriterFactory::CreateWriter(LIBPCAP,(char *)"exp/udp_64B_vm8.pcap");
+    assert(lpWriter);
+
+
+    StreamDPVmInstructionsRunner runner;
+
+    int i;
+    for (i=0; i<20; i++) {
+        runner.run(program_size, 
+                   vm.get_dp_instruction_buffer()->get_program(),
+                   vm.get_bss_ptr(),
+                   (uint8_t*)pcap.m_raw.raw);
+
+        assert(lpWriter->write_packet(&pcap.m_raw));
+    }
+
+    delete lpWriter;
+
+    CErfCmp cmp;
+
+    bool res1=cmp.compare("exp/udp_64B_vm8.pcap","exp/udp_64B_vm8-ex.pcap");
+    EXPECT_EQ(1, res1?1:0);
+}
+
+TEST_F(basic_vm, vm9) {
+
+
+
+    StreamVm vm;
+
+    vm.add_instruction( new StreamVmInstructionFlowClient( "cl1",
+                                                           0x10000001,
+                                                           0x10000006,
+                                                           1025,
+                                                           1027,
+                                                           20,
+                                                           0) );
+
+    /* src ip */
+    vm.add_instruction( new StreamVmInstructionWriteToPkt( "cl1.ip",26, 0,true)
+                        );
+
+    vm.add_instruction( new StreamVmInstructionFixChecksumIpv4(14) );
+
+    /* src port */
+    vm.add_instruction( new StreamVmInstructionWriteToPkt( "cl1.port",34, 0,true)
+                        );
+
+
+    vm.set_packet_size(128);
+
+    vm.compile_next();
+
+
+    uint32_t program_size=vm.get_dp_instruction_buffer()->get_program_size();
+
+    printf (" program size : %lu \n",(ulong)program_size);
+
+
+    vm.Dump(stdout);
+
+    CPcapLoader pcap;
+    pcap.load_pcap_file("cap2/udp_64B.pcap",0);
+
+    
+
+    CFileWriterBase * lpWriter=CCapWriterFactory::CreateWriter(LIBPCAP,(char *)"exp/udp_64B_vm9.pcap");
+    assert(lpWriter);
+
+
+    StreamDPVmInstructionsRunner runner;
+
+    int i;
+    for (i=0; i<30; i++) {
+        runner.run(program_size, 
+                   vm.get_dp_instruction_buffer()->get_program(),
+                   vm.get_bss_ptr(),
+                   (uint8_t*)pcap.m_raw.raw);
+
+        assert(lpWriter->write_packet(&pcap.m_raw));
+    }
+
+    delete lpWriter;
+
+    CErfCmp cmp;
+
+    bool res1=cmp.compare("exp/udp_64B_vm9.pcap","exp/udp_64B_vm9-ex.pcap");
+    EXPECT_EQ(1, res1?1:0);
+}
+
 
 
 //////////////////////////////////////////////////////
diff --git a/src/stateless/cp/trex_stream_vm.cpp b/src/stateless/cp/trex_stream_vm.cpp
index 9fc2f049..188be2d8 100644
--- a/src/stateless/cp/trex_stream_vm.cpp
+++ b/src/stateless/cp/trex_stream_vm.cpp
@@ -113,6 +113,14 @@ void StreamVmInstructionWriteToPkt::Dump(FILE *fd){
 
 
 
+void StreamVmInstructionFlowClient::Dump(FILE *fd){
+
+    fprintf(fd," client_var ,%s , ",m_var_name.c_str());
+
+    fprintf(fd," ip:(%x-%x) port:(%x-%x)  flow_limit:%lu  flags: %x\n",m_client_min,m_client_max, m_port_min,m_port_max,(ulong)m_limit_num_flows,m_flags);
+}
+
+
 
 
 /***************************
@@ -200,18 +208,68 @@ void StreamVm::build_flow_var_table() {
                 err(ss.str());
             }else{
                 var.m_offset=m_cur_var_offset;
-                var.m_instruction = ins_man;
+                var.m_ins.m_ins_flowv = ins_man;
+                var.m_size_bytes = ins_man->m_size_bytes;
                 var_add(ins_man->m_var_name,var);
                 m_cur_var_offset += ins_man->m_size_bytes;
 
-                /* limit the flow var size */
-                if (m_cur_var_offset > StreamVm::svMAX_FLOW_VAR ) {
-                    std::stringstream ss;
-                    ss << "too many flow varibles current size is :" << m_cur_var_offset << " maximum support is " << StreamVm::svMAX_FLOW_VAR;
-                    err(ss.str());
-                }
             }
         }
+
+        if ( inst->get_instruction_type() == StreamVmInstruction::itFLOW_CLIENT ){
+            StreamVmInstructionFlowClient * ins_man=(StreamVmInstructionFlowClient *)inst;
+
+            VmFlowVarRec var;
+            /* if this is the first time */ 
+            if ( var_lookup( ins_man->m_var_name+".ip",var) == true){
+                std::stringstream ss;
+                ss << "instruction id '" << ins_id << "' client variable name " << ins_man->m_var_name << " already exists";
+                err(ss.str());
+            }
+            if ( var_lookup( ins_man->m_var_name+".port",var) == true){
+                std::stringstream ss;
+                ss << "instruction id '" << ins_id << "' client variable name " << ins_man->m_var_name << " already exists";
+                err(ss.str());
+            }
+
+            if ( var_lookup( ins_man->m_var_name+".flow_limit",var) == true){
+                std::stringstream ss;
+                ss << "instruction id '" << ins_id << "' client variable name " << ins_man->m_var_name << " already exists";
+                err(ss.str());
+            }
+
+            var.m_offset = m_cur_var_offset;
+            var.m_ins.m_ins_flow_client = ins_man;
+            var.m_size_bytes =4;
+
+            VmFlowVarRec var_port;
+
+            var_port.m_offset = m_cur_var_offset+4;
+            var_port.m_ins.m_ins_flow_client = ins_man;
+            var_port.m_size_bytes =2;
+
+            VmFlowVarRec var_flow_limit;
+
+            var_flow_limit.m_offset = m_cur_var_offset+6;
+            var_flow_limit.m_ins.m_ins_flow_client = ins_man;
+            var_flow_limit.m_size_bytes =4;
+
+
+            var_add(ins_man->m_var_name+".ip",var);
+            var_add(ins_man->m_var_name+".port",var_port);
+            var_add(ins_man->m_var_name+".flow_limit",var_flow_limit);
+
+            m_cur_var_offset += StreamVmInstructionFlowClient::get_flow_var_size(); 
+
+            assert(sizeof(StreamDPFlowClient)==StreamVmInstructionFlowClient::get_flow_var_size());
+        }
+
+        /* limit the flow var size */
+        if (m_cur_var_offset > StreamVm::svMAX_FLOW_VAR ) {
+            std::stringstream ss;
+            ss << "too many flow varibles current size is :" << m_cur_var_offset << " maximum support is " << StreamVm::svMAX_FLOW_VAR;
+            err(ss.str());
+        }
         ins_id++;
     }
 
@@ -382,15 +440,15 @@ void StreamVm::build_program(){
                 err(ss.str());
             }
 
-            if (lpPkt->m_pkt_offset + var.m_instruction->m_size_bytes > m_pkt_size ) {
+            if (lpPkt->m_pkt_offset + var.m_size_bytes > m_pkt_size ) {
                 std::stringstream ss;
-                ss << "instruction id '" << ins_id << "' packet write with packet_offset   " << lpPkt->m_pkt_offset + var.m_instruction->m_size_bytes  << "   bigger than packet size   "<< m_pkt_size;
+                ss << "instruction id '" << ins_id << "' packet write with packet_offset   " << lpPkt->m_pkt_offset + var.m_size_bytes  << "   bigger than packet size   "<< m_pkt_size;
                 err(ss.str());
             }
-            add_field_cnt(lpPkt->m_pkt_offset + var.m_instruction->m_size_bytes);
+            add_field_cnt(lpPkt->m_pkt_offset + var.m_size_bytes);
 
 
-            uint8_t       op_size=var.m_instruction->m_size_bytes;
+            uint8_t       op_size=var.m_size_bytes;
             bool is_big    = lpPkt->m_is_big_endian;
             uint8_t       flags = (is_big?StreamDPOpPktWrBase::PKT_WR_IS_BIG:0);
             uint8_t       flow_offset = get_var_offset(lpPkt->m_flow_var_name);
@@ -437,6 +495,37 @@ void StreamVm::build_program(){
 
         }
 
+
+        if (ins_type == StreamVmInstruction::itFLOW_CLIENT) {
+            StreamVmInstructionFlowClient *lpMan =(StreamVmInstructionFlowClient *)inst;
+
+            if ( lpMan->is_unlimited_flows() ){
+                StreamDPOpClientsUnLimit  client_cmd;
+                client_cmd.m_op =  StreamDPVmInstructions::itCLIENT_VAR_UNLIMIT;
+
+                client_cmd.m_flow_offset = get_var_offset(lpMan->m_var_name+".ip"); /* start offset */
+                client_cmd.m_flags       = 0; /* not used */
+                client_cmd.m_pad         = 0;
+                client_cmd.m_min_ip      = lpMan->m_client_min;
+                client_cmd.m_max_ip      = lpMan->m_client_max;
+                m_instructions.add_command(&client_cmd,sizeof(client_cmd));
+
+            }else{
+                StreamDPOpClientsLimit  client_cmd;
+                client_cmd.m_op =  StreamDPVmInstructions::itCLIENT_VAR;
+
+                client_cmd.m_flow_offset = get_var_offset(lpMan->m_var_name+".ip"); /* start offset */
+                client_cmd.m_flags       = 0; /* not used */
+                client_cmd.m_pad         = 0;
+                client_cmd.m_min_port    = lpMan->m_port_min;
+                client_cmd.m_max_port    = lpMan->m_port_max;
+                client_cmd.m_min_ip      = lpMan->m_client_min;
+                client_cmd.m_max_ip      = lpMan->m_client_max;
+                client_cmd.m_limit_flows = lpMan->m_limit_num_flows;
+                m_instructions.add_command(&client_cmd,sizeof(client_cmd));
+            }
+        }
+
         ins_id++;
     }
 }
@@ -473,6 +562,28 @@ void StreamVm::build_bss() {
                 assert(0);
             }
         }
+
+        if ( inst->get_instruction_type() == StreamVmInstruction::itFLOW_CLIENT ){
+
+            StreamVmInstructionFlowClient * ins_man=(StreamVmInstructionFlowClient *)inst;
+            if (ins_man->m_client_min>0) {
+                *((uint32_t*)p)=(uint32_t)(ins_man->m_client_min-1);
+            }else{
+                *((uint32_t*)p)=(uint32_t)ins_man->m_client_min;
+            }
+            p+=4;
+
+            if (ins_man->is_unlimited_flows() ) {
+                *((uint16_t*)p)=StreamDPOpClientsUnLimit::CLIENT_UNLIMITED_MIN_PORT;
+            }else{
+                *((uint16_t*)p)=(uint16_t)ins_man->m_port_min;
+            }
+            p+=2;
+
+            *((uint32_t*)p)=0;
+            p+=4;
+        }
+
     }
 }
 
@@ -570,6 +681,9 @@ void StreamDPVmInstructions::Dump(FILE *fd){
     StreamDPOpPktWr16    *lpw16;
     StreamDPOpPktWr32    *lpw32;
     StreamDPOpPktWr64    *lpw64;
+    StreamDPOpClientsLimit   *lp_client;
+    StreamDPOpClientsUnLimit  *lp_client_unlimited;
+
 
     while ( p < p_end) {
         uint8_t op_code=*p;
@@ -667,6 +781,20 @@ void StreamDPVmInstructions::Dump(FILE *fd){
             lpw64->dump(fd,"Wr64");
             p+=sizeof(StreamDPOpPktWr64);
             break;
+
+        case  itCLIENT_VAR :      
+            lp_client =(StreamDPOpClientsLimit *)p;
+            lp_client->dump(fd,"Client");
+            p+=sizeof(StreamDPOpClientsLimit);
+            break;
+
+        case  itCLIENT_VAR_UNLIMIT :      
+            lp_client_unlimited =(StreamDPOpClientsUnLimit *)p;
+            lp_client_unlimited->dump(fd,"ClientUnlimted");
+            p+=sizeof(StreamDPOpClientsUnLimit);
+            break;
+
+
         default:
             assert(0);
         }
@@ -712,5 +840,12 @@ void StreamDPOpIpv4Fix::dump(FILE *fd,std::string opt){
 }
 
 
+void StreamDPOpClientsLimit::dump(FILE *fd,std::string opt){
+    fprintf(fd," %10s  op:%lu, flow_offset: %lu (%x-%x) (%x-%x) flow_limit :%lu flags:%x \n",  opt.c_str(),(ulong)m_op,(ulong)m_flow_offset,m_min_ip,m_max_ip,m_min_port,m_max_port,(ulong)m_limit_flows,m_flags);
+}
+
+void StreamDPOpClientsUnLimit::dump(FILE *fd,std::string opt){
+    fprintf(fd," %10s  op:%lu, flow_offset: %lu (%x-%x) flags:%x \n",  opt.c_str(),(ulong)m_op,(ulong)m_flow_offset,m_min_ip,m_max_ip,m_flags);
+}
 
 
diff --git a/src/stateless/cp/trex_stream_vm.h b/src/stateless/cp/trex_stream_vm.h
index 843ec726..08a8621a 100644
--- a/src/stateless/cp/trex_stream_vm.h
+++ b/src/stateless/cp/trex_stream_vm.h
@@ -269,6 +269,82 @@ public:
 } __attribute__((packed));
 
 
+/* flow varible of Client command */
+struct StreamDPFlowClient {
+    uint32_t cur_ip;
+    uint16_t cur_port;
+    uint32_t cur_flow_id;
+} __attribute__((packed));
+
+
+struct StreamDPOpClientsLimit {
+    uint8_t m_op;
+    uint8_t m_flow_offset; /* offset into the flow var  bytes */
+    uint8_t m_flags;
+    uint8_t m_pad;
+    uint16_t    m_min_port;
+    uint16_t    m_max_port;
+
+    uint32_t    m_min_ip;
+    uint32_t    m_max_ip;
+    uint32_t    m_limit_flows; /* limit the number of flows */
+
+public:
+    void dump(FILE *fd,std::string opt);
+    inline void run(uint8_t * flow_var_base) {
+        StreamDPFlowClient * lp= (StreamDPFlowClient *)(flow_var_base+m_flow_offset);
+        lp->cur_ip++;
+        if (lp->cur_ip > m_max_ip ) {
+            lp->cur_ip= m_min_ip;
+            lp->cur_port++;
+            if (lp->cur_port > m_max_port) {
+                lp->cur_port = m_min_port;
+            }
+        }
+
+        if (m_limit_flows) {
+            lp->cur_flow_id++;
+            if ( lp->cur_flow_id > m_limit_flows ){
+                /* reset to the first flow */
+                lp->cur_flow_id = 1;
+                lp->cur_ip      =  m_min_ip;
+                lp->cur_port    =  m_min_port;
+            }
+        }
+    }
+
+
+} __attribute__((packed));
+
+struct StreamDPOpClientsUnLimit {
+    enum __MIN_PORT {
+        CLIENT_UNLIMITED_MIN_PORT = 1025
+    };
+
+    uint8_t m_op;
+    uint8_t m_flow_offset; /* offset into the flow var  bytes */
+    uint8_t m_flags;
+    uint8_t m_pad;
+    uint32_t    m_min_ip;
+    uint32_t    m_max_ip;
+
+public:
+    void dump(FILE *fd,std::string opt);
+    inline void run(uint8_t * flow_var_base) {
+        StreamDPFlowClient * lp= (StreamDPFlowClient *)(flow_var_base+m_flow_offset);
+        lp->cur_ip++;
+        if (lp->cur_ip > m_max_ip ) {
+            lp->cur_ip= m_min_ip;
+            lp->cur_port++;
+            if (lp->cur_port == 0) {
+                lp->cur_port = StreamDPOpClientsUnLimit::CLIENT_UNLIMITED_MIN_PORT;
+            }
+        }
+    }
+
+} __attribute__((packed));
+
+
 /* datapath instructions */
 class StreamDPVmInstructions {
 public:
@@ -293,7 +369,9 @@ public:
         itPKT_WR8       ,
         itPKT_WR16       ,
         itPKT_WR32       ,
-        itPKT_WR64       
+        itPKT_WR64       ,
+        itCLIENT_VAR       ,
+        itCLIENT_VAR_UNLIMIT      
     };
 
 
@@ -342,12 +420,26 @@ inline void StreamDPVmInstructionsRunner::run(uint32_t program_size,
         StreamDPOpPktWr16    *lpw16;
         StreamDPOpPktWr32    *lpw32;
         StreamDPOpPktWr64    *lpw64;
+        StreamDPOpClientsLimit      *lpcl;
+        StreamDPOpClientsUnLimit    *lpclu;
     } ua ;
 
     while ( p < p_end) {
         uint8_t op_code=*p;
         switch (op_code) {
 
+        case  StreamDPVmInstructions::itCLIENT_VAR :      
+            ua.lpcl =(StreamDPOpClientsLimit *)p;
+            ua.lpcl->run(flow_var);
+            p+=sizeof(StreamDPOpClientsLimit);
+            break;
+
+        case  StreamDPVmInstructions::itCLIENT_VAR_UNLIMIT :      
+            ua.lpclu =(StreamDPOpClientsUnLimit *)p;
+            ua.lpclu->run(flow_var);
+            p+=sizeof(StreamDPOpClientsUnLimit);
+            break;
+
         case   StreamDPVmInstructions::ditINC8  :
             ua.lpv8 =(StreamDPOpFlowVar8 *)p;
             ua.lpv8->run_inc(flow_var);
@@ -461,7 +553,9 @@ public:
         itNONE         = 0,
         itFIX_IPV4_CS  = 4,
         itFLOW_MAN     = 5,
-        itPKT_WR       = 6
+        itPKT_WR       = 6,
+        itFLOW_CLIENT  = 7 
+
     };
 
     typedef uint8_t instruction_type_t ;
@@ -564,10 +658,76 @@ public:
 };
 
 
+/**
+ * flow client instruction - save state for client range and port range 
+ * 
+ * @author hhaim
+ */
+class StreamVmInstructionFlowClient : public StreamVmInstruction {
+
+public:
+    enum client_flags_e {
+        CLIENT_F_UNLIMITED_FLOWS=1, /* unlimited number of flow, don't care about ports  */
+    };
+
+
+    virtual instruction_type_t get_instruction_type(){
+        return ( StreamVmInstruction::itFLOW_CLIENT);
+    }
+
+
+    StreamVmInstructionFlowClient(const std::string &var_name,
+                               uint32_t client_min_value,
+                               uint32_t client_max_value,
+                               uint16_t port_min,
+                               uint16_t port_max,
+                               uint32_t limit_num_flows, /* zero means don't limit */
+                               uint16_t flags
+                               ) { 
+        m_var_name   = var_name;
+        m_client_min = client_min_value;
+        m_client_max = client_max_value;
+
+        m_port_min   = port_min;
+        m_port_max   = port_max;
+
+        m_limit_num_flows = limit_num_flows;
+        m_flags = flags;
+    }
+
+    virtual void Dump(FILE *fd);
+
+    static uint8_t get_flow_var_size(){
+        return  (4+2+4);
+    }
+
+    bool is_unlimited_flows(){
+        return ( (m_flags &   StreamVmInstructionFlowClient::CLIENT_F_UNLIMITED_FLOWS ) ==StreamVmInstructionFlowClient::CLIENT_F_UNLIMITED_FLOWS);
+    }
+public:
+
+
+    /* flow var name */
+    std::string   m_var_name;
+
+    uint32_t m_client_min;  // min ip 
+    uint32_t m_client_max;  // max ip 
+    uint16_t m_port_min;  // start port 
+    uint16_t m_port_max;  // start port 
+    uint32_t m_limit_num_flows;   // number of flows
+    uint16_t m_flags;
+};
+
+
+
 class VmFlowVarRec {
 public:
     uint32_t    m_offset;
-    StreamVmInstructionFlowMan * m_instruction;
+    union {
+        StreamVmInstructionFlowMan * m_ins_flowv;
+        StreamVmInstructionFlowClient * m_ins_flow_client;
+    } m_ins;
+    uint8_t    m_size_bytes;
 };