From 19b2640e89465c1c57f1bbea0274d52d97151f60 Mon Sep 17 00:00:00 2001 From: Bas van Kervel Date: Wed, 16 Dec 2015 10:58:01 +0100 Subject: rpc: migrated the RPC insterface to a new reflection based RPC layer --- rpc/comms/comms.go | 150 ---------- rpc/comms/http.go | 345 ----------------------- rpc/comms/inproc.go | 82 ------ rpc/comms/ipc.go | 158 ----------- rpc/comms/ipc_unix.go | 82 ------ rpc/comms/ipc_windows.go | 697 ----------------------------------------------- 6 files changed, 1514 deletions(-) delete mode 100644 rpc/comms/comms.go delete mode 100644 rpc/comms/http.go delete mode 100644 rpc/comms/inproc.go delete mode 100644 rpc/comms/ipc.go delete mode 100644 rpc/comms/ipc_unix.go delete mode 100644 rpc/comms/ipc_windows.go (limited to 'rpc/comms') diff --git a/rpc/comms/comms.go b/rpc/comms/comms.go deleted file mode 100644 index 61fba5722..000000000 --- a/rpc/comms/comms.go +++ /dev/null @@ -1,150 +0,0 @@ -// Copyright 2015 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package comms - -import ( - "io" - "net" - - "fmt" - "strings" - - "strconv" - - "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" -) - -const ( - maxHttpSizeReqLength = 1024 * 1024 // 1MB -) - -var ( - // List with all API's which are offered over the in proc interface by default - DefaultInProcApis = shared.AllApis - - // List with all API's which are offered over the IPC interface by default - DefaultIpcApis = shared.AllApis - - // List with API's which are offered over thr HTTP/RPC interface by default - DefaultHttpRpcApis = strings.Join([]string{ - shared.DbApiName, shared.EthApiName, shared.NetApiName, shared.Web3ApiName, - }, ",") -) - -type EthereumClient interface { - // Close underlying connection - Close() - // Send request - Send(interface{}) error - // Receive response - Recv() (interface{}, error) - // List with modules this client supports - SupportedModules() (map[string]string, error) -} - -func handle(id int, conn net.Conn, api shared.EthereumApi, c codec.Codec) { - codec := c.New(conn) - - defer func() { - if r := recover(); r != nil { - glog.Errorf("panic: %v\n", r) - } - codec.Close() - }() - - for { - requests, isBatch, err := codec.ReadRequest() - if err == io.EOF { - return - } else if err != nil { - glog.V(logger.Debug).Infof("Closed IPC Conn %06d recv err - %v\n", id, err) - return - } - - if isBatch { - responses := make([]*interface{}, len(requests)) - responseCount := 0 - for _, req := range requests { - res, err := api.Execute(req) - if req.Id != nil { - rpcResponse := shared.NewRpcResponse(req.Id, req.Jsonrpc, res, err) - responses[responseCount] = rpcResponse - responseCount += 1 - } - } - - err = codec.WriteResponse(responses[:responseCount]) - if err != nil { - glog.V(logger.Debug).Infof("Closed IPC Conn %06d send err - %v\n", id, err) - return - } - } else { - var rpcResponse interface{} - res, err := api.Execute(requests[0]) - - rpcResponse = shared.NewRpcResponse(requests[0].Id, requests[0].Jsonrpc, res, err) - err = codec.WriteResponse(rpcResponse) - if err != nil { - glog.V(logger.Debug).Infof("Closed IPC Conn %06d send err - %v\n", id, err) - return - } - } - } -} - -// Endpoint must be in the form of: -// ${protocol}:${path} -// e.g. ipc:/tmp/geth.ipc -// rpc:localhost:8545 -func ClientFromEndpoint(endpoint string, c codec.Codec) (EthereumClient, error) { - if strings.HasPrefix(endpoint, "ipc:") { - cfg := IpcConfig{ - Endpoint: endpoint[4:], - } - return NewIpcClient(cfg, codec.JSON) - } - - if strings.HasPrefix(endpoint, "rpc:") { - parts := strings.Split(endpoint, ":") - addr := "http://localhost" - port := uint(8545) - if len(parts) >= 3 { - addr = parts[1] + ":" + parts[2] - } - - if len(parts) >= 4 { - p, err := strconv.Atoi(parts[3]) - - if err != nil { - return nil, err - } - port = uint(p) - } - - cfg := HttpConfig{ - ListenAddress: addr, - ListenPort: port, - } - - return NewHttpClient(cfg, codec.JSON), nil - } - - return nil, fmt.Errorf("Invalid endpoint") -} diff --git a/rpc/comms/http.go b/rpc/comms/http.go deleted file mode 100644 index f4a930d0e..000000000 --- a/rpc/comms/http.go +++ /dev/null @@ -1,345 +0,0 @@ -// Copyright 2015 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package comms - -import ( - "encoding/json" - "fmt" - "net" - "net/http" - "strings" - "sync" - "time" - - "bytes" - "io" - "io/ioutil" - - "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/rs/cors" -) - -const ( - serverIdleTimeout = 10 * time.Second // idle keep-alive connections - serverReadTimeout = 15 * time.Second // per-request read timeout - serverWriteTimeout = 15 * time.Second // per-request read timeout -) - -var ( - httpServerMu sync.Mutex - httpServer *stopServer -) - -type HttpConfig struct { - ListenAddress string - ListenPort uint - CorsDomain string -} - -// stopServer augments http.Server with idle connection tracking. -// Idle keep-alive connections are shut down when Close is called. -type stopServer struct { - *http.Server - l net.Listener - // connection tracking state - mu sync.Mutex - shutdown bool // true when Stop has returned - idle map[net.Conn]struct{} -} - -type handler struct { - codec codec.Codec - api shared.EthereumApi -} - -// StartHTTP starts listening for RPC requests sent via HTTP. -func StartHttp(cfg HttpConfig, codec codec.Codec, api shared.EthereumApi) error { - httpServerMu.Lock() - defer httpServerMu.Unlock() - - addr := fmt.Sprintf("%s:%d", cfg.ListenAddress, cfg.ListenPort) - if httpServer != nil { - if addr != httpServer.Addr { - return fmt.Errorf("RPC service already running on %s ", httpServer.Addr) - } - return nil // RPC service already running on given host/port - } - // Set up the request handler, wrapping it with CORS headers if configured. - handler := http.Handler(&handler{codec, api}) - if len(cfg.CorsDomain) > 0 { - opts := cors.Options{ - AllowedMethods: []string{"POST"}, - AllowedOrigins: strings.Split(cfg.CorsDomain, " "), - } - handler = cors.New(opts).Handler(handler) - } - // Start the server. - s, err := listenHTTP(addr, handler) - if err != nil { - glog.V(logger.Error).Infof("Can't listen on %s:%d: %v", cfg.ListenAddress, cfg.ListenPort, err) - return err - } - httpServer = s - return nil -} - -func (h *handler) ServeHTTP(w http.ResponseWriter, req *http.Request) { - w.Header().Set("Content-Type", "application/json") - - // Limit request size to resist DoS - if req.ContentLength > maxHttpSizeReqLength { - err := fmt.Errorf("Request too large") - response := shared.NewRpcErrorResponse(-1, shared.JsonRpcVersion, -32700, err) - sendJSON(w, &response) - return - } - - defer req.Body.Close() - payload, err := ioutil.ReadAll(req.Body) - if err != nil { - err := fmt.Errorf("Could not read request body") - response := shared.NewRpcErrorResponse(-1, shared.JsonRpcVersion, -32700, err) - sendJSON(w, &response) - return - } - - c := h.codec.New(nil) - var rpcReq shared.Request - if err = c.Decode(payload, &rpcReq); err == nil { - reply, err := h.api.Execute(&rpcReq) - res := shared.NewRpcResponse(rpcReq.Id, rpcReq.Jsonrpc, reply, err) - sendJSON(w, &res) - return - } - - var reqBatch []shared.Request - if err = c.Decode(payload, &reqBatch); err == nil { - resBatch := make([]*interface{}, len(reqBatch)) - resCount := 0 - for i, rpcReq := range reqBatch { - reply, err := h.api.Execute(&rpcReq) - if rpcReq.Id != nil { // this leaves nil entries in the response batch for later removal - resBatch[i] = shared.NewRpcResponse(rpcReq.Id, rpcReq.Jsonrpc, reply, err) - resCount += 1 - } - } - // make response omitting nil entries - sendJSON(w, resBatch[:resCount]) - return - } - - // invalid request - err = fmt.Errorf("Could not decode request") - res := shared.NewRpcErrorResponse(-1, shared.JsonRpcVersion, -32600, err) - sendJSON(w, res) -} - -func sendJSON(w io.Writer, v interface{}) { - if glog.V(logger.Detail) { - if payload, err := json.MarshalIndent(v, "", "\t"); err == nil { - glog.Infof("Sending payload: %s", payload) - } - } - if err := json.NewEncoder(w).Encode(v); err != nil { - glog.V(logger.Error).Infoln("Error sending JSON:", err) - } -} - -// Stop closes all active HTTP connections and shuts down the server. -func StopHttp() { - httpServerMu.Lock() - defer httpServerMu.Unlock() - if httpServer != nil { - httpServer.Close() - httpServer = nil - } -} - -func listenHTTP(addr string, h http.Handler) (*stopServer, error) { - l, err := net.Listen("tcp", addr) - if err != nil { - return nil, err - } - s := &stopServer{l: l, idle: make(map[net.Conn]struct{})} - s.Server = &http.Server{ - Addr: addr, - Handler: h, - ReadTimeout: serverReadTimeout, - WriteTimeout: serverWriteTimeout, - ConnState: s.connState, - } - go s.Serve(l) - return s, nil -} - -func (s *stopServer) connState(c net.Conn, state http.ConnState) { - s.mu.Lock() - defer s.mu.Unlock() - // Close c immediately if we're past shutdown. - if s.shutdown { - if state != http.StateClosed { - c.Close() - } - return - } - if state == http.StateIdle { - s.idle[c] = struct{}{} - } else { - delete(s.idle, c) - } -} - -func (s *stopServer) Close() { - s.mu.Lock() - defer s.mu.Unlock() - // Shut down the acceptor. No new connections can be created. - s.l.Close() - // Drop all idle connections. Non-idle connections will be - // closed by connState as soon as they become idle. - s.shutdown = true - for c := range s.idle { - glog.V(logger.Detail).Infof("closing idle connection %v", c.RemoteAddr()) - c.Close() - delete(s.idle, c) - } -} - -type httpClient struct { - address string - port uint - codec codec.ApiCoder - lastRes interface{} - lastErr error -} - -// Create a new in process client -func NewHttpClient(cfg HttpConfig, c codec.Codec) *httpClient { - return &httpClient{ - address: cfg.ListenAddress, - port: cfg.ListenPort, - codec: c.New(nil), - } -} - -func (self *httpClient) Close() { - // do nothing -} - -func (self *httpClient) Send(req interface{}) error { - var body []byte - var err error - - self.lastRes = nil - self.lastErr = nil - - if body, err = self.codec.Encode(req); err != nil { - return err - } - - httpReq, err := http.NewRequest("POST", fmt.Sprintf("%s:%d", self.address, self.port), bytes.NewBuffer(body)) - if err != nil { - return err - } - httpReq.Header.Set("Content-Type", "application/json") - - client := http.Client{} - resp, err := client.Do(httpReq) - if err != nil { - return err - } - - defer resp.Body.Close() - - if resp.Status == "200 OK" { - reply, _ := ioutil.ReadAll(resp.Body) - var rpcSuccessResponse shared.SuccessResponse - if err = self.codec.Decode(reply, &rpcSuccessResponse); err == nil { - self.lastRes = &rpcSuccessResponse - self.lastErr = err - return nil - } else { - var rpcErrorResponse shared.ErrorResponse - if err = self.codec.Decode(reply, &rpcErrorResponse); err == nil { - self.lastRes = &rpcErrorResponse - self.lastErr = err - return nil - } else { - return err - } - } - } - - return fmt.Errorf("Not implemented") -} - -func (self *httpClient) Recv() (interface{}, error) { - return self.lastRes, self.lastErr -} - -func (self *httpClient) SupportedModules() (map[string]string, error) { - var body []byte - var err error - - payload := shared.Request{ - Id: 1, - Jsonrpc: "2.0", - Method: "modules", - } - - if body, err = self.codec.Encode(payload); err != nil { - return nil, err - } - - req, err := http.NewRequest("POST", fmt.Sprintf("%s:%d", self.address, self.port), bytes.NewBuffer(body)) - if err != nil { - return nil, err - } - req.Header.Set("Content-Type", "application/json") - - client := http.Client{} - resp, err := client.Do(req) - if err != nil { - return nil, err - } - - defer resp.Body.Close() - - if resp.Status == "200 OK" { - reply, _ := ioutil.ReadAll(resp.Body) - var rpcRes shared.SuccessResponse - if err = self.codec.Decode(reply, &rpcRes); err != nil { - return nil, err - } - - result := make(map[string]string) - if modules, ok := rpcRes.Result.(map[string]interface{}); ok { - for a, v := range modules { - result[a] = fmt.Sprintf("%s", v) - } - return result, nil - } - err = fmt.Errorf("Unable to parse module response - %v", rpcRes.Result) - } else { - fmt.Printf("resp.Status = %s\n", resp.Status) - fmt.Printf("err = %v\n", err) - } - - return nil, err -} diff --git a/rpc/comms/inproc.go b/rpc/comms/inproc.go deleted file mode 100644 index e8058e32b..000000000 --- a/rpc/comms/inproc.go +++ /dev/null @@ -1,82 +0,0 @@ -// Copyright 2015 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package comms - -import ( - "fmt" - - "github.com/ethereum/go-ethereum/rpc/codec" - "github.com/ethereum/go-ethereum/rpc/shared" -) - -type InProcClient struct { - api shared.EthereumApi - codec codec.Codec - lastId interface{} - lastJsonrpc string - lastErr error - lastRes interface{} -} - -// Create a new in process client -func NewInProcClient(codec codec.Codec) *InProcClient { - return &InProcClient{ - codec: codec, - } -} - -func (self *InProcClient) Close() { - // do nothing -} - -// Need to setup api support -func (self *InProcClient) Initialize(offeredApi shared.EthereumApi) { - self.api = offeredApi -} - -func (self *InProcClient) Send(req interface{}) error { - if r, ok := req.(*shared.Request); ok { - self.lastId = r.Id - self.lastJsonrpc = r.Jsonrpc - self.lastRes, self.lastErr = self.api.Execute(r) - return self.lastErr - } - - return fmt.Errorf("Invalid request (%T)", req) -} - -func (self *InProcClient) Recv() (interface{}, error) { - return *shared.NewRpcResponse(self.lastId, self.lastJsonrpc, self.lastRes, self.lastErr), nil -} - -func (self *InProcClient) SupportedModules() (map[string]string, error) { - req := shared.Request{ - Id: 1, - Jsonrpc: "2.0", - Method: "modules", - } - - if res, err := self.api.Execute(&req); err == nil { - if result, ok := res.(map[string]string); ok { - return result, nil - } - } else { - return nil, err - } - - return nil, fmt.Errorf("Invalid response") -} diff --git a/rpc/comms/ipc.go b/rpc/comms/ipc.go deleted file mode 100644 index 3ba747b1d..000000000 --- a/rpc/comms/ipc.go +++ /dev/null @@ -1,158 +0,0 @@ -// Copyright 2015 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package comms - -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 -} - -type ipcClient struct { - endpoint string - c net.Conn - codec codec.Codec - coder codec.ApiCoder -} - -func (self *ipcClient) Close() { - self.coder.Close() -} - -func (self *ipcClient) Send(msg interface{}) error { - var err error - if err = self.coder.WriteResponse(msg); err != nil { - if err = self.reconnect(); err == nil { - err = self.coder.WriteResponse(msg) - } - } - return err -} - -func (self *ipcClient) Recv() (interface{}, error) { - return self.coder.ReadResponse() -} - -func (self *ipcClient) SupportedModules() (map[string]string, error) { - req := shared.Request{ - Id: 1, - Jsonrpc: "2.0", - Method: "rpc_modules", - } - - if err := self.coder.WriteResponse(req); err != nil { - return nil, err - } - - res, _ := self.coder.ReadResponse() - if sucRes, ok := res.(*shared.SuccessResponse); ok { - data, _ := json.Marshal(sucRes.Result) - modules := make(map[string]string) - if err := json.Unmarshal(data, &modules); err == nil { - return modules, nil - } - } - - // old version uses modules instead of rpc_modules, this can be removed after full migration - req.Method = "modules" - if err := self.coder.WriteResponse(req); err != nil { - return nil, err - } - - res, err := self.coder.ReadResponse() - if err != nil { - return nil, err - } - - if sucRes, ok := res.(*shared.SuccessResponse); ok { - data, _ := json.Marshal(sucRes.Result) - modules := make(map[string]string) - err = json.Unmarshal(data, &modules) - if err == nil { - return modules, nil - } - } - - return nil, fmt.Errorf("Invalid response") -} - -// Create a new IPC client, UNIX domain socket on posix, named pipe on Windows -func NewIpcClient(cfg IpcConfig, codec codec.Codec) (*ipcClient, error) { - return newIpcClient(cfg, codec) -} - -// Start IPC server -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 -} - -// CreateListener creates an listener, on Unix platforms this is a unix socket, on Windows this is a named pipe -func CreateListener(cfg IpcConfig) (net.Listener, error) { - return ipcListen(cfg) -} - -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 { - return rand.Int() % 1000000 -} diff --git a/rpc/comms/ipc_unix.go b/rpc/comms/ipc_unix.go deleted file mode 100644 index 4b839572a..000000000 --- a/rpc/comms/ipc_unix.go +++ /dev/null @@ -1,82 +0,0 @@ -// Copyright 2015 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -// +build darwin dragonfly freebsd linux nacl netbsd openbsd solaris - -package comms - -import ( - "net" - "os" - "path/filepath" - - "github.com/ethereum/go-ethereum/rpc/codec" - "github.com/ethereum/go-ethereum/rpc/shared" - "github.com/ethereum/go-ethereum/rpc/useragent" -) - -func newIpcClient(cfg IpcConfig, codec codec.Codec) (*ipcClient, error) { - c, err := net.DialUnix("unix", nil, &net.UnixAddr{cfg.Endpoint, "unix"}) - if err != nil { - return nil, err - } - - coder := codec.New(c) - msg := shared.Request{ - Id: 0, - Method: useragent.EnableUserAgentMethod, - Jsonrpc: shared.JsonRpcVersion, - Params: []byte("[]"), - } - - coder.WriteResponse(msg) - coder.Recv() - - return &ipcClient{cfg.Endpoint, c, codec, coder}, nil -} - -func (self *ipcClient) reconnect() error { - self.coder.Close() - c, err := net.DialUnix("unix", nil, &net.UnixAddr{self.endpoint, "unix"}) - if err == nil { - self.coder = self.codec.New(c) - - msg := shared.Request{ - Id: 0, - Method: useragent.EnableUserAgentMethod, - Jsonrpc: shared.JsonRpcVersion, - Params: []byte("[]"), - } - self.coder.WriteResponse(msg) - self.coder.Recv() - } - - return err -} - -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 nil, err - } - os.Remove(cfg.Endpoint) - l, err := net.Listen("unix", cfg.Endpoint) - if err != nil { - return nil, err - } - os.Chmod(cfg.Endpoint, 0600) - return l, nil -} diff --git a/rpc/comms/ipc_windows.go b/rpc/comms/ipc_windows.go deleted file mode 100644 index e25fba253..000000000 --- a/rpc/comms/ipc_windows.go +++ /dev/null @@ -1,697 +0,0 @@ -// Copyright 2015 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -// +build windows - -package comms - -import ( - "fmt" - "io" - "net" - "os" - "sync" - "syscall" - "time" - "unsafe" - - "github.com/ethereum/go-ethereum/rpc/codec" - "github.com/ethereum/go-ethereum/rpc/shared" - "github.com/ethereum/go-ethereum/rpc/useragent" -) - -var ( - modkernel32 = syscall.NewLazyDLL("kernel32.dll") - - procCreateNamedPipeW = modkernel32.NewProc("CreateNamedPipeW") - procConnectNamedPipe = modkernel32.NewProc("ConnectNamedPipe") - procDisconnectNamedPipe = modkernel32.NewProc("DisconnectNamedPipe") - procWaitNamedPipeW = modkernel32.NewProc("WaitNamedPipeW") - procCreateEventW = modkernel32.NewProc("CreateEventW") - procGetOverlappedResult = modkernel32.NewProc("GetOverlappedResult") - procCancelIoEx = modkernel32.NewProc("CancelIoEx") -) - -func createNamedPipe(name *uint16, openMode uint32, pipeMode uint32, maxInstances uint32, outBufSize uint32, inBufSize uint32, defaultTimeout uint32, sa *syscall.SecurityAttributes) (handle syscall.Handle, err error) { - r0, _, e1 := syscall.Syscall9(procCreateNamedPipeW.Addr(), 8, uintptr(unsafe.Pointer(name)), uintptr(openMode), uintptr(pipeMode), uintptr(maxInstances), uintptr(outBufSize), uintptr(inBufSize), uintptr(defaultTimeout), uintptr(unsafe.Pointer(sa)), 0) - handle = syscall.Handle(r0) - if handle == syscall.InvalidHandle { - if e1 != 0 { - err = error(e1) - } else { - err = syscall.EINVAL - } - } - return -} - -func cancelIoEx(handle syscall.Handle, overlapped *syscall.Overlapped) (err error) { - r1, _, e1 := syscall.Syscall(procCancelIoEx.Addr(), 2, uintptr(handle), uintptr(unsafe.Pointer(overlapped)), 0) - if r1 == 0 { - if e1 != 0 { - err = error(e1) - } else { - err = syscall.EINVAL - } - } - return -} - -func connectNamedPipe(handle syscall.Handle, overlapped *syscall.Overlapped) (err error) { - r1, _, e1 := syscall.Syscall(procConnectNamedPipe.Addr(), 2, uintptr(handle), uintptr(unsafe.Pointer(overlapped)), 0) - if r1 == 0 { - if e1 != 0 { - err = error(e1) - } else { - err = syscall.EINVAL - } - } - return -} - -func disconnectNamedPipe(handle syscall.Handle) (err error) { - r1, _, e1 := syscall.Syscall(procDisconnectNamedPipe.Addr(), 1, uintptr(handle), 0, 0) - if r1 == 0 { - if e1 != 0 { - err = error(e1) - } else { - err = syscall.EINVAL - } - } - return -} - -func waitNamedPipe(name *uint16, timeout uint32) (err error) { - r1, _, e1 := syscall.Syscall(procWaitNamedPipeW.Addr(), 2, uintptr(unsafe.Pointer(name)), uintptr(timeout), 0) - if r1 == 0 { - if e1 != 0 { - err = error(e1) - } else { - err = syscall.EINVAL - } - } - return -} - -func createEvent(sa *syscall.SecurityAttributes, manualReset bool, initialState bool, name *uint16) (handle syscall.Handle, err error) { - var _p0 uint32 - if manualReset { - _p0 = 1 - } else { - _p0 = 0 - } - var _p1 uint32 - if initialState { - _p1 = 1 - } else { - _p1 = 0 - } - r0, _, e1 := syscall.Syscall6(procCreateEventW.Addr(), 4, uintptr(unsafe.Pointer(sa)), uintptr(_p0), uintptr(_p1), uintptr(unsafe.Pointer(name)), 0, 0) - handle = syscall.Handle(r0) - if handle == syscall.InvalidHandle { - if e1 != 0 { - err = error(e1) - } else { - err = syscall.EINVAL - } - } - return -} - -func getOverlappedResult(handle syscall.Handle, overlapped *syscall.Overlapped, transferred *uint32, wait bool) (err error) { - var _p0 uint32 - if wait { - _p0 = 1 - } else { - _p0 = 0 - } - r1, _, e1 := syscall.Syscall6(procGetOverlappedResult.Addr(), 4, uintptr(handle), uintptr(unsafe.Pointer(overlapped)), uintptr(unsafe.Pointer(transferred)), uintptr(_p0), 0, 0) - if r1 == 0 { - if e1 != 0 { - err = error(e1) - } else { - err = syscall.EINVAL - } - } - return -} - -const ( - // openMode - pipe_access_duplex = 0x3 - pipe_access_inbound = 0x1 - pipe_access_outbound = 0x2 - - // openMode write flags - file_flag_first_pipe_instance = 0x00080000 - file_flag_write_through = 0x80000000 - file_flag_overlapped = 0x40000000 - - // openMode ACL flags - write_dac = 0x00040000 - write_owner = 0x00080000 - access_system_security = 0x01000000 - - // pipeMode - pipe_type_byte = 0x0 - pipe_type_message = 0x4 - - // pipeMode read mode flags - pipe_readmode_byte = 0x0 - pipe_readmode_message = 0x2 - - // pipeMode wait mode flags - pipe_wait = 0x0 - pipe_nowait = 0x1 - - // pipeMode remote-client mode flags - pipe_accept_remote_clients = 0x0 - pipe_reject_remote_clients = 0x8 - - pipe_unlimited_instances = 255 - - nmpwait_wait_forever = 0xFFFFFFFF - - // the two not-an-errors below occur if a client connects to the pipe between - // the server's CreateNamedPipe and ConnectNamedPipe calls. - error_no_data syscall.Errno = 0xE8 - error_pipe_connected syscall.Errno = 0x217 - error_pipe_busy syscall.Errno = 0xE7 - error_sem_timeout syscall.Errno = 0x79 - - error_bad_pathname syscall.Errno = 0xA1 - error_invalid_name syscall.Errno = 0x7B - - error_io_incomplete syscall.Errno = 0x3e4 -) - -var _ net.Conn = (*PipeConn)(nil) -var _ net.Listener = (*PipeListener)(nil) - -// ErrClosed is the error returned by PipeListener.Accept when Close is called -// on the PipeListener. -var ErrClosed = PipeError{"Pipe has been closed.", false} - -// PipeError is an error related to a call to a pipe -type PipeError struct { - msg string - timeout bool -} - -// Error implements the error interface -func (e PipeError) Error() string { - return e.msg -} - -// Timeout implements net.AddrError.Timeout() -func (e PipeError) Timeout() bool { - return e.timeout -} - -// Temporary implements net.AddrError.Temporary() -func (e PipeError) Temporary() bool { - return false -} - -// Dial connects to a named pipe with the given address. If the specified pipe is not available, -// it will wait indefinitely for the pipe to become available. -// -// The address must be of the form \\.\\pipe\ for local pipes and \\\pipe\ -// for remote pipes. -// -// Dial will return a PipeError if you pass in a badly formatted pipe name. -// -// Examples: -// // local pipe -// conn, err := Dial(`\\.\pipe\mypipename`) -// -// // remote pipe -// conn, err := Dial(`\\othercomp\pipe\mypipename`) -func Dial(address string) (*PipeConn, error) { - for { - conn, err := dial(address, nmpwait_wait_forever) - if err == nil { - return conn, nil - } - if isPipeNotReady(err) { - <-time.After(100 * time.Millisecond) - continue - } - return nil, err - } -} - -// DialTimeout acts like Dial, but will time out after the duration of timeout -func DialTimeout(address string, timeout time.Duration) (*PipeConn, error) { - deadline := time.Now().Add(timeout) - - now := time.Now() - for now.Before(deadline) { - millis := uint32(deadline.Sub(now) / time.Millisecond) - conn, err := dial(address, millis) - if err == nil { - return conn, nil - } - if err == error_sem_timeout { - // This is WaitNamedPipe's timeout error, so we know we're done - return nil, PipeError{fmt.Sprintf( - "Timed out waiting for pipe '%s' to come available", address), true} - } - if isPipeNotReady(err) { - left := deadline.Sub(time.Now()) - retry := 100 * time.Millisecond - if left > retry { - <-time.After(retry) - } else { - <-time.After(left - time.Millisecond) - } - now = time.Now() - continue - } - return nil, err - } - return nil, PipeError{fmt.Sprintf( - "Timed out waiting for pipe '%s' to come available", address), true} -} - -// isPipeNotReady checks the error to see if it indicates the pipe is not ready -func isPipeNotReady(err error) bool { - // Pipe Busy means another client just grabbed the open pipe end, - // and the server hasn't made a new one yet. - // File Not Found means the server hasn't created the pipe yet. - // Neither is a fatal error. - - return err == syscall.ERROR_FILE_NOT_FOUND || err == error_pipe_busy -} - -// newOverlapped creates a structure used to track asynchronous -// I/O requests that have been issued. -func newOverlapped() (*syscall.Overlapped, error) { - event, err := createEvent(nil, true, true, nil) - if err != nil { - return nil, err - } - return &syscall.Overlapped{HEvent: event}, nil -} - -// waitForCompletion waits for an asynchronous I/O request referred to by overlapped to complete. -// This function returns the number of bytes transferred by the operation and an error code if -// applicable (nil otherwise). -func waitForCompletion(handle syscall.Handle, overlapped *syscall.Overlapped) (uint32, error) { - _, err := syscall.WaitForSingleObject(overlapped.HEvent, syscall.INFINITE) - if err != nil { - return 0, err - } - var transferred uint32 - err = getOverlappedResult(handle, overlapped, &transferred, true) - return transferred, err -} - -// dial is a helper to initiate a connection to a named pipe that has been started by a server. -// The timeout is only enforced if the pipe server has already created the pipe, otherwise -// this function will return immediately. -func dial(address string, timeout uint32) (*PipeConn, error) { - name, err := syscall.UTF16PtrFromString(string(address)) - if err != nil { - return nil, err - } - // If at least one instance of the pipe has been created, this function - // will wait timeout milliseconds for it to become available. - // It will return immediately regardless of timeout, if no instances - // of the named pipe have been created yet. - // If this returns with no error, there is a pipe available. - if err := waitNamedPipe(name, timeout); err != nil { - if err == error_bad_pathname { - // badly formatted pipe name - return nil, badAddr(address) - } - return nil, err - } - pathp, err := syscall.UTF16PtrFromString(address) - if err != nil { - return nil, err - } - handle, err := syscall.CreateFile(pathp, syscall.GENERIC_READ|syscall.GENERIC_WRITE, - uint32(syscall.FILE_SHARE_READ|syscall.FILE_SHARE_WRITE), nil, syscall.OPEN_EXISTING, - syscall.FILE_FLAG_OVERLAPPED, 0) - if err != nil { - return nil, err - } - return &PipeConn{handle: handle, addr: PipeAddr(address)}, nil -} - -// Listen returns a new PipeListener that will listen on a pipe with the given -// address. The address must be of the form \\.\pipe\ -// -// Listen will return a PipeError for an incorrectly formatted pipe name. -func Listen(address string) (*PipeListener, error) { - handle, err := createPipe(address, true) - if err == error_invalid_name { - return nil, badAddr(address) - } - if err != nil { - return nil, err - } - return &PipeListener{ - addr: PipeAddr(address), - handle: handle, - }, nil -} - -// PipeListener is a named pipe listener. Clients should typically -// use variables of type net.Listener instead of assuming named pipe. -type PipeListener struct { - addr PipeAddr - handle syscall.Handle - closed bool - - // acceptHandle contains the current handle waiting for - // an incoming connection or nil. - acceptHandle syscall.Handle - // acceptOverlapped is set before waiting on a connection. - // If not waiting, it is nil. - acceptOverlapped *syscall.Overlapped - // acceptMutex protects the handle and overlapped structure. - acceptMutex sync.Mutex -} - -// Accept implements the Accept method in the net.Listener interface; it -// waits for the next call and returns a generic net.Conn. -func (l *PipeListener) Accept() (net.Conn, error) { - c, err := l.AcceptPipe() - for err == error_no_data { - // Ignore clients that connect and immediately disconnect. - c, err = l.AcceptPipe() - } - if err != nil { - return nil, err - } - return c, nil -} - -// AcceptPipe accepts the next incoming call and returns the new connection. -// It might return an error if a client connected and immediately cancelled -// the connection. -func (l *PipeListener) AcceptPipe() (*PipeConn, error) { - if l == nil || l.addr == "" || l.closed { - return nil, syscall.EINVAL - } - - // the first time we call accept, the handle will have been created by the Listen - // call. This is to prevent race conditions where the client thinks the server - // isn't listening because it hasn't actually called create yet. After the first time, we'll - // have to create a new handle each time - handle := l.handle - if handle == 0 { - var err error - handle, err = createPipe(string(l.addr), false) - if err != nil { - return nil, err - } - } else { - l.handle = 0 - } - - overlapped, err := newOverlapped() - if err != nil { - return nil, err - } - defer syscall.CloseHandle(overlapped.HEvent) - if err := connectNamedPipe(handle, overlapped); err != nil && err != error_pipe_connected { - if err == error_io_incomplete || err == syscall.ERROR_IO_PENDING { - l.acceptMutex.Lock() - l.acceptOverlapped = overlapped - l.acceptHandle = handle - l.acceptMutex.Unlock() - defer func() { - l.acceptMutex.Lock() - l.acceptOverlapped = nil - l.acceptHandle = 0 - l.acceptMutex.Unlock() - }() - - _, err = waitForCompletion(handle, overlapped) - } - if err == syscall.ERROR_OPERATION_ABORTED { - // Return error compatible to net.Listener.Accept() in case the - // listener was closed. - return nil, ErrClosed - } - if err != nil { - return nil, err - } - } - return &PipeConn{handle: handle, addr: l.addr}, nil -} - -// Close stops listening on the address. -// Already Accepted connections are not closed. -func (l *PipeListener) Close() error { - if l.closed { - return nil - } - l.closed = true - if l.handle != 0 { - err := disconnectNamedPipe(l.handle) - if err != nil { - return err - } - err = syscall.CloseHandle(l.handle) - if err != nil { - return err - } - l.handle = 0 - } - l.acceptMutex.Lock() - defer l.acceptMutex.Unlock() - if l.acceptOverlapped != nil && l.acceptHandle != 0 { - // Cancel the pending IO. This call does not block, so it is safe - // to hold onto the mutex above. - if err := cancelIoEx(l.acceptHandle, l.acceptOverlapped); err != nil { - return err - } - err := syscall.CloseHandle(l.acceptOverlapped.HEvent) - if err != nil { - return err - } - l.acceptOverlapped.HEvent = 0 - err = syscall.CloseHandle(l.acceptHandle) - if err != nil { - return err - } - l.acceptHandle = 0 - } - return nil -} - -// Addr returns the listener's network address, a PipeAddr. -func (l *PipeListener) Addr() net.Addr { return l.addr } - -// PipeConn is the implementation of the net.Conn interface for named pipe connections. -type PipeConn struct { - handle syscall.Handle - addr PipeAddr - - // these aren't actually used yet - readDeadline *time.Time - writeDeadline *time.Time -} - -type iodata struct { - n uint32 - err error -} - -// completeRequest looks at iodata to see if a request is pending. If so, it waits for it to either complete or to -// abort due to hitting the specified deadline. Deadline may be set to nil to wait forever. If no request is pending, -// the content of iodata is returned. -func (c *PipeConn) completeRequest(data iodata, deadline *time.Time, overlapped *syscall.Overlapped) (int, error) { - if data.err == error_io_incomplete || data.err == syscall.ERROR_IO_PENDING { - var timer <-chan time.Time - if deadline != nil { - if timeDiff := deadline.Sub(time.Now()); timeDiff > 0 { - timer = time.After(timeDiff) - } - } - done := make(chan iodata) - go func() { - n, err := waitForCompletion(c.handle, overlapped) - done <- iodata{n, err} - }() - select { - case data = <-done: - case <-timer: - syscall.CancelIoEx(c.handle, overlapped) - data = iodata{0, timeout(c.addr.String())} - } - } - // Windows will produce ERROR_BROKEN_PIPE upon closing - // a handle on the other end of a connection. Go RPC - // expects an io.EOF error in this case. - if data.err == syscall.ERROR_BROKEN_PIPE { - data.err = io.EOF - } - return int(data.n), data.err -} - -// Read implements the net.Conn Read method. -func (c *PipeConn) Read(b []byte) (int, error) { - // Use ReadFile() rather than Read() because the latter - // contains a workaround that eats ERROR_BROKEN_PIPE. - overlapped, err := newOverlapped() - if err != nil { - return 0, err - } - defer syscall.CloseHandle(overlapped.HEvent) - var n uint32 - err = syscall.ReadFile(c.handle, b, &n, overlapped) - return c.completeRequest(iodata{n, err}, c.readDeadline, overlapped) -} - -// Write implements the net.Conn Write method. -func (c *PipeConn) Write(b []byte) (int, error) { - overlapped, err := newOverlapped() - if err != nil { - return 0, err - } - defer syscall.CloseHandle(overlapped.HEvent) - var n uint32 - err = syscall.WriteFile(c.handle, b, &n, overlapped) - return c.completeRequest(iodata{n, err}, c.writeDeadline, overlapped) -} - -// Close closes the connection. -func (c *PipeConn) Close() error { - return syscall.CloseHandle(c.handle) -} - -// LocalAddr returns the local network address. -func (c *PipeConn) LocalAddr() net.Addr { - return c.addr -} - -// RemoteAddr returns the remote network address. -func (c *PipeConn) RemoteAddr() net.Addr { - // not sure what to do here, we don't have remote addr.... - return c.addr -} - -// SetDeadline implements the net.Conn SetDeadline method. -// Note that timeouts are only supported on Windows Vista/Server 2008 and above -func (c *PipeConn) SetDeadline(t time.Time) error { - c.SetReadDeadline(t) - c.SetWriteDeadline(t) - return nil -} - -// SetReadDeadline implements the net.Conn SetReadDeadline method. -// Note that timeouts are only supported on Windows Vista/Server 2008 and above -func (c *PipeConn) SetReadDeadline(t time.Time) error { - c.readDeadline = &t - return nil -} - -// SetWriteDeadline implements the net.Conn SetWriteDeadline method. -// Note that timeouts are only supported on Windows Vista/Server 2008 and above -func (c *PipeConn) SetWriteDeadline(t time.Time) error { - c.writeDeadline = &t - return nil -} - -// PipeAddr represents the address of a named pipe. -type PipeAddr string - -// Network returns the address's network name, "pipe". -func (a PipeAddr) Network() string { return "pipe" } - -// String returns the address of the pipe -func (a PipeAddr) String() string { - return string(a) -} - -// createPipe is a helper function to make sure we always create pipes -// with the same arguments, since subsequent calls to create pipe need -// to use the same arguments as the first one. If first is set, fail -// if the pipe already exists. -func createPipe(address string, first bool) (syscall.Handle, error) { - n, err := syscall.UTF16PtrFromString(address) - if err != nil { - return 0, err - } - mode := uint32(pipe_access_duplex | syscall.FILE_FLAG_OVERLAPPED) - if first { - mode |= file_flag_first_pipe_instance - } - return createNamedPipe(n, - mode, - pipe_type_byte, - pipe_unlimited_instances, - 512, 512, 0, nil) -} - -func badAddr(addr string) PipeError { - return PipeError{fmt.Sprintf("Invalid pipe address '%s'.", addr), false} -} -func timeout(addr string) PipeError { - return PipeError{fmt.Sprintf("Pipe IO timed out waiting for '%s'", addr), true} -} - -func newIpcClient(cfg IpcConfig, codec codec.Codec) (*ipcClient, error) { - c, err := Dial(cfg.Endpoint) - if err != nil { - return nil, err - } - - coder := codec.New(c) - msg := shared.Request{ - Id: 0, - Method: useragent.EnableUserAgentMethod, - Jsonrpc: shared.JsonRpcVersion, - Params: []byte("[]"), - } - - coder.WriteResponse(msg) - coder.Recv() - - return &ipcClient{cfg.Endpoint, c, codec, coder}, nil -} - -func (self *ipcClient) reconnect() error { - c, err := Dial(self.endpoint) - if err == nil { - self.coder = self.codec.New(c) - - req := shared.Request{ - Id: 0, - Method: useragent.EnableUserAgentMethod, - Jsonrpc: shared.JsonRpcVersion, - Params: []byte("[]"), - } - self.coder.WriteResponse(req) - self.coder.Recv() - } - return err -} - -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 nil, err - } - os.Chmod(cfg.Endpoint, 0600) - return l, nil -} -- cgit v1.2.3