fix: initialize config once per startup (#9851)

fixes #9857

cmd/bucket-handlers.go

@@ -47,7 +47,7 @@ import (
 const (
 	getBucketVersioningResponse = `<VersioningConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"/>`
 	objectLockConfig            = "object-lock.xml"
-	bucketTaggingConfigFile     = "tagging.xml"
+	bucketTaggingConfig         = "tagging.xml"
 )
 
 // Check if there are buckets on server without corresponding entry in etcd backend and
@@ -1123,7 +1123,7 @@ func (api objectAPIHandlers) PutBucketTaggingHandler(w http.ResponseWriter, r *h
 		return
 	}
 
-	if err = globalBucketMetadataSys.Update(bucket, bucketTaggingConfigFile, configData); err != nil {
+	if err = globalBucketMetadataSys.Update(bucket, bucketTaggingConfig, configData); err != nil {
 		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
 		return
 	}
@@ -1191,7 +1191,7 @@ func (api objectAPIHandlers) DeleteBucketTaggingHandler(w http.ResponseWriter, r
 		return
 	}
 
-	if err := globalBucketMetadataSys.Update(bucket, bucketTaggingConfigFile, nil); err != nil {
+	if err := globalBucketMetadataSys.Update(bucket, bucketTaggingConfig, nil); err != nil {
 		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
 		return
 	}

cmd/bucket-metadata-sys.go

@@ -76,7 +76,44 @@ func (sys *BucketMetadataSys) Update(bucket string, configFile string, configDat
 
 	if globalIsGateway {
 		// This code is needed only for gateway implementations.
-		if configFile == bucketPolicyConfig {
+		switch configFile {
+		case bucketSSEConfig:
+			if globalGatewayName == "nas" {
+				meta, err := loadBucketMetadata(GlobalContext, objAPI, bucket)
+				if err != nil {
+					return err
+				}
+				meta.EncryptionConfigXML = configData
+				return meta.Save(GlobalContext, objAPI)
+			}
+		case bucketLifecycleConfig:
+			if globalGatewayName == "nas" {
+				meta, err := loadBucketMetadata(GlobalContext, objAPI, bucket)
+				if err != nil {
+					return err
+				}
+				meta.LifecycleConfigXML = configData
+				return meta.Save(GlobalContext, objAPI)
+			}
+		case bucketTaggingConfig:
+			if globalGatewayName == "nas" {
+				meta, err := loadBucketMetadata(GlobalContext, objAPI, bucket)
+				if err != nil {
+					return err
+				}
+				meta.TaggingConfigXML = configData
+				return meta.Save(GlobalContext, objAPI)
+			}
+		case bucketNotificationConfig:
+			if globalGatewayName == "nas" {
+				meta, err := loadBucketMetadata(GlobalContext, objAPI, bucket)
+				if err != nil {
+					return err
+				}
+				meta.NotificationConfigXML = configData
+				return meta.Save(GlobalContext, objAPI)
+			}
+		case bucketPolicyConfig:
 			if configData == nil {
 				return objAPI.DeleteBucketPolicy(GlobalContext, bucket)
 			}
@@ -107,7 +144,7 @@ func (sys *BucketMetadataSys) Update(bucket string, configFile string, configDat
 		meta.LifecycleConfigXML = configData
 	case bucketSSEConfig:
 		meta.EncryptionConfigXML = configData
-	case bucketTaggingConfigFile:
+	case bucketTaggingConfig:
 		meta.TaggingConfigXML = configData
 	case objectLockConfig:
 		meta.ObjectLockConfigXML = configData

cmd/bucket-metadata.go

@@ -206,7 +206,7 @@ func (b *BucketMetadata) convertLegacyConfigs(ctx context.Context, objectAPI Obj
 		bucketLifecycleConfig,
 		bucketQuotaConfigFile,
 		bucketSSEConfig,
-		bucketTaggingConfigFile,
+		bucketTaggingConfig,
 		objectLockConfig,
 	}
 
@@ -255,7 +255,7 @@ func (b *BucketMetadata) convertLegacyConfigs(ctx context.Context, objectAPI Obj
 			b.LifecycleConfigXML = configData
 		case bucketSSEConfig:
 			b.EncryptionConfigXML = configData
-		case bucketTaggingConfigFile:
+		case bucketTaggingConfig:
 			b.TaggingConfigXML = configData
 		case objectLockConfig:
 			b.ObjectLockConfigXML = configData

cmd/config-current.go

@@ -325,13 +325,37 @@ func lookupConfigs(s config.Config) {
 
 	etcdCfg, err := etcd.LookupConfig(s[config.EtcdSubSys][config.Default], globalRootCAs)
 	if err != nil {
-		logger.LogIf(ctx, fmt.Errorf("Unable to initialize etcd config: %w", err))
+		if globalIsGateway {
+			logger.FatalIf(err, "Unable to initialize etcd config")
+		} else {
+			logger.LogIf(ctx, fmt.Errorf("Unable to initialize etcd config: %w", err))
+		}
 	}
 
 	if etcdCfg.Enabled {
 		globalEtcdClient, err = etcd.New(etcdCfg)
 		if err != nil {
-			logger.LogIf(ctx, fmt.Errorf("Unable to initialize etcd config: %w", err))
+			if globalIsGateway {
+				logger.FatalIf(err, "Unable to initialize etcd config")
+			} else {
+				logger.LogIf(ctx, fmt.Errorf("Unable to initialize etcd config: %w", err))
+			}
+		}
+		if len(globalDomainNames) != 0 && !globalDomainIPs.IsEmpty() && globalEtcdClient != nil {
+			globalDNSConfig, err = dns.NewCoreDNS(etcdCfg.Config,
+				dns.DomainNames(globalDomainNames),
+				dns.DomainIPs(globalDomainIPs),
+				dns.DomainPort(globalMinioPort),
+				dns.CoreDNSPath(etcdCfg.CoreDNSPath),
+			)
+			if err != nil {
+				if globalIsGateway {
+					logger.FatalIf(err, "Unable to initialize DNS config")
+				} else {
+					logger.LogIf(ctx, fmt.Errorf("Unable to initialize DNS config for %s: %w",
+						globalDomainNames, err))
+				}
+			}
 		}
 	}
 
@@ -342,19 +366,6 @@ func lookupConfigs(s config.Config) {
 	// but not federation.
 	globalBucketFederation = etcdCfg.PathPrefix == "" && etcdCfg.Enabled
 
-	if len(globalDomainNames) != 0 && !globalDomainIPs.IsEmpty() && globalEtcdClient != nil {
-		globalDNSConfig, err = dns.NewCoreDNS(etcdCfg.Config,
-			dns.DomainNames(globalDomainNames),
-			dns.DomainIPs(globalDomainIPs),
-			dns.DomainPort(globalMinioPort),
-			dns.CoreDNSPath(etcdCfg.CoreDNSPath),
-		)
-		if err != nil {
-			logger.LogIf(ctx, fmt.Errorf("Unable to initialize DNS config for %s: %w",
-				globalDomainNames, err))
-		}
-	}
-
 	globalServerRegion, err = config.LookupRegion(s[config.RegionSubSys][config.Default])
 	if err != nil {
 		logger.LogIf(ctx, fmt.Errorf("Invalid region configuration: %w", err))
@@ -377,7 +388,11 @@ func lookupConfigs(s config.Config) {
 
 	globalCacheConfig, err = cache.LookupConfig(s[config.CacheSubSys][config.Default])
 	if err != nil {
-		logger.LogIf(ctx, fmt.Errorf("Unable to setup cache: %w", err))
+		if globalIsGateway {
+			logger.FatalIf(err, "Unable to setup cache")
+		} else {
+			logger.LogIf(ctx, fmt.Errorf("Unable to setup cache: %w", err))
+		}
 	}
 
 	if globalCacheConfig.Enabled {
@@ -555,6 +570,8 @@ func newServerConfig() config.Config {
 	return config.New()
 }
 
+var lookupConfigOnce sync.Once
+
 // newSrvConfig - initialize a new server config, saves env parameters if
 // found, otherwise use default parameters
 func newSrvConfig(objAPI ObjectLayer) error {
@@ -562,7 +579,9 @@ func newSrvConfig(objAPI ObjectLayer) error {
 	srvCfg := newServerConfig()
 
 	// Override any values from ENVs.
-	lookupConfigs(srvCfg)
+	lookupConfigOnce.Do(func() {
+		lookupConfigs(srvCfg)
+	})
 
 	// hold the mutex lock before a new config is assigned.
 	globalServerConfigMu.Lock()
@@ -586,7 +605,9 @@ func loadConfig(objAPI ObjectLayer) error {
 	}
 
 	// Override any values from ENVs.
-	lookupConfigs(srvCfg)
+	lookupConfigOnce.Do(func() {
+		lookupConfigs(srvCfg)
+	})
 
 	// hold the mutex lock before a new config is assigned.
 	globalServerConfigMu.Lock()

cmd/disk-cache-backend.go

@@ -125,12 +125,14 @@ func (m *cacheMeta) ToObjectInfo(bucket, object string) (o ObjectInfo) {
 
 // represents disk cache struct
 type diskCache struct {
-	gcCounter uint64 // ref: https://golang.org/pkg/sync/atomic/#pkg-note-BUG
 	// is set to 0 if drive is offline
-	online uint32
+	online       uint32 // ref: https://golang.org/pkg/sync/atomic/#pkg-note-BUG
+	purgeRunning int32
 
-	dir           string // caching directory
+	triggerGC     chan struct{}
-	quotaPct      int    // max usage in %
+	dir           string         // caching directory
+	stats         CacheDiskStats // disk cache stats for prometheus
+	quotaPct      int            // max usage in %
 	pool          sync.Pool
 	after         int // minimum accesses before an object is cached.
 	lowWatermark  int
@@ -142,12 +144,14 @@ type diskCache struct {
 }
 
 // Inits the disk cache dir if it is not initialized already.
-func newDiskCache(dir string, quotaPct, after, lowWatermark, highWatermark int) (*diskCache, error) {
+func newDiskCache(ctx context.Context, dir string, quotaPct, after, lowWatermark, highWatermark int) (*diskCache, error) {
 	if err := os.MkdirAll(dir, 0777); err != nil {
 		return nil, fmt.Errorf("Unable to initialize '%s' dir, %w", dir, err)
 	}
 	cache := diskCache{
 		dir:           dir,
+		triggerGC:     make(chan struct{}),
+		stats:         CacheDiskStats{Dir: dir},
 		quotaPct:      quotaPct,
 		after:         after,
 		lowWatermark:  lowWatermark,
@@ -161,6 +165,8 @@ func newDiskCache(dir string, quotaPct, after, lowWatermark, highWatermark int)
 		},
 		nsMutex: newNSLock(false),
 	}
+	go cache.purgeWait(ctx)
+	cache.diskUsageHigh() // update if cache usage is already high.
 	cache.NewNSLockFn = func(ctx context.Context, cachePath string) RWLocker {
 		return cache.nsMutex.NewNSLock(ctx, nil, cachePath, "")
 	}
@@ -181,7 +187,12 @@ func (c *diskCache) diskUsageLow() bool {
 		return false
 	}
 	usedPercent := (di.Total - di.Free) * 100 / di.Total
-	return int(usedPercent) < gcStopPct
+	low := int(usedPercent) < gcStopPct
+	atomic.StoreUint64(&c.stats.UsagePercent, usedPercent)
+	if low {
+		atomic.StoreInt32(&c.stats.UsageState, 0)
+	}
+	return low
 }
 
 // Returns if the disk usage reaches high water mark w.r.t the configured cache quota.
@@ -196,7 +207,12 @@ func (c *diskCache) diskUsageHigh() bool {
 		return false
 	}
 	usedPercent := (di.Total - di.Free) * 100 / di.Total
-	return int(usedPercent) >= gcTriggerPct
+	high := int(usedPercent) >= gcTriggerPct
+	atomic.StoreUint64(&c.stats.UsagePercent, usedPercent)
+	if high {
+		atomic.StoreInt32(&c.stats.UsageState, 1)
+	}
+	return high
 }
 
 // Returns if size space can be allocated without exceeding
@@ -230,24 +246,36 @@ var (
 	errDoneForNow = errors.New("done for now")
 )
 
+func (c *diskCache) purgeWait(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+		case <-c.triggerGC: // wait here until someone triggers.
+			c.purge(ctx)
+		}
+	}
+}
+
 // Purge cache entries that were not accessed.
 func (c *diskCache) purge(ctx context.Context) {
-	if c.diskUsageLow() {
+	if atomic.LoadInt32(&c.purgeRunning) == 1 || c.diskUsageLow() {
 		return
 	}
+
 	toFree := c.toClear()
 	if toFree == 0 {
 		return
 	}
+
+	atomic.StoreInt32(&c.purgeRunning, 1) // do not run concurrent purge()
+	defer atomic.StoreInt32(&c.purgeRunning, 0)
+
 	// expiry for cleaning up old cache.json files that
 	// need to be cleaned up.
 	expiry := UTCNow().Add(-cacheExpiryDays)
 	// defaulting max hits count to 100
-	scorer, err := newFileScorer(toFree, time.Now().Unix(), 100)
+	// ignore error we know what value we are passing.
-	if err != nil {
+	scorer, _ := newFileScorer(toFree, time.Now().Unix(), 100)
-		logger.LogIf(ctx, err)
-		return
-	}
 
 	// this function returns FileInfo for cached range files and cache data file.
 	fiStatFn := func(ranges map[string]string, dataFile, pathPrefix string) map[string]os.FileInfo {
@@ -326,27 +354,20 @@ func (c *diskCache) purge(ctx context.Context) {
 		return
 	}
 
-	for _, path := range scorer.fileNames() {
+	scorer.purgeFunc(func(qfile queuedFile) {
-		removeAll(path)
+		fileName := qfile.name
-		slashIdx := strings.LastIndex(path, SlashSeparator)
+		removeAll(fileName)
-		pathPrefix := path[0:slashIdx]
+		slashIdx := strings.LastIndex(fileName, SlashSeparator)
-		fname := path[slashIdx+1:]
+		if slashIdx >= 0 {
-		if fname == cacheDataFile {
+			fileNamePrefix := fileName[0:slashIdx]
-			removeAll(pathPrefix)
+			fname := fileName[slashIdx+1:]
+			if fname == cacheDataFile {
+				removeAll(fileNamePrefix)
+			}
 		}
-	}
+	})
-}
 
-func (c *diskCache) incGCCounter() {
+	scorer.reset()
-	atomic.AddUint64(&c.gcCounter, 1)
-}
-
-func (c *diskCache) resetGCCounter() {
-	atomic.StoreUint64(&c.gcCounter, 0)
-}
-
-func (c *diskCache) gcCount() uint64 {
-	return atomic.LoadUint64(&c.gcCounter)
 }
 
 // sets cache drive status
@@ -630,7 +651,7 @@ func newCacheEncryptMetadata(bucket, object string, metadata map[string]string)
 // Caches the object to disk
 func (c *diskCache) Put(ctx context.Context, bucket, object string, data io.Reader, size int64, rs *HTTPRangeSpec, opts ObjectOptions, incHitsOnly bool) error {
 	if c.diskUsageHigh() {
-		c.incGCCounter()
+		c.triggerGC <- struct{}{}
 		io.Copy(ioutil.Discard, data)
 		return errDiskFull
 	}

cmd/disk-cache-stats.go

@@ -1,5 +1,5 @@
 /*
- * MinIO Cloud Storage, (C) 2019 MinIO, Inc.
+ * MinIO Cloud Storage, (C) 2019, 2020 MinIO, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,14 +16,27 @@
 
 package cmd
 
-import "sync/atomic"
+import (
+	"sync/atomic"
+)
+
+// CacheDiskStats represents cache disk statistics
+// such as current disk usage and available.
+type CacheDiskStats struct {
+	// indicates if usage is high or low, if high value is '1', if low its '0'
+	UsageState int32
+	// indicates the current usage percentage of this cache disk
+	UsagePercent uint64
+	Dir          string
+}
 
 // CacheStats - represents bytes served from cache,
 // cache hits and cache misses.
 type CacheStats struct {
-	BytesServed uint64
+	BytesServed  uint64
-	Hits        uint64
+	Hits         uint64
-	Misses      uint64
+	Misses       uint64
+	GetDiskStats func() []CacheDiskStats
 }
 
 // Increase total bytes served from cache

cmd/disk-cache-utils.go

@@ -404,6 +404,14 @@ func (f *fileScorer) trimQueue() {
 	}
 }
 
+func (f *fileScorer) purgeFunc(p func(qfile queuedFile)) {
+	e := f.queue.Front()
+	for e != nil {
+		p(e.Value.(queuedFile))
+		e = e.Next()
+	}
+}
+
 // fileNames returns all queued file names.
 func (f *fileScorer) fileNames() []string {
 	res := make([]string, 0, f.queue.Len())

cmd/disk-cache.go

@@ -24,6 +24,7 @@ import (
 	"net/http"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/minio/minio/cmd/config/cache"
@@ -264,10 +265,11 @@ func (c *cacheObjects) GetObjectNInfo(ctx context.Context, bucket, object string
 
 	// Reaching here implies cache miss
 	c.cacheStats.incMiss()
+
 	// Since we got here, we are serving the request from backend,
 	// and also adding the object to the cache.
 	if dcache.diskUsageHigh() {
-		dcache.incGCCounter()
+		dcache.triggerGC <- struct{}{} // this is non-blocking
 	}
 
 	bkReader, bkErr := c.GetObjectNInfoFn(ctx, bucket, object, rs, h, lockType, opts)
@@ -526,7 +528,7 @@ func newCache(config cache.Config) ([]*diskCache, bool, error) {
 		if quota == 0 {
 			quota = config.Quota
 		}
-		cache, err := newDiskCache(dir, quota, config.After, config.WatermarkLow, config.WatermarkHigh)
+		cache, err := newDiskCache(ctx, dir, quota, config.After, config.WatermarkLow, config.WatermarkHigh)
 		if err != nil {
 			return nil, false, err
 		}
@@ -666,7 +668,17 @@ func newServerCacheObjects(ctx context.Context, config cache.Config) (CacheObjec
 			return newObjectLayerFn().CopyObject(ctx, srcBucket, srcObject, destBucket, destObject, srcInfo, srcOpts, dstOpts)
 		},
 	}
-
+	c.cacheStats.GetDiskStats = func() []CacheDiskStats {
+		cacheDiskStats := make([]CacheDiskStats, len(c.cache))
+		for i := range c.cache {
+			cacheDiskStats[i] = CacheDiskStats{
+				Dir: c.cache[i].stats.Dir,
+			}
+			atomic.StoreInt32(&cacheDiskStats[i].UsageState, atomic.LoadInt32(&c.cache[i].stats.UsageState))
+			atomic.StoreUint64(&cacheDiskStats[i].UsagePercent, atomic.LoadUint64(&c.cache[i].stats.UsagePercent))
+		}
+		return cacheDiskStats
+	}
 	if migrateSw {
 		go c.migrateCacheFromV1toV2(ctx)
 	}
@@ -686,19 +698,9 @@ func (c *cacheObjects) gc(ctx context.Context) {
 			if c.migrating {
 				continue
 			}
-			var wg sync.WaitGroup
 			for _, dcache := range c.cache {
-				if dcache.gcCount() == 0 {
+				dcache.triggerGC <- struct{}{}
-					continue
-				}
-				wg.Add(1)
-				go func(d *diskCache) {
-					defer wg.Done()
-					d.resetGCCounter()
-					d.purge(ctx)
-				}(dcache)
 			}
-			wg.Wait()
 		}
 	}
 }

cmd/gateway-main.go

@@ -147,7 +147,7 @@ func StartGateway(ctx *cli.Context, gw Gateway) {
 	gatewayHandleEnvVars()
 
 	// Set system resources to maximum.
-	logger.LogIf(GlobalContext, setMaxResources())
+	setMaxResources()
 
 	// Set when gateway is enabled
 	globalIsGateway = true
@@ -162,7 +162,9 @@ func StartGateway(ctx *cli.Context, gw Gateway) {
 	srvCfg := newServerConfig()
 
 	// Override any values from ENVs.
-	lookupConfigs(srvCfg)
+	lookupConfigOnce.Do(func() {
+		lookupConfigs(srvCfg)
+	})
 
 	// hold the mutex lock before a new config is assigned.
 	globalServerConfigMu.Lock()

cmd/metrics.go

@@ -260,6 +260,27 @@ func cacheMetricsPrometheus(ch chan<- prometheus.Metric) {
 		prometheus.CounterValue,
 		float64(cacheObjLayer.CacheStats().getBytesServed()),
 	)
+	for _, cdStats := range cacheObjLayer.CacheStats().GetDiskStats() {
+		// Cache disk usage percentage
+		ch <- prometheus.MustNewConstMetric(
+			prometheus.NewDesc(
+				prometheus.BuildFQName("cache", "usage", "percent"),
+				"Total percentage cache usage",
+				[]string{"disk"}, nil),
+			prometheus.GaugeValue,
+			float64(cdStats.UsagePercent),
+			cdStats.Dir,
+		)
+		ch <- prometheus.MustNewConstMetric(
+			prometheus.NewDesc(
+				prometheus.BuildFQName("cache", "usage", "high"),
+				"Indicates cache usage is high or low, relative to current cache 'quota' settings",
+				[]string{"disk"}, nil),
+			prometheus.GaugeValue,
+			float64(cdStats.UsageState),
+			cdStats.Dir,
+		)
+	}
 }
 
 // collects http metrics for MinIO server in Prometheus specific format

cmd/namespace-lock.go

@@ -79,7 +79,7 @@ func (n *nsLockMap) lock(ctx context.Context, volume string, path string, lockSo
 	nsLk, found := n.lockMap[resource]
 	if !found {
 		nsLk = &nsLock{
-			LRWMutex: lsync.NewLRWMutex(ctx),
+			LRWMutex: lsync.NewLRWMutex(),
 		}
 		// Add a count to indicate that a parallel unlock doesn't clear this entry.
 	}
@@ -89,9 +89,9 @@ func (n *nsLockMap) lock(ctx context.Context, volume string, path string, lockSo
 
 	// Locking here will block (until timeout).
 	if readLock {
-		locked = nsLk.GetRLock(opsID, lockSource, timeout)
+		locked = nsLk.GetRLock(ctx, opsID, lockSource, timeout)
 	} else {
-		locked = nsLk.GetLock(opsID, lockSource, timeout)
+		locked = nsLk.GetLock(ctx, opsID, lockSource, timeout)
 	}
 
 	if !locked { // We failed to get the lock
@@ -139,6 +139,7 @@ func (n *nsLockMap) unlock(volume string, path string, readLock bool) {
 type distLockInstance struct {
 	rwMutex *dsync.DRWMutex
 	opsID   string
+	ctx     context.Context
 }
 
 // Lock - block until write lock is taken or timeout has occurred.
@@ -146,7 +147,7 @@ func (di *distLockInstance) GetLock(timeout *dynamicTimeout) (timedOutErr error)
 	lockSource := getSource(2)
 	start := UTCNow()
 
-	if !di.rwMutex.GetLock(di.opsID, lockSource, timeout.Timeout()) {
+	if !di.rwMutex.GetLock(di.ctx, di.opsID, lockSource, timeout.Timeout()) {
 		timeout.LogFailure()
 		return OperationTimedOut{}
 	}
@@ -163,7 +164,7 @@ func (di *distLockInstance) Unlock() {
 func (di *distLockInstance) GetRLock(timeout *dynamicTimeout) (timedOutErr error) {
 	lockSource := getSource(2)
 	start := UTCNow()
-	if !di.rwMutex.GetRLock(di.opsID, lockSource, timeout.Timeout()) {
+	if !di.rwMutex.GetRLock(di.ctx, di.opsID, lockSource, timeout.Timeout()) {
 		timeout.LogFailure()
 		return OperationTimedOut{}
 	}
@@ -191,10 +192,10 @@ type localLockInstance struct {
 func (n *nsLockMap) NewNSLock(ctx context.Context, lockersFn func() []dsync.NetLocker, volume string, paths ...string) RWLocker {
 	opsID := mustGetUUID()
 	if n.isDistXL {
-		drwmutex := dsync.NewDRWMutex(ctx, &dsync.Dsync{
+		drwmutex := dsync.NewDRWMutex(&dsync.Dsync{
 			GetLockersFn: lockersFn,
 		}, pathsJoinPrefix(volume, paths...)...)
-		return &distLockInstance{drwmutex, opsID}
+		return &distLockInstance{drwmutex, opsID, ctx}
 	}
 	sort.Strings(paths)
 	return &localLockInstance{ctx, n, volume, paths, opsID}

cmd/peer-rest-server.go

@@ -770,7 +770,7 @@ func (s *peerRESTServer) ServerUpdateHandler(w http.ResponseWriter, r *http.Requ
 	var latestReleaseTime time.Time
 	var err error
 	if latestRelease := vars[peerRESTLatestRelease]; latestRelease != "" {
-		latestReleaseTime, err = time.Parse(latestRelease, time.RFC3339)
+		latestReleaseTime, err = time.Parse(time.RFC3339, latestRelease)
 		if err != nil {
 			s.writeErrorResponse(w, err)
 			return

cmd/xl-v1-object.go

@@ -1021,6 +1021,10 @@ func (xl xlObjects) PutObjectTags(ctx context.Context, bucket, object string, ta
 	}
 
 	for i, xlMeta := range metaArr {
+		if errs[i] != nil {
+			// Avoid disks where loading metadata fail
+			continue
+		}
 		// clean xlMeta.Meta of tag key, before updating the new tags
 		delete(xlMeta.Meta, xhttp.AmzObjectTagging)
 		// Don't update for empty tags

cmd/xl-v1.go

@@ -261,9 +261,11 @@ func (xl xlObjects) crawlAndGetDataUsage(ctx context.Context, buckets []BucketIn
 	for _, b := range buckets {
 		e := oldCache.find(b.Name)
 		if e != nil {
-			if bf == nil || bf.containsDir(b.Name) {
+			cache.replace(b.Name, dataUsageRoot, *e)
+			lc, err := globalLifecycleSys.Get(b.Name)
+			activeLC := err == nil && lc.HasActiveRules("", true)
+			if activeLC || bf == nil || bf.containsDir(b.Name) {
 				bucketCh <- b
-				cache.replace(b.Name, dataUsageRoot, *e)
 			} else {
 				if intDataUpdateTracker.debug {
 					logger.Info(color.Green("crawlAndGetDataUsage:")+" Skipping bucket %v, not updated", b.Name)

cmd/xl-zones.go

@@ -1608,7 +1608,7 @@ func (z *xlZones) getZoneAndSet(id string) (int, int, error) {
 			}
 		}
 	}
-	return 0, 0, errDiskNotFound
+	return 0, 0, fmt.Errorf("DiskID(%s) %w", id, errDiskNotFound)
 }
 
 // IsReady - Returns true all the erasure sets are writable.
@@ -1625,6 +1625,7 @@ func (z *xlZones) IsReady(ctx context.Context) bool {
 	for _, id := range diskIDs {
 		zoneIdx, setIdx, err := z.getZoneAndSet(id)
 		if err != nil {
+			logger.LogIf(ctx, err)
 			continue
 		}
 		erasureSetUpCount[zoneIdx][setIdx]++
@@ -1643,6 +1644,8 @@ func (z *xlZones) IsReady(ctx context.Context) bool {
 		}
 		for setIdx := range erasureSetUpCount[zoneIdx] {
 			if erasureSetUpCount[zoneIdx][setIdx] < writeQuorum {
+				logger.LogIf(ctx, fmt.Errorf("Write quorum lost on zone: %d, set: %d, expected write quorum: %d",
+					zoneIdx, setIdx, writeQuorum))
 				return false
 			}
 		}

docs/metrics/prometheus/README.md

@@ -110,7 +110,7 @@ These are the new set of metrics which will be in effect after `RELEASE.2019-10-
     - Metrics that records the http statistics and latencies are labeled to their respective APIs (putobject,getobject etc).
     - Disk usage metrics are distributed and labeled to the respective disk paths.
 
-For more details, please check the `Migration guide for the new set of metrics`. 
+For more details, please check the `Migration guide for the new set of metrics`.
 
 The list of metrics and its definition are as follows. (NOTE: instance here is one MinIO node)
 
@@ -178,14 +178,23 @@ All metrics are labeled by `bucket`, each metric is displayed per bucket. `bucke
 
 MinIO Gateway instances enabled with Disk-Caching expose caching related metrics.
 
-| name                 | description                             |
+#### Global cache metrics
-|:---------------------|:----------------------------------------|
+| name                 | description                                       |
-| `cache_data_served`  | Total number of bytes served from cache |
+|:---------------------|:--------------------------------------------------|
-| `cache_hits_total`   | Total number of cache hits              |
+| `cache_hits_total`   | Total number of cache hits                        |
-| `cache_misses_total` | Total number of cache misses            |
+| `cache_misses_total` | Total number of cache misses                      |
+| `cache_data_served`  | Total number of bytes served from cache           |
 
-### Gateway & Cache specific metrics
+#### Per disk cache metrics
+| `cache_usage_percent` | Total percentage cache usage                                                     |
+| `cache_usage_state`   | Indicates cache usage is high or low, relative to current cache 'quota' settings |
 
+`cache_usage_state` holds only two states
+
+- '1' indicates high disk usage
+- '0' indicates low disk usage
+
+### Gateway specific metrics
 MinIO Gateway instance exposes metrics related to Gateway communication with the cloud backend (S3, Azure & GCS Gateway).
 
 `<gateway_type>` changes based on the gateway in use can be 's3', 'gcs' or 'azure'. Other metrics are labeled with `method` that identifies HTTP GET, HEAD, PUT and POST requests to the backend.

docs/orchestration/docker-compose/docker-compose.yaml

@@ -5,7 +5,7 @@ version: '3.7'
 # 9001 through 9004.
 services:
   minio1:
-    image: minio/minio:RELEASE.2020-06-12T00-06-19Z
+    image: minio/minio:RELEASE.2020-06-14T18-32-17Z
     volumes:
       - data1-1:/data1
       - data1-2:/data2
@@ -22,7 +22,7 @@ services:
       retries: 3
 
   minio2:
-    image: minio/minio:RELEASE.2020-06-12T00-06-19Z
+    image: minio/minio:RELEASE.2020-06-14T18-32-17Z
     volumes:
       - data2-1:/data1
       - data2-2:/data2
@@ -39,7 +39,7 @@ services:
       retries: 3
 
   minio3:
-    image: minio/minio:RELEASE.2020-06-12T00-06-19Z
+    image: minio/minio:RELEASE.2020-06-14T18-32-17Z
     volumes:
       - data3-1:/data1
       - data3-2:/data2
@@ -56,7 +56,7 @@ services:
       retries: 3
 
   minio4:
-    image: minio/minio:RELEASE.2020-06-12T00-06-19Z
+    image: minio/minio:RELEASE.2020-06-14T18-32-17Z
     volumes:
       - data4-1:/data1
       - data4-2:/data2

docs/orchestration/docker-swarm/docker-compose-secrets.yaml

@@ -2,7 +2,7 @@ version: '3.7'
 
 services:
   minio1:
-    image: minio/minio:RELEASE.2020-06-12T00-06-19Z
+    image: minio/minio:RELEASE.2020-06-14T18-32-17Z
     hostname: minio1
     volumes:
       - minio1-data:/export
@@ -29,7 +29,7 @@ services:
       retries: 3
 
   minio2:
-    image: minio/minio:RELEASE.2020-06-12T00-06-19Z
+    image: minio/minio:RELEASE.2020-06-14T18-32-17Z
     hostname: minio2
     volumes:
       - minio2-data:/export
@@ -56,7 +56,7 @@ services:
       retries: 3
 
   minio3:
-    image: minio/minio:RELEASE.2020-06-12T00-06-19Z
+    image: minio/minio:RELEASE.2020-06-14T18-32-17Z
     hostname: minio3
     volumes:
       - minio3-data:/export
@@ -83,7 +83,7 @@ services:
       retries: 3
 
   minio4:
-    image: minio/minio:RELEASE.2020-06-12T00-06-19Z
+    image: minio/minio:RELEASE.2020-06-14T18-32-17Z
     hostname: minio4
     volumes:
       - minio4-data:/export

docs/orchestration/docker-swarm/docker-compose.yaml

@@ -2,7 +2,7 @@ version: '3.7'
 
 services:
   minio1:
-    image: minio/minio:RELEASE.2020-06-12T00-06-19Z
+    image: minio/minio:RELEASE.2020-06-14T18-32-17Z
     hostname: minio1
     volumes:
       - minio1-data:/export
@@ -33,7 +33,7 @@ services:
       retries: 3
 
   minio2:
-    image: minio/minio:RELEASE.2020-06-12T00-06-19Z
+    image: minio/minio:RELEASE.2020-06-14T18-32-17Z
     hostname: minio2
     volumes:
       - minio2-data:/export
@@ -64,7 +64,7 @@ services:
       retries: 3
 
   minio3:
-    image: minio/minio:RELEASE.2020-06-12T00-06-19Z
+    image: minio/minio:RELEASE.2020-06-14T18-32-17Z
     hostname: minio3
     volumes:
       - minio3-data:/export
@@ -95,7 +95,7 @@ services:
       retries: 3
 
   minio4:
-    image: minio/minio:RELEASE.2020-06-12T00-06-19Z
+    image: minio/minio:RELEASE.2020-06-14T18-32-17Z
     hostname: minio4
     volumes:
       - minio4-data:/export

pkg/dsync/drwmutex.go

@@ -54,7 +54,6 @@ type DRWMutex struct {
 	readersLocks [][]string // Array of array of nodes that granted reader locks
 	m            sync.Mutex // Mutex to prevent multiple simultaneous locks from this node
 	clnt         *Dsync
-	ctx          context.Context
 }
 
 // Granted - represents a structure of a granted lock.
@@ -72,12 +71,11 @@ func isLocked(uid string) bool {
 }
 
 // NewDRWMutex - initializes a new dsync RW mutex.
-func NewDRWMutex(ctx context.Context, clnt *Dsync, names ...string) *DRWMutex {
+func NewDRWMutex(clnt *Dsync, names ...string) *DRWMutex {
 	return &DRWMutex{
 		writeLocks: make([]string, len(clnt.GetLockersFn())),
 		Names:      names,
 		clnt:       clnt,
-		ctx:        ctx,
 	}
 }
 
@@ -88,7 +86,7 @@ func NewDRWMutex(ctx context.Context, clnt *Dsync, names ...string) *DRWMutex {
 func (dm *DRWMutex) Lock(id, source string) {
 
 	isReadLock := false
-	dm.lockBlocking(drwMutexInfinite, id, source, isReadLock)
+	dm.lockBlocking(context.Background(), drwMutexInfinite, id, source, isReadLock)
 }
 
 // GetLock tries to get a write lock on dm before the timeout elapses.
@@ -96,10 +94,10 @@ func (dm *DRWMutex) Lock(id, source string) {
 // If the lock is already in use, the calling go routine
 // blocks until either the mutex becomes available and return success or
 // more time has passed than the timeout value and return false.
-func (dm *DRWMutex) GetLock(id, source string, timeout time.Duration) (locked bool) {
+func (dm *DRWMutex) GetLock(ctx context.Context, id, source string, timeout time.Duration) (locked bool) {
 
 	isReadLock := false
-	return dm.lockBlocking(timeout, id, source, isReadLock)
+	return dm.lockBlocking(ctx, timeout, id, source, isReadLock)
 }
 
 // RLock holds a read lock on dm.
@@ -109,7 +107,7 @@ func (dm *DRWMutex) GetLock(id, source string, timeout time.Duration) (locked bo
 func (dm *DRWMutex) RLock(id, source string) {
 
 	isReadLock := true
-	dm.lockBlocking(drwMutexInfinite, id, source, isReadLock)
+	dm.lockBlocking(context.Background(), drwMutexInfinite, id, source, isReadLock)
 }
 
 // GetRLock tries to get a read lock on dm before the timeout elapses.
@@ -118,10 +116,10 @@ func (dm *DRWMutex) RLock(id, source string) {
 // Otherwise the calling go routine blocks until either the mutex becomes
 // available and return success or more time has passed than the timeout
 // value and return false.
-func (dm *DRWMutex) GetRLock(id, source string, timeout time.Duration) (locked bool) {
+func (dm *DRWMutex) GetRLock(ctx context.Context, id, source string, timeout time.Duration) (locked bool) {
 
 	isReadLock := true
-	return dm.lockBlocking(timeout, id, source, isReadLock)
+	return dm.lockBlocking(ctx, timeout, id, source, isReadLock)
 }
 
 // lockBlocking will try to acquire either a read or a write lock
@@ -129,10 +127,10 @@ func (dm *DRWMutex) GetRLock(id, source string, timeout time.Duration) (locked b
 // The function will loop using a built-in timing randomized back-off
 // algorithm until either the lock is acquired successfully or more
 // time has elapsed than the timeout value.
-func (dm *DRWMutex) lockBlocking(timeout time.Duration, id, source string, isReadLock bool) (locked bool) {
+func (dm *DRWMutex) lockBlocking(ctx context.Context, timeout time.Duration, id, source string, isReadLock bool) (locked bool) {
 	restClnts := dm.clnt.GetLockersFn()
 
-	retryCtx, cancel := context.WithTimeout(dm.ctx, timeout)
+	retryCtx, cancel := context.WithTimeout(ctx, timeout)
 
 	defer cancel()
 

pkg/dsync/drwmutex_test.go

@@ -14,8 +14,6 @@
  * limitations under the License.
  */
 
-// GOMAXPROCS=10 go test
-
 package dsync_test
 
 import (
@@ -36,14 +34,14 @@ const (
 
 func testSimpleWriteLock(t *testing.T, duration time.Duration) (locked bool) {
 
-	drwm := NewDRWMutex(context.Background(), ds, "simplelock")
+	drwm := NewDRWMutex(ds, "simplelock")
 
-	if !drwm.GetRLock(id, source, time.Second) {
+	if !drwm.GetRLock(context.Background(), id, source, time.Second) {
 		panic("Failed to acquire read lock")
 	}
 	// fmt.Println("1st read lock acquired, waiting...")
 
-	if !drwm.GetRLock(id, source, time.Second) {
+	if !drwm.GetRLock(context.Background(), id, source, time.Second) {
 		panic("Failed to acquire read lock")
 	}
 	// fmt.Println("2nd read lock acquired, waiting...")
@@ -61,7 +59,7 @@ func testSimpleWriteLock(t *testing.T, duration time.Duration) (locked bool) {
 	}()
 
 	// fmt.Println("Trying to acquire write lock, waiting...")
-	locked = drwm.GetLock(id, source, duration)
+	locked = drwm.GetLock(context.Background(), id, source, duration)
 	if locked {
 		// fmt.Println("Write lock acquired, waiting...")
 		time.Sleep(time.Second)
@@ -92,10 +90,10 @@ func TestSimpleWriteLockTimedOut(t *testing.T) {
 
 func testDualWriteLock(t *testing.T, duration time.Duration) (locked bool) {
 
-	drwm := NewDRWMutex(context.Background(), ds, "duallock")
+	drwm := NewDRWMutex(ds, "duallock")
 
 	// fmt.Println("Getting initial write lock")
-	if !drwm.GetLock(id, source, time.Second) {
+	if !drwm.GetLock(context.Background(), id, source, time.Second) {
 		panic("Failed to acquire initial write lock")
 	}
 
@@ -106,7 +104,7 @@ func testDualWriteLock(t *testing.T, duration time.Duration) (locked bool) {
 	}()
 
 	// fmt.Println("Trying to acquire 2nd write lock, waiting...")
-	locked = drwm.GetLock(id, source, duration)
+	locked = drwm.GetLock(context.Background(), id, source, duration)
 	if locked {
 		// fmt.Println("2nd write lock acquired, waiting...")
 		time.Sleep(time.Second)
@@ -140,8 +138,8 @@ func TestDualWriteLockTimedOut(t *testing.T) {
 // Test cases below are copied 1 to 1 from sync/rwmutex_test.go (adapted to use DRWMutex)
 
 // Borrowed from rwmutex_test.go
-func parallelReader(m *DRWMutex, clocked, cunlock, cdone chan bool) {
+func parallelReader(ctx context.Context, m *DRWMutex, clocked, cunlock, cdone chan bool) {
-	if m.GetRLock(id, source, time.Second) {
+	if m.GetRLock(ctx, id, source, time.Second) {
 		clocked <- true
 		<-cunlock
 		m.RUnlock()
@@ -152,13 +150,13 @@ func parallelReader(m *DRWMutex, clocked, cunlock, cdone chan bool) {
 // Borrowed from rwmutex_test.go
 func doTestParallelReaders(numReaders, gomaxprocs int) {
 	runtime.GOMAXPROCS(gomaxprocs)
-	m := NewDRWMutex(context.Background(), ds, "test-parallel")
+	m := NewDRWMutex(ds, "test-parallel")
 
 	clocked := make(chan bool)
 	cunlock := make(chan bool)
 	cdone := make(chan bool)
 	for i := 0; i < numReaders; i++ {
-		go parallelReader(m, clocked, cunlock, cdone)
+		go parallelReader(context.Background(), m, clocked, cunlock, cdone)
 	}
 	// Wait for all parallel RLock()s to succeed.
 	for i := 0; i < numReaders; i++ {
@@ -184,7 +182,7 @@ func TestParallelReaders(t *testing.T) {
 // Borrowed from rwmutex_test.go
 func reader(rwm *DRWMutex, numIterations int, activity *int32, cdone chan bool) {
 	for i := 0; i < numIterations; i++ {
-		if rwm.GetRLock(id, source, time.Second) {
+		if rwm.GetRLock(context.Background(), id, source, time.Second) {
 			n := atomic.AddInt32(activity, 1)
 			if n < 1 || n >= 10000 {
 				panic(fmt.Sprintf("wlock(%d)\n", n))
@@ -201,7 +199,7 @@ func reader(rwm *DRWMutex, numIterations int, activity *int32, cdone chan bool)
 // Borrowed from rwmutex_test.go
 func writer(rwm *DRWMutex, numIterations int, activity *int32, cdone chan bool) {
 	for i := 0; i < numIterations; i++ {
-		if rwm.GetLock(id, source, time.Second) {
+		if rwm.GetLock(context.Background(), id, source, time.Second) {
 			n := atomic.AddInt32(activity, 10000)
 			if n != 10000 {
 				panic(fmt.Sprintf("wlock(%d)\n", n))
@@ -220,7 +218,7 @@ func HammerRWMutex(gomaxprocs, numReaders, numIterations int) {
 	runtime.GOMAXPROCS(gomaxprocs)
 	// Number of active readers + 10000 * number of active writers.
 	var activity int32
-	rwm := NewDRWMutex(context.Background(), ds, "test")
+	rwm := NewDRWMutex(ds, "test")
 	cdone := make(chan bool)
 	go writer(rwm, numIterations, &activity, cdone)
 	var i int
@@ -263,7 +261,7 @@ func TestUnlockPanic(t *testing.T) {
 			t.Fatalf("unlock of unlocked RWMutex did not panic")
 		}
 	}()
-	mu := NewDRWMutex(context.Background(), ds, "test")
+	mu := NewDRWMutex(ds, "test")
 	mu.Unlock()
 }
 
@@ -274,7 +272,7 @@ func TestUnlockPanic2(t *testing.T) {
 			t.Fatalf("unlock of unlocked RWMutex did not panic")
 		}
 	}()
-	mu := NewDRWMutex(context.Background(), ds, "test-unlock-panic-2")
+	mu := NewDRWMutex(ds, "test-unlock-panic-2")
 	mu.RLock(id, source)
 	mu.Unlock()
 }
@@ -286,7 +284,7 @@ func TestRUnlockPanic(t *testing.T) {
 			t.Fatalf("read unlock of unlocked RWMutex did not panic")
 		}
 	}()
-	mu := NewDRWMutex(context.Background(), ds, "test")
+	mu := NewDRWMutex(ds, "test")
 	mu.RUnlock()
 }
 
@@ -297,14 +295,14 @@ func TestRUnlockPanic2(t *testing.T) {
 			t.Fatalf("read unlock of unlocked RWMutex did not panic")
 		}
 	}()
-	mu := NewDRWMutex(context.Background(), ds, "test-runlock-panic-2")
+	mu := NewDRWMutex(ds, "test-runlock-panic-2")
 	mu.Lock(id, source)
 	mu.RUnlock()
 }
 
 // Borrowed from rwmutex_test.go
 func benchmarkRWMutex(b *testing.B, localWork, writeRatio int) {
-	rwm := NewDRWMutex(context.Background(), ds, "test")
+	rwm := NewDRWMutex(ds, "test")
 	b.RunParallel(func(pb *testing.PB) {
 		foo := 0
 		for pb.Next() {

pkg/dsync/dsync_test.go

@@ -19,7 +19,6 @@
 package dsync_test
 
 import (
-	"context"
 	"fmt"
 	"log"
 	"math/rand"
@@ -89,7 +88,7 @@ func TestMain(m *testing.M) {
 
 func TestSimpleLock(t *testing.T) {
 
-	dm := NewDRWMutex(context.Background(), ds, "test")
+	dm := NewDRWMutex(ds, "test")
 
 	dm.Lock(id, source)
 
@@ -101,7 +100,7 @@ func TestSimpleLock(t *testing.T) {
 
 func TestSimpleLockUnlockMultipleTimes(t *testing.T) {
 
-	dm := NewDRWMutex(context.Background(), ds, "test")
+	dm := NewDRWMutex(ds, "test")
 
 	dm.Lock(id, source)
 	time.Sleep(time.Duration(10+(rand.Float32()*50)) * time.Millisecond)
@@ -127,8 +126,8 @@ func TestSimpleLockUnlockMultipleTimes(t *testing.T) {
 // Test two locks for same resource, one succeeds, one fails (after timeout)
 func TestTwoSimultaneousLocksForSameResource(t *testing.T) {
 
-	dm1st := NewDRWMutex(context.Background(), ds, "aap")
+	dm1st := NewDRWMutex(ds, "aap")
-	dm2nd := NewDRWMutex(context.Background(), ds, "aap")
+	dm2nd := NewDRWMutex(ds, "aap")
 
 	dm1st.Lock(id, source)
 
@@ -151,9 +150,9 @@ func TestTwoSimultaneousLocksForSameResource(t *testing.T) {
 // Test three locks for same resource, one succeeds, one fails (after timeout)
 func TestThreeSimultaneousLocksForSameResource(t *testing.T) {
 
-	dm1st := NewDRWMutex(context.Background(), ds, "aap")
+	dm1st := NewDRWMutex(ds, "aap")
-	dm2nd := NewDRWMutex(context.Background(), ds, "aap")
+	dm2nd := NewDRWMutex(ds, "aap")
-	dm3rd := NewDRWMutex(context.Background(), ds, "aap")
+	dm3rd := NewDRWMutex(ds, "aap")
 
 	dm1st.Lock(id, source)
 
@@ -216,8 +215,8 @@ func TestThreeSimultaneousLocksForSameResource(t *testing.T) {
 // Test two locks for different resources, both succeed
 func TestTwoSimultaneousLocksForDifferentResources(t *testing.T) {
 
-	dm1 := NewDRWMutex(context.Background(), ds, "aap")
+	dm1 := NewDRWMutex(ds, "aap")
-	dm2 := NewDRWMutex(context.Background(), ds, "noot")
+	dm2 := NewDRWMutex(ds, "noot")
 
 	dm1.Lock(id, source)
 	dm2.Lock(id, source)
@@ -247,7 +246,7 @@ func TestMutex(t *testing.T) {
 		loops = 5
 	}
 	c := make(chan bool)
-	m := NewDRWMutex(context.Background(), ds, "test")
+	m := NewDRWMutex(ds, "test")
 	for i := 0; i < 10; i++ {
 		go HammerMutex(m, loops, c)
 	}
@@ -261,7 +260,7 @@ func BenchmarkMutexUncontended(b *testing.B) {
 		*DRWMutex
 	}
 	b.RunParallel(func(pb *testing.PB) {
-		var mu = PaddedMutex{NewDRWMutex(context.Background(), ds, "")}
+		var mu = PaddedMutex{NewDRWMutex(ds, "")}
 		for pb.Next() {
 			mu.Lock(id, source)
 			mu.Unlock()
@@ -270,7 +269,7 @@ func BenchmarkMutexUncontended(b *testing.B) {
 }
 
 func benchmarkMutex(b *testing.B, slack, work bool) {
-	mu := NewDRWMutex(context.Background(), ds, "")
+	mu := NewDRWMutex(ds, "")
 	if slack {
 		b.SetParallelism(10)
 	}
@@ -313,7 +312,7 @@ func BenchmarkMutexNoSpin(b *testing.B) {
 	// These goroutines yield during local work, so that switching from
 	// a blocked goroutine to other goroutines is profitable.
 	// As a matter of fact, this benchmark still triggers some spinning in the mutex.
-	m := NewDRWMutex(context.Background(), ds, "")
+	m := NewDRWMutex(ds, "")
 	var acc0, acc1 uint64
 	b.SetParallelism(4)
 	b.RunParallel(func(pb *testing.PB) {
@@ -345,7 +344,7 @@ func BenchmarkMutexSpin(b *testing.B) {
 	// profitable. To achieve this we create a goroutine per-proc.
 	// These goroutines access considerable amount of local data so that
 	// unnecessary rescheduling is penalized by cache misses.
-	m := NewDRWMutex(context.Background(), ds, "")
+	m := NewDRWMutex(ds, "")
 	var acc0, acc1 uint64
 	b.RunParallel(func(pb *testing.PB) {
 		var data [16 << 10]uint64

pkg/lsync/lrwmutex.go

@@ -32,12 +32,11 @@ type LRWMutex struct {
 	isWriteLock bool
 	ref         int
 	m           sync.Mutex // Mutex to prevent multiple simultaneous locks
-	ctx         context.Context
 }
 
 // NewLRWMutex - initializes a new lsync RW mutex.
-func NewLRWMutex(ctx context.Context) *LRWMutex {
+func NewLRWMutex() *LRWMutex {
-	return &LRWMutex{ctx: ctx}
+	return &LRWMutex{}
 }
 
 // Lock holds a write lock on lm.
@@ -47,14 +46,14 @@ func NewLRWMutex(ctx context.Context) *LRWMutex {
 func (lm *LRWMutex) Lock() {
 
 	const isWriteLock = true
-	lm.lockLoop(lm.id, lm.source, time.Duration(math.MaxInt64), isWriteLock)
+	lm.lockLoop(context.Background(), lm.id, lm.source, time.Duration(math.MaxInt64), isWriteLock)
 }
 
 // GetLock tries to get a write lock on lm before the timeout occurs.
-func (lm *LRWMutex) GetLock(id string, source string, timeout time.Duration) (locked bool) {
+func (lm *LRWMutex) GetLock(ctx context.Context, id string, source string, timeout time.Duration) (locked bool) {
 
 	const isWriteLock = true
-	return lm.lockLoop(id, source, timeout, isWriteLock)
+	return lm.lockLoop(ctx, id, source, timeout, isWriteLock)
 }
 
 // RLock holds a read lock on lm.
@@ -64,60 +63,55 @@ func (lm *LRWMutex) GetLock(id string, source string, timeout time.Duration) (lo
 func (lm *LRWMutex) RLock() {
 
 	const isWriteLock = false
-	lm.lockLoop(lm.id, lm.source, time.Duration(1<<63-1), isWriteLock)
+	lm.lockLoop(context.Background(), lm.id, lm.source, time.Duration(1<<63-1), isWriteLock)
 }
 
 // GetRLock tries to get a read lock on lm before the timeout occurs.
-func (lm *LRWMutex) GetRLock(id string, source string, timeout time.Duration) (locked bool) {
+func (lm *LRWMutex) GetRLock(ctx context.Context, id string, source string, timeout time.Duration) (locked bool) {
 
 	const isWriteLock = false
-	return lm.lockLoop(id, source, timeout, isWriteLock)
+	return lm.lockLoop(ctx, id, source, timeout, isWriteLock)
+}
+
+func (lm *LRWMutex) lock(id, source string, isWriteLock bool) (locked bool) {
+	lm.m.Lock()
+	lm.id = id
+	lm.source = source
+	if isWriteLock {
+		if lm.ref == 0 && !lm.isWriteLock {
+			lm.ref = 1
+			lm.isWriteLock = true
+			locked = true
+		}
+	} else {
+		if !lm.isWriteLock {
+			lm.ref++
+			locked = true
+		}
+	}
+	lm.m.Unlock()
+
+	return locked
 }
 
 // lockLoop will acquire either a read or a write lock
 //
 // The call will block until the lock is granted using a built-in
 // timing randomized back-off algorithm to try again until successful
-func (lm *LRWMutex) lockLoop(id, source string, timeout time.Duration, isWriteLock bool) bool {
+func (lm *LRWMutex) lockLoop(ctx context.Context, id, source string, timeout time.Duration, isWriteLock bool) (locked bool) {
-	retryCtx, cancel := context.WithTimeout(lm.ctx, timeout)
+	retryCtx, cancel := context.WithTimeout(ctx, timeout)
-
 	defer cancel()
 
 	// We timed out on the previous lock, incrementally wait
 	// for a longer back-off time and try again afterwards.
 	for range retry.NewTimer(retryCtx) {
-		// Try to acquire the lock.
+		if lm.lock(id, source, isWriteLock) {
-		var success bool
-		{
-			lm.m.Lock()
-
-			lm.id = id
-			lm.source = source
-
-			if isWriteLock {
-				if lm.ref == 0 && !lm.isWriteLock {
-					lm.ref = 1
-					lm.isWriteLock = true
-					success = true
-				}
-			} else {
-				if !lm.isWriteLock {
-					lm.ref++
-					success = true
-				}
-			}
-
-			lm.m.Unlock()
-		}
-
-		if success {
 			return true
 		}
 	}
 
 	// We timed out on the previous lock, incrementally wait
 	// for a longer back-off time and try again afterwards.
-
 	return false
 }
 

pkg/lsync/lrwmutex_test.go

@@ -33,14 +33,15 @@ import (
 
 func testSimpleWriteLock(t *testing.T, duration time.Duration) (locked bool) {
 
-	lrwm := NewLRWMutex(context.Background())
+	ctx := context.Background()
+	lrwm := NewLRWMutex()
 
-	if !lrwm.GetRLock("", "object1", time.Second) {
+	if !lrwm.GetRLock(ctx, "", "object1", time.Second) {
 		panic("Failed to acquire read lock")
 	}
 	// fmt.Println("1st read lock acquired, waiting...")
 
-	if !lrwm.GetRLock("", "object1", time.Second) {
+	if !lrwm.GetRLock(ctx, "", "object1", time.Second) {
 		panic("Failed to acquire read lock")
 	}
 	// fmt.Println("2nd read lock acquired, waiting...")
@@ -58,7 +59,7 @@ func testSimpleWriteLock(t *testing.T, duration time.Duration) (locked bool) {
 	}()
 
 	// fmt.Println("Trying to acquire write lock, waiting...")
-	locked = lrwm.GetLock("", "", duration)
+	locked = lrwm.GetLock(ctx, "", "", duration)
 	if locked {
 		// fmt.Println("Write lock acquired, waiting...")
 		time.Sleep(1 * time.Second)
@@ -90,10 +91,11 @@ func TestSimpleWriteLockTimedOut(t *testing.T) {
 
 func testDualWriteLock(t *testing.T, duration time.Duration) (locked bool) {
 
-	lrwm := NewLRWMutex(context.Background())
+	ctx := context.Background()
+	lrwm := NewLRWMutex()
 
 	// fmt.Println("Getting initial write lock")
-	if !lrwm.GetLock("", "", time.Second) {
+	if !lrwm.GetLock(ctx, "", "", time.Second) {
 		panic("Failed to acquire initial write lock")
 	}
 
@@ -104,7 +106,7 @@ func testDualWriteLock(t *testing.T, duration time.Duration) (locked bool) {
 	}()
 
 	// fmt.Println("Trying to acquire 2nd write lock, waiting...")
-	locked = lrwm.GetLock("", "", duration)
+	locked = lrwm.GetLock(ctx, "", "", duration)
 	if locked {
 		// fmt.Println("2nd write lock acquired, waiting...")
 		time.Sleep(time.Second)
@@ -139,8 +141,8 @@ func TestDualWriteLockTimedOut(t *testing.T) {
 // Test cases below are copied 1 to 1 from sync/rwmutex_test.go (adapted to use LRWMutex)
 
 // Borrowed from rwmutex_test.go
-func parallelReader(m *LRWMutex, clocked, cunlock, cdone chan bool) {
+func parallelReader(ctx context.Context, m *LRWMutex, clocked, cunlock, cdone chan bool) {
-	if m.GetRLock("", "", time.Second) {
+	if m.GetRLock(ctx, "", "", time.Second) {
 		clocked <- true
 		<-cunlock
 		m.RUnlock()
@@ -151,13 +153,13 @@ func parallelReader(m *LRWMutex, clocked, cunlock, cdone chan bool) {
 // Borrowed from rwmutex_test.go
 func doTestParallelReaders(numReaders, gomaxprocs int) {
 	runtime.GOMAXPROCS(gomaxprocs)
-	m := NewLRWMutex(context.Background())
+	m := NewLRWMutex()
 
 	clocked := make(chan bool)
 	cunlock := make(chan bool)
 	cdone := make(chan bool)
 	for i := 0; i < numReaders; i++ {
-		go parallelReader(m, clocked, cunlock, cdone)
+		go parallelReader(context.Background(), m, clocked, cunlock, cdone)
 	}
 	// Wait for all parallel RLock()s to succeed.
 	for i := 0; i < numReaders; i++ {
@@ -183,7 +185,7 @@ func TestParallelReaders(t *testing.T) {
 // Borrowed from rwmutex_test.go
 func reader(rwm *LRWMutex, numIterations int, activity *int32, cdone chan bool) {
 	for i := 0; i < numIterations; i++ {
-		if rwm.GetRLock("", "", time.Second) {
+		if rwm.GetRLock(context.Background(), "", "", time.Second) {
 			n := atomic.AddInt32(activity, 1)
 			if n < 1 || n >= 10000 {
 				panic(fmt.Sprintf("wlock(%d)\n", n))
@@ -200,7 +202,7 @@ func reader(rwm *LRWMutex, numIterations int, activity *int32, cdone chan bool)
 // Borrowed from rwmutex_test.go
 func writer(rwm *LRWMutex, numIterations int, activity *int32, cdone chan bool) {
 	for i := 0; i < numIterations; i++ {
-		if rwm.GetLock("", "", time.Second) {
+		if rwm.GetLock(context.Background(), "", "", time.Second) {
 			n := atomic.AddInt32(activity, 10000)
 			if n != 10000 {
 				panic(fmt.Sprintf("wlock(%d)\n", n))
@@ -219,7 +221,7 @@ func HammerRWMutex(gomaxprocs, numReaders, numIterations int) {
 	runtime.GOMAXPROCS(gomaxprocs)
 	// Number of active readers + 10000 * number of active writers.
 	var activity int32
-	rwm := NewLRWMutex(context.Background())
+	rwm := NewLRWMutex()
 	cdone := make(chan bool)
 	go writer(rwm, numIterations, &activity, cdone)
 	var i int
@@ -257,7 +259,7 @@ func TestRWMutex(t *testing.T) {
 
 // Borrowed from rwmutex_test.go
 func TestDRLocker(t *testing.T) {
-	wl := NewLRWMutex(context.Background())
+	wl := NewLRWMutex()
 	var rl sync.Locker
 	wlocked := make(chan bool, 1)
 	rlocked := make(chan bool, 1)
@@ -298,7 +300,7 @@ func TestUnlockPanic(t *testing.T) {
 			t.Fatalf("unlock of unlocked RWMutex did not panic")
 		}
 	}()
-	mu := NewLRWMutex(context.Background())
+	mu := NewLRWMutex()
 	mu.Unlock()
 }
 
@@ -309,7 +311,7 @@ func TestUnlockPanic2(t *testing.T) {
 			t.Fatalf("unlock of unlocked RWMutex did not panic")
 		}
 	}()
-	mu := NewLRWMutex(context.Background())
+	mu := NewLRWMutex()
 	mu.RLock()
 	mu.Unlock()
 }
@@ -321,7 +323,7 @@ func TestRUnlockPanic(t *testing.T) {
 			t.Fatalf("read unlock of unlocked RWMutex did not panic")
 		}
 	}()
-	mu := NewLRWMutex(context.Background())
+	mu := NewLRWMutex()
 	mu.RUnlock()
 }
 
@@ -332,7 +334,7 @@ func TestRUnlockPanic2(t *testing.T) {
 			t.Fatalf("read unlock of unlocked RWMutex did not panic")
 		}
 	}()
-	mu := NewLRWMutex(context.Background())
+	mu := NewLRWMutex()
 	mu.Lock()
 	mu.RUnlock()
 }

pkg/sys/rlimit-file_nix.go

@@ -18,7 +18,10 @@
 
 package sys
 
-import "syscall"
+import (
+	"runtime"
+	"syscall"
+)
 
 // GetMaxOpenFileLimit - returns maximum file descriptor number that can be opened by this process.
 func GetMaxOpenFileLimit() (curLimit, maxLimit uint64, err error) {
@@ -33,6 +36,13 @@ func GetMaxOpenFileLimit() (curLimit, maxLimit uint64, err error) {
 
 // SetMaxOpenFileLimit - sets maximum file descriptor number that can be opened by this process.
 func SetMaxOpenFileLimit(curLimit, maxLimit uint64) error {
+	if runtime.GOOS == "darwin" && curLimit > 10240 {
+		// The max file limit is 10240, even though
+		// the max returned by Getrlimit is 1<<63-1.
+		// This is OPEN_MAX in sys/syslimits.h.
+		// refer https://github.com/golang/go/issues/30401
+		curLimit = 10240
+	}
 	rlimit := syscall.Rlimit{Cur: curLimit, Max: maxLimit}
 	return syscall.Setrlimit(syscall.RLIMIT_NOFILE, &rlimit)
 }