Add --watch-interval and rand permuatation for nodes for sending tx

bosnet · Nov 30, 2018 · c6b7850 · c6b7850
1 parent 2f1b1af
commit c6b7850
Show file tree

Hide file tree

Showing 4 changed files with 29 additions and 3 deletions.
diff --git a/cmd/sebak/cmd/run.go b/cmd/sebak/cmd/run.go
@@ -80,7 +80,8 @@ var (
 	flagOperationsInBallotLimit string = common.GetENVValue("SEBAK_OPERATIONS_IN_BALLOT_LIMIT", strconv.Itoa(common.DefaultOperationsInBallotLimit))
 	flagTxPoolLimit             string = common.GetENVValue("SEBAK_TX_POOL_LIMIT", strconv.Itoa(common.DefaultTxPoolLimit))
 
-	flagWatcherMode bool = common.GetENVValue("SEBAK_WATCHER_MODE", "0") == "1"
+	flagWatcherMode   bool   = common.GetENVValue("SEBAK_WATCHER_MODE", "0") == "1"
+	flagWatchInterval string = common.GetENVValue("SEBAK_WATCH_INTERVAL", "5s")
 )
 
 var (
@@ -114,6 +115,7 @@ var (
 	txPoolNodeLimit         uint64
 	syncCheckPrevBlock      time.Duration
 	jsonrpcbindEndpoint     *common.Endpoint
+	watchInterval           time.Duration
 
 	logLevel logging.Lvl
 	log      logging.Logger = logging.New("module", "main")
@@ -223,6 +225,7 @@ func init() {
 
 	nodeCmd.Flags().StringVar(&flagCongressAddress, "set-congress-address", flagCongressAddress, "set congress address")
 	nodeCmd.Flags().BoolVar(&flagWatcherMode, "watcher-mode", flagWatcherMode, "watcher mode")
+	nodeCmd.Flags().StringVar(&flagWatchInterval, "watch-interval", flagWatchInterval, "watch interval")
 
 	rootCmd.AddCommand(nodeCmd)
 }
@@ -447,6 +450,7 @@ func parseFlagsNode() {
 	syncFetchTimeout = getTimeDuration(flagSyncFetchTimeout, sync.FetchTimeout, "--sync-fetch-timeout")
 	syncCheckInterval = getTimeDuration(flagSyncCheckInterval, sync.CheckBlockHeightInterval, "--sync-check-interval")
 	syncCheckPrevBlock = getTimeDuration(flagSyncCheckPrevBlockInterval, sync.CheckPrevBlockInterval, "--sync-check-prevblock")
+	watchInterval = getTimeDuration(flagWatchInterval, sync.WatchInterval, "--watch-interval")
 
 	{
 		if ok := common.HTTPCacheAdapterNames[flagHTTPCacheAdapter]; !ok {
@@ -704,6 +708,7 @@ func runNode() error {
 	c.RetryInterval = syncRetryInterval
 	c.CheckBlockHeightInterval = syncCheckInterval
 	c.CheckPrevBlockInterval = syncCheckPrevBlock
+	c.WatchInterval = watchInterval
 
 	syncer := c.NewSyncer()
 

diff --git a/lib/node/runner/checker_message.go b/lib/node/runner/checker_message.go
@@ -13,6 +13,8 @@
 package runner
 
 import (
+	"math/rand"
+
 	logging "github.com/inconshreveable/log15"
 
 	"encoding/json"
@@ -197,8 +199,14 @@ func BroadcastTransactionFromWatcher(c common.Checker, args ...interface{}) erro
 		return errors.AllValidatorsNotConnected
 	}
 
+	raddrs := make([]string, len(addrs))
+	perm := rand.Perm(len(addrs))
+	for i, v := range perm {
+		raddrs[v] = addrs[i]
+	}
+
 	var err error
-	for _, a := range addrs {
+	for _, a := range raddrs {
 		client := cm.GetConnection(a)
 		_, err = client.SendTransaction(checker.Transaction)
 		if err == nil {

diff --git a/lib/sync/config.go b/lib/sync/config.go
@@ -17,6 +17,7 @@ const (
 	RetryInterval                   = 10 * time.Second
 	CheckBlockHeightInterval        = 30 * time.Second
 	CheckPrevBlockInterval          = 30 * time.Second
+	WatchInterval                   = 5 * time.Second
 )
 
 type Config struct {
@@ -34,6 +35,7 @@ type Config struct {
 	RetryInterval            time.Duration
 	CheckBlockHeightInterval time.Duration
 	CheckPrevBlockInterval   time.Duration
+	WatchInterval            time.Duration
 }
 
 func NewConfig(localNode *node.LocalNode,
@@ -104,12 +106,17 @@ func (c *Config) NewValidator() Validator {
 }
 
 func (c *Config) NewWatcher(s SyncController) *Watcher {
+	c.logger.Info("watcher config", "watchInterval", c.WatchInterval)
+
 	client := c.NewHTTP2Client()
 	w := NewWatcher(
 		s, client,
 		c.connectionManager,
 		c.storage,
 		c.localNode,
+		func(w *Watcher) {
+			w.interval = c.WatchInterval
+		},
 	)
 	w.SetLogger(c.logger.New("submodule", "watcher"))
 	return w

diff --git a/lib/sync/watcher.go b/lib/sync/watcher.go
@@ -21,6 +21,8 @@ import (
 	"github.com/inconshreveable/log15"
 )
 
+type WatcherOption func(*Watcher)
+
 type Watcher struct {
 	syncer    SyncController
 	cm        network.ConnectionManager
@@ -40,7 +42,8 @@ func NewWatcher(
 	client Doer,
 	cm network.ConnectionManager,
 	st *storage.LevelDBBackend,
-	ln *node.LocalNode) *Watcher {
+	ln *node.LocalNode,
+	opts ...WatcherOption) *Watcher {
 	ctx, cancel := context.WithCancel(context.Background())
 	w := &Watcher{
 		syncer:    syncer,
@@ -54,6 +57,9 @@ func NewWatcher(
 		cancel:    cancel,
 		logger:    common.NopLogger(),
 	}
+	for _, o := range opts {
+		o(w)
+	}
 	return w
 }