feat(): add concurrent submitData function

Author: zyjblockchain

example/bundle_test.go

@@ -1,6 +1,7 @@
 package example
 
 import (
+	"context"
 	"github.com/everFinance/everpay-go/sdk"
 	"github.com/everFinance/goar"
 	"github.com/everFinance/goar/types"
@@ -70,7 +71,8 @@ func TestBundleToArweave(t *testing.T) {
 
 	// send to arweave
 	wal, err := goar.NewWalletFromPath("jwkKey.json", "https://arweave.net")
-	tx, err := wal.SendBundleTx(bundle.BundleBinary, []types.Tag{
+	assert.NoError(t, err)
+	tx, err := wal.SendBundleTx(context.TODO(), 0, bundle.BundleBinary, []types.Tag{
 		{Name: "App", Value: "goar"},
 	})
 	assert.NoError(t, err)

types/const.go

@@ -12,8 +12,7 @@ const (
 	HASH_SIZE      = 32
 
 	// concurrent submit chunks min size
-	CONCURRENT_MIN_DATA_SIZE     = 30 * 1024 * 1024 // 30 MB
-	DEFAULT_CHUNK_CONCURRENT_NUM = 100              // default concurrent number
+	DEFAULT_CHUNK_CONCURRENT_NUM = 50 // default concurrent number
 
 	// number of bits in a big.Word
 	WordBits = 32 << (uint64(^big.Word(0)) >> 63)

uploader.go

@@ -1,11 +1,11 @@
 package goar
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"github.com/panjf2000/ants/v2"
 	"math"
-	"math/big"
 	"math/rand"
 	"strconv"
 	"sync"
@@ -111,13 +111,6 @@ func CreateUploader(api *Client, upload interface{}, data []byte) (*TransactionU
 }
 
 func (tt *TransactionUploader) Once() (err error) {
-	dataSize, ok := new(big.Int).SetString(tt.Transaction.DataSize, 10)
-	if ok && dataSize.Cmp(big.NewInt(types.CONCURRENT_MIN_DATA_SIZE)) >= 0 {
-		log.Debug("Using concurrent upload chunks", "dataSize", dataSize.String())
-		return tt.ConcurrentUploadChunks()
-	}
-
-	// not use concurrent submit chunks
 	for !tt.IsComplete() {
 		if err = tt.UploadChunk(); err != nil {
 			return
@@ -158,7 +151,7 @@ func (tt *TransactionUploader) PctComplete() float64 {
 	return math.Trunc(fval * 100)
 }
 
-func (tt *TransactionUploader) ConcurrentUploadChunks() error {
+func (tt *TransactionUploader) ConcurrentOnce(ctx context.Context, concurrentNum int) error {
 	// post tx info
 	if err := tt.postTransaction(); err != nil {
 		return err
@@ -169,10 +162,20 @@ func (tt *TransactionUploader) ConcurrentUploadChunks() error {
 	}
 
 	var wg sync.WaitGroup
-	p, _ := ants.NewPoolWithFunc(types.DEFAULT_CHUNK_CONCURRENT_NUM, func(i interface{}) {
+	if concurrentNum <= 0 {
+		concurrentNum = types.DEFAULT_CHUNK_CONCURRENT_NUM
+	}
+	p, _ := ants.NewPoolWithFunc(concurrentNum, func(i interface{}) {
 		defer wg.Done()
 		// process submit chunk
 		idx := i.(int)
+
+		select {
+		case <-ctx.Done():
+			log.Warn("ctx.done", "chunkIdx", idx)
+			return
+		default:
+		}
 		chunk, err := utils.GetChunk(*tt.Transaction, idx, tt.Data)
 		if err != nil {
 			log.Error("GetChunk error", "err", err, "idx", idx)
@@ -187,8 +190,14 @@ func (tt *TransactionUploader) ConcurrentUploadChunks() error {
 		// try again
 		retryCount := 0
 		for {
-			retryCount++
+			select {
+			case <-ctx.Done():
+				log.Warn("ctx.done", "chunkIdx", idx)
+				return
+			default:
+			}
 
+			retryCount++
 			if statusCode == 429 {
 				time.Sleep(1 * time.Second)
 			} else {

wallet.go

@@ -1,6 +1,7 @@
 package goar
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"io/ioutil"
@@ -101,26 +102,57 @@ func (w *Wallet) SendDataSpeedUp(data []byte, tags []types.Tag, speedFactor int6
 	return w.SendTransaction(tx)
 }
 
-// SendTransaction: if send success, should return pending
-func (w *Wallet) SendTransaction(tx *types.Transaction) (types.Transaction, error) {
-	anchor, err := w.Client.GetTransactionAnchor()
+func (w *Wallet) SendDataConcurrentSpeedUp(ctx context.Context, concurrentNum int, data []byte, tags []types.Tag, speedFactor int64) (types.Transaction, error) {
+	reward, err := w.Client.GetTransactionPrice(data, nil)
 	if err != nil {
 		return types.Transaction{}, err
 	}
-	tx.LastTx = anchor
-	tx.Owner = w.Owner()
-	if err = w.Signer.SignTx(tx); err != nil {
-		return types.Transaction{}, err
+
+	tx := &types.Transaction{
+		Format:   2,
+		Target:   "",
+		Quantity: "0",
+		Tags:     utils.TagsEncode(tags),
+		Data:     utils.Base64Encode(data),
+		DataSize: fmt.Sprintf("%d", len(data)),
+		Reward:   fmt.Sprintf("%d", reward*(100+speedFactor)/100),
 	}
 
-	uploader, err := CreateUploader(w.Client, tx, nil)
+	return w.SendTransactionConcurrent(ctx, concurrentNum, tx)
+}
+
+// SendTransaction: if send success, should return pending
+func (w *Wallet) SendTransaction(tx *types.Transaction) (types.Transaction, error) {
+	uploader, err := w.getUploader(tx)
 	if err != nil {
 		return types.Transaction{}, err
 	}
 	err = uploader.Once()
 	return *tx, err
 }
 
+func (w *Wallet) SendTransactionConcurrent(ctx context.Context, concurrentNum int, tx *types.Transaction) (types.Transaction, error) {
+	uploader, err := w.getUploader(tx)
+	if err != nil {
+		return types.Transaction{}, err
+	}
+	err = uploader.ConcurrentOnce(ctx, concurrentNum)
+	return *tx, err
+}
+
+func (w *Wallet) getUploader(tx *types.Transaction) (*TransactionUploader, error) {
+	anchor, err := w.Client.GetTransactionAnchor()
+	if err != nil {
+		return nil, err
+	}
+	tx.LastTx = anchor
+	tx.Owner = w.Owner()
+	if err = w.Signer.SignTx(tx); err != nil {
+		return nil, err
+	}
+	return CreateUploader(w.Client, tx, nil)
+}
+
 func (w *Wallet) SendPst(contractId string, target string, qty *big.Int, customTags []types.Tag, speedFactor int64) (types.Transaction, error) {
 	maxQty := big.NewInt(9007199254740991) // swc support max js integer
 	if qty.Cmp(maxQty) > 0 {

wallet_bundle.go

@@ -1,11 +1,12 @@
 package goar
 
 import (
+	"context"
 	"errors"
 	"github.com/everFinance/goar/types"
 )
 
-func (w *Wallet) SendBundleTxSpeedUp(bundleBinary []byte, tags []types.Tag, txSpeed int64) (types.Transaction, error) {
+func (w *Wallet) SendBundleTxSpeedUp(ctx context.Context, concurrentNum int, bundleBinary []byte, tags []types.Tag, txSpeed int64) (types.Transaction, error) {
 	bundleTags := []types.Tag{
 		{Name: "Bundle-Format", Value: "binary"},
 		{Name: "Bundle-Version", Value: "2.0.0"},
@@ -23,9 +24,9 @@ func (w *Wallet) SendBundleTxSpeedUp(bundleBinary []byte, tags []types.Tag, txSp
 
 	txTags := make([]types.Tag, 0)
 	txTags = append(bundleTags, tags...)
-	return w.SendDataSpeedUp(bundleBinary, txTags, txSpeed)
+	return w.SendDataConcurrentSpeedUp(ctx, concurrentNum, bundleBinary, txTags, txSpeed)
 }
 
-func (w *Wallet) SendBundleTx(bundleBinary []byte, tags []types.Tag) (types.Transaction, error) {
-	return w.SendBundleTxSpeedUp(bundleBinary, tags, 0)
+func (w *Wallet) SendBundleTx(ctx context.Context, concurrentNum int, bundleBinary []byte, tags []types.Tag) (types.Transaction, error) {
+	return w.SendBundleTxSpeedUp(ctx, concurrentNum, bundleBinary, tags, 0)
 }

wallet_test.go

@@ -4,10 +4,9 @@ import (
 	"encoding/base64"
 	"github.com/everFinance/goar/types"
 	"github.com/everFinance/goar/utils"
+	"github.com/stretchr/testify/assert"
 	"io/ioutil"
 	"testing"
-
-	"github.com/stretchr/testify/assert"
 )
 
 var testWallet *Wallet
@@ -285,7 +284,9 @@ func TestTransactionUploader_ConcurrentUploadChunks(t *testing.T) {
 
 	t.Log(len(bundle.BundleBinary))
 	// send to arweave
-	// tx, err := w.SendBundleTx(bundle.BundleBinary, []types.Tag{
+	// ctx ,cancel := context.WithTimeout(context.Background(),100*time.Millisecond)
+	// defer cancel()
+	// tx, err := w.SendBundleTx(ctx, 0,bundle.BundleBinary, []types.Tag{
 	// 	{Name: "APP", Value: "Goar"},
 	// 	{Name: "Protocol-Name", Value: "BAR"},
 	// 	{Name: "Action", Value: "Burn"},