Merge pull request #971 from fjl/p2p-limit-tweaks

p2p: tweak connection limits

p2p/discover/database.go

@@ -17,6 +17,7 @@ import (
 	"github.com/syndtr/goleveldb/leveldb"
 	"github.com/syndtr/goleveldb/leveldb/errors"
 	"github.com/syndtr/goleveldb/leveldb/iterator"
+	"github.com/syndtr/goleveldb/leveldb/opt"
 	"github.com/syndtr/goleveldb/leveldb/storage"
 	"github.com/syndtr/goleveldb/leveldb/util"
 )
@@ -72,8 +73,8 @@ func newMemoryNodeDB() (*nodeDB, error) {
 // newPersistentNodeDB creates/opens a leveldb backed persistent node database,
 // also flushing its contents in case of a version mismatch.
 func newPersistentNodeDB(path string, version int) (*nodeDB, error) {
-	// Try to open the cache, recovering any corruption
-	db, err := leveldb.OpenFile(path, nil)
+	opts := &opt.Options{OpenFilesCacheCapacity: 5}
+	db, err := leveldb.OpenFile(path, opts)
 	if _, iscorrupted := err.(*errors.ErrCorrupted); iscorrupted {
 		db, err = leveldb.RecoverFile(path, nil)
 	}

p2p/discover/table.go

@@ -25,7 +25,7 @@ const (
 	hashBits   = len(common.Hash{}) * 8
 	nBuckets   = hashBits + 1 // Number of buckets
 
-	maxBondingPingPongs = 10
+	maxBondingPingPongs = 16
 )
 
 type Table struct {

p2p/peer.go

@@ -131,10 +131,11 @@ func (p *Peer) run() DiscReason {
 	case err := <-p.protoErr:
 		reason = discReasonForError(err)
 	case reason = <-p.disc:
+		p.politeDisconnect(reason)
+		reason = DiscRequested
 	}
 
 	close(p.closed)
-	p.politeDisconnect(reason)
 	p.wg.Wait()
 	glog.V(logger.Debug).Infof("%v: Disconnected: %v\n", p, reason)
 	return reason
@@ -191,7 +192,7 @@ func (p *Peer) handle(msg Msg) error {
 		// check errors because, the connection will be closed after it.
 		rlp.Decode(msg.Payload, &reason)
 		glog.V(logger.Debug).Infof("%v: Disconnect Requested: %v\n", p, reason[0])
-		return DiscRequested
+		return reason[0]
 	case msg.Code < baseProtocolLength:
 		// ignore other base protocol messages
 		return msg.Discard()

p2p/peer_test.go

@@ -50,8 +50,6 @@ func testPeer(protos []Protocol) (func(), *conn, *Peer, <-chan DiscReason) {
 }
 
 func TestPeerProtoReadMsg(t *testing.T) {
-	defer testlog(t).detach()
-
 	done := make(chan struct{})
 	proto := Protocol{
 		Name:   "a",
@@ -88,8 +86,6 @@ func TestPeerProtoReadMsg(t *testing.T) {
 }
 
 func TestPeerProtoEncodeMsg(t *testing.T) {
-	defer testlog(t).detach()
-
 	proto := Protocol{
 		Name:   "a",
 		Length: 2,
@@ -112,8 +108,6 @@ func TestPeerProtoEncodeMsg(t *testing.T) {
 }
 
 func TestPeerWriteForBroadcast(t *testing.T) {
-	defer testlog(t).detach()
-
 	closer, rw, peer, peerErr := testPeer([]Protocol{discard})
 	defer closer()
 
@@ -152,8 +146,6 @@ func TestPeerWriteForBroadcast(t *testing.T) {
 }
 
 func TestPeerPing(t *testing.T) {
-	defer testlog(t).detach()
-
 	closer, rw, _, _ := testPeer(nil)
 	defer closer()
 	if err := SendItems(rw, pingMsg); err != nil {
@@ -165,26 +157,24 @@ func TestPeerPing(t *testing.T) {
 }
 
 func TestPeerDisconnect(t *testing.T) {
-	defer testlog(t).detach()
-
 	closer, rw, _, disc := testPeer(nil)
 	defer closer()
 	if err := SendItems(rw, discMsg, DiscQuitting); err != nil {
 		t.Fatal(err)
 	}
-	if err := ExpectMsg(rw, discMsg, []interface{}{DiscRequested}); err != nil {
-		t.Error(err)
-	}
-	closer()
-	if reason := <-disc; reason != DiscRequested {
+	select {
+	case reason := <-disc:
+		if reason != DiscQuitting {
 			t.Errorf("run returned wrong reason: got %v, want %v", reason, DiscRequested)
 		}
+	case <-time.After(500 * time.Millisecond):
+		t.Error("peer did not return")
+	}
 }
 
 // This test is supposed to verify that Peer can reliably handle
 // multiple causes of disconnection occurring at the same time.
 func TestPeerDisconnectRace(t *testing.T) {
-	defer testlog(t).detach()
 	maybe := func() bool { return rand.Intn(1) == 1 }
 
 	for i := 0; i < 1000; i++ {

p2p/server.go

@@ -18,12 +18,12 @@ import (
 )
 
 const (
-	defaultDialTimeout      = 10 * time.Second
+	defaultDialTimeout      = 15 * time.Second
 	refreshPeersInterval    = 30 * time.Second
 	staticPeerCheckInterval = 15 * time.Second
 
 	// Maximum number of concurrently handshaking inbound connections.
-	maxAcceptConns = 10
+	maxAcceptConns = 50
 
 	// Maximum number of concurrently dialing outbound connections.
 	maxDialingConns = 10

p2p/server_test.go

@@ -46,8 +46,6 @@ func startTestServer(t *testing.T, pf newPeerHook) *Server {
 }
 
 func TestServerListen(t *testing.T) {
-	defer testlog(t).detach()
-
 	// start the test server
 	connected := make(chan *Peer)
 	srv := startTestServer(t, func(p *Peer) {
@@ -78,8 +76,6 @@ func TestServerListen(t *testing.T) {
 }
 
 func TestServerDial(t *testing.T) {
-	defer testlog(t).detach()
-
 	// run a one-shot TCP server to handle the connection.
 	listener, err := net.Listen("tcp", "127.0.0.1:0")
 	if err != nil {
@@ -126,8 +122,6 @@ func TestServerDial(t *testing.T) {
 }
 
 func TestServerBroadcast(t *testing.T) {
-	defer testlog(t).detach()
-
 	var connected sync.WaitGroup
 	srv := startTestServer(t, func(p *Peer) {
 		p.running = matchProtocols([]Protocol{discard}, []Cap{discard.cap()}, p.rw)
@@ -172,8 +166,6 @@ func TestServerBroadcast(t *testing.T) {
 //
 // It also serves as a light-weight integration test.
 func TestServerDisconnectAtCap(t *testing.T) {
-	defer testlog(t).detach()
-
 	started := make(chan *Peer)
 	srv := &Server{
 		ListenAddr: "127.0.0.1:0",
@@ -224,8 +216,6 @@ func TestServerDisconnectAtCap(t *testing.T) {
 
 // Tests that static peers are (re)connected, and done so even above max peers.
 func TestServerStaticPeers(t *testing.T) {
-	defer testlog(t).detach()
-
 	// Create a test server with limited connection slots
 	started := make(chan *Peer)
 	server := &Server{
@@ -312,7 +302,6 @@ func TestServerStaticPeers(t *testing.T) {
 
 // Tests that trusted peers and can connect above max peer caps.
 func TestServerTrustedPeers(t *testing.T) {
-	defer testlog(t).detach()
 
 	// Create a trusted peer to accept connections from
 	key := newkey()
@@ -374,8 +363,6 @@ func TestServerTrustedPeers(t *testing.T) {
 
 // Tests that a failed dial will temporarily throttle a peer.
 func TestServerMaxPendingDials(t *testing.T) {
-	defer testlog(t).detach()
-
 	// Start a simple test server
 	server := &Server{
 		ListenAddr:      "127.0.0.1:0",
@@ -443,8 +430,6 @@ func TestServerMaxPendingDials(t *testing.T) {
 }
 
 func TestServerMaxPendingAccepts(t *testing.T) {
-	defer testlog(t).detach()
-
 	// Start a test server and a peer sink for synchronization
 	started := make(chan *Peer)
 	server := &Server{

p2p/testlog_test.go

@@ -1,25 +0,0 @@
-package p2p
-
-import (
-	"testing"
-
-	"github.com/ethereum/go-ethereum/logger"
-)
-
-type testLogger struct{ t *testing.T }
-
-func testlog(t *testing.T) testLogger {
-	logger.Reset()
-	l := testLogger{t}
-	logger.AddLogSystem(l)
-	return l
-}
-
-func (l testLogger) LogPrint(msg logger.LogMsg) {
-	l.t.Logf("%s", msg.String())
-}
-
-func (testLogger) detach() {
-	logger.Flush()
-	logger.Reset()
-}