choose starting concurrency based on number of local disks

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 {
+	var 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 {