Merge pull request #1394 from 0chain/sprint-1.13

Sprint 1.13

code/go/0chain.net/blobbercore/allocation/allocationchange.go

@@ -14,7 +14,7 @@ import (
  "github.com/0chain/blobber/code/go/0chain.net/core/common"
  "github.com/0chain/blobber/code/go/0chain.net/core/logging"
  "github.com/0chain/gosdk/constants"
- "github.com/remeh/sizedwaitgroup"
+ "golang.org/x/sync/errgroup"
 
  "go.uber.org/zap"
  "gorm.io/gorm"
@@ -236,35 +236,19 @@ func (cc *AllocationChangeCollector) ApplyChanges(ctx context.Context, allocatio
 }
 
 func (a *AllocationChangeCollector) CommitToFileStore(ctx context.Context) error {
- commitCtx, cancel := context.WithCancel(ctx)
- defer cancel()
- // Can be configured at runtime, this number will depend on the number of active allocations
- swg := sizedwaitgroup.New(5)
+ // Limit can be configured at runtime, this number will depend on the number of active allocations
+ eg, _ := errgroup.WithContext(ctx)
+ eg.SetLimit(5)
  mut := &sync.Mutex{}
- var (
- commitError error
- errorMutex sync.Mutex
- )
  for _, change := range a.AllocationChanges {
- select {
- case <-commitCtx.Done():
- return fmt.Errorf("commit to filestore failed: %s", commitError.Error())
- default:
- }
- swg.Add()
- go func(change AllocationChangeProcessor) {
- err := change.CommitToFileStore(ctx, mut)
- if err != nil && !errors.Is(common.ErrFileWasDeleted, err) {
- cancel()
- errorMutex.Lock()
- commitError = err
- errorMutex.Unlock()
- }
- swg.Done()
- }(change)
+ allocChange := change
+ eg.Go(func() error {
+ return allocChange.CommitToFileStore(ctx, mut)
+ })
  }
- swg.Wait()
- return commitError
+ logging.Logger.Info("Waiting for commit to filestore", zap.String("allocation_id", a.AllocationID))
+
+ return eg.Wait()
 }
 
 func (a *AllocationChangeCollector) DeleteChanges(ctx context.Context) {

code/go/0chain.net/blobbercore/allocation/common_test.go

@@ -29,6 +29,10 @@ func (mfs *MockFileStore) WriteFile(allocID, connID string,
  }, nil
 }
 
+func (mfs *MockFileStore) WriteDataToTree(allocID, connID, fileName, filePathHash string, hasher *filestore.CommitHasher) error {
+ return nil
+}
+
 func (mfs *MockFileStore) CommitWrite(allocID, connID string, fileData *filestore.FileInputData) (bool, error) {
  return true, nil
 }
@@ -64,6 +68,10 @@ func (mfs *MockFileStore) GetFilePathSize(allocID, contentHash, thumbHash string
  return 0, 0, nil
 }
 
+func (mfs *MockFileStore) GetTempFilePath(allocID, connID, fileName, filePathHash string) string {
+ return ""
+}
+
 func (mfs *MockFileStore) GetBlocksMerkleTreeForChallenge(cir *filestore.ChallengeReadBlockInput) (*filestore.ChallengeResponse, error) {
  return nil, nil
 }