choose starting concurrency based on number of local disks (#15428)

smaller setups may have less drives per server choosing
the concurrency based on number of local drives, and let
the MinIO server change the overall concurrency as
necessary.

cmd/admin-handlers.go

@@ -1206,18 +1206,6 @@ func (a adminAPIHandlers) ObjectSpeedTestHandler(w http.ResponseWriter, r *http.
 		concurrent = 32
 	}
 
-	if runtime.GOMAXPROCS(0) < concurrent {
-		concurrent = runtime.GOMAXPROCS(0)
-	}
-
-	// if we have less drives than concurrency then choose
-	// only the concurrency to be number of drives to start
-	// with - since default '32' might be big and may not
-	// complete in total time of 10s.
-	if globalEndpoints.NEndpoints() < concurrent {
-		concurrent = globalEndpoints.NEndpoints()
-	}
-
 	duration, err := time.ParseDuration(durationStr)
 	if err != nil {
 		duration = time.Second * 10
@@ -2225,17 +2213,6 @@ func (a adminAPIHandlers) HealthInfoHandler(w http.ResponseWriter, r *http.Reque
 	getAndWriteObjPerfInfo := func() {
 		if query.Get(string(madmin.HealthDataTypePerfObj)) == "true" {
 			concurrent := 32
-			if runtime.GOMAXPROCS(0) < concurrent {
-				concurrent = runtime.GOMAXPROCS(0)
-			}
-
-			// if we have less drives than concurrency then choose
-			// only the concurrency to be number of drives to start
-			// with - since default '32' might be big and may not
-			// complete in total time of 10s.
-			if globalEndpoints.NEndpoints() < concurrent {
-				concurrent = globalEndpoints.NEndpoints()
-			}
 
 			storageInfo, _ := objectAPI.StorageInfo(ctx)
 

cmd/endpoint.go

@@ -292,6 +292,19 @@ func (l EndpointServerPools) LocalDisksPaths() []string {
 	return disks
 }
 
+// NLocalDisksPathsPerPool returns the disk paths of the local disks per pool
+func (l EndpointServerPools) NLocalDisksPathsPerPool() []int {
+	localDisksCount := make([]int, len(l))
+	for i, ep := range l {
+		for _, endpoint := range ep.Endpoints {
+			if endpoint.IsLocal {
+				localDisksCount[i]++
+			}
+		}
+	}
+	return localDisksCount
+}
+
 // FirstLocal returns true if the first endpoint is local.
 func (l EndpointServerPools) FirstLocal() bool {
 	return l[0].Endpoints[0].IsLocal

cmd/speedtest.go

@@ -21,6 +21,7 @@ import (
 	"context"
 	"fmt"
 	"net/url"
+	"runtime"
 	"sort"
 	"time"
 
@@ -48,6 +49,38 @@ func objectSpeedTest(ctx context.Context, opts speedTestOpts) chan madmin.SpeedT
 
 		concurrency := opts.concurrencyStart
 
+		if opts.autotune {
+			// if we have less drives than concurrency then choose
+			// only the concurrency to be number of drives to start
+			// with - since default '32' might be big and may not
+			// complete in total time of 10s.
+			if globalEndpoints.NEndpoints() < concurrency {
+				concurrency = globalEndpoints.NEndpoints()
+			}
+
+			// Check if we have local disks per pool less than
+			// the concurrency make sure we choose only the "start"
+			// concurrency to be equal to the lowest number of
+			// local disks per server.
+			for _, localDiskCount := range globalEndpoints.NLocalDisksPathsPerPool() {
+				if localDiskCount < concurrency {
+					concurrency = localDiskCount
+				}
+			}
+
+			// Any concurrency less than '4' just stick to '4' concurrent
+			// operations for now to begin with.
+			if concurrency < 4 {
+				concurrency = 4
+			}
+
+			// if GOMAXPROCS is set to a lower value then choose to use
+			// concurrency == GOMAXPROCS instead.
+			if runtime.GOMAXPROCS(0) < concurrency {
+				concurrency = runtime.GOMAXPROCS(0)
+			}
+		}
+
 		throughputHighestGet := uint64(0)
 		throughputHighestPut := uint64(0)
 		var throughputHighestResults []SpeedTestResult
@@ -97,7 +130,11 @@ func objectSpeedTest(ctx context.Context, opts speedTestOpts) chan madmin.SpeedT
 			result.Version = Version
 			result.Concurrent = concurrency
 
-			ch <- result
+			select {
+			case ch <- result:
+			case <-ctx.Done():
+				return
+			}
 		}
 
 		for {