Merge pull request #42 from everFinance/feature/concurency-get-chunks

Feature/concurency get chunks

Author: zyjblockchain

README.md

@@ -96,6 +96,7 @@ fmt.Println(tx.ID, err)
 - [x] GetUnconfirmedTx
 - [x] GetPendingTxIds
 - [x] GetBlockHashList
+- [x] ConcurrentDownloadChunkData
 
 Initialize the instance:
 

client.go

@@ -6,13 +6,15 @@ import (
 	"errors"
 	"fmt"
 	"github.com/inconshreveable/log15"
+	"github.com/panjf2000/ants/v2"
 	"io/ioutil"
 	"math/big"
 	"net/http"
 	"net/url"
 	"path"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/everFinance/goar/types"
@@ -503,6 +505,109 @@ func (c *Client) DownloadChunkData(id string) ([]byte, error) {
 	return data, nil
 }
 
+func (c *Client) ConcurrentDownloadChunkData(id string, concurrentNum int) ([]byte, error) {
+	offsetResponse, err := c.getTransactionOffset(id)
+	if err != nil {
+		return nil, err
+	}
+	size, err := strconv.ParseInt(offsetResponse.Size, 10, 64)
+	if err != nil {
+		return nil, err
+	}
+	endOffset, err := strconv.ParseInt(offsetResponse.Offset, 10, 64)
+	if err != nil {
+		return nil, err
+	}
+	startOffset := endOffset - size + 1
+
+	offsetArr := make([]int64, 0, 5)
+	for i := 0; int64(i)+startOffset < endOffset; {
+		offsetArr = append(offsetArr, int64(i)+startOffset)
+		i += types.MAX_CHUNK_SIZE
+	}
+
+	if len(offsetArr) <= 3 { // not need concurrent get chunks
+		return c.DownloadChunkData(id)
+	}
+
+	log.Debug("need download chunks length", "length", len(offsetArr))
+	// concurrent get chunks
+	type OffsetSort struct {
+		Idx    int
+		Offset int64
+	}
+
+	chunkArr := make([][]byte, len(offsetArr)-2)
+	var (
+		lock sync.Mutex
+		wg   sync.WaitGroup
+	)
+	if concurrentNum <= 0 {
+		concurrentNum = types.DEFAULT_CHUNK_CONCURRENT_NUM
+	}
+	p, _ := ants.NewPoolWithFunc(concurrentNum, func(i interface{}) {
+		defer wg.Done()
+		oss := i.(OffsetSort)
+		chunkData, err := c.getChunkData(oss.Offset)
+		if err != nil {
+			count := 0
+			for count < 50 {
+				time.Sleep(2 * time.Second)
+				chunkData, err = c.getChunkData(oss.Offset)
+				if err == nil {
+					break
+				}
+				log.Error("retry getChunkData failed and try again...", "err", err, "offset", oss.Offset, "retryCount", count)
+				count++
+			}
+		}
+		lock.Lock()
+		chunkArr[oss.Idx] = chunkData
+		lock.Unlock()
+	})
+
+	defer p.Release()
+
+	for i, offset := range offsetArr[:len(offsetArr)-2] {
+		wg.Add(1)
+		if err := p.Invoke(OffsetSort{Idx: i, Offset: offset}); err != nil {
+			log.Error("p.Invoke(i)", "err", err, "i", i)
+			return nil, err
+		}
+	}
+	wg.Wait()
+
+	// add latest 2 chunks
+	start := offsetArr[len(offsetArr)-3] + types.MAX_CHUNK_SIZE
+	for i := 0; int64(i)+start < endOffset; {
+		chunkData, err := c.getChunkData(int64(i) + start)
+		if err != nil {
+			count := 0
+			for count < 50 {
+				time.Sleep(2 * time.Second)
+				chunkData, err = c.getChunkData(int64(i) + start)
+				if err == nil {
+					break
+				}
+				log.Error("retry getChunkData failed and try again...", "err", err, "offset", int64(i)+start, "retryCount", count)
+				count++
+			}
+		}
+		chunkArr = append(chunkArr, chunkData)
+		i += len(chunkData)
+	}
+
+	// assemble data
+	data := make([]byte, 0, size)
+	for _, chunk := range chunkArr {
+		if chunk == nil {
+			return nil, errors.New("concurrent get chunk failed")
+		}
+		data = append(data, chunk...)
+	}
+	return data, nil
+}
+
 func (c *Client) GetUnconfirmedTx(arId string) (*types.Transaction, error) {
 	_path := fmt.Sprintf("unconfirmed_tx/%s", arId)
 	body, statusCode, err := c.httpGet(_path)

client_test.go

@@ -305,3 +305,15 @@ func TestClient_GetBlockHashList2(t *testing.T) {
 	// 	t.Log(list)
 	// }
 }
+
+func TestClient_ConcurrentDownloadChunkData(t *testing.T) {
+	c := NewClient("https://arweave.net")
+	arId := "hG478oYzU0ReX8fCsRMcsm4ywuO4eiElCqDUAsm0nDk"
+	data, err := c.ConcurrentDownloadChunkData(arId, 0)
+	assert.NoError(t, err)
+	// ioutil.WriteFile("nannan.jpeg",data,0777)
+	chunks := utils.GenerateChunks(data)
+	dataRoot := utils.Base64Encode(chunks.DataRoot)
+	t.Log(dataRoot)
+	t.Log(len(data))
+}