Add Fanout benchmark

README.md

@@ -528,6 +528,40 @@ The analysis throughput represents the object count and sizes as they are writte
 
 Request times shown with `--analyze.v` represents request time for each snowball.
 
+## FANOUT
+
+The Fanout benchmark will test uploading a single object that is copied to multiple individual objects.
+This feature is only available on a recent MinIO server.
+
+Parameters:
+
+* `--obj.size=N` controls the size of each object that is uploaded. Default is 1MiB.
+* `--copies=N` controls the number of object copies per request. Default is 100.
+
+Size is calculated as `--obj.size` * `--copies`.
+
+Example: Use 8 concurrent uploads to copy a 512KB objects to 50 locations. 
+
+```
+λ warp fanout --copies=50 --obj.size=512KiB --concurrent=8
+warp: Benchmark data written to "warp-fanout-2023-06-15[105151]-j3qb.csv.zst"
+
+----------------------------------------
+Operation: POST
+* Average: 113.06 MiB/s, 226.12 obj/s
+
+Throughput, split into 57 x 1s:
+ * Fastest: 178.4MiB/s, 356.74 obj/s
+ * 50% Median: 113.9MiB/s, 227.76 obj/s
+ * Slowest: 56.3MiB/s, 112.53 obj/s
+warp: Cleanup Done.
+```
+
+The analysis throughput represents the object count and sizes as they are written when extracted.
+
+Request times shown with `--analyze.v` represents request time for each fan-out call.
+
+
 # Analysis
 
 When benchmarks have finished all request data will be saved to a file and an analysis will be shown.

cli/cli.go

@@ -97,6 +97,7 @@ func init() {
 		multipartCmd,
 		zipCmd,
 		snowballCmd,
+		fanoutCmd,
 	}
 	b := []cli.Command{
 		analyzeCmd,

cli/fanout.go

@@ -0,0 +1,78 @@
+/*
+ * Warp (C) 2019-2023 MinIO, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package cli
+
+import (
+	"github.com/minio/cli"
+	"github.com/minio/pkg/console"
+	"github.com/minio/warp/pkg/bench"
+)
+
+var fanoutFlags = []cli.Flag{
+	cli.StringFlag{
+		Name:  "obj.size",
+		Value: "1MiB",
+		Usage: "Size of each generated object. Can be a number or 10KiB/MiB/GiB. All sizes are base 2 binary.",
+	},
+	cli.IntFlag{
+		Name:   "copies",
+		Value:  100,
+		Usage:  "Number of copies per uploaded object",
+		Hidden: true,
+	},
+}
+
+// Fanout command.
+var fanoutCmd = cli.Command{
+	Name:   "fanout",
+	Usage:  "benchmark fan-out of objects on MinIO servers",
+	Action: mainFanout,
+	Before: setGlobalsFromContext,
+	Flags:  combineFlags(globalFlags, ioFlags, fanoutFlags, genFlags, benchFlags, analyzeFlags),
+	CustomHelpTemplate: `NAME:
+  {{.HelpName}} - {{.Usage}}
+
+USAGE:
+  {{.HelpName}} [FLAGS]
+  -> see https://github.com/minio/warp#fanout
+
+FLAGS:
+  {{range .VisibleFlags}}{{.}}
+  {{end}}`,
+}
+
+// mainFanout is the entry point for cp command.
+func mainFanout(ctx *cli.Context) error {
+	checkFanoutSyntax(ctx)
+	b := bench.Fanout{
+		Copies: ctx.Int("copies"),
+		Common: getCommon(ctx, newGenSource(ctx, "obj.size")),
+	}
+	return runBench(ctx, &b)
+}
+
+func checkFanoutSyntax(ctx *cli.Context) {
+	if ctx.NArg() > 0 {
+		console.Fatal("Command takes no arguments")
+	}
+	if ctx.Int("copies") <= 0 {
+		console.Fatal("Copies must be bigger than 0")
+	}
+	checkAnalyze(ctx)
+	checkBenchmark(ctx)
+}

go.mod

@@ -8,12 +8,12 @@ require (
 	github.com/dustin/go-humanize v1.0.1
 	github.com/fatih/color v1.14.1
 	github.com/influxdata/influxdb-client-go/v2 v2.12.3
-	github.com/klauspost/compress v1.16.4
+	github.com/klauspost/compress v1.16.5
 	github.com/minio/cli v1.24.2
 	github.com/minio/madmin-go/v2 v2.0.14
 	github.com/minio/mc v0.0.0-20230221173238-1843bab50013
 	github.com/minio/md5-simd v1.1.2
-	github.com/minio/minio-go/v7 v7.0.50
+	github.com/minio/minio-go/v7 v7.0.56
 	github.com/minio/pkg v1.6.3
 	github.com/minio/websocket v1.6.0
 	github.com/posener/complete v1.2.3
@@ -42,7 +42,7 @@ require (
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.17 // indirect
 	github.com/mattn/go-runewidth v0.0.14 // indirect
-	github.com/minio/sha256-simd v1.0.0 // indirect
+	github.com/minio/sha256-simd v1.0.1 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/philhofer/fwd v1.1.2 // indirect
@@ -51,9 +51,9 @@ require (
 	github.com/prometheus/procfs v0.9.0 // indirect
 	github.com/rivo/uniseg v0.4.4 // indirect
 	github.com/rjeczalik/notify v0.9.3 // indirect
-	github.com/rs/xid v1.4.0 // indirect
+	github.com/rs/xid v1.5.0 // indirect
 	github.com/shirou/gopsutil/v3 v3.23.1 // indirect
-	github.com/sirupsen/logrus v1.9.0 // indirect
+	github.com/sirupsen/logrus v1.9.2 // indirect
 	github.com/tinylib/msgp v1.1.8 // indirect
 	github.com/tklauser/go-sysconf v0.3.11 // indirect
 	github.com/tklauser/numcpus v0.6.0 // indirect

go.sum

@@ -47,10 +47,9 @@ github.com/influxdata/line-protocol v0.0.0-20200327222509-2487e7298839 h1:W9WBk7
 github.com/influxdata/line-protocol v0.0.0-20200327222509-2487e7298839/go.mod h1:xaLFMmpvUxqXtVkUJfg9QmT88cDaCJ3ZKgdZ78oO8Qo=
 github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
-github.com/klauspost/compress v1.16.4 h1:91KN02FnsOYhuunwU4ssRe8lc2JosWmizWa91B5v1PU=
-github.com/klauspost/compress v1.16.4/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
+github.com/klauspost/compress v1.16.5 h1:IFV2oUNUzZaz+XyusxpLzpzS8Pt5rh0Z16For/djlyI=
+github.com/klauspost/compress v1.16.5/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
 github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
-github.com/klauspost/cpuid/v2 v2.0.4/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.2.4 h1:acbojRNwl3o09bUq+yDCtZFc1aiwaAAxtcn8YkZXnvk=
 github.com/klauspost/cpuid/v2 v2.2.4/go.mod h1:RVVoqg1df56z8g3pUjL/3lE5UfnlrJX8tyFgg4nqhuY=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
@@ -103,13 +102,13 @@ github.com/minio/mc v0.0.0-20230221173238-1843bab50013 h1:+4Zrme5jLaQ7o8O3CMMZbO
 github.com/minio/mc v0.0.0-20230221173238-1843bab50013/go.mod h1:Yuxzld65ajOb1I89IfxInz0wqSIgHv+8NTiuoX+O+o8=
 github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34=
 github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM=
-github.com/minio/minio-go/v7 v7.0.50 h1:4IL4V8m/kI90ZL6GupCARZVrBv8/XrcKcJhaJ3iz68k=
-github.com/minio/minio-go/v7 v7.0.50/go.mod h1:IbbodHyjUAguneyucUaahv+VMNs/EOTV9du7A7/Z3HU=
+github.com/minio/minio-go/v7 v7.0.56 h1:pkZplIEHu8vinjkmhsexcXpWth2tjVLphrTZx6fBVZY=
+github.com/minio/minio-go/v7 v7.0.56/go.mod h1:NUDy4A4oXPq1l2yK6LTSvCEzAMeIcoz9lcj5dbzSrRE=
 github.com/minio/mux v1.8.2 h1:r9oVDFM09y+u8CF4HPLanguAG41niXgYwZAFkVHce9M=
 github.com/minio/pkg v1.6.3 h1:8XTM8pmlR5WZyy0rYxKj7nieRgwns07Vq4FejUsg+SM=
 github.com/minio/pkg v1.6.3/go.mod h1:ijZyWsfvtS0AcY6WT86AJ9VcK8gSsu++U28qlNCy9A0=
-github.com/minio/sha256-simd v1.0.0 h1:v1ta+49hkWZyvaKwrQB8elexRqm6Y0aMLjCNsrYxo6g=
-github.com/minio/sha256-simd v1.0.0/go.mod h1:OuYzVNI5vcoYIAmbIvHPl3N3jUzVedXbKy5RFepssQM=
+github.com/minio/sha256-simd v1.0.1 h1:6kaan5IFmwTNynnKKpDHe6FWHohJOHhCPchzK49dzMM=
+github.com/minio/sha256-simd v1.0.1/go.mod h1:Pz6AKMiUdngCLpeTL/RJY1M9rUuPMYujV5xJjtbRSN8=
 github.com/minio/websocket v1.6.0 h1:CPvnQvNvlVaQmvw5gtJNyYQhg4+xRmrPNhBbv8BdpAE=
 github.com/minio/websocket v1.6.0/go.mod h1:COH1CePZfHT9Ec1O7vZjTlX5uEPpyYnrifPNbu665DM=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
@@ -136,14 +135,14 @@ github.com/rivo/uniseg v0.4.4 h1:8TfxU8dW6PdqD27gjM8MVNuicgxIjxpm4K7x4jp8sis=
 github.com/rivo/uniseg v0.4.4/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
 github.com/rjeczalik/notify v0.9.3 h1:6rJAzHTGKXGj76sbRgDiDcYj/HniypXmSJo1SWakZeY=
 github.com/rjeczalik/notify v0.9.3/go.mod h1:gF3zSOrafR9DQEWSE8TjfI9NkooDxbyT4UgRGKZA0lc=
-github.com/rs/xid v1.4.0 h1:qd7wPTDkN6KQx2VmMBLrpHkiyQwgFXRnkOLacUiaSNY=
-github.com/rs/xid v1.4.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
+github.com/rs/xid v1.5.0 h1:mKX4bl4iPYJtEIxp6CYiUuLQ/8DYMoz0PUdtGgMFRVc=
+github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
 github.com/secure-io/sio-go v0.3.1 h1:dNvY9awjabXTYGsTF1PiCySl9Ltofk9GA3VdWlo7rRc=
 github.com/secure-io/sio-go v0.3.1/go.mod h1:+xbkjDzPjwh4Axd07pRKSNriS9SCiYksWnZqdnfpQxs=
 github.com/shirou/gopsutil/v3 v3.23.1 h1:a9KKO+kGLKEvcPIs4W62v0nu3sciVDOOOPUD0Hz7z/4=
 github.com/shirou/gopsutil/v3 v3.23.1/go.mod h1:NN6mnm5/0k8jw4cBfCnJtr5L7ErOTg18tMNpgFkn0hA=
-github.com/sirupsen/logrus v1.9.0 h1:trlNQbNUG3OdDrDil03MCb1H2o9nJ1x4/5LYw7byDE0=
-github.com/sirupsen/logrus v1.9.0/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
+github.com/sirupsen/logrus v1.9.2 h1:oxx1eChJGI6Uks2ZC4W1zpLlVgqB8ner4EuQwV4Ik1Y=
+github.com/sirupsen/logrus v1.9.2/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=

pkg/bench/fanout.go

@@ -0,0 +1,131 @@
+/*
+ * Warp (C) 2019-2023 MinIO, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package bench
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/minio/minio-go/v7"
+)
+
+// Fanout benchmarks upload speed.
+type Fanout struct {
+	Common
+	Copies   int
+	prefixes map[string]struct{}
+}
+
+// Prepare will create an empty bucket ot delete any content already there.
+func (u *Fanout) Prepare(ctx context.Context) error {
+	return u.createEmptyBucket(ctx)
+}
+
+// Start will execute the main benchmark.
+// Operations should begin executing when the start channel is closed.
+func (u *Fanout) Start(ctx context.Context, wait chan struct{}) (Operations, error) {
+	var wg sync.WaitGroup
+	wg.Add(u.Concurrency)
+	u.addCollector()
+	c := u.Collector
+	if u.AutoTermDur > 0 {
+		ctx = c.AutoTerm(ctx, http.MethodPost, u.AutoTermScale, autoTermCheck, autoTermSamples, u.AutoTermDur)
+	}
+	u.prefixes = make(map[string]struct{}, u.Concurrency)
+
+	// Non-terminating context.
+	nonTerm := context.Background()
+
+	for i := 0; i < u.Concurrency; i++ {
+		src := u.Source()
+		u.prefixes[src.Prefix()] = struct{}{}
+		go func(i int) {
+			rcv := c.Receiver()
+			defer wg.Done()
+			opts := minio.PutObjectFanOutRequest{
+				Entries:  make([]minio.PutObjectFanOutEntry, u.Copies),
+				Checksum: minio.Checksum{},
+				SSE:      nil,
+			}
+			done := ctx.Done()
+
+			<-wait
+			for {
+				select {
+				case <-done:
+					return
+				default:
+				}
+				obj := src.Object()
+				for i := range opts.Entries {
+					opts.Entries[i] = minio.PutObjectFanOutEntry{
+						Key:                fmt.Sprintf("%s/copy-%d.ext", obj.Name, i),
+						UserMetadata:       u.PutOpts.UserMetadata,
+						UserTags:           u.PutOpts.UserTags,
+						ContentType:        obj.ContentType,
+						ContentEncoding:    u.PutOpts.ContentEncoding,
+						ContentDisposition: u.PutOpts.ContentDisposition,
+						ContentLanguage:    u.PutOpts.ContentLanguage,
+						CacheControl:       u.PutOpts.CacheControl,
+					}
+				}
+				client, cldone := u.Client()
+				op := Operation{
+					OpType:   http.MethodPost,
+					Thread:   uint16(i),
+					Size:     obj.Size * int64(u.Copies),
+					ObjPerOp: u.Copies,
+					File:     obj.Name,
+					Endpoint: client.EndpointURL().String(),
+				}
+
+				op.Start = time.Now()
+				res, err := client.PutObjectFanOut(nonTerm, u.Bucket, obj.Reader, opts)
+				op.End = time.Now()
+				if err != nil {
+					u.Error("upload error: ", err)
+					op.Err = err.Error()
+				}
+
+				if len(res) != u.Copies && op.Err == "" {
+					err := fmt.Sprint("short upload. want:", u.Copies, " copies, got:", len(res))
+					if op.Err == "" {
+						op.Err = err
+					}
+					u.Error(err)
+				}
+				cldone()
+				rcv <- op
+			}
+		}(i)
+	}
+	wg.Wait()
+	return c.Close(), nil
+}
+
+// Cleanup deletes everything uploaded to the bucket.
+func (u *Fanout) Cleanup(ctx context.Context) {
+	pf := make([]string, 0, len(u.prefixes))
+	for p := range u.prefixes {
+		pf = append(pf, p)
+	}
+	u.deleteAllInBucket(ctx, pf...)
+}