Some fixes and improvements

cmd/batch-handlers.go

@@ -1771,22 +1771,22 @@ func (a adminAPIHandlers) StartBatchJob(w http.ResponseWriter, r *http.Request)
 	// Fill with default values
 	if job.Replicate != nil {
 		if job.Replicate.Source.Snowball.Disable == nil {
-			job.Replicate.Source.Snowball.Disable = ptr(false)
+			job.Replicate.Source.Snowball.Disable = new(false)
 		}
 		if job.Replicate.Source.Snowball.Batch == nil {
-			job.Replicate.Source.Snowball.Batch = ptr(100)
+			job.Replicate.Source.Snowball.Batch = new(100)
 		}
 		if job.Replicate.Source.Snowball.InMemory == nil {
-			job.Replicate.Source.Snowball.InMemory = ptr(true)
+			job.Replicate.Source.Snowball.InMemory = new(true)
 		}
 		if job.Replicate.Source.Snowball.Compress == nil {
-			job.Replicate.Source.Snowball.Compress = ptr(false)
+			job.Replicate.Source.Snowball.Compress = new(false)
 		}
 		if job.Replicate.Source.Snowball.SmallerThan == nil {
-			job.Replicate.Source.Snowball.SmallerThan = ptr("5MiB")
+			job.Replicate.Source.Snowball.SmallerThan = new("5MiB")
 		}
 		if job.Replicate.Source.Snowball.SkipErrs == nil {
-			job.Replicate.Source.Snowball.SkipErrs = ptr(true)
+			job.Replicate.Source.Snowball.SkipErrs = new(true)
 		}
 	}
 

cmd/data-usage-cache.go

@@ -635,11 +635,12 @@ func (d *dataUsageCache) forceCompact(limit int) {
 // StringAll returns a detailed string representation of all entries in the cache.
 func (d *dataUsageCache) StringAll() string {
 	// Remove bloom filter from print.
-	s := fmt.Sprintf("info:%+v\n", d.Info)
+	var s strings.Builder
+	s.WriteString(fmt.Sprintf("info:%+v\n", d.Info))
 	for k, v := range d.Cache {
-		s += fmt.Sprintf("\t%v: %+v\n", k, v)
+		s.WriteString(fmt.Sprintf("\t%v: %+v\n", k, v))
 	}
-	return strings.TrimSpace(s)
+	return strings.TrimSpace(s.String())
 }
 
 // String returns a human readable representation of the string.

cmd/dynamic-timeouts.go

@@ -129,13 +129,10 @@ func (dt *dynamicTimeout) adjust(entries [dynamicTimeoutLogSize]time.Duration) {
 
 	if failPct > dynamicTimeoutIncreaseThresholdPct {
 		// We are hitting the timeout too often, so increase the timeout by 25%
-		timeout := min(
+		timeout := max(
 			// Set upper cap.
-			atomic.LoadInt64(&dt.timeout)*125/100, int64(maxDynamicTimeout))
-		// Safety, shouldn't happen
-		if timeout < dt.minimum {
-			timeout = dt.minimum
-		}
+			min(atomic.LoadInt64(&dt.timeout)*125/100, int64(maxDynamicTimeout)),
+			dt.minimum)
 		atomic.StoreInt64(&dt.timeout, timeout)
 	} else if failPct < dynamicTimeoutDecreaseThresholdPct {
 		// We are hitting the timeout relatively few times,

cmd/erasure-common.go

@@ -26,17 +26,21 @@ import (
 
 func (er erasureObjects) getOnlineDisks() (newDisks []StorageAPI) {
 	disks := er.getDisks()
+
 	var wg sync.WaitGroup
 	var mu sync.Mutex
+	newDisks = make([]StorageAPI, 0, len(disks))
+
 	r := rand.New(rand.NewSource(time.Now().UnixNano()))
+
 	for _, i := range r.Perm(len(disks)) {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			if disks[i] == nil {
-				return
-			}
-			di, err := disks[i].DiskInfo(context.Background(), DiskInfoOptions{})
+		disk := disks[i] // avoids repeated indexing
+		if disk == nil {
+			// avoids spawning goroutines that immediately return
+			continue
+		}
+		wg.Go(func() {
+			di, err := disk.DiskInfo(context.Background(), DiskInfoOptions{})
 			if err != nil || di.Healing {
 				// - Do not consume disks which are not reachable
 				//   unformatted or simply not accessible for some reason.
@@ -48,10 +52,11 @@ func (er erasureObjects) getOnlineDisks() (newDisks []StorageAPI) {
 			}
 
 			mu.Lock()
-			newDisks = append(newDisks, disks[i])
+			newDisks = append(newDisks, disk)
 			mu.Unlock()
-		}()
+		})
 	}
+
 	wg.Wait()
 	return newDisks
 }

cmd/erasure.go

@@ -285,24 +285,23 @@ func (er erasureObjects) getOnlineDisksWithHealingAndInfo(inclHealing bool) (new
 	infos := make([]DiskInfo, len(disks))
 	r := rand.New(rand.NewSource(time.Now().UnixNano()))
 	for _, i := range r.Perm(len(disks)) {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
+		disk := disks[i]
+		if disk == nil {
+			infos[i].Error = errDiskNotFound.Error()
+			continue
+		}
 
-			disk := disks[i]
-			if disk == nil {
-				infos[i].Error = errDiskNotFound.Error()
-				return
-			}
+		i, disk := i, disk
 
+		wg.Go(func() {
 			di, err := disk.DiskInfo(context.Background(), DiskInfoOptions{})
 			infos[i] = di
 			if err != nil {
 				// - Do not consume disks which are not reachable
 				//   unformatted or simply not accessible for some reason.
 				infos[i].Error = err.Error()
 			}
-		}()
+		})
 	}
 	wg.Wait()
 

cmd/iam-object-store.go

@@ -576,7 +576,7 @@ func (iamOS *IAMObjectStore) loadAllFromObjStore(ctx context.Context, cache *iam
 	if took := time.Since(listStartTime); took > maxIAMLoadOpTime {
 		var s strings.Builder
 		for k, v := range listedConfigItems {
-			fmt.Fprintf(&s, "    %s: %d items\n", k, len(v))
+			s.WriteString(fmt.Sprintf("    %s: %d items\n", k, len(v)))
 		}
 		logger.Info("listAllIAMConfigItems took %.2fs with contents:\n%s", took.Seconds(), s.String())
 	}

cmd/metacache-bucket.go

@@ -96,6 +96,14 @@ func (b *bucketMetacache) findCache(o listPathOptions) metacache {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
+	if b.caches == nil {
+		b.caches = make(map[string]metacache)
+	}
+
+	if b.cachesRoot == nil {
+		b.cachesRoot = make(map[string][]string)
+	}
+
 	// Check if exists already.
 	if c, ok := b.caches[o.ID]; ok {
 		c.lastHandout = time.Now()
@@ -157,16 +165,16 @@ func (b *bucketMetacache) cleanup() {
 			}
 			remainCaches = append(remainCaches, cache)
 		}
-		if len(remainCaches) > metacacheMaxEntries {
-			// Sort oldest last...
-			sort.Slice(remainCaches, func(i, j int) bool {
-				return remainCaches[i].lastHandout.Before(remainCaches[j].lastHandout)
-			})
-			// Keep first metacacheMaxEntries...
-			for _, cache := range remainCaches[metacacheMaxEntries:] {
-				if time.Since(cache.lastHandout) > metacacheMaxClientWait {
-					remove[cache.id] = struct{}{}
-				}
+
+		// Sort oldest last...
+		sort.Slice(remainCaches, func(i, j int) bool {
+			return remainCaches[i].lastHandout.Before(remainCaches[j].lastHandout)
+		})
+		// Remove oldest entries above the limit.
+		excess := len(remainCaches) - metacacheMaxEntries
+		for _, cache := range remainCaches[:excess] {
+			if time.Since(cache.lastHandout) > metacacheMaxClientWait {
+				remove[cache.id] = struct{}{}
 			}
 		}
 	}
@@ -181,6 +189,9 @@ func (b *bucketMetacache) cleanup() {
 func (b *bucketMetacache) updateCacheEntry(update metacache) (metacache, error) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
+	if b.caches == nil {
+		return update, errFileNotFound
+	}
 	existing, ok := b.caches[update.id]
 	if !ok {
 		return update, errFileNotFound
@@ -227,32 +238,46 @@ func (b *bucketMetacache) deleteAll() {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
-	b.updated = true
 	// Delete all.
 	ez.deleteAll(ctx, minioMetaBucket, metacachePrefixForID(b.bucket, slashSeparator))
+	b.updated = true
 	b.caches = make(map[string]metacache, 10)
 	b.cachesRoot = make(map[string][]string, 10)
 }
 
+// for test purposes
+var deleteMetacacheFiles = func(c metacache, ctx context.Context) {
+	c.delete(ctx)
+}
+
 // deleteCache will delete a specific cache and all files related to it across the cluster.
 func (b *bucketMetacache) deleteCache(id string) {
 	b.mu.Lock()
+	defer b.mu.Unlock()
 	c, ok := b.caches[id]
-	if ok {
-		// Delete from root map.
-		list := b.cachesRoot[c.root]
-		for i, lid := range list {
-			if id == lid {
-				list = append(list[:i], list[i+1:]...)
-				break
-			}
+	if !ok {
+		return
+	}
+
+	// Delete from root map.
+	list := b.cachesRoot[c.root]
+	n := 0
+	for _, lid := range list {
+		if lid != id {
+			list[n] = lid
+			n++
 		}
-		b.cachesRoot[c.root] = list
-		delete(b.caches, id)
-		b.updated = true
 	}
-	b.mu.Unlock()
-	if ok {
-		c.delete(context.Background())
+	list = list[:n]
+
+	if len(list) == 0 {
+		delete(b.cachesRoot, c.root)
+	} else {
+		b.cachesRoot[c.root] = list
 	}
+
+	delete(b.caches, id)
+	b.updated = true
+
+	deleteMetacacheFiles(c, context.Background())
 }