From 66fcd02aa22278b63e4bd3fa93ce5d9dba8f64d1 Mon Sep 17 00:00:00 2001
From: Harshavardhana <harsha@minio.io>
Date: Fri, 17 Sep 2021 12:14:12 -0700
Subject: [PATCH] de-couple walkMu and walkReadMu for some granularity (#13231)

This commit brings two locks instead of single lock for
WalkDir() calls on top of c25816eabc7712893b177b29b4820df0cb81da58.

The main reason is to avoid contention between readMetadata()
and ListDir() calls, ListDir() can take time on prefixes that
are huge for readdir() but this shouldn't end up blocking
all readMetadata() operations, this allows for more room for
I/O while not overly penalizing all listing operations.
---
 cmd/metacache-walk.go | 15 +++++++--------
 cmd/xl-storage.go     |  3 ++-
 2 files changed, 9 insertions(+), 9 deletions(-)

diff --git a/cmd/metacache-walk.go b/cmd/metacache-walk.go
index 6350179d3..f2022d742 100644
--- a/cmd/metacache-walk.go
+++ b/cmd/metacache-walk.go
@@ -121,6 +121,7 @@ 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()
@@ -173,9 +174,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()
+				s.walkReadMu.Lock()
 				meta.metadata, err = s.readMetadata(pathJoin(volumeDir, current, entry))
-				s.walkMu.Unlock()
+				s.walkReadMu.Unlock()
 				if err != nil {
 					logger.LogIf(ctx, err)
 					continue
@@ -190,9 +191,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()
+				s.walkReadMu.Lock()
 				meta.metadata, err = xioutil.ReadFile(pathJoin(volumeDir, current, entry))
-				s.walkMu.Unlock()
+				s.walkReadMu.Unlock()
 				if err != nil {
 					logger.LogIf(ctx, err)
 					continue
@@ -248,9 +249,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()
+			s.walkReadMu.Lock()
 			meta.metadata, err = s.readMetadata(pathJoin(volumeDir, meta.name, xlStorageFormatFile))
-			s.walkMu.Unlock()
+			s.walkReadMu.Unlock()
 			switch {
 			case err == nil:
 				// It was an object
@@ -259,9 +260,7 @@ func (s *xlStorage) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writ
 				}
 				out <- meta
 			case osIsNotExist(err), isSysErrIsDir(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 133d96292..e991c8082 100644
--- a/cmd/xl-storage.go
+++ b/cmd/xl-storage.go
@@ -142,7 +142,8 @@ type xlStorage struct {
 	sync.RWMutex
 
 	// mutex to prevent concurrent read operations overloading walks.
-	walkMu sync.Mutex
+	walkMu     sync.Mutex
+	walkReadMu sync.Mutex
 }
 
 // checkPathLength - returns error if given path name length more than 255