Sched batch

batch_manager.go

@@ -0,0 +1,82 @@
+package main
+
+import (
+	"golang.org/x/net/ipv4"
+	"runtime"
+	"sync/atomic"
+	"time"
+)
+
+func RunNewPacketCollector(bucketSize int) *PacketCollector {
+	collector := &PacketCollector{
+		atomicLock:  0,
+		BucketSize:  bucketSize,
+		activeBatch: &Batch{make([]ipv4.Message, 0), 0},
+		dstChannel:  make(chan *Batch),
+	}
+	go collector.Run()
+	return collector
+}
+
+type PacketCollector struct {
+	atomicLock  int32
+	BucketSize  int
+	activeBatch *Batch
+	dstChannel  chan *Batch
+}
+
+func (pc *PacketCollector) PushOne(message ipv4.Message) {
+	for !pc.TryLock() {
+		runtime.Gosched()
+	}
+	pc.activeBatch.msgCount = pc.activeBatch.msgCount + 1
+	pc.activeBatch.messages = append(pc.activeBatch.messages, message)
+	pc.Unlock()
+}
+
+func (pc *PacketCollector) Push(messages []ipv4.Message) {
+	for !pc.TryLock() {
+		runtime.Gosched()
+	}
+	pc.activeBatch.msgCount = pc.activeBatch.msgCount + len(messages)
+	pc.activeBatch.messages = append(pc.activeBatch.messages, messages...)
+	pc.Unlock()
+}
+
+func (pc *PacketCollector) Run() {
+	ticker := time.NewTicker(1 * time.Millisecond)
+	for {
+		select {
+		case <-ticker.C:
+			pc.sendActiveBucket()
+		}
+	}
+}
+
+func (pc *PacketCollector) sendActiveBucket() {
+	for !pc.TryLock() {
+		runtime.Gosched()
+	}
+	pc.dstChannel <- pc.activeBatch
+	pc.activeBatch = &Batch{make([]ipv4.Message, 0), 0}
+	pc.Unlock()
+}
+
+func (pc *PacketCollector) TryLock() bool {
+	return atomic.CompareAndSwapInt32(&pc.atomicLock, 0, 1)
+}
+
+func (pc *PacketCollector) Unlock() {
+	atomic.CompareAndSwapInt32(&pc.atomicLock, 1, 0)
+}
+
+func getMessageBuffer(size int) []ipv4.Message {
+	ms := make([]ipv4.Message, size)
+	for i := 0; i < size; i++ {
+		ms[i] = ipv4.Message{
+			OOB:     make([]byte, 10),
+			Buffers: [][]byte{make([]byte, RCVR_BUF_SIZE)},
+		}
+	}
+	return ms
+}

batch_monitor.go

@@ -0,0 +1,46 @@
+package main
+
+import (
+	"log"
+	"sync/atomic"
+	"time"
+)
+
+var udpTunReceivedMgs int64
+var udpTunProcessedMsg int64
+
+var tunUdpReceivedMgs int64
+var tunUdpProcessedMsg int64
+
+var addFromUDPToTunCounter int64
+var addFromTunToUDPCounters int64
+
+func AddFromUDPToTunCounters(receivedMgs, processedMsg int32) {
+	atomic.AddInt64(&udpTunReceivedMgs, int64(receivedMgs))
+	atomic.AddInt64(&udpTunProcessedMsg, int64(processedMsg))
+	atomic.AddInt64(&addFromUDPToTunCounter, 1)
+}
+
+func AddFromTunToUDPCounters(receivedMgs, processedMsg int32) {
+	atomic.AddInt64(&tunUdpReceivedMgs, int64(receivedMgs))
+	atomic.AddInt64(&tunUdpProcessedMsg, int64(processedMsg))
+	atomic.AddInt64(&addFromTunToUDPCounters, 1)
+}
+
+func RunMonitor() {
+	ticker := time.NewTicker(1 * time.Second)
+	for {
+		select {
+		case <-ticker.C:
+			log.Printf("From TUN to UDP: %d  -  %d   -  %d",
+				atomic.SwapInt64(&tunUdpReceivedMgs, 0),
+				atomic.SwapInt64(&tunUdpProcessedMsg, 0),
+				atomic.SwapInt64(&addFromTunToUDPCounters, 0))
+			log.Printf("From UDP to TUN: %d  -  %d   -  %d",
+				atomic.SwapInt64(&udpTunReceivedMgs, 0),
+				atomic.SwapInt64(&udpTunProcessedMsg, 0),
+				atomic.SwapInt64(&addFromUDPToTunCounter, 0))
+		}
+
+	}
+}

batch_processing.go

@@ -0,0 +1,104 @@
+package main
+
+import (
+	"log"
+	"runtime"
+	"sync/atomic"
+	"syscall"
+	"time"
+)
+
+func runTunReadBatchThread(pc *PacketCollector) {
+	go func(pc *PacketCollector) {
+		runtime.LockOSThread()
+		var packet = make([]byte, mtu)
+		for {
+			plen, err := tunInterface.Read(packet)
+			if err != nil {
+				log.Fatalf("Tun Interface Read: type unknown %+v\n", err)
+			}
+			dataPacket := getDataPacket()
+			copy(dataPacket.data, packet[:plen])
+			dataPacket.packetLen = plen
+			pc.PushOne(dataPacket.ToMessage(dstIP))
+		}
+	}(pc)
+}
+
+func runTunWriteBatchThread(pc *PacketCollector) {
+	go func(pc *PacketCollector) {
+		runtime.LockOSThread()
+		for {
+			select {
+			case batch := <-pc.dstChannel:
+				msgCount := int32(batch.msgCount)
+				processedMsg := int32(0)
+				time.AfterFunc(1*time.Millisecond, func() {
+					AddFromUDPToTunCounters(msgCount, processedMsg)
+				})
+				if msgCount > 0 {
+					for i := 0; i < batch.msgCount; i++ {
+						message := batch.messages[i]
+						bytes := message.Buffers[0]
+						//log.Printf("Tun Interface Write: %+v\n", batch.msgCount)
+						_, err := tunInterface.Write(bytes)
+						if err != nil {
+							log.Fatalf("Tun Interface Write: type unknown %+v\n", err)
+						}
+						processedMsg += 1
+					}
+					messagesPool.Put(batch)
+				}
+			}
+		}
+	}(pc)
+}
+
+func runUDPReadBatchThread(pc *PacketCollector) {
+	go func(pc *PacketCollector) {
+		runtime.LockOSThread()
+		for {
+			batch := messagesPool.Get().(*Batch)
+			count, err := udpListenConn.ReadBatch(batch.messages, syscall.MSG_WAITFORONE)
+			if err != nil {
+				log.Fatalf("UDP Interface Read: type unknown %+v\n", err)
+			}
+			batch.msgCount = count
+			pc.Push(batch.messages[:count])
+		}
+	}(pc)
+}
+
+func runUDPBatchWriteThread(pc *PacketCollector) {
+	go func(pc *PacketCollector) {
+		runtime.LockOSThread()
+		for {
+			select {
+			case batch := <-pc.dstChannel:
+				msgCount := int32(batch.msgCount)
+				processedMsg := int32(0)
+				time.AfterFunc(1*time.Millisecond, func() {
+					AddFromTunToUDPCounters(msgCount, atomic.LoadInt32(&processedMsg))
+				})
+				if msgCount > 0 {
+					offset := int32(0)
+					bucketSize := int32(0)
+					for offset < msgCount {
+						if (msgCount-offset)/int32(RCVR_MSG_PACK) >= 1 {
+							bucketSize = offset + int32(RCVR_MSG_PACK)
+						} else {
+							bucketSize = offset + ((msgCount - offset) % int32(RCVR_MSG_PACK))
+						}
+						//log.Printf("offset: %d, count: %d, bucket: %d", offset, msgCount, bucketSize)
+						n, err := udpWriterConn.WriteBatch(batch.messages[offset:bucketSize], syscall.MSG_WAITFORONE)
+						if err != nil {
+							log.Fatalf("UDP Interface Write error: %+v\n", err)
+						}
+						atomic.AddInt32(&processedMsg, int32(n))
+						offset = bucketSize
+					}
+				}
+			}
+		}
+	}(pc)
+}

model.go

@@ -0,0 +1,25 @@
+package main
+
+import (
+	"golang.org/x/net/ipv4"
+	"net"
+)
+
+type DataPacket struct {
+	data      []byte
+	packetLen int
+}
+
+func (packet *DataPacket) ToMessage(address *net.UDPAddr) ipv4.Message {
+	return ipv4.Message{
+		OOB:     nil,
+		Buffers: [][]byte{packet.data},
+		N:       packet.packetLen,
+		Addr:    address,
+	}
+}
+
+type Batch struct {
+	messages []ipv4.Message
+	msgCount int
+}