A thread safe way to take a free port (#485)

* initial commit

* safe latest error

* remove json listener

* fix: debug block unit test

* remove jsonrpc listener from server

* fix: consume network's find port in heimdall client test

* update key in log while startup

* linters

* fix

Co-authored-by: Manav Darji <manavdarji.india@gmail.com>

common/network/port.go

@@ -0,0 +1,36 @@
+package network
+
+import (
+	"errors"
+	"fmt"
+	"net"
+)
+
+const (
+	maxPortCheck = 100
+
+	emptyPort = "127.0.0.1:0"
+)
+
+var (
+	ErrCantFindAPort = errors.New("no available port found")
+)
+
+// FindAvailablePort returns the an available port
+func FindAvailablePort() (int, net.Listener, error) {
+	var (
+		listener net.Listener
+		err      error
+	)
+
+	for i := uint(0); i < maxPortCheck; i++ {
+		listener, err = net.Listen("tcp", emptyPort)
+		if err != nil {
+			continue
+		}
+
+		return listener.Addr().(*net.TCPAddr).Port, listener, nil
+	}
+
+	return 0, nil, fmt.Errorf("%w: %s", ErrCantFindAPort, err)
+}

consensus/bor/heimdall/client_test.go

@@ -9,18 +9,16 @@ import (
 	"net"
 	"net/http"
 	"sync"
-	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/network"
 	"github.com/ethereum/go-ethereum/consensus/bor/heimdall/checkpoint"
 
 	"github.com/stretchr/testify/require"
 )
 
-var maxPortCheck int32 = 100
-
 // HttpHandlerFake defines the handler functions required to serve
 // requests to the mock heimdal server for specific functions. Add more handlers
 // according to requirements.
@@ -34,7 +32,7 @@ func (h *HttpHandlerFake) GetCheckpointHandler() http.HandlerFunc {
 	}
 }
 
-func CreateMockHeimdallServer(wg *sync.WaitGroup, port int32, handler *HttpHandlerFake) (*http.Server, error) {
+func CreateMockHeimdallServer(wg *sync.WaitGroup, port int, listener net.Listener, handler *HttpHandlerFake) (*http.Server, error) {
 	// Create a new server mux
 	mux := http.NewServeMux()
 
@@ -51,6 +49,12 @@ func CreateMockHeimdallServer(wg *sync.WaitGroup, port int32, handler *HttpHandl
 		Handler: mux,
 	}
 
+	// Close the listener using the port and immediately consume it below
+	err := listener.Close()
+	if err != nil {
+		return nil, err
+	}
+
 	go func() {
 		defer wg.Done()
 
@@ -93,12 +97,12 @@ func TestFetchCheckpointFromMockHeimdall(t *testing.T) {
 		}
 	}
 
-	// Fetch available port starting from 50000
+	// Fetch available port
-	port, err := findAvailablePort(50000, 0)
+	port, listener, err := network.FindAvailablePort()
 	require.NoError(t, err, "expect no error in finding available port")
 
 	// Create mock heimdall server and pass handler instance for setting up the routes
-	srv, err := CreateMockHeimdallServer(wg, port, handler)
+	srv, err := CreateMockHeimdallServer(wg, port, listener, handler)
 	require.NoError(t, err, "expect no error in starting mock heimdall server")
 
 	// Create a new heimdall client and use same port for connection
@@ -150,12 +154,12 @@ func TestFetchShutdown(t *testing.T) {
 		}
 	}
 
-	// Fetch available port starting from 50000
+	// Fetch available port
-	port, err := findAvailablePort(50000, 0)
+	port, listener, err := network.FindAvailablePort()
 	require.NoError(t, err, "expect no error in finding available port")
 
 	// Create mock heimdall server and pass handler instance for setting up the routes
-	srv, err := CreateMockHeimdallServer(wg, port, handler)
+	srv, err := CreateMockHeimdallServer(wg, port, listener, handler)
 	require.NoError(t, err, "expect no error in starting mock heimdall server")
 
 	// Create a new heimdall client and use same port for connection
@@ -216,26 +220,6 @@ func TestFetchShutdown(t *testing.T) {
 	wg.Wait()
 }
 
-// findAvailablePort returns the next available port starting from `from`
-func findAvailablePort(from int32, count int32) (int32, error) {
-	if count == maxPortCheck {
-		return 0, fmt.Errorf("no available port found")
-	}
-
-	port := atomic.AddInt32(&from, 1)
-	addr := fmt.Sprintf("localhost:%d", port)
-
-	count++
-
-	lis, err := net.Listen("tcp", addr)
-	if err == nil {
-		lis.Close()
-		return port, nil
-	} else {
-		return findAvailablePort(from, count)
-	}
-}
-
 // TestContext includes bunch of simple tests to verify the working of timeout
 // based context and cancellation.
 func TestContext(t *testing.T) {

internal/cli/debug_test.go

@@ -13,7 +13,7 @@ import (
 	"github.com/ethereum/go-ethereum/internal/cli/server"
 )
 
-var currentDir string = ""
+var currentDir string
 
 func TestCommand_DebugBlock(t *testing.T) {
 	t.Parallel()

internal/cli/server/command.go

@@ -72,7 +72,7 @@ func (c *Command) extractFlags(args []string) error {
 	}
 	// read if config file is provided, this will overwrite the cli flags, if provided
 	if c.configFile != "" {
-		log.Warn("Config File provided, this will overwrite the cli flags.", "configFile:", c.configFile)
+		log.Warn("Config File provided, this will overwrite the cli flags", "path", c.configFile)
 		cfg, err := readConfigFile(c.configFile)
 		if err != nil {
 			c.UI.Error(err.Error())
@@ -101,7 +101,7 @@ func (c *Command) Run(args []string) int {
 		return 1
 	}
 
-	srv, err := NewServer(c.config)
+	srv, err := NewServer(c.config, WithGRPCAddress())
 	if err != nil {
 		c.UI.Error(err.Error())
 		return 1

internal/cli/server/helper.go

@@ -2,75 +2,36 @@ package server
 
 import (
 	"fmt"
-	"math/rand"
-	"net"
 	"os"
-	"sync/atomic"
+
-	"time"
+	"github.com/ethereum/go-ethereum/common/network"
 )
 
-var maxPortCheck int32 = 100
-
-// findAvailablePort returns the next available port starting from `from`
-func findAvailablePort(from int32, count int32) (int32, error) {
-	if count == maxPortCheck {
-		return 0, fmt.Errorf("no available port found")
-	}
-
-	port := atomic.AddInt32(&from, 1)
-	addr := fmt.Sprintf("localhost:%d", port)
-
-	count++
-
-	lis, err := net.Listen("tcp", addr)
-	if err == nil {
-		lis.Close()
-		return port, nil
-	} else {
-		return findAvailablePort(from, count)
-	}
-}
-
 func CreateMockServer(config *Config) (*Server, error) {
 	if config == nil {
 		config = DefaultConfig()
 	}
 
-	// find available port for grpc server
+	// get grpc port and listener
-	rand.Seed(time.Now().UnixNano())
+	grpcPort, gRPCListener, err := network.FindAvailablePort()
-
-	var (
-		from int32 = 60000 // the min port to start checking from
-		to   int32 = 61000 // the max port to start checking from
-	)
-
-	//nolint: gosec
-	port, err := findAvailablePort(rand.Int31n(to-from+1)+from, 0)
 	if err != nil {
 		return nil, err
 	}
 
-	// grpc port
+	// The test uses grpc port from config so setting it here.
-	config.GRPC.Addr = fmt.Sprintf(":%d", port)
+	config.GRPC.Addr = fmt.Sprintf(":%d", grpcPort)
 
 	// datadir
-	datadir, _ := os.MkdirTemp("/tmp", "bor-cli-test")
+	datadir, err := os.MkdirTemp("", "bor-cli-test")
-	config.DataDir = datadir
-
-	// find available port for http server
-	from = 8545
-	to = 9545
-
-	//nolint: gosec
-	port, err = findAvailablePort(rand.Int31n(to-from+1)+from, 0)
 	if err != nil {
 		return nil, err
 	}
 
-	config.JsonRPC.Http.Port = uint64(port)
+	config.DataDir = datadir
+	config.JsonRPC.Http.Port = 0 // It will choose a free/available port
 
 	// start the server
-	return NewServer(config)
+	return NewServer(config, WithGRPCListener(gRPCListener))
 }
 
 func CloseMockServer(server *Server) {

internal/cli/server/server.go

@@ -50,7 +50,22 @@ type Server struct {
 	tracerAPI *tracers.API
 }
 
-func NewServer(config *Config) (*Server, error) {
+type serverOption func(srv *Server, config *Config) error
+
+func WithGRPCAddress() serverOption {
+	return func(srv *Server, config *Config) error {
+		return srv.gRPCServerByAddress(config.GRPC.Addr)
+	}
+}
+
+func WithGRPCListener(lis net.Listener) serverOption {
+	return func(srv *Server, _ *Config) error {
+		return srv.gRPCServerByListener(lis)
+	}
+}
+
+//nolint:gocognit
+func NewServer(config *Config, opts ...serverOption) (*Server, error) {
 	srv := &Server{
 		config: config,
 	}
@@ -58,12 +73,17 @@ func NewServer(config *Config) (*Server, error) {
 	// start the logger
 	setupLogger(config.LogLevel)
 
-	if err := srv.setupGRPCServer(config.GRPC.Addr); err != nil {
+	var err error
+
+	for _, opt := range opts {
+		err = opt(srv, config)
+		if err != nil {
 			return nil, err
 		}
+	}
 
 	// load the chain genesis
-	if err := config.loadChain(); err != nil {
+	if err = config.loadChain(); err != nil {
 		return nil, err
 	}
 
@@ -151,7 +171,6 @@ func NewServer(config *Config) (*Server, error) {
 				wallet, err := accountManager.Find(accounts.Account{Address: eb})
 				if wallet == nil || err != nil {
 					log.Error("Etherbase account unavailable locally", "err", err)
-
 					return nil, fmt.Errorf("signer missing: %v", err)
 				}
 
@@ -217,8 +236,13 @@ func NewServer(config *Config) (*Server, error) {
 }
 
 func (s *Server) Stop() {
+	if s.node != nil {
 		s.node.Close()
+	}
+
+	if s.grpcServer != nil {
 		s.grpcServer.Stop()
+	}
 
 	// shutdown the tracer
 	if s.tracer != nil {
@@ -327,22 +351,26 @@ func (s *Server) setupMetrics(config *TelemetryConfig, serviceName string) error
 	return nil
 }
 
-func (s *Server) setupGRPCServer(addr string) error {
+func (s *Server) gRPCServerByAddress(addr string) error {
-	s.grpcServer = grpc.NewServer(s.withLoggingUnaryInterceptor())
-	proto.RegisterBorServer(s.grpcServer, s)
-
 	lis, err := net.Listen("tcp", addr)
 	if err != nil {
 		return err
 	}
 
+	return s.gRPCServerByListener(lis)
+}
+
+func (s *Server) gRPCServerByListener(listener net.Listener) error {
+	s.grpcServer = grpc.NewServer(s.withLoggingUnaryInterceptor())
+	proto.RegisterBorServer(s.grpcServer, s)
+
 	go func() {
-		if err := s.grpcServer.Serve(lis); err != nil {
+		if err := s.grpcServer.Serve(listener); err != nil {
 			log.Error("failed to serve grpc server", "err", err)
 		}
 	}()
 
-	log.Info("GRPC Server started", "addr", addr)
+	log.Info("GRPC Server started", "addr", listener.Addr())
 
 	return nil
 }

node/node.go

@@ -20,6 +20,7 @@ import (
 	crand "crypto/rand"
 	"errors"
 	"fmt"
+	"net"
 	"net/http"
 	"os"
 	"path/filepath"
@@ -27,6 +28,8 @@ import (
 	"strings"
 	"sync"
 
+	"github.com/prometheus/tsdb/fileutil"
+
 	"github.com/ethereum/go-ethereum/accounts"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
@@ -36,7 +39,6 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/rpc"
-	"github.com/prometheus/tsdb/fileutil"
 )
 
 // Node is a container on which services can be registered.
@@ -466,6 +468,12 @@ func (n *Node) startRPC() error {
 		if err := initHttp(n.http, open, n.config.HTTPPort); err != nil {
 			return err
 		}
+
+		defer func() {
+			if n.http.listener != nil {
+				n.config.HTTPPort = n.http.listener.Addr().(*net.TCPAddr).Port
+			}
+		}()
 	}
 	// Configure WebSocket.
 	if n.config.WSHost != "" {
@@ -473,6 +481,12 @@ func (n *Node) startRPC() error {
 		if err := initWS(open, n.config.WSPort); err != nil {
 			return err
 		}
+
+		defer func() {
+			if n.ws.listener != nil {
+				n.config.WSPort = n.ws.listener.Addr().(*net.TCPAddr).Port
+			}
+		}()
 	}
 	// Configure authenticated API
 	if len(open) != len(all) {
@@ -480,16 +494,19 @@ func (n *Node) startRPC() error {
 		if err != nil {
 			return err
 		}
-		if err := initAuth(all, n.config.AuthPort, jwtSecret); err != nil {
+
+		if err = initAuth(all, n.config.AuthPort, jwtSecret); err != nil {
 			return err
 		}
 	}
+
 	// Start the servers
 	for _, server := range servers {
 		if err := server.start(); err != nil {
 			return err
 		}
 	}
+
 	return nil
 }
 

node/rpcstack.go

@@ -29,9 +29,10 @@ import (
 	"sync"
 	"sync/atomic"
 
+	"github.com/rs/cors"
+
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/rpc"
-	"github.com/rs/cors"
 )
 
 // httpConfig is the JSON-RPC/HTTP configuration.