Performance testing

zboxcore/sdk/allocation.go

@@ -38,6 +38,16 @@ import (
 	"go.uber.org/zap"
 )
 
+// timing logs for upload
+var (
+	numRequest         int
+	totalChunkRead     int64
+	totalChunkWrite    int64
+	totalChunkUpload   int64
+	totalStatusWM      int64
+	totalCommitBlobber int64
+)
+
 var (
 	noBLOBBERS       = errors.New("", "No Blobbers set in this allocation")
 	notInitialized   = errors.New("sdk_not_initialized", "Please call InitStorageSDK Init and use GetAllocation to get the allocation object")
@@ -325,7 +335,9 @@ func (a *Allocation) dispatchWork(ctx context.Context) {
 		case downloadReq := <-a.downloadChan:
 			l.Logger.Info(fmt.Sprintf("received a download request for %v\n", downloadReq.remotefilepath))
 			go func() {
+				start := time.Now()
 				downloadReq.processDownload(ctx)
+				l.Logger.Info("[processDownload]", time.Since(start).Seconds())
 			}()
 		case repairReq := <-a.repairChan:
 
@@ -468,6 +480,7 @@ func (a *Allocation) EncryptAndUploadFileWithThumbnail(
 }
 
 func (a *Allocation) StartMultiUpload(workdir string, localPaths []string, fileNames []string, thumbnailPaths []string, encrypts []bool, chunkNumbers []int, remotePaths []string, isUpdate []bool, isWebstreaming []bool, status StatusCallback) error {
+	now := time.Now()
 	if len(localPaths) != len(thumbnailPaths) {
 		return errors.New("invalid_value", "length of localpaths and thumbnailpaths must be equal")
 	}
@@ -571,6 +584,7 @@ func (a *Allocation) StartMultiUpload(workdir string, localPaths []string, fileN
 		logger.Logger.Error("Error in multi upload ", err.Error())
 		return err
 	}
+	logger.Logger.Info("Multi upload completed", zap.Float64("time", time.Since(now).Seconds()))
 	return nil
 }
 
@@ -796,6 +810,7 @@ func (a *Allocation) DoMultiOperation(operations []OperationRequest) error {
 		return notInitialized
 	}
 	connectionID := zboxutil.NewConnectionId()
+	printLog := true
 	var mo MultiOperation
 	for i := 0; i < len(operations); {
 		// resetting multi operation and previous paths for every batch
@@ -879,6 +894,7 @@ func (a *Allocation) DoMultiOperation(operations []OperationRequest) error {
 				operation, newConnectionID, err = NewUploadOperation(op.Workdir, mo.allocationObj, mo.connectionID, op.FileMeta, op.FileReader, false, op.IsWebstreaming, op.Opts...)
 
 			case constants.FileOperationDelete:
+				printLog = false
 				operation = NewDeleteOperation(op.RemotePath, mo.operationMask, mo.maskMU, mo.consensusThresh, mo.fullconsensus, mo.ctx)
 
 			case constants.FileOperationUpdate:
@@ -916,8 +932,18 @@ func (a *Allocation) DoMultiOperation(operations []OperationRequest) error {
 				return err
 			}
 			mo.operations = nil
+			mo.operations = nil
+		}
+	}
+	l.Logger.Info("-----------------MultiOperation completed-----------------")
+	if printLog {
+		if numRequest > 0 {
+			l.Logger.Info(fmt.Sprintf("numRequests: %v avgReadChunk: %v avgWriteChunk: %v avgUploadTime: %v lockWMStatus: %v commitBlobber: %v", numRequest, totalChunkRead/int64(numRequest), totalChunkWrite/int64(numRequest), totalChunkUpload/int64(numRequest), totalStatusWM, totalCommitBlobber))
+		} else {
+			l.Logger.Info(fmt.Sprintf("statusCheck: %v,commitBlobber:%v ", totalStatusWM, totalCommitBlobber))
 		}
 	}
+	// "numRequests", numRequest, "avgReadChunk", totalChunkRead/int64(numRequest), "avgWriteChunk", totalChunkWrite/int64(numRequest), "avgUploadTime", totalChunkUpload/int64(numRequest), "lockWMStatus", totalStatusWM), ("commitBlobber", totalCommitBlobber))
 	return nil
 }
 
@@ -1150,7 +1176,7 @@ func (a *Allocation) processReadMarker(drs []*DownloadRequest) {
 				a.downloadChan <- dr
 			}(dr)
 		}
-		l.Logger.Info("[processReadMarker]", zap.String("allocation_id", a.ID),
+		l.Logger.Info("[processReadMarker]",
 			zap.Int("num of download requests", len(drs)),
 			zap.Duration("processDownloadRequest", elapsedProcessDownloadRequest))
 		return
@@ -1178,7 +1204,7 @@ func (a *Allocation) processReadMarker(drs []*DownloadRequest) {
 	wg.Wait()
 	elapsedSubmitReadmarker := time.Since(now) - elapsedProcessDownloadRequest
 
-	l.Logger.Info("[processReadMarker]", zap.String("allocation_id", a.ID),
+	l.Logger.Info("[processReadMarker]",
 		zap.Int("num of download requests", len(drs)),
 		zap.Duration("processDownloadRequest", elapsedProcessDownloadRequest),
 		zap.Duration("submitReadmarker", elapsedSubmitReadmarker))

zboxcore/sdk/blockdownloadworker.go

@@ -16,6 +16,7 @@ import (
 	"github.com/0chain/gosdk/zboxcore/blockchain"
 	"github.com/0chain/gosdk/zboxcore/client"
 	"github.com/0chain/gosdk/zboxcore/fileref"
+	l "github.com/0chain/gosdk/zboxcore/logger"
 	zlogger "github.com/0chain/gosdk/zboxcore/logger"
 	"github.com/0chain/gosdk/zboxcore/marker"
 	"github.com/0chain/gosdk/zboxcore/zboxutil"
@@ -116,7 +117,7 @@ func (req *BlockDownloadRequest) downloadBlobberBlock() {
 		if len(req.remotefilepath) > 0 {
 			req.remotefilepathhash = fileref.GetReferenceLookup(req.allocationID, req.remotefilepath)
 		}
-
+		start := time.Now()
 		var httpreq *http.Request
 		httpreq, err = zboxutil.NewDownloadRequest(req.blobber.Baseurl, req.allocationID, req.allocationTx)
 		if err != nil {
@@ -143,19 +144,18 @@ func (req *BlockDownloadRequest) downloadBlobberBlock() {
 
 		header.ToHeader(httpreq)
 
-		zlogger.Logger.Debug(fmt.Sprintf("downloadBlobberBlock - blobberID: %v, clientID: %v, blockNum: %d", req.blobber.ID, client.GetClientID(), header.BlockNum))
-
 		err = zboxutil.HttpDo(ctx, cncl, httpreq, func(resp *http.Response, err error) error {
 			if err != nil {
 				return err
 			}
 			if resp.Body != nil {
 				defer resp.Body.Close()
 			}
+			elapsedDownloadReqBlobber := time.Since(start).Milliseconds()
+			var rspData downloadBlock
 			if req.chunkSize == 0 {
 				req.chunkSize = CHUNK_SIZE
 			}
-			var rspData downloadBlock
 			respLen := resp.Header.Get("Content-Length")
 			var respBody []byte
 			if respLen != "" {
@@ -171,12 +171,13 @@ func (req *BlockDownloadRequest) downloadBlobberBlock() {
 					return err
 				}
 			} else {
-				respBody, err = readBody(resp.Body, int(req.numBlocks)*req.chunkSize)
+				respBody, err = readBody(resp.Body, 32*1024*1024)
 				if err != nil {
 					zlogger.Logger.Error("respBody read error: ", err)
 					return err
 				}
 			}
+			elapsedReadBody := time.Since(start).Milliseconds() - elapsedDownloadReqBlobber
 			if resp.StatusCode != http.StatusOK {
 				zlogger.Logger.Debug(fmt.Sprintf("downloadBlobberBlock FAIL - blobberID: %v, clientID: %v, blockNum: %d, retry: %d, response: %v", req.blobber.ID, client.GetClientID(), header.BlockNum, retry, string(respBody)))
 				if err = json.Unmarshal(respBody, &rspData); err == nil {
@@ -187,27 +188,30 @@ func (req *BlockDownloadRequest) downloadBlobberBlock() {
 
 			dR := downloadResponse{}
 			contentType := resp.Header.Get("Content-Type")
+			zlogger.Logger.Info("contentType", contentType)
 			if contentType == "application/json" {
 				err = json.Unmarshal(respBody, &dR)
 				if err != nil {
+					zlogger.Logger.Error("respBody unmarshal error: ", err)
 					return err
 				}
 			} else {
 				dR.Data = respBody
 			}
+			elapsedUnmarshal := time.Since(start).Milliseconds() - elapsedReadBody - elapsedDownloadReqBlobber
 			if req.contentMode == DOWNLOAD_CONTENT_FULL && req.shouldVerify {
-
+				now := time.Now()
 				vmp := util.MerklePathForMultiLeafVerification{
 					Nodes:    dR.Nodes,
 					Index:    dR.Indexes,
 					RootHash: req.blobberFile.validationRoot,
 					DataSize: req.blobberFile.size,
 				}
-				zlogger.Logger.Info("verifying multiple blocks")
 				err = vmp.VerifyMultipleBlocks(dR.Data)
 				if err != nil {
 					return errors.New("merkle_path_verification_error", err.Error())
 				}
+				l.Logger.Info("[verifyMultiBlock]", time.Since(now).Milliseconds())
 			}
 
 			rspData.idx = req.blobberIdx
@@ -221,13 +225,10 @@ func (req *BlockDownloadRequest) downloadBlobberBlock() {
 					rspData.BlockChunks = req.splitData(dR.Data, req.chunkSize)
 				}
 			} else {
-				if req.chunkSize == 0 {
-					req.chunkSize = CHUNK_SIZE
-				}
 				rspData.BlockChunks = req.splitData(dR.Data, req.chunkSize)
 			}
 
-			zlogger.Logger.Debug(fmt.Sprintf("downloadBlobberBlock 200 OK: blobberID: %v, clientID: %v, blockNum: %d", req.blobber.ID, client.GetClientID(), header.BlockNum))
+			zlogger.Logger.Debug(fmt.Sprintf("downloadBlobberBlock 200 OK: blobberID: %v, clientID: %v, blockNum: %d, downloadReqBlobber: %v,readBody: %v,unmarshalJSON: %v, totalTime: %v", req.blobber.ID, client.GetClientID(), header.BlockNum, elapsedDownloadReqBlobber, elapsedReadBody, elapsedUnmarshal, time.Since(start).Milliseconds()))
 
 			req.result <- &rspData
 			return nil
@@ -260,14 +261,17 @@ func AddBlockDownloadReq(req *BlockDownloadRequest) {
 
 func readBody(r io.Reader, size int) ([]byte, error) {
 	b := make([]byte, 0, size)
+	x := 0
 	for {
+		x++
 		if len(b) == cap(b) {
 			// Add more capacity (let append pick how much).
 			b = append(b, 0)[:len(b)]
 		}
 		n, err := r.Read(b[len(b):cap(b)])
 		b = b[:len(b)+n]
 		if err != nil {
+			zlogger.Logger.Info("readBodyTimes", x)
 			if err == io.EOF {
 				err = nil
 			}

zboxcore/sdk/chunked_upload.go

@@ -393,8 +393,10 @@ func (su *ChunkedUpload) process() error {
 	defer close(su.uploadChan)
 	defer su.ctxCncl(nil)
 	for {
-
+		numRequest++
+		start := time.Now()
 		chunks, err := su.readChunks(su.chunkNumber)
+		totalChunkRead += time.Since(start).Milliseconds()
 
 		// chunk, err := su.chunkReader.Next()
 		if err != nil {
@@ -534,7 +536,7 @@ func (su *ChunkedUpload) readChunks(num int) (*batchChunksData, error) {
 		chunkStartIndex: -1,
 		chunkEndIndex:   -1,
 	}
-
+	now := time.Now()
 	for i := 0; i < num; i++ {
 		chunk, err := su.chunkReader.Next()
 
@@ -579,6 +581,7 @@ func (su *ChunkedUpload) readChunks(num int) (*batchChunksData, error) {
 			break
 		}
 	}
+	logger.Logger.Info("readChunks", time.Since(now).Milliseconds())
 	return data, nil
 }
 
@@ -603,6 +606,7 @@ func (su *ChunkedUpload) processUpload(chunkStartIndex, chunkEndIndex int,
 	}
 	wg := &sync.WaitGroup{}
 	var pos uint64
+	start := time.Now()
 	for i := su.uploadMask; !i.Equals64(0); i = i.And(zboxutil.NewUint128(1).Lsh(pos).Not()) {
 		pos = uint64(i.TrailingZeros())
 		blobber := su.blobbers[pos]
@@ -637,7 +641,7 @@ func (su *ChunkedUpload) processUpload(chunkStartIndex, chunkEndIndex int,
 
 	wg.Wait()
 	close(wgErrors)
-
+	totalChunkWrite += time.Since(start).Milliseconds()
 	for err := range wgErrors {
 		su.removeProgress()
 		return thrown.New("upload_failed", fmt.Sprintf("Upload failed. %s", err))
@@ -746,6 +750,7 @@ func (su *ChunkedUpload) uploadProcessor() {
 			if !ok {
 				return
 			}
+			start := time.Now()
 			wgErrors := make(chan error, len(su.blobbers))
 			ctx, cancel := context.WithCancel(context.TODO())
 			defer cancel()
@@ -793,6 +798,7 @@ func (su *ChunkedUpload) uploadProcessor() {
 				su.progress.ChunkIndex = uploadData.chunkEndIndex
 				su.saveProgress()
 			}
+			totalChunkUpload += time.Since(start).Milliseconds()
 			if uploadData.isFinal {
 				return
 			}

zboxcore/sdk/chunked_upload_blobber.go

@@ -72,7 +72,8 @@ func (sb *ChunkedUploadBlobber) sendUploadRequest(
 	if err != nil {
 		return err
 	}
-
+	logger.Logger.Info("Uploading to blobber. ", body.Len())
+	req.Header.Set("Content-Length", fmt.Sprintf("%d", body.Len()))
 	req.Header.Add("Content-Type", formData.ContentType)
 
 	var (
@@ -85,6 +86,7 @@ func (sb *ChunkedUploadBlobber) sendUploadRequest(
 		err, shouldContinue = func() (err error, shouldContinue bool) {
 			reqCtx, ctxCncl := context.WithTimeout(ctx, su.uploadTimeOut)
 			var resp *http.Response
+			start := time.Now()
 			err = zboxutil.HttpDo(reqCtx, ctxCncl, req, func(r *http.Response, err error) error {
 				resp = r
 				return err
@@ -95,7 +97,7 @@ func (sb *ChunkedUploadBlobber) sendUploadRequest(
 				logger.Logger.Error("Upload : ", err)
 				return fmt.Errorf("Error while doing reqeust. Error %s", err), false
 			}
-
+			logger.Logger.Info("[chunkedUploadBlobber]", time.Since(start).Milliseconds())
 			if resp.Body != nil {
 				defer resp.Body.Close()
 			}

zboxcore/sdk/chunked_upload_chunk_reader.go

@@ -85,7 +85,9 @@ func createChunkReader(fileReader io.Reader, size, chunkSize int64, dataShards i
 	if hasher == nil {
 		return nil, errors.Throw(constants.ErrInvalidParameter, "hasher")
 	}
-
+	if chunkNumber <= 0 {
+		chunkNumber = 10
+	}
 	r := &chunkedUploadChunkReader{
 		fileReader:      fileReader,
 		size:            size,