mknet/benchmarks/s3concurrent/main.go

package main

import (
	"context"
	"crypto/tls"
	"fmt"
	"time"
	"io"
	"log"
	"math/rand"
	"net/http"
	"os"

	"github.com/google/uuid"
	"github.com/minio/minio-go/v7"
	"github.com/minio/minio-go/v7/pkg/credentials"
)

func buildMc() (*minio.Client, error) {
	_, isSSL := os.LookupEnv("SSL")
	opts := minio.Options{
		Creds:  credentials.NewStaticV4(os.Getenv("AWS_ACCESS_KEY_ID"), os.Getenv("AWS_SECRET_ACCESS_KEY"), ""),
		Secure: isSSL,
	}

	if region, ok := os.LookupEnv("REGION"); ok {
		opts.Region = region
	}

	if _, ok := os.LookupEnv("SSL_INSECURE"); ok {
		opts.Transport = &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}
	}

	mc, err := minio.New(os.Getenv("ENDPOINT"), &opts)
	return mc, err
}

type PRNG struct {
	rem int64
}

func (r *PRNG) Read(p []byte) (n int, err error) {
	//log.Printf("rem=%d, buf=%d\n", r.rem, len(p))
	if int64(len(p)) > r.rem {
		p = p[:r.rem]
	}

	if int64(len(p)) > r.rem {
		log.Fatal("LOGIC ERROR")
	}

	n, err = rand.Read(p)
	if err != nil {
		return
	}
	r.rem -= int64(n)
	if r.rem <= 0 {
		err = io.EOF
		//log.Printf("PRNG file has been fully read. rem=%d,n=%d,err=%s\n", r.rem, n, err)
	}
	return
}

func putObj(buck string, size int64) error {
	mc, err := buildMc()
	if err != nil {
		return err
	}

	prng := new(PRNG)
	prng.rem = size

	key := uuid.New().String()

	_, err = mc.PutObject(
		context.Background(),
		buck,
		key,
		prng,
		size,
		minio.PutObjectOptions{ContentType: "application/octet-stream"},
	)

	return err
}

func main() {
	minio.MaxRetry = 1
	mc, err := buildMc()
	if err != nil {
		log.Fatal("failed connect", err)
		return
	}

	// Create Bucket
	buck := uuid.New().String()
	err = mc.MakeBucket(context.Background(), buck, minio.MakeBucketOptions{})
	if err != nil {
		log.Fatal(err)
		return
	}
	log.Printf("created bucket %s\n", buck)

	fmt.Println("sent,success,elapsed,elapsed_per_req")

	// Send to bucket
	for i := 1; i <= 16; i++ {
		log.Printf("start concurrent loop with %d coroutines\n", i)
		syn := make(chan error)
		
		start := time.Now()
		for j := 1; j <= i; j++ {
			go func() {
				syn <- putObj(buck, 1024*1024)
			}()
		}

		errCount := 0
		for j := 1; j <= i; j++ {
			cerr := <-syn
			if cerr != nil {
				errCount += 1
				log.Printf("%d/%d failed with %s\n", j, i, cerr)
			}
		}
		elapsed := time.Since(start)
		fmt.Printf("%d,%d,%v,%v\n", i, i - errCount, elapsed.Nanoseconds(), elapsed.Nanoseconds() / int64(i))
		log.Printf("done, %d coroutines returned\n", i)
	}

	log.Println("bench done")
}