You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

167 lines
4.0 KiB

limit number of /subscribe clients and queries per client (#3269) * limit number of /subscribe clients and queries per client Add the following config variables (under [rpc] section): * max_subscription_clients * max_subscriptions_per_client * timeout_broadcast_tx_commit Fixes #2826 new HTTPClient interface for subscriptions finalize HTTPClient events interface remove EventSubscriber fix data race ``` WARNING: DATA RACE Read at 0x00c000a36060 by goroutine 129: github.com/tendermint/tendermint/rpc/client.(*Local).Subscribe.func1() /go/src/github.com/tendermint/tendermint/rpc/client/localclient.go:168 +0x1f0 Previous write at 0x00c000a36060 by goroutine 132: github.com/tendermint/tendermint/rpc/client.(*Local).Subscribe() /go/src/github.com/tendermint/tendermint/rpc/client/localclient.go:191 +0x4e0 github.com/tendermint/tendermint/rpc/client.WaitForOneEvent() /go/src/github.com/tendermint/tendermint/rpc/client/helpers.go:64 +0x178 github.com/tendermint/tendermint/rpc/client_test.TestTxEventsSentWithBroadcastTxSync.func1() /go/src/github.com/tendermint/tendermint/rpc/client/event_test.go:139 +0x298 testing.tRunner() /usr/local/go/src/testing/testing.go:827 +0x162 Goroutine 129 (running) created at: github.com/tendermint/tendermint/rpc/client.(*Local).Subscribe() /go/src/github.com/tendermint/tendermint/rpc/client/localclient.go:164 +0x4b7 github.com/tendermint/tendermint/rpc/client.WaitForOneEvent() /go/src/github.com/tendermint/tendermint/rpc/client/helpers.go:64 +0x178 github.com/tendermint/tendermint/rpc/client_test.TestTxEventsSentWithBroadcastTxSync.func1() /go/src/github.com/tendermint/tendermint/rpc/client/event_test.go:139 +0x298 testing.tRunner() /usr/local/go/src/testing/testing.go:827 +0x162 Goroutine 132 (running) created at: testing.(*T).Run() /usr/local/go/src/testing/testing.go:878 +0x659 github.com/tendermint/tendermint/rpc/client_test.TestTxEventsSentWithBroadcastTxSync() /go/src/github.com/tendermint/tendermint/rpc/client/event_test.go:119 +0x186 testing.tRunner() /usr/local/go/src/testing/testing.go:827 +0x162 ================== ``` lite client works (tested manually) godoc comments httpclient: do not close the out channel use TimeoutBroadcastTxCommit no timeout for unsubscribe but 1s Local (5s HTTP) timeout for resubscribe format code change Subscribe#out cap to 1 and replace config vars with RPCConfig TimeoutBroadcastTxCommit can't be greater than rpcserver.WriteTimeout rpc: Context as first parameter to all functions reformat code fixes after my own review fixes after Ethan's review add test stubs fix config.toml * fixes after manual testing - rpc: do not recommend to use BroadcastTxCommit because it's slow and wastes Tendermint resources (pubsub) - rpc: better error in Subscribe and BroadcastTxCommit - HTTPClient: do not resubscribe if err = ErrAlreadySubscribed * fixes after Ismail's review * Update rpc/grpc/grpc_test.go Co-Authored-By: melekes <anton.kalyaev@gmail.com>
6 years ago
limit number of /subscribe clients and queries per client (#3269) * limit number of /subscribe clients and queries per client Add the following config variables (under [rpc] section): * max_subscription_clients * max_subscriptions_per_client * timeout_broadcast_tx_commit Fixes #2826 new HTTPClient interface for subscriptions finalize HTTPClient events interface remove EventSubscriber fix data race ``` WARNING: DATA RACE Read at 0x00c000a36060 by goroutine 129: github.com/tendermint/tendermint/rpc/client.(*Local).Subscribe.func1() /go/src/github.com/tendermint/tendermint/rpc/client/localclient.go:168 +0x1f0 Previous write at 0x00c000a36060 by goroutine 132: github.com/tendermint/tendermint/rpc/client.(*Local).Subscribe() /go/src/github.com/tendermint/tendermint/rpc/client/localclient.go:191 +0x4e0 github.com/tendermint/tendermint/rpc/client.WaitForOneEvent() /go/src/github.com/tendermint/tendermint/rpc/client/helpers.go:64 +0x178 github.com/tendermint/tendermint/rpc/client_test.TestTxEventsSentWithBroadcastTxSync.func1() /go/src/github.com/tendermint/tendermint/rpc/client/event_test.go:139 +0x298 testing.tRunner() /usr/local/go/src/testing/testing.go:827 +0x162 Goroutine 129 (running) created at: github.com/tendermint/tendermint/rpc/client.(*Local).Subscribe() /go/src/github.com/tendermint/tendermint/rpc/client/localclient.go:164 +0x4b7 github.com/tendermint/tendermint/rpc/client.WaitForOneEvent() /go/src/github.com/tendermint/tendermint/rpc/client/helpers.go:64 +0x178 github.com/tendermint/tendermint/rpc/client_test.TestTxEventsSentWithBroadcastTxSync.func1() /go/src/github.com/tendermint/tendermint/rpc/client/event_test.go:139 +0x298 testing.tRunner() /usr/local/go/src/testing/testing.go:827 +0x162 Goroutine 132 (running) created at: testing.(*T).Run() /usr/local/go/src/testing/testing.go:878 +0x659 github.com/tendermint/tendermint/rpc/client_test.TestTxEventsSentWithBroadcastTxSync() /go/src/github.com/tendermint/tendermint/rpc/client/event_test.go:119 +0x186 testing.tRunner() /usr/local/go/src/testing/testing.go:827 +0x162 ================== ``` lite client works (tested manually) godoc comments httpclient: do not close the out channel use TimeoutBroadcastTxCommit no timeout for unsubscribe but 1s Local (5s HTTP) timeout for resubscribe format code change Subscribe#out cap to 1 and replace config vars with RPCConfig TimeoutBroadcastTxCommit can't be greater than rpcserver.WriteTimeout rpc: Context as first parameter to all functions reformat code fixes after my own review fixes after Ethan's review add test stubs fix config.toml * fixes after manual testing - rpc: do not recommend to use BroadcastTxCommit because it's slow and wastes Tendermint resources (pubsub) - rpc: better error in Subscribe and BroadcastTxCommit - HTTPClient: do not resubscribe if err = ErrAlreadySubscribed * fixes after Ismail's review * Update rpc/grpc/grpc_test.go Co-Authored-By: melekes <anton.kalyaev@gmail.com>
6 years ago
  1. package core
  2. import (
  3. "fmt"
  4. "time"
  5. cfg "github.com/tendermint/tendermint/config"
  6. "github.com/tendermint/tendermint/consensus"
  7. "github.com/tendermint/tendermint/crypto"
  8. "github.com/tendermint/tendermint/libs/log"
  9. mempl "github.com/tendermint/tendermint/mempool"
  10. "github.com/tendermint/tendermint/p2p"
  11. "github.com/tendermint/tendermint/proxy"
  12. sm "github.com/tendermint/tendermint/state"
  13. "github.com/tendermint/tendermint/state/txindex"
  14. "github.com/tendermint/tendermint/types"
  15. )
  16. const (
  17. // see README
  18. defaultPerPage = 30
  19. maxPerPage = 100
  20. // SubscribeTimeout is the maximum time we wait to subscribe for an event.
  21. // must be less than the server's write timeout (see rpcserver.DefaultConfig)
  22. SubscribeTimeout = 5 * time.Second
  23. )
  24. var (
  25. // set by Node
  26. env *Environment
  27. )
  28. // SetEnvironment sets up the given Environment.
  29. // It will race if multiple Node call SetEnvironment.
  30. func SetEnvironment(e *Environment) {
  31. env = e
  32. }
  33. //----------------------------------------------
  34. // These interfaces are used by RPC and must be thread safe
  35. type Consensus interface {
  36. GetState() sm.State
  37. GetValidators() (int64, []*types.Validator)
  38. GetLastHeight() int64
  39. GetRoundStateJSON() ([]byte, error)
  40. GetRoundStateSimpleJSON() ([]byte, error)
  41. }
  42. type transport interface {
  43. Listeners() []string
  44. IsListening() bool
  45. NodeInfo() p2p.NodeInfo
  46. }
  47. type peers interface {
  48. AddPersistentPeers([]string) error
  49. AddUnconditionalPeerIDs([]string) error
  50. AddPrivatePeerIDs([]string) error
  51. DialPeersAsync([]string) error
  52. Peers() p2p.IPeerSet
  53. }
  54. //----------------------------------------------
  55. // Environment contains objects and interfaces used by the RPC. It is expected
  56. // to be setup once during startup.
  57. type Environment struct {
  58. // external, thread safe interfaces
  59. ProxyAppQuery proxy.AppConnQuery
  60. ProxyAppMempool proxy.AppConnMempool
  61. // interfaces defined in types and above
  62. StateStore sm.Store
  63. BlockStore sm.BlockStore
  64. EvidencePool sm.EvidencePool
  65. ConsensusState Consensus
  66. P2PPeers peers
  67. P2PTransport transport
  68. // objects
  69. PubKey crypto.PubKey
  70. GenDoc *types.GenesisDoc // cache the genesis structure
  71. TxIndexer txindex.TxIndexer
  72. ConsensusReactor *consensus.Reactor
  73. EventBus *types.EventBus // thread safe
  74. Mempool mempl.Mempool
  75. Logger log.Logger
  76. Config cfg.RPCConfig
  77. }
  78. //----------------------------------------------
  79. func validatePage(pagePtr *int, perPage, totalCount int) (int, error) {
  80. if perPage < 1 {
  81. panic(fmt.Sprintf("zero or negative perPage: %d", perPage))
  82. }
  83. if pagePtr == nil { // no page parameter
  84. return 1, nil
  85. }
  86. pages := ((totalCount - 1) / perPage) + 1
  87. if pages == 0 {
  88. pages = 1 // one page (even if it's empty)
  89. }
  90. page := *pagePtr
  91. if page <= 0 || page > pages {
  92. return 1, fmt.Errorf("page should be within [1, %d] range, given %d", pages, page)
  93. }
  94. return page, nil
  95. }
  96. func validatePerPage(perPagePtr *int) int {
  97. if perPagePtr == nil { // no per_page parameter
  98. return defaultPerPage
  99. }
  100. perPage := *perPagePtr
  101. if perPage < 1 {
  102. return defaultPerPage
  103. } else if perPage > maxPerPage {
  104. return maxPerPage
  105. }
  106. return perPage
  107. }
  108. func validateSkipCount(page, perPage int) int {
  109. skipCount := (page - 1) * perPage
  110. if skipCount < 0 {
  111. return 0
  112. }
  113. return skipCount
  114. }
  115. // latestHeight can be either latest committed or uncommitted (+1) height.
  116. func getHeight(latestHeight int64, heightPtr *int64) (int64, error) {
  117. if heightPtr != nil {
  118. height := *heightPtr
  119. if height <= 0 {
  120. return 0, fmt.Errorf("height must be greater than 0, but got %d", height)
  121. }
  122. if height > latestHeight {
  123. return 0, fmt.Errorf("height %d must be less than or equal to the current blockchain height %d",
  124. height, latestHeight)
  125. }
  126. base := env.BlockStore.Base()
  127. if height < base {
  128. return 0, fmt.Errorf("height %v is not available, lowest height is %v",
  129. height, base)
  130. }
  131. return height, nil
  132. }
  133. return latestHeight, nil
  134. }
  135. func latestUncommittedHeight() int64 {
  136. nodeIsSyncing := env.ConsensusReactor.WaitSync()
  137. if nodeIsSyncing {
  138. return env.BlockStore.Height()
  139. }
  140. return env.BlockStore.Height() + 1
  141. }