diff --git a/connmgr/connmanager.go b/connmgr/connmanager.go index 8b6b8391..41000045 100644 --- a/connmgr/connmanager.go +++ b/connmgr/connmanager.go @@ -176,6 +176,12 @@ type handleFailed struct { err error } +// handleCancelPending is used to remove failing connections from retries. +type handleCancelPending struct { + addr net.Addr + done chan struct{} +} + // ConnManager provides a manager to handle network connections. type ConnManager struct { // The following variables must only be used atomically. @@ -346,6 +352,28 @@ out: log.Debugf("Failed to connect to %v: %v", connReq, msg.err) cm.handleFailedConn(connReq) + + case handleCancelPending: + found := false + var idToRemove uint64 + connReq := &ConnReq{} + for id, req := range pending { + if msg.addr.String() == req.Addr.String() { + idToRemove = id + connReq = req + found = true + break + } + + } + if found { + delete(pending, idToRemove) + connReq.updateState(ConnCanceled) + log.Debugf("Canceled pending connection to %v", msg.addr) + } else { + log.Errorf("Did not find connection to cancel at address %v", msg.addr) + } + close(msg.done) } case <-cm.quit: @@ -496,6 +524,28 @@ func (cm *ConnManager) Remove(id uint64) { } } +// CancelPending removes the connection corresponding to the given address +// from the list of pending failed connections. +func (cm *ConnManager) CancelPending(addr net.Addr) { + if atomic.LoadInt32(&cm.stop) != 0 { + return + } + done := make(chan struct{}) + + select { + case cm.requests <- handleCancelPending{addr, done}: + case <-cm.quit: + } + + // Wait for the connection to be removed from the conn manager's + // internal state. + select { + case <-done: + case <-cm.quit: + return + } +} + // listenHandler accepts incoming connections on a given listener. It must be // run as a goroutine. func (cm *ConnManager) listenHandler(listener net.Listener) { diff --git a/server.go b/server.go index a7a56751..25119283 100644 --- a/server.go +++ b/server.go @@ -1865,6 +1865,11 @@ type removeNodeMsg struct { reply chan error } +type cancelPendingMsg struct { + addr string + reply chan error +} + // handleQuery is the central handler for all queries and commands from other // goroutines related to peer state. func (s *server) handleQuery(state *peerState, querymsg interface{}) { @@ -1940,6 +1945,14 @@ func (s *server) handleQuery(state *peerState, querymsg interface{}) { } else { msg.reply <- errors.New("peer not found") } + case cancelPendingMsg: + netAddr, err := addrStringToNetAddr(msg.addr) + if err != nil { + msg.reply <- err + return + } + s.connManager.CancelPending(netAddr) + msg.reply <- nil case getOutboundGroup: count, ok := state.outboundGroups[msg.key] if ok { @@ -2316,6 +2329,24 @@ func (s *server) RemoveNodeByAddr(addr string) error { reply: replyChan, } + err := <-replyChan + if err != nil { + // This connection may still be pending, cancel it. + return s.cancelPendingConnection(addr) + } + return nil +} + +// cancelPendingConnection removes an address from the list of +// pending connections. +func (s *server) cancelPendingConnection(addr string) error { + replyChan := make(chan error) + + s.query <- cancelPendingMsg{ + addr: addr, + reply: replyChan, + } + return <-replyChan }