From c25816eabc7712893b177b29b4820df0cb81da58 Mon Sep 17 00:00:00 2001
From: Klaus Post <klauspost@gmail.com>
Date: Thu, 19 Aug 2021 03:10:36 +0200
Subject: [PATCH] xl walk: Limit walk concurrent IO (#12885)

We are observing heavy system loads, potentially
locking the system up for periods when concurrent
listing operations are performed.

We place a per-disk lock on walk IO operations.
This will minimize the impact of concurrent listing
operations on the entire system and de-prioritize
them compared to other operations.

Single list operations should remain largely unaffected.
---
 cmd/metacache-walk.go | 10 ++++++++++
 cmd/xl-storage.go     |  3 +++
 2 files changed, 13 insertions(+)

diff --git a/cmd/metacache-walk.go b/cmd/metacache-walk.go
index e11d244fa..476e01f5a 100644
--- a/cmd/metacache-walk.go
+++ b/cmd/metacache-walk.go
@@ -121,7 +121,9 @@ func (s *xlStorage) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writ
 		if contextCanceled(ctx) {
 			return ctx.Err()
 		}
+		s.walkMu.Lock()
 		entries, err := s.ListDir(ctx, opts.Bucket, current, -1)
+		s.walkMu.Unlock()
 		if err != nil {
 			// Folder could have gone away in-between
 			if err != errVolumeNotFound && err != errFileNotFound {
@@ -171,7 +173,9 @@ func (s *xlStorage) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writ
 			// If root was an object return it as such.
 			if HasSuffix(entry, xlStorageFormatFile) {
 				var meta metaCacheEntry
+				s.walkMu.Lock()
 				meta.metadata, err = s.readMetadata(pathJoin(volumeDir, current, entry))
+				s.walkMu.Unlock()
 				if err != nil {
 					logger.LogIf(ctx, err)
 					continue
@@ -186,7 +190,9 @@ func (s *xlStorage) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writ
 			// Check legacy.
 			if HasSuffix(entry, xlStorageFormatFileV1) {
 				var meta metaCacheEntry
+				s.walkMu.Lock()
 				meta.metadata, err = xioutil.ReadFile(pathJoin(volumeDir, current, entry))
+				s.walkMu.Unlock()
 				if err != nil {
 					logger.LogIf(ctx, err)
 					continue
@@ -242,7 +248,9 @@ func (s *xlStorage) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writ
 				meta.name = meta.name[:len(meta.name)-1] + globalDirSuffixWithSlash
 			}
 
+			s.walkMu.Lock()
 			meta.metadata, err = s.readMetadata(pathJoin(volumeDir, meta.name, xlStorageFormatFile))
+			s.walkMu.Unlock()
 			switch {
 			case err == nil:
 				// It was an object
@@ -251,7 +259,9 @@ func (s *xlStorage) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writ
 				}
 				out <- meta
 			case osIsNotExist(err):
+				s.walkMu.Lock()
 				meta.metadata, err = xioutil.ReadFile(pathJoin(volumeDir, meta.name, xlStorageFormatFileV1))
+				s.walkMu.Unlock()
 				if err == nil {
 					// It was an object
 					out <- meta
diff --git a/cmd/xl-storage.go b/cmd/xl-storage.go
index a818408ee..b6cde37b6 100644
--- a/cmd/xl-storage.go
+++ b/cmd/xl-storage.go
@@ -139,6 +139,9 @@ type xlStorage struct {
 
 	ctx context.Context
 	sync.RWMutex
+
+	// mutex to prevent concurrent read operations overloading walks.
+	walkMu sync.Mutex
 }
 
 // checkPathLength - returns error if given path name length more than 255