ssc-test.go

// Copyright (C) 2023 Storj Labs, Inc.
// See LICENSE for copying information.

package main

import (
	"context"
	"fmt"
	"os"
	"sync"

	"github.com/zeebo/errs"

	"storj.io/common/memory"
	"storj.io/common/sync2"
	"storj.io/common/testrand"
	"storj.io/uplink"
)

func main() {
	accessString := os.Args[1]

	access, err := uplink.ParseAccess(accessString)
	if err != nil {
		panic(err)
	}

	ctx := context.Background()
	err = sscTest(ctx, access)
	if err != nil {
		panic(err)
	}
}

func sscTest(ctx context.Context, access *uplink.Access) error {
	bucketName := "michal-test"
	objectKey := "ssc-test/10k-segments"

	project, err := uplink.OpenProject(ctx, access)
	if err != nil {
		return err
	}
	defer project.Close()

	uploadInfo, err := project.BeginUpload(ctx, bucketName, objectKey, nil)
	if err != nil {
		return err
	}

	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	var mu sync.Mutex
	var es errs.Group

	addError := func(err error) {
		if err == nil {
			return
		}

		mu.Lock()
		defer mu.Unlock()

		es.Add(err)

		cancel()
	}

	limiter := sync2.NewLimiter(15)
	for i := 0; i < 10000; i++ {
		i := i
		ok := limiter.Go(ctx, func() {
			upload, err := project.UploadPart(ctx, bucketName, objectKey, uploadInfo.UploadID, uint32(i+1))
			if err != nil {
				addError(err)
				return
			}

			_, err = upload.Write(testrand.Bytes(5 * memory.MiB))
			if err != nil {
				addError(err)
				return
			}

			err = upload.Commit()
			if err != nil {
				addError(err)
				return
			}
		})
		if i%100 == 0 {
			fmt.Println("Parts", i)
		}
		if !ok {
			break
		}
	}

	limiter.Wait()

	if es.Err() != nil {
		return es.Err()
	}

	_, err = project.CommitUpload(ctx, bucketName, objectKey, uploadInfo.UploadID, nil)

	return err
}