From 3f047be5aa93b6222506445414ca909dd59c7eeb Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Wed, 22 Jul 2015 18:48:40 +0200 Subject: all: update license headers to distiguish GPL/LGPL All code outside of cmd/ is licensed as LGPL. The headers now reflect this by calling the whole work "the go-ethereum library". --- rpc/comms/http.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) (limited to 'rpc/comms/http.go') diff --git a/rpc/comms/http.go b/rpc/comms/http.go index 3fb429e65..91eacca91 100644 --- a/rpc/comms/http.go +++ b/rpc/comms/http.go @@ -1,18 +1,18 @@ // Copyright 2015 The go-ethereum Authors -// This file is part of go-ethereum. +// This file is part of the go-ethereum library. // // go-ethereum 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. // -// go-ethereum is distributed in the hope that it will be useful, +// 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 +// 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 go-ethereum. If not, see . +// along with the go-ethereum library. If not, see . package comms -- cgit v1.2.3 From bfbcfbe4a9dd9125391b56d6a13158cc5ec71c89 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Thu, 23 Jul 2015 18:35:11 +0200 Subject: all: fix license headers one more time I forgot to update one instance of "go-ethereum" in commit 3f047be5a. --- rpc/comms/http.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'rpc/comms/http.go') diff --git a/rpc/comms/http.go b/rpc/comms/http.go index 91eacca91..108ba0c5f 100644 --- a/rpc/comms/http.go +++ b/rpc/comms/http.go @@ -1,7 +1,7 @@ // Copyright 2015 The go-ethereum Authors // This file is part of the go-ethereum library. // -// go-ethereum is free software: you can redistribute it and/or modify +// 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. -- cgit v1.2.3 From ba14957cf3ce8415bd930310a0edd1fce76ddb22 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Mon, 3 Aug 2015 02:29:59 +0200 Subject: rpc/comms: use ConnState to track HTTP connections The JSON-RPC server wraps TCP connections in order to be able to shut down them down when RPC is stopped. This is rather scary code. Go 1.3 introduced the http.Server.ConnState hook for such purposes. We can use this facility now that we depend on Go 1.4. There are multiple reasons for the switch apart from making the code less scary: * the TCP listener no longer ticks every second to check a channel * pending requests are allowed to finish after stopping the server * we can time out idle keep-alive connections --- rpc/comms/http.go | 185 +++++++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 162 insertions(+), 23 deletions(-) (limited to 'rpc/comms/http.go') diff --git a/rpc/comms/http.go b/rpc/comms/http.go index 108ba0c5f..c165aa27e 100644 --- a/rpc/comms/http.go +++ b/rpc/comms/http.go @@ -17,11 +17,16 @@ package comms import ( + "encoding/json" "fmt" + "net" "net/http" "strings" + "sync" + "time" "bytes" + "io" "io/ioutil" "github.com/ethereum/go-ethereum/logger" @@ -31,10 +36,15 @@ import ( "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 ( - // main HTTP rpc listener - httpListener *stoppableTCPListener - listenerStoppedError = fmt.Errorf("Listener has stopped") + httpServerMu sync.Mutex + httpServer *stopServer ) type HttpConfig struct { @@ -43,42 +53,171 @@ type HttpConfig struct { 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 { - if httpListener != nil { - if fmt.Sprintf("%s:%d", cfg.ListenAddress, cfg.ListenPort) != httpListener.Addr().String() { - return fmt.Errorf("RPC service already running on %s ", httpListener.Addr().String()) + 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 } - - l, err := newStoppableTCPListener(fmt.Sprintf("%s:%d", cfg.ListenAddress, cfg.ListenPort)) + // 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 } - httpListener = l + httpServer = s + return nil +} - var handler http.Handler - if len(cfg.CorsDomain) > 0 { - var opts cors.Options - opts.AllowedMethods = []string{"POST"} - opts.AllowedOrigins = strings.Split(cfg.CorsDomain, " ") +func (h *handler) ServeHTTP(w http.ResponseWriter, req *http.Request) { + w.Header().Set("Content-Type", "application/json") - c := cors.New(opts) - handler = newStoppableHandler(c.Handler(gethHttpHandler(codec, api)), l.stop) - } else { - handler = newStoppableHandler(gethHttpHandler(codec, api), l.stop) + // 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 } - go http.Serve(l, handler) + 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 + } - return nil + 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() { - if httpListener != nil { - httpListener.Stop() - httpListener = nil + 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) } } -- cgit v1.2.3 From 5c949d3b3ba81ea0563575b19a7b148aeac4bf61 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Mon, 3 Aug 2015 02:45:33 +0200 Subject: fdtrack: temporary hack for tracking file descriptor usage Package fdtrack logs statistics about open file descriptors. This should help identify the source of #1549. --- rpc/comms/http.go | 2 ++ 1 file changed, 2 insertions(+) (limited to 'rpc/comms/http.go') diff --git a/rpc/comms/http.go b/rpc/comms/http.go index c165aa27e..c08b744a1 100644 --- a/rpc/comms/http.go +++ b/rpc/comms/http.go @@ -29,6 +29,7 @@ import ( "io" "io/ioutil" + "github.com/ethereum/go-ethereum/fdtrack" "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/rpc/codec" @@ -177,6 +178,7 @@ func listenHTTP(addr string, h http.Handler) (*stopServer, error) { if err != nil { return nil, err } + l = fdtrack.WrapListener("rpc", l) s := &stopServer{l: l, idle: make(map[net.Conn]struct{})} s.Server = &http.Server{ Addr: addr, -- cgit v1.2.3