5
0
mirror of https://github.com/cwinfo/yggdrasil-go.git synced 2024-11-29 15:31:35 +00:00

send an ack if we receive a packet and don't have any return traffic, keeping a legacy 4-second keep-alive in case there's no traffic at all to send (to be removed later, after nodes have upgraded), ideally we should either remove ReadTimeout or use it for the switch idle timeout instead

This commit is contained in:
Arceliar 2019-02-02 22:18:55 -06:00
parent 6d83d970bb
commit b44a0f29f3
2 changed files with 65 additions and 32 deletions

View File

@ -3,6 +3,7 @@ package util
// These are misc. utility functions that didn't really fit anywhere else // These are misc. utility functions that didn't really fit anywhere else
import "runtime" import "runtime"
import "time"
// A wrapper around runtime.Gosched() so it doesn't need to be imported elsewhere. // A wrapper around runtime.Gosched() so it doesn't need to be imported elsewhere.
func Yield() { func Yield() {
@ -44,3 +45,14 @@ func PutBytes(bs []byte) {
default: default:
} }
} }
// This is a workaround to go's broken timer implementation
func TimerStop(t *time.Timer) bool {
if !t.Stop() {
select {
case <-t.C:
default:
}
}
return true
}

View File

@ -20,6 +20,7 @@ type link struct {
mutex sync.RWMutex // protects interfaces below mutex sync.RWMutex // protects interfaces below
interfaces map[linkInfo]*linkInterface interfaces map[linkInfo]*linkInterface
awdl awdl // AWDL interface support awdl awdl // AWDL interface support
// TODO timeout (to remove from switch), read from config.ReadTimeout
} }
type linkInfo struct { type linkInfo struct {
@ -78,8 +79,6 @@ func (l *link) create(msgIO linkInterfaceMsgIO, name, linkType, local, remote st
incoming: incoming, incoming: incoming,
force: force, force: force,
} }
//l.interfaces[intf.name] = &intf
//go intf.start()
return &intf, nil return &intf, nil
} }
@ -172,41 +171,49 @@ func (intf *linkInterface) handler() error {
go intf.peer.linkLoop() go intf.peer.linkLoop()
// Start the writer // Start the writer
signalReady := make(chan struct{}, 1) signalReady := make(chan struct{}, 1)
signalSent := make(chan struct{}, 1)
sendAck := make(chan struct{}, 1)
go func() { go func() {
defer close(signalReady) defer close(signalReady)
defer close(signalSent)
interval := 4 * time.Second interval := 4 * time.Second
timer := time.NewTimer(interval) tcpTimer := time.NewTimer(interval) // used for backwards compat with old tcp
clearTimer := func() { defer util.TimerStop(tcpTimer)
if !timer.Stop() { send := func(bs []byte) {
select { intf.msgIO.writeMsg(bs)
case <-timer.C: select {
default: case signalSent <- struct{}{}:
} default:
} }
} }
defer clearTimer()
for { for {
// First try to send any link protocol traffic // First try to send any link protocol traffic
select { select {
case msg := <-intf.peer.linkOut: case msg := <-intf.peer.linkOut:
intf.msgIO.writeMsg(msg) send(msg)
continue continue
default: default:
} }
// No protocol traffic to send, so reset the timer // No protocol traffic to send, so reset the timer
clearTimer() util.TimerStop(tcpTimer)
timer.Reset(interval) tcpTimer.Reset(interval)
// Now block until something is ready or the timer triggers keepalive traffic // Now block until something is ready or the timer triggers keepalive traffic
select { select {
case <-timer.C: case <-tcpTimer.C:
intf.msgIO.writeMsg(nil) intf.link.core.log.Debugf("Sending (legacy) keep-alive to %s: %s, source %s",
strings.ToUpper(intf.info.linkType), themString, intf.info.local)
send(nil)
case <-sendAck:
intf.link.core.log.Debugf("Sending ack to %s: %s, source %s",
strings.ToUpper(intf.info.linkType), themString, intf.info.local)
send(nil)
case msg := <-intf.peer.linkOut: case msg := <-intf.peer.linkOut:
intf.msgIO.writeMsg(msg) intf.msgIO.writeMsg(msg)
case msg, ok := <-out: case msg, ok := <-out:
if !ok { if !ok {
return return
} }
intf.msgIO.writeMsg(msg) send(msg)
util.PutBytes(msg) util.PutBytes(msg)
select { select {
case signalReady <- struct{}{}: case signalReady <- struct{}{}:
@ -217,27 +224,23 @@ func (intf *linkInterface) handler() error {
}() }()
//intf.link.core.switchTable.idleIn <- intf.peer.port // notify switch that we're idle //intf.link.core.switchTable.idleIn <- intf.peer.port // notify switch that we're idle
// Used to enable/disable activity in the switch // Used to enable/disable activity in the switch
signalAlive := make(chan struct{}, 1) signalAlive := make(chan bool, 1) // True = real packet, false = keep-alive
defer close(signalAlive) defer close(signalAlive)
go func() { go func() {
var isAlive bool var isAlive bool
var isReady bool var isReady bool
interval := 6 * time.Second // TODO set to ReadTimeout from the config, reset if it gets changed var ackTimerRunning bool
timer := time.NewTimer(interval) timeout := 6 * time.Second // TODO set to ReadTimeout from the config, reset if it gets changed
clearTimer := func() { ackTime := time.Second
if !timer.Stop() { timer := time.NewTimer(timeout)
select { defer util.TimerStop(timer)
case <-timer.C: ackTimer := time.NewTimer(ackTime)
default: defer util.TimerStop(ackTimer)
}
}
}
defer clearTimer()
for { for {
clearTimer() util.TimerStop(timer)
timer.Reset(interval) timer.Reset(timeout)
select { select {
case _, ok := <-signalAlive: case gotMsg, ok := <-signalAlive:
if !ok { if !ok {
return return
} }
@ -249,6 +252,24 @@ func (intf *linkInterface) handler() error {
intf.link.core.switchTable.idleIn <- intf.peer.port intf.link.core.switchTable.idleIn <- intf.peer.port
} }
} }
if gotMsg && !ackTimerRunning {
util.TimerStop(ackTimer)
ackTimer.Reset(ackTime)
ackTimerRunning = true
}
case _, ok := <-signalSent:
// Stop any running ack timer
if !ok {
return
}
util.TimerStop(ackTimer)
ackTimerRunning = false
case <-ackTimer.C:
// We haven't sent anything in the past ackTimeout, so signal a send of a nil packet
select {
case sendAck <- struct{}{}:
default:
}
case _, ok := <-signalReady: case _, ok := <-signalReady:
if !ok { if !ok {
return return
@ -275,7 +296,7 @@ func (intf *linkInterface) handler() error {
return err return err
} }
select { select {
case signalAlive <- struct{}{}: case signalAlive <- len(msg) > 0:
default: default:
} }
} }