From fbdb44dcc17240a01b45e55d3aa4e4b8db0868cd Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Thu, 29 Oct 2015 13:28:00 +0100 Subject: cmd/utils, rpc/comms: stop XEth when IPC connection ends There are a bunch of changes required to make this work: - in miner: allow unregistering agents, fix RemoteAgent.Stop - in eth/filters: make FilterSystem.Stop not crash - in rpc/comms: move listen loop to platform-independent code Fixes #1930. I ran the shell loop there for a few minutes and didn't see any changes in the memory profile. --- rpc/comms/ipc.go | 43 +++++++++++++++++++++++++++++++++++++++++-- rpc/comms/ipc_unix.go | 40 +++++----------------------------------- rpc/comms/ipc_windows.go | 36 +++--------------------------------- 3 files changed, 49 insertions(+), 70 deletions(-) (limited to 'rpc') diff --git a/rpc/comms/ipc.go b/rpc/comms/ipc.go index 3de659b65..882d62ab4 100644 --- a/rpc/comms/ipc.go +++ b/rpc/comms/ipc.go @@ -20,13 +20,22 @@ import ( "fmt" "math/rand" "net" + "os" "encoding/json" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/rpc/codec" "github.com/ethereum/go-ethereum/rpc/shared" ) +type Stopper interface { + Stop() +} + +type InitFunc func(conn net.Conn) (Stopper, shared.EthereumApi, error) + type IpcConfig struct { Endpoint string } @@ -90,8 +99,38 @@ func NewIpcClient(cfg IpcConfig, codec codec.Codec) (*ipcClient, error) { } // Start IPC server -func StartIpc(cfg IpcConfig, codec codec.Codec, initializer func(conn net.Conn) (shared.EthereumApi, error)) error { - return startIpc(cfg, codec, initializer) +func StartIpc(cfg IpcConfig, codec codec.Codec, initializer InitFunc) error { + l, err := ipcListen(cfg) + if err != nil { + return err + } + go ipcLoop(cfg, codec, initializer, l) + return nil +} + +func ipcLoop(cfg IpcConfig, codec codec.Codec, initializer InitFunc, l net.Listener) { + glog.V(logger.Info).Infof("IPC service started (%s)\n", cfg.Endpoint) + defer os.Remove(cfg.Endpoint) + defer l.Close() + for { + conn, err := l.Accept() + if err != nil { + glog.V(logger.Debug).Infof("accept: %v", err) + return + } + id := newIpcConnId() + go func() { + defer conn.Close() + glog.V(logger.Debug).Infof("new connection with id %06d started", id) + stopper, api, err := initializer(conn) + if err != nil { + glog.V(logger.Error).Infof("Unable to initialize IPC connection: %v", err) + return + } + defer stopper.Stop() + handle(id, conn, api, codec) + }() + } } func newIpcConnId() int { diff --git a/rpc/comms/ipc_unix.go b/rpc/comms/ipc_unix.go index d68363a45..4b839572a 100644 --- a/rpc/comms/ipc_unix.go +++ b/rpc/comms/ipc_unix.go @@ -23,8 +23,6 @@ import ( "os" "path/filepath" - "github.com/ethereum/go-ethereum/logger" - "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/rpc/codec" "github.com/ethereum/go-ethereum/rpc/shared" "github.com/ethereum/go-ethereum/rpc/useragent" @@ -69,44 +67,16 @@ func (self *ipcClient) reconnect() error { return err } -func startIpc(cfg IpcConfig, codec codec.Codec, initializer func(conn net.Conn) (shared.EthereumApi, error)) error { +func ipcListen(cfg IpcConfig) (net.Listener, error) { // Ensure the IPC path exists and remove any previous leftover if err := os.MkdirAll(filepath.Dir(cfg.Endpoint), 0751); err != nil { - return err + return nil, err } os.Remove(cfg.Endpoint) - - l, err := net.ListenUnix("unix", &net.UnixAddr{Name: cfg.Endpoint, Net: "unix"}) + l, err := net.Listen("unix", cfg.Endpoint) if err != nil { - return err + return nil, err } os.Chmod(cfg.Endpoint, 0600) - - go func() { - for { - conn, err := l.AcceptUnix() - if err != nil { - glog.V(logger.Error).Infof("Error accepting ipc connection - %v\n", err) - continue - } - - id := newIpcConnId() - glog.V(logger.Debug).Infof("New IPC connection with id %06d started\n", id) - - api, err := initializer(conn) - if err != nil { - glog.V(logger.Error).Infof("Unable to initialize IPC connection - %v\n", err) - conn.Close() - continue - } - - go handle(id, conn, api, codec) - } - - os.Remove(cfg.Endpoint) - }() - - glog.V(logger.Info).Infof("IPC service started (%s)\n", cfg.Endpoint) - - return nil + return l, nil } diff --git a/rpc/comms/ipc_windows.go b/rpc/comms/ipc_windows.go index 47edd9e5b..e25fba253 100644 --- a/rpc/comms/ipc_windows.go +++ b/rpc/comms/ipc_windows.go @@ -28,8 +28,6 @@ import ( "time" "unsafe" - "github.com/ethereum/go-ethereum/logger" - "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/rpc/codec" "github.com/ethereum/go-ethereum/rpc/shared" "github.com/ethereum/go-ethereum/rpc/useragent" @@ -688,40 +686,12 @@ func (self *ipcClient) reconnect() error { return err } -func startIpc(cfg IpcConfig, codec codec.Codec, initializer func(conn net.Conn) (shared.EthereumApi, error)) error { +func ipcListen(cfg IpcConfig) (net.Listener, error) { os.Remove(cfg.Endpoint) // in case it still exists from a previous run - l, err := Listen(cfg.Endpoint) if err != nil { - return err + return nil, err } os.Chmod(cfg.Endpoint, 0600) - - go func() { - for { - conn, err := l.Accept() - if err != nil { - glog.V(logger.Error).Infof("Error accepting ipc connection - %v\n", err) - continue - } - - id := newIpcConnId() - glog.V(logger.Debug).Infof("New IPC connection with id %06d started\n", id) - - api, err := initializer(conn) - if err != nil { - glog.V(logger.Error).Infof("Unable to initialize IPC connection - %v\n", err) - conn.Close() - continue - } - - go handle(id, conn, api, codec) - } - - os.Remove(cfg.Endpoint) - }() - - glog.V(logger.Info).Infof("IPC service started (%s)\n", cfg.Endpoint) - - return nil + return l, nil } -- cgit v1.2.3