diff --git a/src/yggdrasil/router.go b/src/yggdrasil/router.go index 3246f63..ddb4848 100644 --- a/src/yggdrasil/router.go +++ b/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: diff --git a/src/yggdrasil/switch.go b/src/yggdrasil/switch.go index 67d9d8f..de79f4a 100644 --- a/src/yggdrasil/switch.go +++ b/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) } diff --git a/src/yggdrasil/tcp.go b/src/yggdrasil/tcp.go index 6564dfd..e7b682c 100644 --- a/src/yggdrasil/tcp.go +++ b/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 {