connmgr: Allow pending outbound conn removal.

Modify `dcrctl addnode <ip_addr:port> remove` to remove the address from
the list of pending connections, preventing unnecessary retries.

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) {

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
 }