rpc: clean up IPC handler

This avoids logging accept errors on shutdown and removes a bit of
duplication.

Author: fjl

cmd/clef/main.go

@@ -23,17 +23,18 @@ import (
 	"context"
 	"crypto/rand"
 	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"io"
 	"io/ioutil"
 	"os"
+	"os/signal"
 	"os/user"
 	"path/filepath"
 	"runtime"
 	"strings"
 
-	"encoding/hex"
 	"github.com/ethereum/go-ethereum/cmd/utils"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
@@ -44,7 +45,6 @@ import (
 	"github.com/ethereum/go-ethereum/signer/rules"
 	"github.com/ethereum/go-ethereum/signer/storage"
 	"gopkg.in/urfave/cli.v1"
-	"os/signal"
 )
 
 // ExternalApiVersion -- see extapi_changelog.md
@@ -435,7 +435,7 @@ func signer(c *cli.Context) error {
 			ipcApiUrl = filepath.Join(configDir, "clef.ipc")
 		}
 
-		listener, _, err := rpc.StartIPCEndpoint(func() bool { return true }, ipcApiUrl, rpcApi)
+		listener, _, err := rpc.StartIPCEndpoint(ipcApiUrl, rpcApi)
 		if err != nil {
 			utils.Fatalf("Could not start IPC api: %v", err)
 		}

node/node.go

@@ -303,23 +303,13 @@ func (n *Node) stopInProc() {
 
 // 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
-
-	}
-	isClosed := func() bool {
-		n.lock.RLock()
-		defer n.lock.RUnlock()
-		return n.ipcListener == nil
+		return nil // IPC disabled.
 	}
-
-	listener, handler, err := rpc.StartIPCEndpoint(isClosed, n.ipcEndpoint, apis)
+	listener, handler, err := rpc.StartIPCEndpoint(n.ipcEndpoint, apis)
 	if err != nil {
 		return err
 	}
-
-	// All listeners booted successfully
 	n.ipcListener = listener
 	n.ipcHandler = handler
 	n.log.Info("IPC endpoint opened", "url", n.ipcEndpoint)

rpc/endpoints.go

@@ -17,8 +17,9 @@
 package rpc
 
 import (
-	"github.com/ethereum/go-ethereum/log"
 	"net"
+
+	"github.com/ethereum/go-ethereum/log"
 )
 
 // StartHTTPEndpoint starts the HTTP RPC endpoint, configured with cors/vhosts/modules
@@ -81,40 +82,21 @@ func StartWSEndpoint(endpoint string, apis []API, modules []string, wsOrigins []
 
 }
 
-// StartIPCEndpoint starts an IPC endpoint
-func StartIPCEndpoint(isClosedFn func() bool, ipcEndpoint string, apis []API) (net.Listener, *Server, error) {
-	// Register all the APIs exposed by the services
+// StartIPCEndpoint starts an IPC endpoint.
+func StartIPCEndpoint(ipcEndpoint string, apis []API) (net.Listener, *Server, error) {
+	// Register all the APIs exposed by the services.
 	handler := NewServer()
 	for _, api := range apis {
 		if err := handler.RegisterName(api.Namespace, api.Service); err != nil {
 			return nil, nil, err
 		}
 		log.Debug("IPC registered", "namespace", api.Namespace)
 	}
-	// All APIs registered, start the IPC listener
-	var (
-		listener net.Listener
-		err      error
-	)
-	if listener, err = CreateIPCListener(ipcEndpoint); err != nil {
+	// All APIs registered, start the IPC listener.
+	listener, err := ipcListen(ipcEndpoint)
+	if err != nil {
 		return nil, nil, err
 	}
-	go func() {
-		for {
-			conn, err := listener.Accept()
-			if err != nil {
-				// Terminate if the listener was closed
-				if isClosedFn() {
-					log.Info("IPC closed", "err", err)
-				} else {
-					// Not closed, just some error; report and continue
-					log.Error("IPC accept failed", "err", err)
-				}
-				continue
-			}
-			go handler.ServeCodec(NewJSONCodec(conn), OptionMethodInvocation|OptionSubscriptions)
-		}
-	}()
-
+	go handler.ServeListener(listener)
 	return listener, handler, nil
 }

rpc/ipc.go

@@ -18,26 +18,23 @@ package rpc
 
 import (
 	"context"
-	"fmt"
 	"net"
 
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/p2p/netutil"
 )
 
-// CreateIPCListener creates an listener, on Unix platforms this is a unix socket, on
-// 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 (srv *Server) ServeListener(l net.Listener) error {
 	for {
 		conn, err := l.Accept()
-		if err != nil {
+		if netutil.IsTemporaryError(err) {
+			log.Warn("RPC accept error", "err", err)
+			continue
+		} else if err != nil {
 			return err
 		}
-		log.Trace(fmt.Sprint("accepted conn", conn.RemoteAddr()))
+		log.Trace("Accepted connection", "addr", conn.RemoteAddr())
 		go srv.ServeCodec(NewJSONCodec(conn), OptionMethodInvocation|OptionSubscriptions)
 	}
 }