rpc: clean up IPC handler (#16524)

node/node.go

@@ -140,7 +140,7 @@ func (n *Node) Register(constructor ServiceConstructor) error {
 	return nil
 }
 
-// Start create a live P2P node and starts running it.
+// Start creates a live P2P node and starts running it.
 func (n *Node) Start() error {
 	n.lock.Lock()
 	defer n.lock.Unlock()
@@ -217,7 +217,7 @@ func (n *Node) Start() error {
 		// Mark the service started for potential cleanup
 		started = append(started, kind)
 	}
-	// Lastly start the configured RPC interfaces
+	// Lastly, start the configured RPC interfaces
 	if err := n.startRPC(services); err != nil {
 		for _, service := range services {
 			service.Stop()
@@ -252,7 +252,7 @@ func (n *Node) openDataDir() error {
 	return nil
 }
 
-// startRPC is a helper method to start all the various RPC endpoint during node
+// startRPC is a helper method to start all the various RPC endpoints during node
 // startup. It's not meant to be called at any time afterwards as it makes certain
 // assumptions about the state of the node.
 func (n *Node) startRPC(services map[reflect.Type]Service) error {
@@ -293,7 +293,7 @@ func (n *Node) startInProc(apis []rpc.API) error {
 		if err := handler.RegisterName(api.Namespace, api.Service); err != nil {
 			return err
 		}
-		n.log.Debug("InProc registered", "service", api.Service, "namespace", api.Namespace)
+		n.log.Debug("InProc registered", "namespace", api.Namespace)
 	}
 	n.inprocHandler = handler
 	return nil
@@ -320,22 +320,13 @@ func (n *Node) RegisterAPIs(apis []rpc.API) {
 
 // startIPC initializes and starts the IPC RPC endpoint.
 func (n *Node) startIPC(apis []rpc.API) error {
-	// Short circuit if the IPC endpoint isn't being exposed
 	if n.ipcEndpoint == "" {
-		return nil
+		return nil // IPC disabled.
 	}
-	isClosed := func() bool {
+	listener, handler, err := rpc.StartIPCEndpoint(n.ipcEndpoint, apis)
-		n.lock.RLock()
-		defer n.lock.RUnlock()
-		return n.ipcListener == nil
-	}
-
-	listener, handler, err := rpc.StartIPCEndpoint(isClosed, n.ipcEndpoint, apis)
 	if err != nil {
 		return err
 	}
-
-	// All listeners booted successfully
 	n.ipcListener = listener
 	n.ipcHandler = handler
 	log.Info("IPC endpoint opened", "url", n.ipcEndpoint)
@@ -348,7 +339,7 @@ func (n *Node) stopIPC() {
 		n.ipcListener.Close()
 		n.ipcListener = nil
 
-		n.log.Info("IPC endpoint closed", "endpoint", n.ipcEndpoint)
+		n.log.Info("IPC endpoint closed", "url", n.ipcEndpoint)
 	}
 	if n.ipcHandler != nil {
 		n.ipcHandler.Stop()

rpc/endpoints.go

@@ -80,11 +80,10 @@ func StartWSEndpoint(endpoint string, apis []API, modules []string, wsOrigins []
 	}
 	go NewWSServer(wsOrigins, handler).Serve(listener)
 	return listener, handler, err
-
 }
 
 // StartIPCEndpoint starts an IPC endpoint.
-func StartIPCEndpoint(isClosedFn func() bool, ipcEndpoint string, apis []API) (net.Listener, *Server, error) {
+func StartIPCEndpoint(ipcEndpoint string, apis []API) (net.Listener, *Server, error) {
 	// Register all the APIs exposed by the services.
 	var (
 		handler    = NewServer()
@@ -104,30 +103,10 @@ func StartIPCEndpoint(isClosedFn func() bool, ipcEndpoint string, apis []API) (n
 	}
 	log.Debug("IPCs registered", "namespaces", strings.Join(registered, ","))
 	// All APIs registered, start the IPC listener.
-	var (
+	listener, err := ipcListen(ipcEndpoint)
-		listener net.Listener
+	if err != nil {
-		err      error
-	)
-	if listener, err = CreateIPCListener(ipcEndpoint); err != nil {
 		return nil, nil, err
 	}
-	go func() {
+	go handler.ServeListener(listener)
-		for {
-			conn, err := listener.Accept()
-			if err != nil {
-				// Terminate if the listener was closed
-				if isClosedFn() {
-					log.Info("IPC closed", "err", err)
-					return
-				}
-				// Not closed, just some error; report and continue
-				log.Error("IPC accept failed", "err", err)
-				continue
-			}
-			log.Trace("Accepted RPC connection", "conn", conn.RemoteAddr())
-			go handler.ServeCodec(NewCodec(conn), 0)
-		}
-	}()
-
 	return listener, handler, nil
 }

rpc/ipc.go

@@ -24,23 +24,17 @@ import (
 	"github.com/XinFinOrg/XDPoSChain/p2p/netutil"
 )
 
-// CreateIPCListener creates an listener, on Unix platforms this is a unix socket, on
+// ServeListener accepts connections on l, serving IPC-RPC on them.
-// Windows this is a named pipe
-func CreateIPCListener(endpoint string) (net.Listener, error) {
-	return ipcListen(endpoint)
-}
-
-// ServeListener accepts connections on l, serving JSON-RPC on them.
 func (s *Server) ServeListener(l net.Listener) error {
 	for {
 		conn, err := l.Accept()
 		if netutil.IsTemporaryError(err) {
-			log.Warn("RPC accept error", "err", err)
+			log.Warn("IPC accept error", "err", err)
 			continue
 		} else if err != nil {
 			return err
 		}
-		log.Trace("Accepted RPC connection", "conn", conn.RemoteAddr())
+		log.Trace("IPC accepted connection")
 		go s.ServeCodec(NewCodec(conn), 0)
 	}
 }