Merge pull request #212 from irisnet/develop

R4R: limit max num for batch insert while create catch up task

conf/server/types.go

@@ -12,7 +12,7 @@ import (
 var (
 	BlockChainMonitorUrl = []string{"tcp://35.201.147.145:30657"}
 
-	WorkerNumCreateTask  = 2
+	WorkerNumCreateTask  = 1
 	WorkerNumExecuteTask = 60
 
 	InitConnectionNum  = 50              // fast init num of tendermint client pool
@@ -33,11 +33,13 @@ func init() {
 
 	workerNumCreateTask, found := os.LookupEnv(constant.EnvNameWorkerNumCreateTask)
 	if found {
-		var err error
-		WorkerNumCreateTask, err = strconv.Atoi(workerNumCreateTask)
-		if err != nil {
-			logger.Fatal("Can't convert str to int", logger.String(constant.EnvNameWorkerNumCreateTask, workerNumCreateTask))
-		}
+		//var err error
+		//WorkerNumCreateTask, err = strconv.Atoi(workerNumCreateTask)
+		//if err != nil {
+		//	logger.Fatal("Can't convert str to int", logger.String(constant.EnvNameWorkerNumCreateTask, workerNumCreateTask))
+		//}
+		// worker num for create task no longer set by env value
+		WorkerNumCreateTask = 1
 	}
 	logger.Info("Env Value", logger.Int(constant.EnvNameWorkerNumCreateTask, WorkerNumCreateTask))
 

cron/cron_task.go

@@ -1,15 +1,15 @@
 package cron
 
 import (
-	"time"
-	"os"
-	"os/signal"
 	"github.com/irisnet/irishub-sync/logger"
 	"github.com/irisnet/irishub-sync/store"
 	"github.com/irisnet/irishub-sync/store/document"
 	"github.com/irisnet/irishub-sync/util/helper"
 	"gopkg.in/mgo.v2"
 	"gopkg.in/mgo.v2/bson"
+	"os"
+	"os/signal"
+	"time"
 )
 
 type CronService struct{}
@@ -149,4 +149,4 @@ func UpdateUnknowOrEmptyTypeTxs(commontx []*document.CommonTx) error {
 	}
 
 	return nil
-}
+}

cron/cron_task_test.go

@@ -1,9 +1,9 @@
 package cron
 
 import (
+	"github.com/irisnet/irishub-sync/store"
 	"testing"
 	"time"
-	"github.com/irisnet/irishub-sync/store"
 )
 
 func TestUpdateUnknownOrEmptyTypeTxsByPage(t *testing.T) {
@@ -28,4 +28,3 @@ func TestCronService_StartCronService(t *testing.T) {
 	new(CronService).StartCronService()
 	time.Sleep(1 * time.Minute)
 }
-

go.mod

@@ -1,26 +1,26 @@
 module github.com/irisnet/irishub-sync
 
-go 1.13
+        go 1.13
 
-require (
-	github.com/go-kit/kit v0.9.0
-	github.com/irisnet/irishub v0.16.0-rc0
-	github.com/jolestar/go-commons-pool v2.0.0+incompatible
-	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
-	github.com/modern-go/reflect2 v1.0.1 // indirect
-	github.com/pkg/errors v0.8.1
-	github.com/prometheus/client_golang v1.2.1
-	github.com/robfig/cron v1.2.0
-	github.com/stretchr/testify v1.4.0
-	github.com/tendermint/tendermint v0.32.7
-	go.uber.org/zap v1.12.0
-	gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22
-	gopkg.in/natefinch/lumberjack.v2 v2.0.0
-	gopkg.in/tomb.v2 v2.0.0-20161208151619-d5d1b5820637 // indirect
-)
+        require (
+        github.com/go-kit/kit v0.9.0
+        github.com/irisnet/irishub v0.16.0-rc0
+        github.com/jolestar/go-commons-pool v2.0.0+incompatible
+        github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+        github.com/modern-go/reflect2 v1.0.1 // indirect
+        github.com/pkg/errors v0.8.1
+        github.com/prometheus/client_golang v1.2.1
+        github.com/robfig/cron v1.2.0
+        github.com/stretchr/testify v1.4.0
+        github.com/tendermint/tendermint v0.32.7
+        go.uber.org/zap v1.12.0
+        gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22
+        gopkg.in/natefinch/lumberjack.v2 v2.0.0
+        gopkg.in/tomb.v2 v2.0.0-20161208151619-d5d1b5820637 // indirect
+        )
 
-replace (
-	github.com/tendermint/iavl => github.com/irisnet/iavl v0.12.3
-	github.com/tendermint/tendermint => github.com/irisnet/tendermint v0.32.0
-	golang.org/x/crypto => github.com/tendermint/crypto v0.0.0-20180820045704-3764759f34a5
-)
+        replace (
+        github.com/tendermint/iavl => github.com/irisnet/iavl v0.12.3
+        github.com/tendermint/tendermint => github.com/irisnet/tendermint v0.32.0
+        golang.org/x/crypto => github.com/tendermint/crypto v0.0.0-20180820045704-3764759f34a5
+        )

main.go

@@ -1,12 +1,12 @@
 package main
 
 import (
+	"github.com/irisnet/irishub-sync/cron"
 	"github.com/irisnet/irishub-sync/logger"
 	"github.com/irisnet/irishub-sync/monitor"
 	"github.com/irisnet/irishub-sync/service"
 	"github.com/irisnet/irishub-sync/store"
 	"github.com/irisnet/irishub-sync/util/helper"
-	"github.com/irisnet/irishub-sync/cron"
 	"os"
 	"os/signal"
 	"syscall"

script/mongodb.js

@@ -43,7 +43,7 @@ db.proposal.createIndex({"voting_end_time": 1, "deposit_end_time": 1, "status":
 // db.tx_msg.createIndex({"hash": 1}, {"unique": true});
 
 // init data
-db.sync_conf.insert({"block_num_per_worker_handle": 50, "max_worker_sleep_time": 120});
+db.sync_conf.insert({"block_num_per_worker_handle": 100, "max_worker_sleep_time": 120});
 
 // drop collection
 // db.account.drop();

service/handler/proposal.go

@@ -82,7 +82,7 @@ func IsContainVotingEndTag(blockresult document.ResponseEndBlock) (uint64, bool)
 	if len(tags) > 0 {
 		for _, tag := range tags {
 			if tag.Key == constant.BlockTagProposalId {
-				proposalid,_ := strconv.ParseUint(tag.Value,10,64)
+				proposalid, _ := strconv.ParseUint(tag.Value, 10, 64)
 				return proposalid, true
 			}
 		}

service/handler/tx.go

@@ -33,7 +33,6 @@ func HandleTx(block *types.Block) ([]string, error) {
 		}
 		batch = append(batch, txOp)
 
-
 		// save or update proposal
 		handleProposal(tx)
 		//handleTokenFlow(blockWithTags, tx, &batch)

service/task/task_create.go

@@ -12,6 +12,8 @@ import (
 	"time"
 )
 
+const maxRecordNumForBatchInsert = 1000
+
 func StartCreateTask() {
 	log := logger.GetLogger("StartCreateTask")
 	var (
@@ -180,12 +182,16 @@ func getBlockChainLatestHeight() (int64, error) {
 	return currentBlockHeight, nil
 }
 
+// limit max num for batch insert
 func createCatchUpTask(maxEndHeight, blockNumPerWorker, blockChainLatestHeight int64) []document.SyncTask {
 	var (
 		syncTasks []document.SyncTask
 	)
 
 	for maxEndHeight+blockNumPerWorker <= blockChainLatestHeight {
+		if len(syncTasks) >= maxRecordNumForBatchInsert {
+			break
+		}
 		syncTask := document.SyncTask{
 			StartHeight:    maxEndHeight + 1,
 			EndHeight:      maxEndHeight + blockNumPerWorker,

store/document/proposal.go

@@ -41,10 +41,10 @@ type PVote struct {
 //-----------------------------------------------------------
 // Tally Results
 type PTallyResult struct {
-	Yes        string `json:"yes" bson:"yes"`
-	Abstain    string `json:"abstain" bson:"abstain"`
-	No         string `json:"no" bson:"no"`
-	NoWithVeto string `json:"no_with_veto" bson:"nowithveto"`
+	Yes               string `json:"yes" bson:"yes"`
+	Abstain           string `json:"abstain" bson:"abstain"`
+	No                string `json:"no" bson:"no"`
+	NoWithVeto        string `json:"no_with_veto" bson:"nowithveto"`
 	SystemVotingPower string `json:"system_voting_power" bson:"system_voting_power"`
 }
 

store/document/tx.go

@@ -17,7 +17,6 @@ const (
 	Tx_Field_Type   = "type"
 	Tx_Field_Status = "status"
 	Tx_Field_Height = "height"
-
 )
 
 type CommonTx struct {
@@ -43,7 +42,7 @@ type CommonTx struct {
 	//StakeCreateValidator StakeCreateValidator `bson:"stake_create_validator"`
 	//StakeEditValidator   StakeEditValidator   `bson:"stake_edit_validator"`
 	//Msg                  store.Msg            `bson:"-"`
-	Signers              []Signer             `bson:"signers"`
+	Signers []Signer `bson:"signers"`
 
 	Msgs []DocTxMsg `bson:"msgs"`
 }
@@ -132,7 +131,7 @@ func (d CommonTx) GetUnknownOrEmptyTypeTxs(skip, limit int) (res []CommonTx, err
 		return c.Find(q).Select(selector).Sort(sorts...).Skip(skip).Limit(limit).All(&res)
 	}
 
-	err = store.ExecCollection(CollectionNmCommonTx, fn);
+	err = store.ExecCollection(CollectionNmCommonTx, fn)
 	if err != nil {
 		return nil, err
 	}

types/msg/bank.go

@@ -1,9 +1,9 @@
 package msg
 
 import (
+	"github.com/irisnet/irishub-sync/store"
 	itypes "github.com/irisnet/irishub-sync/types"
 	"github.com/irisnet/irishub-sync/util/constant"
-	"github.com/irisnet/irishub-sync/store"
 )
 
 type DocTxMsgSetMemoRegexp struct {
@@ -56,11 +56,11 @@ func (doctx *DocTxMsgSend) Type() string {
 func (doctx *DocTxMsgSend) BuildMsg(txMsg interface{}) {
 	msg := txMsg.(itypes.MsgTransfer)
 	doctx.Inputs = append(doctx.Inputs, Data{
-		Address:msg.Inputs[0].Address.String(),
-		Coins:itypes.ParseCoins(msg.Inputs[0].Coins.String()),
+		Address: msg.Inputs[0].Address.String(),
+		Coins:   itypes.ParseCoins(msg.Inputs[0].Coins.String()),
 	})
 	doctx.Outputs = append(doctx.Outputs, Data{
-		Address:msg.Outputs[0].Address.String(),
-		Coins:itypes.ParseCoins(msg.Outputs[0].Coins.String()),
+		Address: msg.Outputs[0].Address.String(),
+		Coins:   itypes.ParseCoins(msg.Outputs[0].Coins.String()),
 	})
 }

types/msg/coinswap.go

@@ -1,9 +1,9 @@
 package msg
 
 import (
-	"github.com/irisnet/irishub-sync/util/constant"
 	"github.com/irisnet/irishub-sync/store"
 	itypes "github.com/irisnet/irishub-sync/types"
+	"github.com/irisnet/irishub-sync/util/constant"
 )
 
 type DocTxMsgAddLiquidity struct {

types/msg/distribution.go

@@ -1,8 +1,8 @@
 package msg
 
 import (
-	"github.com/irisnet/irishub-sync/util/constant"
 	itypes "github.com/irisnet/irishub-sync/types"
+	"github.com/irisnet/irishub-sync/util/constant"
 )
 
 // msg struct for changing the withdraw address for a delegator (or validator self-delegation)

types/msg/gov.go

@@ -157,7 +157,6 @@ type DocTxMsgDeposit struct {
 	Amount     store.Coins `bson:"amount"`      // Coins to add to the proposal's deposit
 }
 
-
 func (doctx *DocTxMsgDeposit) Type() string {
 	return constant.TxTypeDeposit
 }

types/msg/htlc.go

@@ -1,10 +1,10 @@
 package msg
 
 import (
-	"github.com/irisnet/irishub-sync/util/constant"
+	"encoding/hex"
 	"github.com/irisnet/irishub-sync/store"
 	itypes "github.com/irisnet/irishub-sync/types"
-	"encoding/hex"
+	"github.com/irisnet/irishub-sync/util/constant"
 )
 
 type DocTxMsgCreateHTLC struct {

types/msg/stake.go

@@ -1,10 +1,10 @@
 package msg
 
 import (
-	"github.com/irisnet/irishub-sync/util/constant"
-	itypes "github.com/irisnet/irishub-sync/types"
 	"github.com/irisnet/irishub-sync/store"
 	"github.com/irisnet/irishub-sync/store/document"
+	itypes "github.com/irisnet/irishub-sync/types"
+	"github.com/irisnet/irishub-sync/util/constant"
 	"github.com/irisnet/irishub/app/v1/stake"
 )
 

types/types.go

@@ -18,13 +18,13 @@ import (
 	"github.com/irisnet/irishub/app/v1/rand"
 	"github.com/irisnet/irishub/app/v1/slashing"
 	"github.com/irisnet/irishub/app/v1/stake"
-	"github.com/irisnet/irishub/app/v2/coinswap"
-	"github.com/irisnet/irishub/app/v2/htlc"
-	"github.com/irisnet/irishub/modules/guardian"
 	stags "github.com/irisnet/irishub/app/v1/stake/tags"
 	staketypes "github.com/irisnet/irishub/app/v1/stake/types"
+	"github.com/irisnet/irishub/app/v2/coinswap"
+	"github.com/irisnet/irishub/app/v2/htlc"
 	"github.com/irisnet/irishub/client/utils"
 	"github.com/irisnet/irishub/codec"
+	"github.com/irisnet/irishub/modules/guardian"
 	"github.com/irisnet/irishub/types"
 	abci "github.com/tendermint/tendermint/abci/types"
 	cmn "github.com/tendermint/tendermint/libs/common"
@@ -34,12 +34,11 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
-
 )
 
 type (
-	MsgTransfer = bank.MsgSend
-	MsgBurn     = bank.MsgBurn
+	MsgTransfer      = bank.MsgSend
+	MsgBurn          = bank.MsgBurn
 	MsgSetMemoRegexp = bank.MsgSetMemoRegexp
 
 	MsgStakeCreate                 = stake.MsgCreateValidator
@@ -65,11 +64,11 @@ type (
 	Proposal                         = gov.Proposal
 	SdkVote                          = gov.Vote
 
-	MsgSwapOrder = coinswap.MsgSwapOrder
-	MsgAddLiquidity = coinswap.MsgAddLiquidity
+	MsgSwapOrder       = coinswap.MsgSwapOrder
+	MsgAddLiquidity    = coinswap.MsgAddLiquidity
 	MsgRemoveLiquidity = coinswap.MsgRemoveLiquidity
 
-	MsgClaimHTLC = htlc.MsgClaimHTLC
+	MsgClaimHTLC  = htlc.MsgClaimHTLC
 	MsgCreateHTLC = htlc.MsgCreateHTLC
 	MsgRefundHTLC = htlc.MsgRefundHTLC
 
@@ -83,10 +82,10 @@ type (
 	AssetEditGateWay          = asset.MsgEditGateway
 	AssetTransferGatewayOwner = asset.MsgTransferGatewayOwner
 
-	MsgAddProfiler = guardian.MsgAddProfiler
-	MsgAddTrustee = guardian.MsgAddTrustee
+	MsgAddProfiler    = guardian.MsgAddProfiler
+	MsgAddTrustee     = guardian.MsgAddTrustee
 	MsgDeleteProfiler = guardian.MsgDeleteProfiler
-	MsgDeleteTrustee = guardian.MsgDeleteTrustee
+	MsgDeleteTrustee  = guardian.MsgDeleteTrustee
 
 	ResponseDeliverTx = abci.ResponseDeliverTx
 
@@ -212,6 +211,7 @@ func ParseCoin(coinStr string) (coin store.Coin) {
 
 func getPrecision(amount string) string {
 	length := len(amount)
+
 	if length > 15 {
 		nums := strings.Split(amount, ".")
 		if len(nums) > 2 {

types/types_test.go

@@ -2,9 +2,9 @@ package types
 
 import (
 	"encoding/json"
-	"testing"
 	"fmt"
 	"strconv"
+	"testing"
 )
 
 func TestParseCoin(t *testing.T) {
@@ -54,4 +54,4 @@ func Test_getPrecision(t *testing.T) {
 	} else {
 		t.Fatal("Failed")
 	}
-}
+}