mirror of
https://github.com/cwinfo/yggdrasil-go.git
synced 2024-11-10 06:20:26 +00:00
more link migration
This commit is contained in:
Arceliar
parent
b5b179904b
commit
bd3eaefb72
@ -554,7 +554,8 @@ func (c *Core) DEBUG_addAllowedEncryptionPublicKey(boxStr string) {
|
||||
func DEBUG_simLinkPeers(p, q *peer) {
|
||||
// Sets q.out() to point to p and starts p.start()
|
||||
goWorkers := func(source, dest *peer) {
|
||||
source.linkOut = make(chan []byte, 1)
|
||||
linkOut := make(chan []byte, 1)
|
||||
source.linkOut = func(bs []byte) { linkOut <- bs }
|
||||
send := make(chan []byte, 1)
|
||||
source.out = func(bss [][]byte) {
|
||||
for _, bs := range bss {
|
||||
@ -566,7 +567,7 @@ func DEBUG_simLinkPeers(p, q *peer) {
|
||||
var packets [][]byte
|
||||
for {
|
||||
select {
|
||||
case packet := <-source.linkOut:
|
||||
case packet := <-linkOut:
|
||||
packets = append(packets, packet)
|
||||
continue
|
||||
case packet := <-send:
|
||||
@ -583,7 +584,7 @@ func DEBUG_simLinkPeers(p, q *peer) {
|
||||
continue
|
||||
}
|
||||
select {
|
||||
case packet := <-source.linkOut:
|
||||
case packet := <-linkOut:
|
||||
packets = append(packets, packet)
|
||||
case packet := <-send:
|
||||
packets = append(packets, packet)
|
||||
|
||||
@ -199,7 +199,7 @@ func (intf *linkInterface) handler() error {
|
||||
intf.link.mutex.Lock()
|
||||
delete(intf.link.interfaces, intf.info)
|
||||
intf.link.mutex.Unlock()
|
||||
//close(intf.closed)
|
||||
close(intf.closed)
|
||||
}()
|
||||
intf.link.core.log.Debugln("DEBUG: registered interface for", intf.name)
|
||||
}
|
||||
@ -214,249 +214,22 @@ func (intf *linkInterface) handler() error {
|
||||
// More cleanup can go here
|
||||
intf.link.core.peers.removePeer(intf.peer.port)
|
||||
}()
|
||||
// Finish setting up the peer struct
|
||||
/*
|
||||
out := make(chan [][]byte, 1)
|
||||
defer close(out)
|
||||
intf.peer.out = func(msgs [][]byte) {
|
||||
defer func() { recover() }()
|
||||
out <- msgs
|
||||
}
|
||||
*/
|
||||
intf.peer.out = func(msgs [][]byte) {
|
||||
intf.writer.sendFrom(intf.peer, msgs, false)
|
||||
}
|
||||
intf.peer.linkOut = make(chan []byte, 1)
|
||||
go func() {
|
||||
// TODO fix this
|
||||
for bs := range intf.peer.linkOut {
|
||||
intf.peer.linkOut = func(bs []byte) {
|
||||
intf.writer.sendFrom(intf.peer, [][]byte{bs}, true)
|
||||
}
|
||||
}()
|
||||
themAddr := address.AddrForNodeID(crypto.GetNodeID(&intf.info.box))
|
||||
themAddrString := net.IP(themAddr[:]).String()
|
||||
themString := fmt.Sprintf("%s@%s", themAddrString, intf.info.remote)
|
||||
intf.link.core.log.Infof("Connected %s: %s, source %s",
|
||||
strings.ToUpper(intf.info.linkType), themString, intf.info.local)
|
||||
// Start the link loop
|
||||
// Start things
|
||||
go intf.peer.start()
|
||||
// Start the writer
|
||||
/*
|
||||
signalReady := make(chan struct{}, 1)
|
||||
signalSent := make(chan bool, 1)
|
||||
sendAck := make(chan struct{}, 1)
|
||||
sendBlocked := time.NewTimer(time.Second)
|
||||
defer util.TimerStop(sendBlocked)
|
||||
util.TimerStop(sendBlocked)
|
||||
go func() {
|
||||
defer close(signalReady)
|
||||
defer close(signalSent)
|
||||
interval := 4 * time.Second
|
||||
tcpTimer := time.NewTimer(interval) // used for backwards compat with old tcp
|
||||
defer util.TimerStop(tcpTimer)
|
||||
send := func(bss [][]byte) {
|
||||
sendBlocked.Reset(time.Second)
|
||||
size, _ := intf.msgIO.writeMsgs(bss)
|
||||
util.TimerStop(sendBlocked)
|
||||
select {
|
||||
case signalSent <- size > 0:
|
||||
default:
|
||||
}
|
||||
}
|
||||
for {
|
||||
// First try to send any link protocol traffic
|
||||
select {
|
||||
case msg := <-intf.peer.linkOut:
|
||||
send([][]byte{msg})
|
||||
continue
|
||||
default:
|
||||
}
|
||||
// No protocol traffic to send, so reset the timer
|
||||
util.TimerStop(tcpTimer)
|
||||
tcpTimer.Reset(interval)
|
||||
// Now block until something is ready or the timer triggers keepalive traffic
|
||||
select {
|
||||
case <-tcpTimer.C:
|
||||
intf.link.core.log.Tracef("Sending (legacy) keep-alive to %s: %s, source %s",
|
||||
strings.ToUpper(intf.info.linkType), themString, intf.info.local)
|
||||
send([][]byte{nil})
|
||||
case <-sendAck:
|
||||
intf.link.core.log.Tracef("Sending ack to %s: %s, source %s",
|
||||
strings.ToUpper(intf.info.linkType), themString, intf.info.local)
|
||||
send([][]byte{nil})
|
||||
case msg := <-intf.peer.linkOut:
|
||||
send([][]byte{msg})
|
||||
case msgs, ok := <-out:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
send(msgs)
|
||||
for _, msg := range msgs {
|
||||
util.PutBytes(msg)
|
||||
}
|
||||
select {
|
||||
case signalReady <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
//intf.link.core.log.Tracef("Sending packet to %s: %s, source %s",
|
||||
// strings.ToUpper(intf.info.linkType), themString, intf.info.local)
|
||||
}
|
||||
}
|
||||
}()
|
||||
//intf.link.core.switchTable.idleIn <- intf.peer.port // notify switch that we're idle
|
||||
// Used to enable/disable activity in the switch
|
||||
signalAlive := make(chan bool, 1) // True = real packet, false = keep-alive
|
||||
defer close(signalAlive)
|
||||
ret := make(chan error, 1) // How we signal the return value when multiple goroutines are involved
|
||||
go func() {
|
||||
var isAlive bool
|
||||
var isReady bool
|
||||
var sendTimerRunning bool
|
||||
var recvTimerRunning bool
|
||||
recvTime := 6 * time.Second // TODO set to ReadTimeout from the config, reset if it gets changed
|
||||
closeTime := 2 * switch_timeout // TODO or maybe this makes more sense for ReadTimeout?...
|
||||
sendTime := time.Second
|
||||
sendTimer := time.NewTimer(sendTime)
|
||||
defer util.TimerStop(sendTimer)
|
||||
recvTimer := time.NewTimer(recvTime)
|
||||
defer util.TimerStop(recvTimer)
|
||||
closeTimer := time.NewTimer(closeTime)
|
||||
defer util.TimerStop(closeTimer)
|
||||
for {
|
||||
//intf.link.core.log.Debugf("State of %s: %s, source %s :: isAlive %t isReady %t sendTimerRunning %t recvTimerRunning %t",
|
||||
// strings.ToUpper(intf.info.linkType), themString, intf.info.local,
|
||||
// isAlive, isReady, sendTimerRunning, recvTimerRunning)
|
||||
select {
|
||||
case gotMsg, ok := <-signalAlive:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
util.TimerStop(closeTimer)
|
||||
closeTimer.Reset(closeTime)
|
||||
util.TimerStop(recvTimer)
|
||||
recvTimerRunning = false
|
||||
isAlive = true
|
||||
if !isReady {
|
||||
// (Re-)enable in the switch
|
||||
intf.link.core.switchTable.RecvFrom(nil, func() {
|
||||
intf.link.core.switchTable._idleIn(intf.peer.port)
|
||||
})
|
||||
isReady = true
|
||||
}
|
||||
if gotMsg && !sendTimerRunning {
|
||||
// We got a message
|
||||
// Start a timer, if it expires then send a 0-sized ack to let them know we're alive
|
||||
util.TimerStop(sendTimer)
|
||||
sendTimer.Reset(sendTime)
|
||||
sendTimerRunning = true
|
||||
}
|
||||
if !gotMsg {
|
||||
intf.link.core.log.Tracef("Received ack from %s: %s, source %s",
|
||||
strings.ToUpper(intf.info.linkType), themString, intf.info.local)
|
||||
}
|
||||
case sentMsg, ok := <-signalSent:
|
||||
// Stop any running ack timer
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
util.TimerStop(sendTimer)
|
||||
sendTimerRunning = false
|
||||
if sentMsg && !recvTimerRunning {
|
||||
// We sent a message
|
||||
// Start a timer, if it expires and we haven't gotten any return traffic (including a 0-sized ack), then assume there's a problem
|
||||
util.TimerStop(recvTimer)
|
||||
recvTimer.Reset(recvTime)
|
||||
recvTimerRunning = true
|
||||
}
|
||||
case _, ok := <-signalReady:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if !isAlive {
|
||||
// Disable in the switch
|
||||
isReady = false
|
||||
} else {
|
||||
// Keep enabled in the switch
|
||||
intf.link.core.switchTable.RecvFrom(nil, func() {
|
||||
intf.link.core.switchTable._idleIn(intf.peer.port)
|
||||
})
|
||||
isReady = true
|
||||
}
|
||||
case <-sendBlocked.C:
|
||||
// We blocked while trying to send something
|
||||
isReady = false
|
||||
intf.link.core.switchTable.blockPeer(intf.peer.port)
|
||||
case <-sendTimer.C:
|
||||
// We haven't sent anything, so signal a send of a 0 packet to let them know we're alive
|
||||
select {
|
||||
case sendAck <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
case <-recvTimer.C:
|
||||
// We haven't received anything, so assume there's a problem and don't return this node to the switch until they start responding
|
||||
isAlive = false
|
||||
intf.link.core.switchTable.blockPeer(intf.peer.port)
|
||||
case <-closeTimer.C:
|
||||
// We haven't received anything in a really long time, so things have died at the switch level and then some...
|
||||
// Just close the connection at this point...
|
||||
select {
|
||||
case ret <- errors.New("timeout"):
|
||||
default:
|
||||
}
|
||||
intf.msgIO.close()
|
||||
}
|
||||
}
|
||||
}()
|
||||
// Run reader loop
|
||||
var helper phony.Inbox
|
||||
done := make(chan struct{})
|
||||
var helperFunc func()
|
||||
helperFunc = func() {
|
||||
// The helper reads in a loop and sends to the peer
|
||||
// It loops by sending itself a message, which can be delayed by backpressure
|
||||
// So if the peer is busy, backpressure will pause reading until the peer catches up
|
||||
msg, err := intf.msgIO.readMsg()
|
||||
if len(msg) > 0 {
|
||||
// TODO rewrite this if the link becomes an actor
|
||||
intf.peer.handlePacketFrom(&helper, msg)
|
||||
}
|
||||
if err != nil {
|
||||
if err != io.EOF {
|
||||
select {
|
||||
case ret <- err:
|
||||
default:
|
||||
}
|
||||
}
|
||||
close(done)
|
||||
return
|
||||
}
|
||||
select {
|
||||
case signalAlive <- len(msg) > 0:
|
||||
default:
|
||||
}
|
||||
// Now try to read again
|
||||
helper.RecvFrom(nil, helperFunc)
|
||||
}
|
||||
|
||||
// Start the read loop
|
||||
helper.RecvFrom(nil, helperFunc)
|
||||
<-done // Wait for the helper to exit
|
||||
////////////////////////////////////////////////////////////////////////////////
|
||||
// Remember to set `err` to something useful before returning
|
||||
select {
|
||||
case err = <-ret:
|
||||
intf.link.core.log.Infof("Disconnected %s: %s, source %s; error: %s",
|
||||
strings.ToUpper(intf.info.linkType), themString, intf.info.local, err)
|
||||
default:
|
||||
err = nil
|
||||
intf.link.core.log.Infof("Disconnected %s: %s, source %s",
|
||||
strings.ToUpper(intf.info.linkType), themString, intf.info.local)
|
||||
}
|
||||
*/
|
||||
// Start the reader
|
||||
intf.reader.RecvFrom(nil, intf.reader._read)
|
||||
// Wait for the reader to finish
|
||||
err = <- intf.reader.err
|
||||
err = <-intf.reader.err
|
||||
if err != nil {
|
||||
intf.link.core.log.Infof("Disconnected %s: %s, source %s; error: %s",
|
||||
strings.ToUpper(intf.info.linkType), themString, intf.info.local, err)
|
||||
@ -469,22 +242,11 @@ func (intf *linkInterface) handler() error {
|
||||
|
||||
////////////////////////////////////////////////////////////////////////////////
|
||||
|
||||
|
||||
/*
|
||||
phony.Inbox // Protects the below
|
||||
sendTimer *time.Timer // Fires to signal that sending is blocked
|
||||
stallTimer *time.Time // Fires to signal that no incoming traffic (including keep-alive) has been seen
|
||||
recvTimer *time.Timer // Fires to send keep-alive traffic
|
||||
closeTimer *time.Timer // Fires when the link has been idle so long we need to close it
|
||||
inSwitch bool // True if the switch is tracking this link
|
||||
stalled bool // True if we haven't been receiving any response traffic
|
||||
*/
|
||||
|
||||
const (
|
||||
sendBlockedTime = time.Second // How long to wait before deciding a send is blocked
|
||||
keepAliveTime = 2*time.Second // How long to wait before sending a keep-alive response if we have no real traffic to send
|
||||
stallTime = 6*time.Second // How long to wait for response traffic before deciding the connection has stalled
|
||||
closeTime = 2*switch_timeout // How long to wait before closing the link
|
||||
keepAliveTime = 2 * time.Second // How long to wait before sending a keep-alive response if we have no real traffic to send
|
||||
stallTime = 6 * time.Second // How long to wait for response traffic before deciding the connection has stalled
|
||||
closeTime = 2 * switch_timeout // How long to wait before closing the link
|
||||
)
|
||||
|
||||
// notify the intf that we're currently sending
|
||||
@ -553,7 +315,6 @@ func (intf *linkInterface) notifyStalled() {
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
// reset the close timer
|
||||
func (intf *linkInterface) notifyReading(from phony.Actor) {
|
||||
intf.RecvFrom(from, func() {
|
||||
@ -567,7 +328,6 @@ func (intf *linkInterface) notifyReading(from phony.Actor) {
|
||||
// wake up the link if it was stalled, and (if size > 0) prepare to send keep-alive traffic
|
||||
func (intf *linkInterface) notifyReadFrom(from phony.Actor, size int) {
|
||||
intf.RecvFrom(from, func() {
|
||||
intf.link.core.log.Printf("DEBUG notifyReadFrom: inSwitch %v, stalled %v\n", intf.inSwitch, intf.stalled)
|
||||
if intf.stallTimer != nil {
|
||||
intf.stallTimer.Stop()
|
||||
intf.stallTimer = nil
|
||||
@ -607,7 +367,6 @@ func (w *linkWriter) sendFrom(from phony.Actor, bss [][]byte, isLinkTraffic bool
|
||||
w.intf.notifySending(size, isLinkTraffic)
|
||||
w.intf.msgIO.writeMsgs(bss)
|
||||
w.intf.notifySent(size, isLinkTraffic)
|
||||
w.intf.link.core.log.Println("DEBUG: wrote something, size:", size, "isLinkTraffic:", isLinkTraffic)
|
||||
})
|
||||
}
|
||||
|
||||
@ -622,14 +381,13 @@ type linkReader struct {
|
||||
func (r *linkReader) _read() {
|
||||
r.intf.notifyReading(r)
|
||||
msg, err := r.intf.msgIO.readMsg()
|
||||
r.intf.link.core.log.Println("DEBUG read something")
|
||||
r.intf.notifyReadFrom(r, len(msg))
|
||||
if len(msg) > 0 {
|
||||
r.intf.peer.handlePacketFrom(r, msg)
|
||||
}
|
||||
if err != nil {
|
||||
if err != io.EOF {
|
||||
r.err<-err
|
||||
r.err <- err
|
||||
}
|
||||
close(r.err)
|
||||
return
|
||||
@ -637,4 +395,3 @@ func (r *linkReader) _read() {
|
||||
// Now try to read again
|
||||
r.RecvFrom(nil, r._read)
|
||||
}
|
||||
|
||||
|
||||
@ -103,7 +103,7 @@ type peer struct {
|
||||
linkShared crypto.BoxSharedKey
|
||||
endpoint string
|
||||
firstSeen time.Time // To track uptime for getPeers
|
||||
linkOut (chan []byte) // used for protocol traffic (to bypass queues)
|
||||
linkOut func([]byte) // used for protocol traffic (bypasses the switch)
|
||||
dinfo *dhtInfo // used to keep the DHT working
|
||||
out func([][]byte) // Set up by whatever created the peers struct, used to send packets to other nodes
|
||||
done (chan struct{}) // closed to exit the linkLoop
|
||||
@ -263,8 +263,6 @@ func (p *peer) _sendPackets(packets [][]byte) {
|
||||
p.out(packets)
|
||||
}
|
||||
|
||||
var peerLinkOutHelper phony.Inbox
|
||||
|
||||
// This wraps the packet in the inner (ephemeral) and outer (permanent) crypto layers.
|
||||
// It sends it to p.linkOut, which bypasses the usual packet queues.
|
||||
func (p *peer) _sendLinkPacket(packet []byte) {
|
||||
@ -280,13 +278,7 @@ func (p *peer) _sendLinkPacket(packet []byte) {
|
||||
Payload: bs,
|
||||
}
|
||||
packet = linkPacket.encode()
|
||||
// TODO replace this with a message send if/when the link becomes an actor
|
||||
peerLinkOutHelper.RecvFrom(nil, func() {
|
||||
select {
|
||||
case p.linkOut <- packet:
|
||||
case <-p.done:
|
||||
}
|
||||
})
|
||||
p.linkOut(packet)
|
||||
}
|
||||
|
||||
// Decrypts the outer (permanent) and inner (ephemeral) crypto layers on link traffic.
|
||||
|
||||
@ -250,7 +250,7 @@ func (t *switchTable) cleanRoot() {
|
||||
t.core.router.reset(nil)
|
||||
}
|
||||
t.data.locator = switchLocator{root: t.key, tstamp: now.Unix()}
|
||||
t.core.peers.sendSwitchMsgs(nil) // TODO update if/when the switch becomes an actor
|
||||
t.core.peers.sendSwitchMsgs(t)
|
||||
}
|
||||
}
|
||||
|
||||
@ -517,7 +517,7 @@ func (t *switchTable) unlockedHandleMsg(msg *switchMsg, fromPort switchPort, rep
|
||||
}
|
||||
t.data.locator = sender.locator
|
||||
t.parent = sender.port
|
||||
t.core.peers.sendSwitchMsgs(nil) // TODO update if/when the switch becomes an actor
|
||||
t.core.peers.sendSwitchMsgs(t)
|
||||
}
|
||||
if doUpdate {
|
||||
t.updater.Store(&sync.Once{})
|
||||
|
||||
Loading…
Reference in New Issue
Block a user