goapcket生成包应该不支持并发，并发之后性能下降很多

Author: cang233

log/logger.go

@@ -10,24 +10,24 @@ func init() {
 }
 
 func Print(v ...interface{}) {
-	logger.Print(v)
+	logger.Print(v...)
 }
 func Println(v ...interface{}) {
-	logger.Println(v)
+	logger.Println(v...)
 }
 
 func Panicf(formatString string, v ...interface{}) {
-	logger.Panicf(formatString, v)
+	logger.Panicf(formatString, v...)
 }
 
 func Panic(v ...interface{}) {
-	logger.Panic(v)
+	logger.Panic(v...)
 }
 
 func Fatal(v ...interface{}) {
-	logger.Fatal(v)
+	logger.Fatal(v...)
 }
 
 func Fatalf(formatString string, v ...interface{}) {
-	logger.Fatalf(formatString, v)
+	logger.Fatalf(formatString, v...)
 }

producer/main/random_packet_generator_simple.go

@@ -0,0 +1,9 @@
+package main
+
+import "flow-generator/producer"
+
+func main() {
+	sfg := producer.SimpleFlowGenerator{}
+	sfg.Init()
+	sfg.Run()
+}

producer/random_flow_generator_simple.go

@@ -0,0 +1,139 @@
+package producer
+
+import (
+	"flag"
+	"flow-generator/util"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/google/gopacket"
+	"github.com/google/gopacket/layers"
+	"github.com/google/gopacket/pcap"
+)
+
+type SimpleFlowGenerator struct {
+	config  map[string]string
+	handler *pcap.Handle
+}
+
+func (s *SimpleFlowGenerator) getOptions() {
+	var nicName string
+	flag.StringVar(&nicName, "i", "enp10s0", "the nic name used for sending packets.")
+
+	flag.Parse()
+	s.config["i"] = nicName
+}
+
+func (s *SimpleFlowGenerator) Init() {
+	//init struct
+	s.config = make(map[string]string, 0)
+	//get options
+	s.getOptions()
+
+	//init other
+	s.initHandler()
+}
+
+func (s *SimpleFlowGenerator) initHandler() {
+	var (
+		snapshotLen int32 = 65535
+		promiscuous bool  = false
+		err         error
+		timeout     time.Duration = 30 * time.Second
+		nicName     string
+	)
+	if v, ok := s.config["i"]; ok {
+		nicName = v
+	} else {
+		log.Panic("nic name needed!")
+	}
+
+	devices, err := pcap.FindAllDevs()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	for _, value := range devices {
+		//这里为了兼容win与linux，同时检查2个
+		if value.Description == nicName || value.Name == nicName {
+			//Open device
+			s.handler, err = pcap.OpenLive(value.Name, snapshotLen, promiscuous, timeout)
+			if err != nil {
+				log.Fatal(err)
+			}
+		}
+		fmt.Println(value.Description, value.Name)
+	}
+
+	if s.handler == nil {
+		log.Panicln("can not find nic:", nicName)
+	}
+}
+
+func (s *SimpleFlowGenerator) Run() {
+	defer s.handler.Close()
+
+	var (
+		buffer      gopacket.SerializeBuffer
+		options     gopacket.SerializeOptions
+		err         error
+		flowPackets int
+		lys         *randomPacketLayer
+	)
+
+	// And create the packet with the layers
+	buffer = gopacket.NewSerializeBuffer()
+
+	count := 0
+	lastTime := time.Now()
+	//loop
+	for {
+		flowPackets = util.RandomInt(6, 32)
+		lys = randPktConfig()
+		//gen a flow
+		for flowPackets > 0 {
+			//change content
+			gopacket.SerializeLayers(buffer, options,
+				lys.ether,
+				lys.ipv4,
+				lys.tcp,
+				gopacket.Payload(util.RandomBytes(20)),
+			)
+			err = s.handler.WritePacketData(buffer.Bytes())
+			if err != nil {
+				log.Fatal(err)
+			}
+			//change flag
+			if flowPackets == 1 {
+				//the last of a flow
+				lys.tcp.SYN = false
+				lys.tcp.PSH = false
+				lys.tcp.FIN = util.RandomBoolean() //have ending or not
+			} else {
+				lys.tcp.SYN = false
+				lys.tcp.FIN = false
+				lys.tcp.PSH = true
+			}
+			//count
+			count++
+			now := time.Now()
+			if now.Sub(lastTime) > time.Second*1 {
+				fmt.Println("pps:", count)
+				lastTime = now
+				count = 0
+			}
+			//
+			flowPackets--
+
+		}
+
+	}
+}
+
+func simpleRandomChangePacket(lys *randomPacketLayer) {
+	lys.ipv4.SrcIP = util.RandomIPv4()
+	lys.ipv4.DstIP = util.RandomIPv4()
+	lys.tcp.SrcPort = layers.TCPPort(util.RandomPort())
+	lys.tcp.DstPort = layers.TCPPort(util.RandomPort())
+}

producer/random_packet_generator.go

@@ -55,9 +55,10 @@ func (r *RandPktGenerator) Run() {
 	wg.Wait()
 }
 
+//TODO 增加worker的数量>=2的时候，吞吐量反而下降巨多，说明gopacket生成包对并发操作支持不好。
 func defaultConfig() map[string]int {
 	return map[string]int{
-		"default_worker_count": 10,
+		"default_worker_count": 1,
 	}
 }
 
@@ -203,6 +204,8 @@ func (w *worker) Run() {
 		logTicker := time.NewTicker(time.Second * 1)
 		defer stopTicker.Stop()
 		defer logTicker.Stop()
+
+		buffer := gopacket.NewSerializeBuffer()
 		for w.runningStatus == WorkStatusRunning {
 			select {
 			case <-stopTicker.C:
@@ -211,12 +214,11 @@ func (w *worker) Run() {
 				w.sts.CountClear()
 			default: //do rand update
 				randomChangePacket(w.tmplate)
-				buffer := gopacket.NewSerializeBuffer()
 				gopacket.SerializeLayers(buffer, gopacket.SerializeOptions{},
 					w.tmplate.ether,
 					w.tmplate.ipv4,
 					w.tmplate.tcp,
-					gopacket.Payload(util.RandomBytes(util.RandomInt(0, 1000))),
+					gopacket.Payload(util.RandomBytes(util.RandomInt(0, 10))),
 				)
 				w.exportChan <- buffer.Bytes()
 				w.sts.CountAdd()
@@ -303,6 +305,10 @@ func (s *sender) Run() {
 					s.father.addWorker()
 					log.Println("[sender]chan is not full,call father adding a worker")
 				} else {
+					//若只有一个worker，则不删除
+					if len(s.father.workers) <= 1 {
+						continue
+					}
 					s.father.removeWorker()
 					log.Println("[sender]chan is full,call father removing a worker")
 				}

test/go_gopkt_test.go

@@ -3,14 +3,15 @@ package test
 import (
 	"bytes"
 	"fmt"
-	"github.com/google/gopacket"
-	"github.com/google/gopacket/layers"
-	"github.com/google/gopacket/pcap"
 	"log"
 	"math/rand"
 	"net"
 	"testing"
 	"time"
+
+	"github.com/google/gopacket"
+	"github.com/google/gopacket/layers"
+	"github.com/google/gopacket/pcap"
 )
 
 func TestUseGopacket(t *testing.T) {
@@ -30,7 +31,7 @@ func TestUseGopacket(t *testing.T) {
 	}
 
 	for _, value := range devices {
-		if value.Description == "Realtek Gaming GbE Family Controller" {
+		if value.Description == "Realtek Gaming GbE Family Controller" || value.Name == "enp10s0" {
 			//Open device
 			handle, err = pcap.OpenLive(value.Name, snapshot_len, promiscuous, timeout)
 			if err != nil {
@@ -39,6 +40,7 @@ func TestUseGopacket(t *testing.T) {
 		}
 		fmt.Println(value.Description, value.Name)
 	}
+	defer handle.Close()
 	// Send raw bytes over wire
 	data := "this is raw data in packets"
 	data = ""
@@ -77,24 +79,31 @@ func TestUseGopacket(t *testing.T) {
 	}
 	// And create the packet with the layers
 	buffer = gopacket.NewSerializeBuffer()
-	gopacket.SerializeLayers(buffer, options,
-		ethernetLayer,
-		ipLayer,
-		tcpLayer,
-		gopacket.Payload(rawBytes),
-	)
-	outgoingPacket := buffer.Bytes()
+
+	count := 0
+	lastTime := time.Now()
 	for {
-		time.Sleep(time.Second * 3)
-		fmt.Println(handle)
-		fmt.Println(len(outgoingPacket))
+		gopacket.SerializeLayers(buffer, options,
+			ethernetLayer,
+			ipLayer,
+			tcpLayer,
+			gopacket.Payload(rawBytes),
+		)
+		outgoingPacket := buffer.Bytes()
+
 		err = handle.WritePacketData(outgoingPacket)
 		if err != nil {
 			log.Fatal(err)
 		}
+		count++
+		now := time.Now()
+		if now.Sub(lastTime) > time.Second*1 {
+			fmt.Println("pps:", count)
+			lastTime = now
+			count = 0
+		}
 	}
 
-	handle.Close()
 }
 
 func TestCountOx(t *testing.T) {

util/random.go

@@ -31,7 +31,7 @@ func RandomSequence() uint32 {
 
 func RandomBoolean() bool {
 	const N = 1 << 10
-	return rand.Intn(N)-N/2 >= 0
+	return rand.Intn(N) >= N/2
 }
 
 func RandomString(n int) string {