Merge pull request #94 from binance-chain/v0.30.1-binance.6-beta.1

[R4R] prepare Release 0.30.1-binance.7

Author: unclezoro

config/config.go

@@ -348,6 +348,22 @@ type RPCConfig struct {
 	// Should be < {ulimit -Sn} - {MaxNumInboundPeers} - {MaxNumOutboundPeers} - {N of wal, db and other open files}
 	// 1024 - 40 - 10 - 50 = 924 = ~900
 	MaxOpenConnections int `mapstructure:"max_open_connections"`
+
+	// Maximum number of go routine to process websocket request.
+	// 1 - process websocket request synchronously.
+	// 10 - default size.
+	// Should be {WebsocketPoolSpawnSize} =< {WebsocketPoolMaxSize}
+	WebsocketPoolMaxSize int `mapstructure:"websocket_pool_size"`
+
+	// The queued buffer for workers to process requests.
+	// 10 -default
+	WebsocketPoolQueueSize int `mapstructure:"websocket_pool_queue_size"`
+
+	// The initial size of goroutines in pool.
+	// 1 - process websocket request synchronously.
+	// 5 - default size
+	// Should be {WebsocketPoolSpawnSize} =< {WebsocketPoolMaxSize}
+	WebsocketPoolSpawnSize int `mapstructure:"websocket_pool_spawn_size"`
 }
 
 // DefaultRPCConfig returns a default configuration for the RPC server
@@ -360,8 +376,11 @@ func DefaultRPCConfig() *RPCConfig {
 		GRPCListenAddress:      "",
 		GRPCMaxOpenConnections: 900,
 
-		Unsafe:             false,
-		MaxOpenConnections: 900,
+		Unsafe:                 false,
+		MaxOpenConnections:     900,
+		WebsocketPoolMaxSize:   10,
+		WebsocketPoolQueueSize: 10,
+		WebsocketPoolSpawnSize: 5,
 	}
 }
 
@@ -383,6 +402,18 @@ func (cfg *RPCConfig) ValidateBasic() error {
 	if cfg.MaxOpenConnections < 0 {
 		return errors.New("max_open_connections can't be negative")
 	}
+	if cfg.WebsocketPoolMaxSize < 1 {
+		return errors.New("websocket_pool_size can't be less than 1")
+	}
+	if cfg.WebsocketPoolQueueSize < 1 {
+		return errors.New("websocket_pool_queue_size can't be less than 1")
+	}
+	if cfg.WebsocketPoolSpawnSize < 1 {
+		return errors.New("websocket_pool_spawn_size can't be less than 1")
+	}
+	if cfg.WebsocketPoolSpawnSize > cfg.WebsocketPoolMaxSize {
+		return errors.New("websocket_pool_spawn_size can't be large than websocket_pool_size")
+	}
 	return nil
 }
 

config/toml.go

@@ -177,6 +177,23 @@ unsafe = {{ .RPC.Unsafe }}
 # 1024 - 40 - 10 - 50 = 924 = ~900
 max_open_connections = {{ .RPC.MaxOpenConnections }}
 
+
+# Maximum number of go routine to process websocket request.
+# 1 - process websocket request synchronously.
+# 10 - default size.
+# Should be {WebsocketPoolSpawnSize} =< {WebsocketPoolMaxSize}
+websocket_pool_size = {{ .RPC.WebsocketPoolMaxSize }}
+	
+# The queued buffer for workers to process requests.
+# 10 -default
+websocket_pool_queue_size = {{ .RPC.WebsocketPoolQueueSize }}
+
+# The initial size of goroutines in pool.
+# 1 - process websocket request synchronously.
+# 5 - default size
+# Should be {WebsocketPoolSpawnSize} =< {WebsocketPoolMaxSize}
+websocket_pool_spawn_size = {{ .RPC.WebsocketPoolSpawnSize }}
+
 ##### peer to peer configuration options #####
 [p2p]
 

libs/gopool/pool.go

@@ -0,0 +1,83 @@
+// Package gopool contains tools for goroutine reuse.
+package gopool
+
+import (
+	"fmt"
+	"runtime/debug"
+	"time"
+
+	"github.com/tendermint/tendermint/libs/common"
+)
+
+// ErrScheduleTimeout returned by Pool to indicate that there no free
+// goroutines during some period of time.
+var ErrScheduleTimeout = fmt.Errorf("schedule error: timed out")
+
+// Pool contains logic of goroutine reuse.
+type Pool struct {
+	common.BaseService
+	sem  chan struct{}
+	work chan func()
+}
+
+// NewPool creates new goroutine pool with given size. It also creates a work
+// queue of given size. Finally, it spawns given amount of goroutines
+// immediately.
+func NewPool(size, queue, spawn int) *Pool {
+	if spawn <= 0 && queue > 0 {
+		panic("dead queue configuration detected")
+	}
+	if spawn > size {
+		panic("spawn > workers")
+	}
+	p := &Pool{
+		sem:  make(chan struct{}, size),
+		work: make(chan func(), queue),
+	}
+	p.BaseService = *common.NewBaseService(nil, "routine-pool", p)
+
+	for i := 0; i < spawn; i++ {
+		p.sem <- struct{}{}
+		go p.worker(func() {})
+	}
+
+	return p
+}
+
+// Schedule schedules task to be executed over pool's workers.
+func (p *Pool) Schedule(task func()) {
+	p.schedule(task, nil)
+}
+
+// ScheduleTimeout schedules task to be executed over pool's workers.
+// It returns ErrScheduleTimeout when no free workers met during given timeout.
+func (p *Pool) ScheduleTimeout(timeout time.Duration, task func()) error {
+	return p.schedule(task, time.After(timeout))
+}
+
+func (p *Pool) schedule(task func(), timeout <-chan time.Time) error {
+	select {
+	case <-timeout:
+		return ErrScheduleTimeout
+	case p.work <- task:
+		return nil
+	case p.sem <- struct{}{}:
+		go p.worker(task)
+		return nil
+	}
+}
+
+func (p *Pool) worker(task func()) {
+	defer func() {
+		if r := recover(); r != nil {
+			p.Logger.Error("failed to process task", "err", r, "stack", string(debug.Stack()))
+		}
+		<-p.sem
+	}()
+
+	task()
+
+	for task := range p.work {
+		task()
+	}
+}

node/node.go

@@ -25,6 +25,7 @@ import (
 	"github.com/tendermint/tendermint/evidence"
 	cmn "github.com/tendermint/tendermint/libs/common"
 	dbm "github.com/tendermint/tendermint/libs/db"
+	"github.com/tendermint/tendermint/libs/gopool"
 	"github.com/tendermint/tendermint/libs/log"
 	mempl "github.com/tendermint/tendermint/mempool"
 	"github.com/tendermint/tendermint/p2p"
@@ -722,10 +723,17 @@ func (n *Node) startRPC() ([]net.Listener, error) {
 
 	// we may expose the rpc over both a unix and tcp socket
 	listeners := make([]net.Listener, len(listenAddrs))
+	var wsWorkerPool *gopool.Pool
+	if n.config.RPC.WebsocketPoolMaxSize > 1{
+		wsWorkerPool = gopool.NewPool(n.config.RPC.WebsocketPoolMaxSize, n.config.RPC.WebsocketPoolQueueSize, n.config.RPC.WebsocketPoolSpawnSize)
+		wsWorkerPool.SetLogger(n.Logger.With("module", "routine-pool"))
+	}
+
 	for i, listenAddr := range listenAddrs {
 		mux := http.NewServeMux()
 		rpcLogger := n.Logger.With("module", "rpc-server")
-		wm := rpcserver.NewWebsocketManager(rpccore.Routes, coreCodec, rpcserver.EventSubscriber(n.eventBus))
+
+		wm := rpcserver.NewWebsocketManager(rpccore.Routes, coreCodec, rpcserver.EventSubscriber(n.eventBus), rpcserver.SetWorkerPool(wsWorkerPool))
 		wm.SetLogger(rpcLogger.With("protocol", "websocket"))
 		mux.HandleFunc("/websocket", wm.WebsocketHandler)
 		rpcserver.RegisterRPCFuncs(mux, rpccore.Routes, coreCodec, rpcLogger)

rpc/lib/server/handlers.go

@@ -19,6 +19,7 @@ import (
 
 	amino "github.com/tendermint/go-amino"
 	cmn "github.com/tendermint/tendermint/libs/common"
+	"github.com/tendermint/tendermint/libs/gopool"
 	"github.com/tendermint/tendermint/libs/log"
 	types "github.com/tendermint/tendermint/rpc/lib/types"
 )
@@ -436,6 +437,8 @@ type wsConnection struct {
 
 	// object that is used to subscribe / unsubscribe from events
 	eventSub types.EventSubscriber
+
+	workerPool *gopool.Pool
 }
 
 // NewWSConnection wraps websocket.Conn.
@@ -477,6 +480,12 @@ func EventSubscriber(eventSub types.EventSubscriber) func(*wsConnection) {
 	}
 }
 
+func SetWorkerPool(workerPool *gopool.Pool) func(*wsConnection) {
+	return func(wsc *wsConnection) {
+		wsc.workerPool = workerPool
+	}
+}
+
 // WriteWait sets the amount of time to wait before a websocket write times out.
 // It should only be used in the constructor - not Goroutine-safe.
 func WriteWait(writeWait time.Duration) func(*wsConnection) {
@@ -611,57 +620,68 @@ func (wsc *wsConnection) readRoutine() {
 				wsc.Stop()
 				return
 			}
-
-			var request types.RPCRequest
-			err = json.Unmarshal(in, &request)
-			if err != nil {
-				wsc.WriteRPCResponse(types.RPCParseError(types.JSONRPCStringID(""), errors.Wrap(err, "Error unmarshaling request")))
-				continue
-			}
-
-			// A Notification is a Request object without an "id" member.
-			// The Server MUST NOT reply to a Notification, including those that are within a batch request.
-			if request.ID == types.JSONRPCStringID("") {
-				wsc.Logger.Debug("WSJSONRPC received a notification, skipping... (please send a non-empty ID if you want to call a method)")
-				continue
+			// Fetch the RPCFunc and execute it.
+			if wsc.workerPool != nil {
+				wsc.workerPool.Schedule(func() {
+					wsc.processRequest(in)
+				})
+			} else {
+				wsc.processRequest(in)
 			}
 
-			// Now, fetch the RPCFunc and execute it.
+		}
+	}
+}
 
-			rpcFunc := wsc.funcMap[request.Method]
-			if rpcFunc == nil {
-				wsc.WriteRPCResponse(types.RPCMethodNotFoundError(request.ID))
-				continue
-			}
-			var args []reflect.Value
-			if rpcFunc.ws {
-				wsCtx := types.WSRPCContext{Request: request, WSRPCConnection: wsc}
-				if len(request.Params) > 0 {
-					args, err = jsonParamsToArgsWS(rpcFunc, wsc.cdc, request.Params, wsCtx)
-				}
-			} else {
-				if len(request.Params) > 0 {
-					args, err = jsonParamsToArgsRPC(rpcFunc, wsc.cdc, request.Params)
-				}
-			}
-			if err != nil {
-				wsc.WriteRPCResponse(types.RPCInternalError(request.ID, errors.Wrap(err, "Error converting json params to arguments")))
-				continue
-			}
-			returns := rpcFunc.f.Call(args)
+func (wsc *wsConnection) processRequest(message []byte) {
+	var request types.RPCRequest
+	err := json.Unmarshal(message, &request)
+	if err != nil {
+		wsc.WriteRPCResponse(types.RPCParseError(types.JSONRPCStringID(""), errors.Wrap(err, "Error unmarshaling request")))
+		return
+	}
 
-			// TODO: Need to encode args/returns to string if we want to log them
-			wsc.Logger.Info("WSJSONRPC", "method", request.Method)
+	// A Notification is a Request object without an "id" member.
+	// The Server MUST NOT reply to a Notification, including those that are within a batch request.
+	if request.ID == types.JSONRPCStringID("") {
+		wsc.Logger.Debug("WSJSONRPC received a notification, skipping... (please send a non-empty ID if you want to call a method)")
+		return
+	}
 
-			result, err := unreflectResult(returns)
-			if err != nil {
-				wsc.WriteRPCResponse(types.RPCInternalError(request.ID, err))
-				continue
-			}
+	// Now, fetch the RPCFunc and execute it.
 
-			wsc.WriteRPCResponse(types.NewRPCSuccessResponse(wsc.cdc, request.ID, result))
+	rpcFunc := wsc.funcMap[request.Method]
+	if rpcFunc == nil {
+		wsc.WriteRPCResponse(types.RPCMethodNotFoundError(request.ID))
+		return
+	}
+	var args []reflect.Value
+	if rpcFunc.ws {
+		wsCtx := types.WSRPCContext{Request: request, WSRPCConnection: wsc}
+		if len(request.Params) > 0 {
+			args, err = jsonParamsToArgsWS(rpcFunc, wsc.cdc, request.Params, wsCtx)
 		}
+	} else {
+		if len(request.Params) > 0 {
+			args, err = jsonParamsToArgsRPC(rpcFunc, wsc.cdc, request.Params)
+		}
+	}
+	if err != nil {
+		wsc.WriteRPCResponse(types.RPCInternalError(request.ID, errors.Wrap(err, "Error converting json params to arguments")))
+		return
+	}
+	returns := rpcFunc.f.Call(args)
+
+	// TODO: Need to encode args/returns to string if we want to log them
+	wsc.Logger.Info("WSJSONRPC", "method", request.Method)
+
+	result, err := unreflectResult(returns)
+	if err != nil {
+		wsc.WriteRPCResponse(types.RPCInternalError(request.ID, err))
+		return
 	}
+
+	wsc.WriteRPCResponse(types.NewRPCSuccessResponse(wsc.cdc, request.ID, result))
 }
 
 // receives on a write channel and writes out on the socket

rpc/test/helpers.go

@@ -89,6 +89,9 @@ func GetConfig() *cfg.Config {
 		globalConfig.RPC.ListenAddress = rpc
 		globalConfig.RPC.CORSAllowedOrigins = []string{"https://tendermint.com/"}
 		globalConfig.RPC.GRPCListenAddress = grpc
+		globalConfig.RPC.WebsocketPoolMaxSize = 1
+		globalConfig.RPC.WebsocketPoolQueueSize = 1
+		globalConfig.RPC.WebsocketPoolSpawnSize = 1
 		globalConfig.TxIndex.IndexTags = "app.creator,tx.height" // see kvstore application
 	}
 	return globalConfig