make data size optional

core/sys/fs_mem.go

@@ -284,23 +284,27 @@ type MemChanFile struct {
 	ModTime        time.Time   // FileInfo.ModTime
 	ChunkWriteSize int         //  0 value means no limit
 	Sys            interface{} // FileInfo.Sys
-	reader         io.Reader
+	ErrChan        chan error
 	data           []byte
 }
 
 func (f *MemChanFile) Stat() (fs.FileInfo, error) {
 	return &MemFileChanInfo{name: f.Name, f: f}, nil
 }
 func (f *MemChanFile) Read(p []byte) (int, error) {
-	recieveData, ok := <-f.Buffer
-	if !ok {
-		return 0, io.EOF
-	}
-	if len(recieveData) > len(p) {
-		return 0, io.ErrShortBuffer
+	select {
+	case err := <-f.ErrChan:
+		return 0, err
+	case recieveData, ok := <-f.Buffer:
+		if !ok {
+			return 0, io.EOF
+		}
+		if len(recieveData) > len(p) {
+			return 0, io.ErrShortBuffer
+		}
+		n := copy(p, recieveData)
+		return n, nil
 	}
-	n := copy(p, recieveData)
-	return n, nil
 }
 
 func (f *MemChanFile) Write(p []byte) (n int, err error) {
@@ -310,7 +314,7 @@ func (f *MemChanFile) Write(p []byte) (n int, err error) {
 		f.Buffer <- data
 	} else {
 		if cap(f.data) == 0 {
-			f.data = make([]byte, 0, f.ChunkWriteSize)
+			f.data = make([]byte, 0, len(p))
 		}
 		f.data = append(f.data, p...)
 	}
@@ -326,15 +330,14 @@ func (f *MemChanFile) Sync() error {
 		}
 		f.Buffer <- f.data[current:end]
 	}
-	f.data = make([]byte, 0, f.ChunkWriteSize)
+	f.data = nil
 	return nil
 }
 func (f *MemChanFile) Seek(offset int64, whence int) (ret int64, err error) {
 	return 0, nil
 }
 
 func (f *MemChanFile) Close() error {
-	f.reader = nil
 	close(f.Buffer)
 	return nil
 }

core/util/validation_tree.go

@@ -17,6 +17,11 @@ const (
 	Right
 )
 
+const (
+	START_LENGTH = 64
+	ADD_LENGTH   = 320
+)
+
 type ValidationTree struct {
 	writeLock      sync.Mutex
 	writeCount     int
@@ -61,7 +66,7 @@ func (v *ValidationTree) Write(b []byte) (int, error) {
 		return 0, nil
 	}
 
-	if v.writtenSize+int64(len(b)) > v.dataSize {
+	if v.dataSize > 0 && v.writtenSize+int64(len(b)) > v.dataSize {
 		return 0, fmt.Errorf("data size overflow. expected %d, got %d", v.dataSize, v.writtenSize+int64(len(b)))
 	}
 
@@ -80,6 +85,12 @@ func (v *ValidationTree) Write(b []byte) (int, error) {
 		n, _ := v.h.Write(b[i:j])
 		v.writeCount += n // update write count
 		if v.writeCount == MaxMerkleLeavesSize {
+			if v.leafIndex >= len(v.leaves) {
+				// increase leaves size
+				leaves := make([][]byte, len(v.leaves)+ADD_LENGTH)
+				copy(leaves, v.leaves)
+				v.leaves = leaves
+			}
 			v.leaves[v.leafIndex] = v.h.Sum(nil)
 			v.leafIndex++
 			v.writeCount = 0 // reset writeCount
@@ -137,21 +148,28 @@ func (v *ValidationTree) Finalize() error {
 	if v.isFinalized {
 		return errors.New("already finalized")
 	}
-	if v.writtenSize != v.dataSize {
+	if v.dataSize > 0 && v.writtenSize != v.dataSize {
 		return fmt.Errorf("invalid size. Expected %d got %d", v.dataSize, v.writtenSize)
 	}
 
 	v.isFinalized = true
 
 	if v.writeCount > 0 {
 		v.leaves[v.leafIndex] = v.h.Sum(nil)
+	} else {
+		v.leafIndex--
+	}
+	if v.leafIndex < len(v.leaves) {
+		v.leaves = v.leaves[:v.leafIndex+1]
 	}
 	return nil
 }
 
 func NewValidationTree(dataSize int64) *ValidationTree {
 	totalLeaves := (dataSize + MaxMerkleLeavesSize - 1) / MaxMerkleLeavesSize
-
+	if totalLeaves == 0 {
+		totalLeaves = START_LENGTH
+	}
 	return &ValidationTree{
 		dataSize: dataSize,
 		h:        sha256.New(),

wasmsdk/bridge.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/0chain/gosdk/zcnbridge"
 	"github.com/0chain/gosdk/zcnbridge/errors"
+	"github.com/0chain/gosdk/zcnbridge/log"
 	"github.com/0chain/gosdk/zcnbridge/transaction"
 	"github.com/0chain/gosdk/zcnbridge/wallet"
 	"github.com/0chain/gosdk/zcncore"
@@ -124,6 +125,7 @@ func getNotProcessedWZCNBurnEvents() string {
 		return errors.New("getNotProcessedWZCNBurnEvents", "failed to retreive last ZCN processed mint nonce").Error()
 	}
 
+	log.Logger.Debug("MintNonce = " + strconv.Itoa(int(mintNonce)))
 	burnEvents, err := bridge.QueryEthereumBurnEvents(strconv.Itoa(int(mintNonce)))
 	if err != nil {
 		return errors.Wrap("getNotProcessedWZCNBurnEvents", "failed to retreive WZCN burn events", err).Error()

zboxcore/sdk/allocation.go

@@ -233,6 +233,7 @@ type OperationRequest struct {
 	FileReader   io.Reader
 	Mask         *zboxutil.Uint128 // Required for delete repair operation
 	DownloadFile bool              // Required for upload repair operation
+	StreamUpload bool              // Required for streaming file when actualSize is not available
 	Opts         []ChunkedUploadOption
 }
 
@@ -402,11 +403,13 @@ func (a *Allocation) RepairFile(file sys.File, remotepath string, statusCallback
 			WithEncrypt(true),
 			WithStatusCallback(statusCallback),
 			WithEncryptedPoint(ref.EncryptedKeyPoint),
+			WithChunkNumber(100),
 		}
 	} else {
 		opts = []ChunkedUploadOption{
 			WithMask(mask),
 			WithStatusCallback(statusCallback),
+			WithChunkNumber(100),
 		}
 	}
 	op := &OperationRequest{
@@ -894,7 +897,7 @@ func (a *Allocation) DoMultiOperation(operations []OperationRequest, opts ...Mul
 				cancelLock.Lock()
 				CancelOpCtx[op.FileMeta.RemotePath] = mo.ctxCncl
 				cancelLock.Unlock()
-				operation, newConnectionID, err = NewUploadOperation(mo.ctx, op.Workdir, mo.allocationObj, mo.connectionID, op.FileMeta, op.FileReader, false, op.IsWebstreaming, op.IsRepair, op.DownloadFile, op.Opts...)
+				operation, newConnectionID, err = NewUploadOperation(mo.ctx, op.Workdir, mo.allocationObj, mo.connectionID, op.FileMeta, op.FileReader, false, op.IsWebstreaming, op.IsRepair, op.DownloadFile, op.StreamUpload, op.Opts...)
 
 			case constants.FileOperationDelete:
 				if op.Mask != nil {
@@ -907,7 +910,7 @@ func (a *Allocation) DoMultiOperation(operations []OperationRequest, opts ...Mul
 				cancelLock.Lock()
 				CancelOpCtx[op.FileMeta.RemotePath] = mo.ctxCncl
 				cancelLock.Unlock()
-				operation, newConnectionID, err = NewUploadOperation(mo.ctx, op.Workdir, mo.allocationObj, mo.connectionID, op.FileMeta, op.FileReader, true, op.IsWebstreaming, op.IsRepair, op.DownloadFile, op.Opts...)
+				operation, newConnectionID, err = NewUploadOperation(mo.ctx, op.Workdir, mo.allocationObj, mo.connectionID, op.FileMeta, op.FileReader, true, op.IsWebstreaming, op.IsRepair, op.DownloadFile, op.StreamUpload, op.Opts...)
 
 			case constants.FileOperationCreateDir:
 				operation = NewDirOperation(op.RemotePath, mo.operationMask, mo.maskMU, mo.consensusThresh, mo.fullconsensus, mo.ctx)
@@ -1106,6 +1109,9 @@ func (a *Allocation) generateDownloadRequest(
 	downloadReq.contentMode = contentMode
 	downloadReq.connectionID = connectionID
 	downloadReq.downloadQueue = make(downloadQueue, len(a.Blobbers))
+	for i := 0; i < len(a.Blobbers); i++ {
+		downloadReq.downloadQueue[i].timeTaken = 1000000
+	}
 
 	return downloadReq, nil
 }
@@ -2219,6 +2225,9 @@ func (a *Allocation) downloadFromAuthTicket(fileHandler sys.File, authTicket str
 	downloadReq.fullconsensus = a.fullconsensus
 	downloadReq.consensusThresh = a.consensusThreshold
 	downloadReq.downloadQueue = make(downloadQueue, len(a.Blobbers))
+	for i := 0; i < len(a.Blobbers); i++ {
+		downloadReq.downloadQueue[i].timeTaken = 1000000
+	}
 	downloadReq.connectionID = zboxutil.NewConnectionId()
 	downloadReq.completedCallback = func(remotepath string, remotepathHash string) {
 		a.mutex.Lock()

zboxcore/sdk/chunked_upload.go

@@ -442,6 +442,7 @@ func (su *ChunkedUpload) process() error {
 			}
 			if su.fileMeta.ActualSize == 0 {
 				su.fileMeta.ActualSize = su.progress.ReadLength
+				su.shardSize = getShardSize(su.fileMeta.ActualSize, su.allocationObj.DataShards, su.encryptOnUpload)
 			} else if su.fileMeta.ActualSize != su.progress.ReadLength && su.thumbnailBytes == nil {
 				if su.statusCallback != nil {
 					su.statusCallback.Error(su.allocationObj.ID, su.fileMeta.RemotePath, su.opCode, thrown.New("upload_failed", "Upload failed. Uploaded size does not match with actual size: "+fmt.Sprintf("%d != %d", su.fileMeta.ActualSize, su.progress.ReadLength)))
@@ -568,7 +569,6 @@ func (su *ChunkedUpload) readChunks(num int) (*batchChunksData, error) {
 
 		// upload entire thumbnail in first chunk request only
 		if chunk.Index == 0 && len(su.thumbnailBytes) > 0 {
-			data.totalReadSize += int64(su.fileMeta.ActualThumbnailSize)
 
 			data.thumbnailShards, err = su.chunkReader.Read(su.thumbnailBytes)
 			if err != nil {

zboxcore/sdk/chunked_upload_blobber.go

@@ -103,11 +103,6 @@ func (sb *ChunkedUploadBlobber) sendUploadRequest(
 					}
 
 					respbody := resp.Body()
-					if err != nil {
-						logger.Logger.Error("Error: Resp ", err)
-						return fmt.Errorf("Error while reading body. Error %s", err), false
-					}
-
 					if resp.StatusCode() == http.StatusTooManyRequests {
 						logger.Logger.Error("Got too many request error")
 						var r int

zboxcore/sdk/dirworker.go

@@ -195,7 +195,7 @@ func (req *DirRequest) createDirInBlobber(blobber *blockchain.StorageNode, pos u
 
 	for i := 0; i < 3; i++ {
 		err, shouldContinue = func() (err error, shouldContinue bool) {
-			ctx, cncl := context.WithTimeout(req.ctx, (time.Second * 30))
+			ctx, cncl := context.WithTimeout(req.ctx, (time.Second * 10))
 			resp, err = zboxutil.Client.Do(httpreq.WithContext(ctx))
 			cncl()
 			if err != nil {

zboxcore/sdk/downloadworker.go

@@ -105,7 +105,7 @@ type DownloadRequest struct {
 	bufferMap          map[int]zboxutil.DownloadBuffer
 	downloadStorer     DownloadProgressStorer
 	workdir            string
-	downloadQueue      downloadQueue
+	downloadQueue      downloadQueue // Always initialize this queue with max time taken
 }
 
 type downloadPriority struct {
@@ -285,7 +285,7 @@ func (req *DownloadRequest) downloadBlock(
 			if req.shouldVerify {
 				go AddBlockDownloadReq(req.ctx, blockDownloadReq, nil, req.effectiveBlockSize)
 			} else {
-				go AddBlockDownloadReq(req.ctx, blockDownloadReq, req.bufferMap[int(pos)], req.effectiveBlockSize)
+				go AddBlockDownloadReq(req.ctx, blockDownloadReq, req.bufferMap[blobberIdx], req.effectiveBlockSize)
 			}
 		}
 
@@ -308,7 +308,9 @@ func (req *DownloadRequest) downloadBlock(
 					downloadErrors[i] = fmt.Sprintf("Error %s from %s",
 						err.Error(), req.blobbers[result.idx].Baseurl)
 					logger.Logger.Error(err)
-					req.bufferMap[result.idx].ReleaseChunk(int(req.startBlock / req.numBlocks))
+					if req.bufferMap != nil {
+						req.bufferMap[result.idx].ReleaseChunk(int(req.startBlock / req.numBlocks))
+					}
 				} else if timeRequest {
 					req.downloadQueue[result.idx].timeTaken = result.timeTaken
 				}
@@ -538,10 +540,11 @@ func (req *DownloadRequest) processDownload() {
 		}
 		for i := req.downloadMask; !i.Equals64(0); i = i.And(zboxutil.NewUint128(1).Lsh(pos).Not()) {
 			pos = uint64(i.TrailingZeros())
+			blobberIdx := req.downloadQueue[pos].blobberIdx
 			if writerAt {
-				req.bufferMap[int(pos)] = zboxutil.NewDownloadBufferWithChan(sz, int(numBlocks), req.effectiveBlockSize)
+				req.bufferMap[blobberIdx] = zboxutil.NewDownloadBufferWithChan(sz, int(numBlocks), req.effectiveBlockSize)
 			} else {
-				req.bufferMap[int(pos)] = zboxutil.NewDownloadBufferWithMask(sz, int(numBlocks), req.effectiveBlockSize)
+				req.bufferMap[blobberIdx] = zboxutil.NewDownloadBufferWithMask(sz, int(numBlocks), req.effectiveBlockSize)
 			}
 		}
 	}
@@ -678,14 +681,19 @@ func (req *DownloadRequest) processDownload() {
 	var progressLock sync.Mutex
 	firstReqWG := sync.WaitGroup{}
 	firstReqWG.Add(1)
-	eg, _ := errgroup.WithContext(ctx)
+	eg, egCtx := errgroup.WithContext(ctx)
 	eg.SetLimit(downloadWorkerCount + EXTRA_COUNT)
 	for i := 0; i < n; i++ {
 		j := i
 		if i == 1 {
 			firstReqWG.Wait()
 			heap.Init(&req.downloadQueue)
 		}
+		select {
+		case <-egCtx.Done():
+			goto breakDownloadLoop
+		default:
+		}
 		eg.Go(func() error {
 
 			if j == 0 {
@@ -735,6 +743,7 @@ func (req *DownloadRequest) processDownload() {
 			}
 			return nil
 		})
+	breakDownloadLoop:
 	}
 	if err := eg.Wait(); err != nil {
 		writeCancel()
@@ -1082,11 +1091,8 @@ func GetFileRefFromBlobber(allocationID, blobberId, remotePath string) (fRef *fi
 	listReq.ctx = ctx
 	listReq.remotefilepath = remotePath
 
-	listReq.wg = &sync.WaitGroup{}
-	listReq.wg.Add(1)
 	rspCh := make(chan *fileMetaResponse, 1)
 	go listReq.getFileMetaInfoFromBlobber(listReq.blobbers[0], 0, rspCh)
-	listReq.wg.Wait()
 	resp := <-rspCh
 	return resp.fileref, resp.err
 }

zboxcore/sdk/filemetaworker.go

@@ -8,7 +8,6 @@ import (
 	"io"
 	"mime/multipart"
 	"net/http"
-	"sync"
 	"time"
 
 	"github.com/0chain/errors"
@@ -25,7 +24,6 @@ type fileMetaResponse struct {
 }
 
 func (req *ListRequest) getFileMetaInfoFromBlobber(blobber *blockchain.StorageNode, blobberIdx int, rspCh chan<- *fileMetaResponse) {
-	defer req.wg.Done()
 	body := new(bytes.Buffer)
 	formWriter := multipart.NewWriter(body)
 
@@ -92,13 +90,10 @@ func (req *ListRequest) getFileMetaInfoFromBlobber(blobber *blockchain.StorageNo
 
 func (req *ListRequest) getFileMetaFromBlobbers() []*fileMetaResponse {
 	numList := len(req.blobbers)
-	req.wg = &sync.WaitGroup{}
-	req.wg.Add(numList)
 	rspCh := make(chan *fileMetaResponse, numList)
 	for i := 0; i < numList; i++ {
 		go req.getFileMetaInfoFromBlobber(req.blobbers[i], i, rspCh)
 	}
-	req.wg.Wait()
 	fileInfos := make([]*fileMetaResponse, len(req.blobbers))
 	for i := 0; i < numList; i++ {
 		ch := <-rspCh

zboxcore/sdk/filemetaworker_test.go

@@ -169,12 +169,9 @@ func TestListRequest_getFileMetaInfoFromBlobber(t *testing.T) {
 				authToken: &marker.AuthTicket{
 					Signature: mockSignature,
 				},
-				wg: &sync.WaitGroup{},
 			}
 			rspCh := make(chan *fileMetaResponse, 1)
-			req.wg.Add(1)
 			go req.getFileMetaInfoFromBlobber(blobber, 73, rspCh)
-			req.wg.Wait()
 
 			var resp *fileMetaResponse
 			select {
@@ -280,7 +277,6 @@ func TestListRequest_getFileConsensusFromBlobbers(t *testing.T) {
 				allocationTx: mockAllocationTxId,
 				ctx:          context.TODO(),
 				blobbers:     []*blockchain.StorageNode{},
-				wg:           &sync.WaitGroup{},
 				Consensus:    tt.consensus, //nolint
 			}
 			for i := 0; i < tt.numBlobbers; i++ {