log all connection lifecycle events on local peer
There were situations, such as deleting another connection as result of it losing a tie-break, where lifecycle events of connections on the local peer weren't logged. Fixing that entails moving the logging closer to (i.e. right after) the state change action. This in turn requires promotion of logging to the Connection interface instead of it LocalConnection. Which in turn requires a change from 'log' to 'Log', for consistency with other interface methods. Fixes #449.
This commit is contained in:
1 parent
123e8bbb28
commit
6f89ead9bd
4 files changed
+30
-38
No files matched your search
+11
-10
@@ -17,6 +17,7 @@ type Connection interface {
|
||||
RemoteTCPAddr() string
|
||||
Established() bool
|
||||
Shutdown(error)
|
||||
Log(args ...interface{})
|
||||
}
|
||||
|
||||
type ConnectionTieBreak int
|
||||
@@ -96,6 +97,10 @@ func (conn *RemoteConnection) Established() bool {
|
||||
func (conn *RemoteConnection) Shutdown(error) {
|
||||
}
|
||||
|
||||
func (conn *RemoteConnection) Log(args ...interface{}) {
|
||||
log.Println(append(append([]interface{}{}, fmt.Sprintf("->[%s]:", conn.remote.Name)), args...)...)
|
||||
}
|
||||
|
||||
func (conn *RemoteConnection) String() string {
|
||||
from := "<nil>"
|
||||
if conn.local != nil {
|
||||
@@ -164,7 +169,7 @@ func (conn *LocalConnection) setEffectivePMTU(pmtu int) {
|
||||
defer conn.Unlock()
|
||||
if conn.effectivePMTU != pmtu {
|
||||
conn.effectivePMTU = pmtu
|
||||
conn.log("Effective PMTU set to", pmtu)
|
||||
conn.Log("Effective PMTU set to", pmtu)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -177,10 +182,6 @@ func (conn *LocalConnection) setStackFrag(frag bool) {
|
||||
conn.stackFrag = frag
|
||||
}
|
||||
|
||||
func (conn *LocalConnection) log(args ...interface{}) {
|
||||
log.Println(append(append([]interface{}{}, fmt.Sprintf("->[%s]:", conn.remote.Name)), args...)...)
|
||||
}
|
||||
|
||||
// Send directly, not via the Actor. If it goes via the Actor we can
|
||||
// get a deadlock where LocalConnection is blocked talking to
|
||||
// LocalPeer and LocalPeer is blocked trying send a ProtocolMsg via
|
||||
@@ -271,7 +272,7 @@ func (conn *LocalConnection) run(queryChan <-chan *ConnectionInteraction, finish
|
||||
|
||||
if conn.remoteUDPAddr != nil {
|
||||
if err := conn.sendFastHeartbeats(); err != nil {
|
||||
conn.log("connection shutting down due to error:", err)
|
||||
conn.Log("connection shutting down due to error:", err)
|
||||
return
|
||||
}
|
||||
}
|
||||
@@ -279,9 +280,9 @@ func (conn *LocalConnection) run(queryChan <-chan *ConnectionInteraction, finish
|
||||
conn.heartbeatTimeout = time.NewTimer(HeartbeatTimeout)
|
||||
|
||||
if err := conn.queryLoop(queryChan); err != nil {
|
||||
conn.log("connection shutting down due to error:", err)
|
||||
conn.Log("connection shutting down due to error:", err)
|
||||
} else {
|
||||
conn.log("connection shutting down")
|
||||
conn.Log("connection shutting down")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -439,7 +440,7 @@ func (conn *LocalConnection) receiveTCP(decoder *gob.Decoder) {
|
||||
break
|
||||
}
|
||||
if len(msg) < 1 {
|
||||
conn.log("ignoring blank msg")
|
||||
conn.Log("ignoring blank msg")
|
||||
continue
|
||||
}
|
||||
if err = conn.handleProtocolMsg(ProtocolTag(msg[0]), msg[1:]); err != nil {
|
||||
@@ -479,7 +480,7 @@ func (conn *LocalConnection) handleProtocolMsg(tag ProtocolTag, payload []byte)
|
||||
case ProtocolGossip:
|
||||
return conn.Router.handleGossip(payload, deliverGossip)
|
||||
default:
|
||||
conn.log("ignoring unknown protocol tag:", tag)
|
||||
conn.Log("ignoring unknown protocol tag:", tag)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
+3
-3
@@ -95,7 +95,7 @@ func (conn *LocalConnection) Forward(df bool, frame *ForwardedFrame, dec *Ethern
|
||||
conn.RUnlock()
|
||||
|
||||
if forwardChan == nil || forwardChanDF == nil {
|
||||
conn.log("Cannot forward frame yet - awaiting contact")
|
||||
conn.Log("Cannot forward frame yet - awaiting contact")
|
||||
return nil
|
||||
}
|
||||
// We could use non-blocking channel sends here, i.e. drop frames
|
||||
@@ -272,7 +272,7 @@ func (fwd *Forwarder) run() {
|
||||
fwd.pmtuVerified = true
|
||||
fwd.maxPayload = epmtu + fwd.effectiveOverhead() - UDPOverhead
|
||||
fwd.conn.setEffectivePMTU(epmtu)
|
||||
fwd.conn.log("Effective PMTU verified at", epmtu)
|
||||
fwd.conn.Log("Effective PMTU verified at", epmtu)
|
||||
}
|
||||
case frame = <-fwd.ch:
|
||||
if !fwd.appendFrame(frame) {
|
||||
@@ -368,5 +368,5 @@ func (fwd *Forwarder) drain() {
|
||||
}
|
||||
|
||||
func (fwd *Forwarder) logDrop(frame *ForwardedFrame) {
|
||||
fwd.conn.log("Dropping too big frame during forwarding: frame len:", len(frame.frame), "; effective PMTU:", fwd.maxPayload+UDPOverhead-fwd.effectiveOverhead())
|
||||
fwd.conn.Log("Dropping too big frame during forwarding: frame len:", len(frame.frame), "; effective PMTU:", fwd.maxPayload+UDPOverhead-fwd.effectiveOverhead())
|
||||
}
|
||||
+15
-24
@@ -162,21 +162,12 @@ func (peer *LocalPeer) queryLoop(queryChan <-chan *PeerInteraction) {
|
||||
}
|
||||
switch query.code {
|
||||
case PAddConnection:
|
||||
conn := query.payload.(*LocalConnection)
|
||||
if peer.handleAddConnection(conn) {
|
||||
conn.log("connection added")
|
||||
}
|
||||
peer.handleAddConnection(query.payload.(*LocalConnection))
|
||||
query.resultChan <- nil
|
||||
case PConnectionEstablished:
|
||||
conn := query.payload.(*LocalConnection)
|
||||
if peer.handleConnectionEstablished(conn) {
|
||||
conn.log("connection fully established")
|
||||
}
|
||||
peer.handleConnectionEstablished(query.payload.(*LocalConnection))
|
||||
case PDeleteConnection:
|
||||
conn := query.payload.(*LocalConnection)
|
||||
if peer.handleDeleteConnection(conn) {
|
||||
conn.log("connection deleted")
|
||||
}
|
||||
peer.handleDeleteConnection(query.payload.(*LocalConnection))
|
||||
query.resultChan <- nil
|
||||
}
|
||||
case <-gossipTimer:
|
||||
@@ -185,7 +176,7 @@ func (peer *LocalPeer) queryLoop(queryChan <-chan *PeerInteraction) {
|
||||
}
|
||||
}
|
||||
|
||||
func (peer *LocalPeer) handleAddConnection(conn Connection) bool {
|
||||
func (peer *LocalPeer) handleAddConnection(conn Connection) {
|
||||
if peer.Peer != conn.Local() {
|
||||
log.Fatal("Attempt made to add connection to peer where peer is not the source of connection")
|
||||
}
|
||||
@@ -197,7 +188,7 @@ func (peer *LocalPeer) handleAddConnection(conn Connection) bool {
|
||||
// deliberately non symmetrical
|
||||
if dupConn, found := peer.connections[toName]; found {
|
||||
if dupConn == conn {
|
||||
return false
|
||||
return
|
||||
}
|
||||
switch conn.BreakTie(dupConn) {
|
||||
case TieBreakWon:
|
||||
@@ -205,39 +196,39 @@ func (peer *LocalPeer) handleAddConnection(conn Connection) bool {
|
||||
peer.handleDeleteConnection(dupConn)
|
||||
case TieBreakLost:
|
||||
conn.Shutdown(dupErr)
|
||||
return false
|
||||
return
|
||||
case TieBreakTied:
|
||||
// oh good grief. Sod it, just kill both of them.
|
||||
conn.Shutdown(dupErr)
|
||||
dupConn.Shutdown(dupErr)
|
||||
peer.handleDeleteConnection(dupConn)
|
||||
return false
|
||||
return
|
||||
}
|
||||
}
|
||||
if err := peer.checkConnectionLimit(); err != nil {
|
||||
conn.Shutdown(err)
|
||||
return false
|
||||
return
|
||||
}
|
||||
peer.addConnection(conn)
|
||||
conn.Log("connection added")
|
||||
peer.Router.SendAllGossipDown(conn)
|
||||
peer.broadcastPeerUpdate(conn.Remote())
|
||||
return true
|
||||
}
|
||||
|
||||
func (peer *LocalPeer) handleConnectionEstablished(conn Connection) bool {
|
||||
func (peer *LocalPeer) handleConnectionEstablished(conn Connection) {
|
||||
if peer.Peer != conn.Local() {
|
||||
log.Fatal("Peer informed of active connection where peer is not the source of connection")
|
||||
}
|
||||
if dupConn, found := peer.connections[conn.Remote().Name]; !found || conn != dupConn {
|
||||
conn.Shutdown(fmt.Errorf("Cannot set unknown connection active"))
|
||||
return false
|
||||
return
|
||||
}
|
||||
peer.connectionEstablished(conn)
|
||||
conn.Log("connection fully established")
|
||||
peer.broadcastPeerUpdate()
|
||||
return true
|
||||
}
|
||||
|
||||
func (peer *LocalPeer) handleDeleteConnection(conn Connection) bool {
|
||||
func (peer *LocalPeer) handleDeleteConnection(conn Connection) {
|
||||
if peer.Peer != conn.Local() {
|
||||
log.Fatal("Attempt made to delete connection from peer where peer is not the source of connection")
|
||||
}
|
||||
@@ -246,14 +237,14 @@ func (peer *LocalPeer) handleDeleteConnection(conn Connection) bool {
|
||||
}
|
||||
toName := conn.Remote().Name
|
||||
if connFound, found := peer.connections[toName]; !found || connFound != conn {
|
||||
return false
|
||||
return
|
||||
}
|
||||
peer.deleteConnection(conn)
|
||||
conn.Log("connection deleted")
|
||||
// Must do garbage collection first to ensure we don't send out an
|
||||
// update with unreachable peers (can cause looping)
|
||||
peer.Router.Peers.GarbageCollect()
|
||||
peer.broadcastPeerUpdate()
|
||||
return true
|
||||
}
|
||||
|
||||
func (peer *LocalPeer) broadcastPeerUpdate(peers ...*Peer) {
|
||||
|
||||
+1
-1
@@ -259,7 +259,7 @@ func (router *Router) udpReader(conn *net.UDPConn, po PacketSink) {
|
||||
if pde.Fatal {
|
||||
relayConn.Shutdown(pde)
|
||||
} else {
|
||||
relayConn.log(pde.Error())
|
||||
relayConn.Log(pde.Error())
|
||||
}
|
||||
} else {
|
||||
checkWarn(err)
|
||||
|
||||
Reference in new issue
Block a user