diff --git a/p2p/server.go b/p2p/server.go index 8f768bdffe..ce108f78ab 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -21,6 +21,7 @@ const ( defaultDialTimeout = 15 * time.Second refreshPeersInterval = 30 * time.Second staticPeerCheckInterval = 15 * time.Second + dialFailureCooldown = 30 * time.Second // Maximum number of concurrently handshaking inbound connections. maxAcceptConns = 50 @@ -113,6 +114,7 @@ type Server struct { lock sync.RWMutex // protects running, peers and the trust fields running bool peers map[discover.NodeID]*Peer + throttle map[discover.NodeID]bool // List of recently failed connections staticNodes map[discover.NodeID]*discover.Node // Map of currently maintained static remote nodes staticDial chan *discover.Node // Dial request channel reserved for the static nodes staticCycle time.Duration // Overrides staticPeerCheckInterval, used for testing @@ -217,6 +219,7 @@ func (srv *Server) Start() (err error) { } srv.quit = make(chan struct{}) srv.peers = make(map[discover.NodeID]*Peer) + srv.throttle = make(map[discover.NodeID]bool) // Create the current trust maps, and the associated dialing channel srv.trustedNodes = make(map[discover.NodeID]bool) @@ -491,12 +494,34 @@ func (srv *Server) dialNode(dest *discover.Node) { // dialNode, so we need to count it down again. startPeer also // does that when an error occurs. srv.peerWG.Done() - glog.V(logger.Detail).Infof("dial error: %v", err) + + // Throttle the node to prevent hammering it with connections + glog.V(logger.Detail).Infof("Dialing %v failed: %v, throttling", dest, err) + srv.throttleNode(dest.ID) + return } srv.startPeer(conn, dest) } +// throttleNode temporarily throttles a node, disabling both egress as well as +// ingress connections to/from that particular node. +func (srv *Server) throttleNode(id discover.NodeID) { + // Mark the node as throttled + srv.lock.Lock() + srv.throttle[id] = true + srv.lock.Unlock() + + // Wait a while, after which discard the throttling + go func() { + <-time.After(dialFailureCooldown) + + srv.lock.Lock() + delete(srv.throttle, id) + srv.lock.Unlock() + }() +} + func (srv *Server) startPeer(fd net.Conn, dest *discover.Node) { // TODO: handle/store session token @@ -556,7 +581,11 @@ func (srv *Server) runPeer(p *Peer) { srv.newPeerHook(p) } discreason := p.run() + + // Drop the node, and throttle it to prevent hammering srv.removePeer(p) + srv.throttleNode(p.ID()) + glog.V(logger.Debug).Infof("Removed %v (%v)\n", p, discreason) srvjslog.LogJson(&logger.P2PDisconnected{ RemoteId: p.ID().String(), @@ -596,6 +625,8 @@ func (srv *Server) checkPeer(id discover.NodeID) (bool, DiscReason) { return false, DiscAlreadyConnected case id == srv.ntab.Self().ID: return false, DiscSelf + case srv.throttle[id]: + return false, DiscUselessPeer default: return true, 0 } diff --git a/p2p/server_test.go b/p2p/server_test.go index 55fc815724..918ea64df4 100644 --- a/p2p/server_test.go +++ b/p2p/server_test.go @@ -79,7 +79,7 @@ func TestServerDial(t *testing.T) { // run a one-shot TCP server to handle the connection. listener, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { - t.Fatalf("could not setup listener: %v") + t.Fatalf("could not setup listener: %v", err) } defer listener.Close() accepted := make(chan net.Conn) @@ -487,6 +487,116 @@ func TestServerMaxPendingAccepts(t *testing.T) { } } +// Tests that a failed dial will temporarily throttle a peer. +func TestServerOutboundThrottling(t *testing.T) { + defer testlog(t).detach() + + // Start a simple test server + server := startTestServer(t, nil) + defer server.Stop() + + // Request a failing dial from the server and check throttling + node := &discover.Node{ + ID: discover.PubkeyID(&newkey().PublicKey), + IP: []byte{127, 0, 0, 1}, + TCP: 65535, + } + server.staticDial <- node + + time.Sleep(100 * time.Millisecond) + if throttle := server.throttle[node.ID]; !throttle { + t.Fatalf("throttling mismatch: have %v, want %v", throttle, true) + } + // Open a one shot TCP server to listen for valid dialing + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("failed to setup listener: %v", err) + } + defer listener.Close() + + connected := make(chan net.Conn) + go func() { + conn, err := listener.Accept() + if err == nil { + conn.Close() + connected <- conn + } + }() + time.Sleep(100 * time.Millisecond) + + // Request a re-dial from the server, make sure this also fails + addr := listener.Addr().(*net.TCPAddr) + node.IP, node.TCP = addr.IP, uint16(addr.Port) + server.staticDial <- node + + select { + case peer := <-connected: + t.Fatalf("throttled peer connected: %v", peer) + + case <-time.After(100 * time.Millisecond): + // Ok, peer not accepted + } +} + +// Tests that a failed peer will get temporarily throttled. +func TestServerInboundThrottling(t *testing.T) { + defer testlog(t).detach() + + // Start a test server and a peer sink for synchronization + started := make(chan *Peer) + server := &Server{ + ListenAddr: "127.0.0.1:0", + PrivateKey: newkey(), + MaxPeers: 10, + NoDial: true, + newPeerHook: func(p *Peer) { started <- p }, + } + if err := server.Start(); err != nil { + t.Fatal("failed to start test server: %v", err) + } + defer server.Stop() + + // Create a new peer identity to check throttling with + dialer := &net.Dialer{Deadline: time.Now().Add(3 * time.Second)} + key := newkey() + + // Connect with a new peer, run the handshake and disconnect + connection, err := dialer.Dial("tcp", server.ListenAddr) + if err != nil { + t.Fatalf("failed to dial server: %v", err) + } + shake := &protoHandshake{Version: baseProtocolVersion, ID: discover.PubkeyID(&key.PublicKey)} + if _, err = setupConn(connection, key, shake, server.Self(), false, server.trustedNodes); err != nil { + t.Fatalf("failed to run handshake: %v", err) + } + <-started + connection.Close() + server.peerWG.Wait() + + // Check that throttling has been enabled + if throttle := server.throttle[shake.ID]; !throttle { + t.Fatalf("throttling mismatch: have %v, want %v", throttle, true) + } + // Try to reconnect with the same peer, run the handshake and verify throttling + connection, err = dialer.Dial("tcp", server.ListenAddr) + if err != nil { + t.Fatalf("failed to re-dial server: %v", err) + } + defer connection.Close() + + shake = &protoHandshake{Version: baseProtocolVersion, ID: discover.PubkeyID(&key.PublicKey)} + if _, err = setupConn(connection, key, shake, server.Self(), false, server.trustedNodes); err != nil { + t.Fatalf("failed to re-run handshake: %v", err) + } + select { + case peer := <-started: + t.Fatalf("throttled peer accepted: %v", peer) + + case <-time.After(100 * time.Millisecond): + // Ok, peer not accepted + } +} + func newkey() *ecdsa.PrivateKey { key, err := crypto.GenerateKey() if err != nil {