delay profiling

tcp/flow.go

@@ -4,6 +4,7 @@ import (
 	"encoding/binary"
 	"fmt"
 	"io"
+	"log"
 	"mpbl3p/proxy"
 	"mpbl3p/shared"
 	"net"
@@ -110,9 +111,9 @@ func (f *InitiatedFlow) Reconnect() error {
 		return err
 	}
 
-	//if err := conn.SetWriteBuffer(20); err != nil {
-	//	return err
-	//}
+	if err := conn.SetWriteBuffer(20); err != nil {
+		return err
+	}
 
 	f.conn = conn
 	f.isAlive = true
@@ -186,7 +187,9 @@ func (f *Flow) Produce(v proxy.MacVerifier) (proxy.Packet, error) {
 
 func (f *Flow) consumeMarshalled() {
 	for {
+		t1 := time.Now()
 		data := <-f.toConsume
+		t2 := time.Now()
 
 		err := f.conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
 		if err != nil {
@@ -198,11 +201,15 @@ func (f *Flow) consumeMarshalled() {
 			f.consumeErrors <- err
 			return
 		}
+
+		t3 := time.Now()
+		log.Printf("consumer: `%dns` spent waiting, `%dns` spent consuming", t2.Sub(t1).Nanoseconds(), t3.Sub(t2).Nanoseconds())
 	}
 }
 
 func (f *Flow) produceMarshalled() {
 	for {
+		t1 := time.Now()
 		lengthBytes := make([]byte, 4)
 		if n, err := io.LimitReader(f.conn, 4).Read(lengthBytes); err != nil {
 			f.produceErrors <- err
@@ -225,6 +232,10 @@ func (f *Flow) produceMarshalled() {
 			}
 		}
 
+		t2 := time.Now()
 		f.produced <- dataBytes
+		t3 := time.Now()
+
+		log.Printf("producer: `%dns` spent producing, `%dns` spent waiting", t2.Sub(t1).Nanoseconds(), t3.Sub(t2).Nanoseconds())
 	}
 }

tcp/listener.go

@@ -24,9 +24,9 @@ func NewListener(p *proxy.Proxy, local string, v func() proxy.MacVerifier, g fun
 				panic(err)
 			}
 
-			//if err := conn.SetWriteBuffer(20); err != nil {
-			//	panic(err)
-			//}
+			if err := conn.SetWriteBuffer(20); err != nil {
+				panic(err)
+			}
 
 			f := NewFlowConn(conn)