5
0
mirror of https://github.com/cwinfo/yggdrasil-go.git synced 2024-11-26 06:01:37 +00:00

adjust link packet logic so they bypass the lifo stack and are delivered first

This commit is contained in:
Arceliar 2018-06-06 17:44:10 -05:00
parent bbae9ff8e8
commit 690d29435d
4 changed files with 36 additions and 29 deletions

View File

@ -453,16 +453,16 @@ func (c *Core) DEBUG_addAllowedEncryptionPublicKey(boxStr string) {
func DEBUG_simLinkPeers(p, q *peer) { func DEBUG_simLinkPeers(p, q *peer) {
// Sets q.out() to point to p and starts p.linkLoop() // Sets q.out() to point to p and starts p.linkLoop()
plinkIn := make(chan []byte, 1) p.linkIn, q.linkIn = make(chan []byte, 32), make(chan []byte, 32)
qlinkIn := make(chan []byte, 1) p.linkOut, q.linkOut = q.linkIn, p.linkIn
p.out = func(bs []byte) { p.out = func(bs []byte) {
go q.handlePacket(bs, qlinkIn) go q.handlePacket(bs)
} }
q.out = func(bs []byte) { q.out = func(bs []byte) {
go p.handlePacket(bs, plinkIn) go p.handlePacket(bs)
} }
go p.linkLoop(plinkIn) go p.linkLoop()
go q.linkLoop(qlinkIn) go q.linkLoop()
} }
func (c *Core) DEBUG_simFixMTU() { func (c *Core) DEBUG_simFixMTU() {

View File

@ -111,6 +111,9 @@ type peer struct {
close func() close func()
// To allow the peer to call close if idle for too long // To allow the peer to call close if idle for too long
lastAnc time.Time lastAnc time.Time
// used for protocol traffic (to bypass queues)
linkIn (chan []byte) // handlePacket sends, linkLoop recvs
linkOut (chan []byte)
} }
const peer_Throttle = 1 const peer_Throttle = 1
@ -123,8 +126,7 @@ func (p *peer) updateQueueSize(delta int64) {
atomic.AddInt64(&p.queueSize, delta) atomic.AddInt64(&p.queueSize, delta)
} }
func (ps *peers) newPeer(box *boxPubKey, func (ps *peers) newPeer(box *boxPubKey, sig *sigPubKey) *peer {
sig *sigPubKey) *peer {
now := time.Now() now := time.Now()
p := peer{box: *box, p := peer{box: *box,
sig: *sig, sig: *sig,
@ -170,14 +172,14 @@ func (ps *peers) removePeer(port switchPort) {
} }
} }
func (p *peer) linkLoop(in <-chan []byte) { func (p *peer) linkLoop() {
ticker := time.NewTicker(time.Second) ticker := time.NewTicker(time.Second)
defer ticker.Stop() defer ticker.Stop()
var counter uint8 var counter uint8
var lastRSeq uint64 var lastRSeq uint64
for { for {
select { select {
case packet, ok := <-in: case packet, ok := <-p.linkIn:
if !ok { if !ok {
return return
} }
@ -214,7 +216,7 @@ func (p *peer) linkLoop(in <-chan []byte) {
} }
} }
func (p *peer) handlePacket(packet []byte, linkIn chan<- []byte) { func (p *peer) handlePacket(packet []byte) {
// TODO See comment in sendPacket about atomics technically being done wrong // TODO See comment in sendPacket about atomics technically being done wrong
atomic.AddUint64(&p.bytesRecvd, uint64(len(packet))) atomic.AddUint64(&p.bytesRecvd, uint64(len(packet)))
pType, pTypeLen := wire_decode_uint64(packet) pType, pTypeLen := wire_decode_uint64(packet)
@ -227,12 +229,7 @@ func (p *peer) handlePacket(packet []byte, linkIn chan<- []byte) {
case wire_ProtocolTraffic: case wire_ProtocolTraffic:
p.handleTraffic(packet, pTypeLen) p.handleTraffic(packet, pTypeLen)
case wire_LinkProtocolTraffic: case wire_LinkProtocolTraffic:
{ p.linkIn <- packet
select {
case linkIn <- packet:
default:
}
}
default: /*panic(pType) ;*/ default: /*panic(pType) ;*/
return return
} }
@ -284,7 +281,7 @@ func (p *peer) sendLinkPacket(packet []byte) {
Payload: bs, Payload: bs,
} }
packet = linkPacket.encode() packet = linkPacket.encode()
p.sendPacket(packet) p.linkOut <- packet
} }
func (p *peer) handleLinkTraffic(bs []byte) { func (p *peer) handleLinkTraffic(bs []byte) {

View File

@ -55,7 +55,7 @@ func (r *router) init(core *Core) {
} }
} }
r.in = in r.in = in
r.out = func(packet []byte) { p.handlePacket(packet, nil) } // The caller is responsible for go-ing if it needs to not block r.out = func(packet []byte) { p.handlePacket(packet) } // The caller is responsible for go-ing if it needs to not block
recv := make(chan []byte, 32) recv := make(chan []byte, 32)
send := make(chan []byte, 32) send := make(chan []byte, 32)
r.recv = recv r.recv = recv

View File

@ -208,10 +208,11 @@ func (iface *tcpInterface) handler(sock net.Conn, incoming bool) {
}() }()
// Note that multiple connections to the same node are allowed // Note that multiple connections to the same node are allowed
// E.g. over different interfaces // E.g. over different interfaces
linkIn := make(chan []byte, 1) p := iface.core.peers.newPeer(&info.box, &info.sig)
p := iface.core.peers.newPeer(&info.box, &info.sig) //, in, out) p.linkIn = make(chan []byte, 1)
p.linkOut = make(chan []byte, 1)
in := func(bs []byte) { in := func(bs []byte) {
p.handlePacket(bs, linkIn) p.handlePacket(bs)
} }
out := make(chan []byte, 32) // TODO? what size makes sense out := make(chan []byte, 32) // TODO? what size makes sense
defer close(out) defer close(out)
@ -221,10 +222,10 @@ func (iface *tcpInterface) handler(sock net.Conn, incoming bool) {
buf.Write(tcp_msg[:]) buf.Write(tcp_msg[:])
buf.Write(msgLen) buf.Write(msgLen)
buf.Write(msg) buf.Write(msg)
p.updateQueueSize(-1)
util_putBytes(msg) util_putBytes(msg)
} }
go func() { go func() {
defer buf.Flush()
var stack [][]byte var stack [][]byte
put := func(msg []byte) { put := func(msg []byte) {
stack = append(stack, msg) stack = append(stack, msg)
@ -234,14 +235,22 @@ func (iface *tcpInterface) handler(sock net.Conn, incoming bool) {
p.updateQueueSize(-1) p.updateQueueSize(-1)
} }
} }
for msg := range out { for {
put(msg) select {
case msg := <-p.linkOut:
send(msg)
case msg, ok := <-out:
if !ok {
return
}
put(msg)
}
for len(stack) > 0 { for len(stack) > 0 {
// Keep trying to fill the stack (LIFO order) while sending
select { select {
case msg := <-p.linkOut:
send(msg)
case msg, ok := <-out: case msg, ok := <-out:
if !ok { if !ok {
buf.Flush()
return return
} }
put(msg) put(msg)
@ -249,6 +258,7 @@ func (iface *tcpInterface) handler(sock net.Conn, incoming bool) {
msg := stack[len(stack)-1] msg := stack[len(stack)-1]
stack = stack[:len(stack)-1] stack = stack[:len(stack)-1]
send(msg) send(msg)
p.updateQueueSize(-1)
} }
} }
buf.Flush() buf.Flush()
@ -265,11 +275,11 @@ func (iface *tcpInterface) handler(sock net.Conn, incoming bool) {
} }
p.close = func() { sock.Close() } p.close = func() { sock.Close() }
setNoDelay(sock, true) setNoDelay(sock, true)
go p.linkLoop(linkIn) go p.linkLoop()
defer func() { defer func() {
// Put all of our cleanup here... // Put all of our cleanup here...
p.core.peers.removePeer(p.port) p.core.peers.removePeer(p.port)
close(linkIn) close(p.linkIn)
}() }()
them, _, _ := net.SplitHostPort(sock.RemoteAddr().String()) them, _, _ := net.SplitHostPort(sock.RemoteAddr().String())
themNodeID := getNodeID(&info.box) themNodeID := getNodeID(&info.box)