Merge pull request #1392 from 0chain/fix/upload-resume

Fix resume upload
0chain · Feb 26, 2024 · ebbb3d9 · ebbb3d9
2 parents 5b5fb10 + 264f6c4
commit ebbb3d9
Show file tree

Hide file tree

Showing 5 changed files with 21 additions and 13 deletions.
diff --git a/code/go/0chain.net/blobbercore/allocation/connection.go b/code/go/0chain.net/blobbercore/allocation/connection.go
@@ -193,7 +193,6 @@ func SaveFileChange(connectionID, pathHash, fileName string, cmd FileCommand, is
  hasher := filestore.GetNewCommitHasher(contentSize)
  change.hasher = hasher
  change.seqPQ = seqpriorityqueue.NewSeqPriorityQueue(contentSize)
- hasher.WG.Add(1)
  go hasher.Start(connectionObj.ctx, connectionID, connectionObj.AllocationID, fileName, pathHash, change.seqPQ)
  saveChange = true
  }
@@ -203,6 +202,7 @@ func SaveFileChange(connectionID, pathHash, fileName string, cmd FileCommand, is
  if change.isFinalized {
  return false, nil
  }
+
  if isFinal {
  change.isFinalized = true
  change.seqPQ.Done(seqpriorityqueue.UploadData{

diff --git a/code/go/0chain.net/blobbercore/filestore/storage.go b/code/go/0chain.net/blobbercore/filestore/storage.go
@@ -276,7 +276,9 @@ func (fs *FileStore) CommitWrite(allocID, conID string, fileData *FileInputData)
  nodeSie := getNodesSize(fileData.Size, util.MaxMerkleLeavesSize)
  fileSize := rStat.Size() - nodeSie - FMTSize
  now := time.Now()
- err = fileData.Hasher.Wait(conID, allocID, fileData.Name, filePathHash)
+ ctx, cancel := context.WithTimeout(context.TODO(), 30*time.Second)
+ defer cancel()
+ err = fileData.Hasher.Wait(ctx, conID, allocID, fileData.Name, filePathHash)
  if err != nil {
  return false, common.NewError("hasher_wait_error", err.Error())
  }

diff --git a/code/go/0chain.net/blobbercore/filestore/store_test.go b/code/go/0chain.net/blobbercore/filestore/store_test.go
@@ -273,7 +273,6 @@ func TestStoreStorageWriteAndCommit(t *testing.T) {
  tF, err := os.Stat(tempFilePath)
  require.Nil(t, err)
  seqPQ := seqpriorityqueue.NewSeqPriorityQueue(int64(size))
- hasher.WG.Add(1)
  go hasher.Start(context.TODO(), test.connID, test.allocID, test.fileName, pathHash, seqPQ)
  seqPQ.Done(seqpriorityqueue.UploadData{
  Offset: 0,
@@ -365,7 +364,6 @@ func TestDeletePreCommitDir(t *testing.T) {
  nodeSize := getNodesSize(int64(size), util.MaxMerkleLeavesSize)
  require.Equal(t, int64(size), tF.Size()-nodeSize-FMTSize)
  seqPQ := seqpriorityqueue.NewSeqPriorityQueue(int64(size))
- hasher.WG.Add(1)
  go hasher.Start(context.TODO(), connID, allocID, fileName, pathHash, seqPQ)
  seqPQ.Done(seqpriorityqueue.UploadData{
  Offset: 0,
@@ -401,7 +399,6 @@ func TestDeletePreCommitDir(t *testing.T) {
  _, err = os.Stat(tempFilePath)
  require.Nil(t, err)
  seqPQ = seqpriorityqueue.NewSeqPriorityQueue(int64(size))
- hasher.WG.Add(1)
  go hasher.Start(context.TODO(), connID, allocID, fileName, pathHash, seqPQ)
  seqPQ.Done(seqpriorityqueue.UploadData{
  Offset: 0,
@@ -476,7 +473,6 @@ func TestStorageUploadUpdate(t *testing.T) {
  nodeSize := getNodesSize(int64(size), util.MaxMerkleLeavesSize)
  require.Equal(t, int64(size), tF.Size()-nodeSize-FMTSize)
  seqPQ := seqpriorityqueue.NewSeqPriorityQueue(int64(size))
- hasher.WG.Add(1)
  go hasher.Start(context.TODO(), connID, allocID, fileName, pathHash, seqPQ)
  seqPQ.Done(seqpriorityqueue.UploadData{
  Offset: 0,

diff --git a/code/go/0chain.net/blobbercore/filestore/tree_validation.go b/code/go/0chain.net/blobbercore/filestore/tree_validation.go
@@ -409,7 +409,7 @@ type CommitHasher struct {
  fmt *fixedMerkleTree
  vt *validationTree
  isInitialized bool
- WG  sync.WaitGroup
+ doneChan chan struct{}
  hashErr error
  dataSize int64
 }
@@ -419,12 +419,13 @@ func GetNewCommitHasher(dataSize int64) *CommitHasher {
  c.fmt = getNewFixedMerkleTree()
  c.vt = getNewValidationTree(dataSize)
  c.isInitialized = true
+ c.doneChan = make(chan struct{})
  c.dataSize = dataSize
  return c
 }
 
 func (c *CommitHasher) Start(ctx context.Context, connID, allocID, fileName, filePathHash string, seqPQ *seqpriorityqueue.SeqPriorityQueue) {
- defer c.WG.Done()
+ defer close(c.doneChan)
  tempFilePath := GetFileStore().GetTempFilePath(allocID, connID, fileName, filePathHash)
  f, err := os.Open(tempFilePath)
  if err != nil {
@@ -453,6 +454,8 @@ func (c *CommitHasher) Start(ctx context.Context, connID, allocID, fileName, fil
  default:
  }
  toFinalize = true
+ } else if pq.DataBytes == 0 {
+ continue
  }
  logging.Logger.Info("hasher_pop", zap.Int64("offset", pq.Offset), zap.Int64("dataBytes", pq.DataBytes), zap.Any("toFinalize", toFinalize), zap.Int64("dataSize", c.dataSize), zap.String("filename", fileName), zap.Int64("totalWritten", totalWritten))
  bufSize := 2 * BufferSize
@@ -487,9 +490,13 @@ func (c *CommitHasher) Start(ctx context.Context, connID, allocID, fileName, fil
  }
 }
 
-func (c *CommitHasher) Wait(connID, allocID, fileName, filePathHash string) error {
- c.WG.Wait()
- return c.hashErr
+func (c *CommitHasher) Wait(ctx context.Context, connID, allocID, fileName, filePathHash string) error {
+ select {
+ case <-c.doneChan:
+ return c.hashErr
+ case <-ctx.Done():
+ return ctx.Err()
+ }
 }
 
 func (c *CommitHasher) Write(b []byte) (int, error) {

diff --git a/code/go/0chain.net/blobbercore/seqpriorityqueue/seqpriorityqueue.go b/code/go/0chain.net/blobbercore/seqpriorityqueue/seqpriorityqueue.go
@@ -78,7 +78,7 @@ func (pq *SeqPriorityQueue) Done(v UploadData) {
 
 func (pq *SeqPriorityQueue) Popup() UploadData {
  pq.lock.Lock()
- for pq.queue.Len() == 0 && !pq.done || (pq.queue.Len() > 0 && pq.queue[0].Offset != pq.next) {
+ for pq.queue.Len() == 0 || (!pq.done && pq.queue[0].Offset > pq.next) {
  pq.cv.Wait()
  }
  if pq.done {
@@ -91,8 +91,11 @@ func (pq *SeqPriorityQueue) Popup() UploadData {
  retItem := UploadData{
  Offset: pq.next,
  }
- for pq.queue.Len() > 0 && pq.queue[0].Offset == pq.next {
+ for pq.queue.Len() > 0 && pq.queue[0].Offset <= pq.next {
  item := heap.Pop(&pq.queue).(UploadData)
+ if item.Offset < pq.next {
+ continue
+ }
  pq.next += item.DataBytes
  }
  retItem.DataBytes = pq.next - retItem.Offset