more tcp debugging

src/yggdrasil/router.go

@@ -93,7 +93,7 @@ func (r *router) mainLoop() {
 				// Any periodic maintenance stuff goes here
 				r.core.switchTable.doMaintenance()
 				r.core.dht.doMaintenance()
-				//r.core.peers.fixSwitchAfterPeerDisconnect() // FIXME makes sure dht peers get added quickly
+				//r.core.peers.sendSwitchMsgs() // FIXME debugging
 				util_getBytes() // To slowly drain things
 			}
 		case f := <-r.admin:

src/yggdrasil/switch.go

@@ -223,6 +223,9 @@ func (t *switchTable) removePeer(port switchPort) {
 	delete(t.data.peers, port)
 	t.updater.Store(&sync.Once{})
 	// TODO if parent, find a new peer to use as parent instead
+	for _, info := range t.data.peers {
+		t.unlockedHandleMsg(&info.msg, info.port)
+	}
 }
 
 func (t *switchTable) cleanDropped() {
@@ -261,9 +264,13 @@ func (t *switchTable) getMsg() *switchMsg {
 }
 
 func (t *switchTable) handleMsg(msg *switchMsg, fromPort switchPort) {
-	// TODO directly use a switchMsg instead of switchMessage + sigs
 	t.mutex.Lock()
 	defer t.mutex.Unlock()
+	t.unlockedHandleMsg(msg, fromPort)
+}
+
+func (t *switchTable) unlockedHandleMsg(msg *switchMsg, fromPort switchPort) {
+	// TODO directly use a switchMsg instead of switchMessage + sigs
 	now := time.Now()
 	// Set up the sender peerInfo
 	var sender peerInfo
@@ -433,6 +440,7 @@ func (t *switchTable) lookup(dest []byte, ttl uint64) (switchPort, uint64) {
 		}
 	}
 	//t.core.log.Println("DEBUG: sending to", best, "bandwidth", getBandwidth(best))
+	//t.core.log.Println("DEBUG: sending to", best, "cost", bestCost)
 	return best, uint64(myDist)
 }
 

src/yggdrasil/tcp.go

@@ -15,7 +15,6 @@ import "time"
 import "errors"
 import "sync"
 import "fmt"
-import "bufio"
 import "golang.org/x/net/proxy"
 
 const tcp_msgSize = 2048 + 65535 // TODO figure out what makes sense
@@ -215,16 +214,7 @@ func (iface *tcpInterface) handler(sock net.Conn, incoming bool) {
 	}
 	out := make(chan []byte, 32) // TODO? what size makes sense
 	defer close(out)
-	buf := bufio.NewWriterSize(sock, tcp_msgSize)
-	send := func(msg []byte) {
-		msgLen := wire_encode_uint64(uint64(len(msg)))
-		buf.Write(tcp_msg[:])
-		buf.Write(msgLen)
-		buf.Write(msg)
-		util_putBytes(msg)
-	}
 	go func() {
-		defer buf.Flush()
 		var shadow uint64
 		var stack [][]byte
 		put := func(msg []byte) {
@@ -235,11 +225,29 @@ func (iface *tcpInterface) handler(sock net.Conn, incoming bool) {
 				shadow++
 			}
 		}
+		send := func(msg []byte) {
+			msgLen := wire_encode_uint64(uint64(len(msg)))
+			buf := net.Buffers{tcp_msg[:], msgLen, msg}
+			buf.WriteTo(sock)
+			util_putBytes(msg)
+		}
+		timerInterval := 4 * time.Second
+		timer := time.NewTimer(timerInterval)
+		defer timer.Stop()
 		for {
 			for ; shadow > 0; shadow-- {
 				p.updateQueueSize(-1)
 			}
+			timer.Stop()
 			select {
+			case <-timer.C:
+			default:
+			}
+			timer.Reset(timerInterval)
+			select {
+			case _ = <-timer.C:
+				//iface.core.log.Println("DEBUG: sending keep-alive:", sock.RemoteAddr().String())
+				send(nil) // TCP keep-alive traffic
 			case msg := <-p.linkOut:
 				send(msg)
 			case msg, ok := <-out:
@@ -264,7 +272,6 @@ func (iface *tcpInterface) handler(sock net.Conn, incoming bool) {
 					p.updateQueueSize(-1)
 				}
 			}
-			buf.Flush()
 		}
 	}()
 	p.out = func(msg []byte) {
@@ -298,7 +305,7 @@ func (iface *tcpInterface) reader(sock net.Conn, in func([]byte)) {
 	bs := make([]byte, 2*tcp_msgSize)
 	frag := bs[:0]
 	for {
-		timeout := time.Now().Add(2 * time.Minute)
+		timeout := time.Now().Add(6 * time.Second)
 		sock.SetReadDeadline(timeout)
 		n, err := sock.Read(bs[len(frag):])
 		if err != nil || n == 0 {