Initial impl

input_raw.go

@@ -83,10 +83,7 @@ func (i *RAWInput) listen(address string) {
 		default:
 		}
 
-		// Receiving TCPMessage object
-		m := <- ch
-
-		i.data <- m
+		i.data <- ch.Read()
 	}
 }
 

raw_socket_listener/listener.go

@@ -51,10 +51,10 @@ type Listener struct {
 	respWithoutReq map[uint32]string
 
 	// Messages ready to be send to client
-	packetsChan chan *TCPPacket
+	packetsQueue *PacketQueue
 
 	// Messages ready to be send to client
-	messagesChan chan *TCPMessage
+	messagesQueue *MessageQueue
 
 	addr string // IP to listen
 	port uint16 // Port to listen
@@ -82,8 +82,9 @@ const (
 func NewListener(addr string, port string, engine int, expire time.Duration) (l *Listener) {
 	l = &Listener{}
 
-	l.packetsChan = make(chan *TCPPacket, 10000)
-	l.messagesChan = make(chan *TCPMessage, 10000)
+	l.messagesQueue = &MessageQueue{lastCommittedIndex : 0, nextFreeIndex : 1, readerIndex : 1}
+	l.packetsQueue = &PacketQueue{lastCommittedIndex : 0, nextFreeIndex : 1, readerIndex : 1}
+
 	l.quit = make(chan bool)
 	l.readyCh = make(chan bool, 1)
 
@@ -124,7 +125,7 @@ func NewListener(addr string, port string, engine int, expire time.Duration) (l
 func (t *Listener) processPackets() {
 	for {
 		// We need to use channels to process each packet to avoid data races
-		packet := <-t.packetsChan
+		packet := t.packetsQueue.Read()
 		// log.Println(packet)
 		t.mu.Lock()
 		t.processTCPPacket(packet)
@@ -207,7 +208,7 @@ func (t *Listener) dispatchMessage(message *TCPMessage) {
 		}
 	}
 
-	t.messagesChan <- message
+	t.messagesQueue.Write(message)
 }
 
 // DeviceNotFoundError raised if user specified wrong ip
@@ -308,9 +309,7 @@ func (t *Listener) readPcap() {
 			newBuf := make([]byte, len(data))
 			copy(newBuf, data)
 
-			go func(newBuf []byte) {
-				t.packetsChan <- ParseTCPPacket(net.IP(srcIP).String(), newBuf)
-			}(newBuf)
+			t.packetsQueue.Write(ParseTCPPacket(net.IP(srcIP).String(), newBuf))
 		}
 	}
 }
@@ -346,9 +345,7 @@ func (t *Listener) readRAWSocket() {
 				newBuf := make([]byte, n)
 				copy(newBuf, buf[:n])
 
-				go func(newBuf []byte) {
-					t.packetsChan <- ParseTCPPacket(addr.String(), newBuf)
-				}(newBuf)
+				t.packetsQueue.Write(ParseTCPPacket(addr.String(), newBuf))
 			}
 		}
 	}
@@ -519,8 +516,8 @@ func (t *Listener) IsReady() bool {
 }
 
 // Receive TCP messages from the listener channel
-func (t *Listener) Receiver() chan *TCPMessage {
-	return t.messagesChan
+func (t *Listener) Receiver() *MessageQueue {
+	return t.messagesQueue
 }
 
 func (t *Listener) Close() {

raw_socket_listener/listener_test.go

@@ -376,7 +376,7 @@ func TestRawListenerBench(t *testing.T) {
 						}
 					}
 
-					l.packetsChan <- p
+					l.packetsQueue.Write(p)
 					time.Sleep(time.Millisecond)
 				}
 
@@ -390,11 +390,12 @@ func TestRawListenerBench(t *testing.T) {
 	var count int32
 
 	for {
+		ch.Read()
 		select {
 			case <- ch:
 				atomic.AddInt32(&count, 1)
 			case <-time.After(2000 * time.Millisecond):
-				log.Println("Emitted 200000 messages, captured: ", count, len(l.ackAliases), len(l.seqWithData), len(l.respAliases), len(l.respWithoutReq), len(l.packetsChan))
+				log.Println("Emitted 200000 messages, captured: ", count, len(l.ackAliases), len(l.seqWithData), len(l.respAliases), len(l.respWithoutReq))
 				return
 		}
 	}

raw_socket_listener/queue.go

@@ -0,0 +1,75 @@
+package rawSocket
+
+const queueSize uint64 = 8192
+const indexMask uint64 = queueSize - 1
+
+type PacketQueue struct {
+    padding1 [8]uint64
+    lastCommittedIndex uint64
+    padding2 [8]uint64
+    nextFreeIndex uint64
+    padding3 [8]uint64
+    readerIndex uint64
+    padding4 [8]uint64
+    contents [queueSize]*TCPPacket
+    padding5 [8]uint64
+}
+
+func (self *PacketQueue) Write(value *TCPPacket) {
+    var myIndex = atomic.AddUint64(&self.nextFreeIndex, 1) - 1
+    //Wait for reader to catch up, so we don't clobber a slot which it is (or will be) reading
+    for myIndex > (self.readerIndex + queueSize - 2) {
+        runtime.Gosched()
+    }
+    //Write the item into it's slot
+    self.contents[myIndex & indexMask] = value
+    //Increment the lastCommittedIndex so the item is available for reading
+    for !atomic.CompareAndSwapUint64(&self.lastCommittedIndex, myIndex - 1, myIndex) {
+        runtime.Gosched()
+    }
+}
+
+func (self *PacketQueue) Read() *TCPPacket {
+    var myIndex = atomic.AddUint64(&self.readerIndex, 1) - 1
+    //If reader has out-run writer, wait for a value to be committed
+    for myIndex > self.lastCommittedIndex {
+        runtime.Gosched()
+    }
+    return self.contents[myIndex & indexMask]
+}
+
+
+type MessageQueue struct {
+    padding1 [8]uint64
+    lastCommittedIndex uint64
+    padding2 [8]uint64
+    nextFreeIndex uint64
+    padding3 [8]uint64
+    readerIndex uint64
+    padding4 [8]uint64
+    contents [queueSize]*TCPMessage
+    padding5 [8]uint64
+}
+
+func (self *MessageQueue) Write(value *TCPMessage) {
+    var myIndex = atomic.AddUint64(&self.nextFreeIndex, 1) - 1
+    //Wait for reader to catch up, so we don't clobber a slot which it is (or will be) reading
+    for myIndex > (self.readerIndex + queueSize - 2) {
+        runtime.Gosched()
+    }
+    //Write the item into it's slot
+    self.contents[myIndex & indexMask] = value
+    //Increment the lastCommittedIndex so the item is available for reading
+    for !atomic.CompareAndSwapUint64(&self.lastCommittedIndex, myIndex - 1, myIndex) {
+        runtime.Gosched()
+    }
+}
+
+func (self *MessageQueue) Read() *TCPMessage {
+    var myIndex = atomic.AddUint64(&self.readerIndex, 1) - 1
+    //If reader has out-run writer, wait for a value to be committed
+    for myIndex > self.lastCommittedIndex {
+        runtime.Gosched()
+    }
+    return self.contents[myIndex & indexMask]
+}