prevent premature peer gc
Caused by a race between adding the connection to our map and receiving a topology gossip on it. We remove the race by making AddConnection synchronous and invoking it prior to entering the tcp receive loop. Fixes #367.
This commit is contained in:
1 parent
5892cae5d0
commit
2d135f3da0
2 files changed
+26
-6
No files matched your search
+21
-3
@@ -237,6 +237,17 @@ func (conn *LocalConnection) run(queryChan <-chan *ConnectionInteraction, accept
|
||||
}
|
||||
log.Printf("->[%s] completed handshake with %s\n", conn.remoteTCPAddr, conn.remote.Name)
|
||||
|
||||
// We invoke AddConnection in the same goroutine that subsequently
|
||||
// becomes the tcp receive loop, rather than outside, because a)
|
||||
// the ordering relative to the receive loop is the only one that
|
||||
// matters [1], b) it prevents unnecessary delays in entering the
|
||||
// main connection loop, and c) it guards against potential
|
||||
// deadlocks.
|
||||
go func () {
|
||||
conn.Router.Ourself.AddConnection(conn)
|
||||
conn.receiveTCP(dec)
|
||||
}()
|
||||
|
||||
heartbeatFrameBytes := make([]byte, EthernetOverhead+8)
|
||||
binary.BigEndian.PutUint64(heartbeatFrameBytes[EthernetOverhead:], conn.uid)
|
||||
conn.heartbeatFrame = &ForwardedFrame{
|
||||
@@ -244,9 +255,6 @@ func (conn *LocalConnection) run(queryChan <-chan *ConnectionInteraction, accept
|
||||
dstPeer: conn.remote,
|
||||
frame: heartbeatFrameBytes}
|
||||
|
||||
go conn.receiveTCP(dec)
|
||||
conn.Router.Ourself.AddConnection(conn)
|
||||
|
||||
if conn.remoteUDPAddr != nil {
|
||||
if err := conn.sendFastHeartbeats(); err != nil {
|
||||
conn.log("connection shutting down due to error:", err)
|
||||
@@ -262,6 +270,16 @@ func (conn *LocalConnection) run(queryChan <-chan *ConnectionInteraction, accept
|
||||
conn.log("connection shutting down")
|
||||
}
|
||||
}
|
||||
// [1] In the absence of any indirect connectivity to the remote peer,
|
||||
// the first we hear about it (and any peers reachable from it) is
|
||||
// through topology gossip it sends us on the connection. We must
|
||||
// ensure that the connection has been added to Ourself prior to
|
||||
// processing any such gossip, otherwise we risk immediately gc'ing
|
||||
// part of that newly received portion of the topology (though not the
|
||||
// remote peer itself, since that will have a positive ref count),
|
||||
// leaving behind dangling references to peers. Therefore we invoke
|
||||
// AddConnection, which is *synchronous*, before entering the tcp
|
||||
// receive loop.
|
||||
|
||||
func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction) (err error) {
|
||||
terminate := false
|
||||
|
||||
@@ -123,12 +123,13 @@ const (
|
||||
PSendProtocolMsg
|
||||
)
|
||||
|
||||
// Async: rely on the peer to shut us down if we shouldn't be adding
|
||||
// ourselves, so therefore this can be async
|
||||
// Sync.
|
||||
func (peer *LocalPeer) AddConnection(conn *LocalConnection) {
|
||||
resultChan := make(chan interface{})
|
||||
peer.queryChan <- &PeerInteraction{
|
||||
Interaction: Interaction{code: PAddConnection},
|
||||
Interaction: Interaction{code: PAddConnection, resultChan: resultChan},
|
||||
payload: conn}
|
||||
<-resultChan
|
||||
}
|
||||
|
||||
// Async.
|
||||
@@ -170,6 +171,7 @@ func (peer *LocalPeer) queryLoop(queryChan <-chan *PeerInteraction) {
|
||||
if peer.handleAddConnection(conn) {
|
||||
conn.log("connection added")
|
||||
}
|
||||
query.resultChan <- nil
|
||||
case PConnectionEstablished:
|
||||
conn := query.payload.(*LocalConnection)
|
||||
if peer.handleConnectionEstablished(conn) {
|
||||
|
||||
Reference in new issue
Block a user