diff --git a/cmd/config/cache/config.go b/cmd/config/cache/config.go
index adaffbef6..a8a20b3e6 100644
--- a/cmd/config/cache/config.go
+++ b/cmd/config/cache/config.go
@@ -28,13 +28,15 @@ import (
 
 // Config represents cache config settings
 type Config struct {
-	Enabled bool     `json:"-"`
-	Drives  []string `json:"drives"`
-	Expiry  int      `json:"expiry"`
-	MaxUse  int      `json:"maxuse"`
-	Quota   int      `json:"quota"`
-	Exclude []string `json:"exclude"`
-	After   int      `json:"after"`
+	Enabled       bool     `json:"-"`
+	Drives        []string `json:"drives"`
+	Expiry        int      `json:"expiry"`
+	MaxUse        int      `json:"maxuse"`
+	Quota         int      `json:"quota"`
+	Exclude       []string `json:"exclude"`
+	After         int      `json:"after"`
+	WatermarkLow  int      `json:"watermark_low"`
+	WatermarkHigh int      `json:"watermark_high"`
 }
 
 // UnmarshalJSON - implements JSON unmarshal interface for unmarshalling
@@ -64,6 +66,15 @@ func (cfg *Config) UnmarshalJSON(data []byte) (err error) {
 	if _cfg.After < 0 {
 		return errors.New("cache after value should not be less than 0")
 	}
+	if _cfg.WatermarkLow < 0 || _cfg.WatermarkLow > 100 {
+		return errors.New("config low watermark value should be between 0 and 100")
+	}
+	if _cfg.WatermarkHigh < 0 || _cfg.WatermarkHigh > 100 {
+		return errors.New("config high watermark value should be between 0 and 100")
+	}
+	if _cfg.WatermarkLow > 0 && (_cfg.WatermarkLow >= _cfg.WatermarkHigh) {
+		return errors.New("config low watermark value should be less than high watermark")
+	}
 	return nil
 }
 
diff --git a/cmd/config/cache/help.go b/cmd/config/cache/help.go
index cb2ba9436..a599d0725 100644
--- a/cmd/config/cache/help.go
+++ b/cmd/config/cache/help.go
@@ -56,5 +56,17 @@ var (
 			Optional:    true,
 			Type:        "number",
 		},
+		config.HelpKV{
+			Key:         WatermarkLow,
+			Description: `% of cache use at which to stop cache eviction`,
+			Optional:    true,
+			Type:        "number",
+		},
+		config.HelpKV{
+			Key:         WatermarkHigh,
+			Description: `% of cache use at which to start cache eviction`,
+			Optional:    true,
+			Type:        "number",
+		},
 	}
 )
diff --git a/cmd/config/cache/lookup.go b/cmd/config/cache/lookup.go
index 1a187c2e9..ed02e0912 100644
--- a/cmd/config/cache/lookup.go
+++ b/cmd/config/cache/lookup.go
@@ -26,25 +26,31 @@ import (
 
 // Cache ENVs
 const (
-	Drives  = "drives"
-	Exclude = "exclude"
-	Expiry  = "expiry"
-	MaxUse  = "maxuse"
-	Quota   = "quota"
-	After   = "after"
+	Drives        = "drives"
+	Exclude       = "exclude"
+	Expiry        = "expiry"
+	MaxUse        = "maxuse"
+	Quota         = "quota"
+	After         = "after"
+	WatermarkLow  = "watermark_low"
+	WatermarkHigh = "watermark_high"
 
-	EnvCacheDrives  = "MINIO_CACHE_DRIVES"
-	EnvCacheExclude = "MINIO_CACHE_EXCLUDE"
-	EnvCacheExpiry  = "MINIO_CACHE_EXPIRY"
-	EnvCacheMaxUse  = "MINIO_CACHE_MAXUSE"
-	EnvCacheQuota   = "MINIO_CACHE_QUOTA"
-	EnvCacheAfter   = "MINIO_CACHE_AFTER"
+	EnvCacheDrives        = "MINIO_CACHE_DRIVES"
+	EnvCacheExclude       = "MINIO_CACHE_EXCLUDE"
+	EnvCacheExpiry        = "MINIO_CACHE_EXPIRY"
+	EnvCacheMaxUse        = "MINIO_CACHE_MAXUSE"
+	EnvCacheQuota         = "MINIO_CACHE_QUOTA"
+	EnvCacheAfter         = "MINIO_CACHE_AFTER"
+	EnvCacheWatermarkLow  = "MINIO_CACHE_WATERMARK_LOW"
+	EnvCacheWatermarkHigh = "MINIO_CACHE_WATERMARK_HIGH"
 
 	EnvCacheEncryptionMasterKey = "MINIO_CACHE_ENCRYPTION_MASTER_KEY"
 
-	DefaultExpiry = "90"
-	DefaultQuota  = "80"
-	DefaultAfter  = "0"
+	DefaultExpiry        = "90"
+	DefaultQuota         = "80"
+	DefaultAfter         = "0"
+	DefaultWaterMarkLow  = "70"
+	DefaultWaterMarkHigh = "80"
 )
 
 // DefaultKVS - default KV settings for caching.
@@ -70,6 +76,14 @@ var (
 			Key:   After,
 			Value: DefaultAfter,
 		},
+		config.KV{
+			Key:   WatermarkLow,
+			Value: DefaultWaterMarkLow,
+		},
+		config.KV{
+			Key:   WatermarkHigh,
+			Value: DefaultWaterMarkHigh,
+		},
 	}
 )
 
@@ -87,7 +101,6 @@ func Enabled(kvs config.KVS) bool {
 // variables and merge them with provided CacheConfiguration.
 func LookupConfig(kvs config.KVS) (Config, error) {
 	cfg := Config{}
-
 	if err := config.CheckValidKeys(config.CacheSubSys, kvs, DefaultKVS); err != nil {
 		return cfg, err
 	}
@@ -154,5 +167,33 @@ func LookupConfig(kvs config.KVS) (Config, error) {
 		}
 	}
 
+	if lowWMStr := env.Get(EnvCacheWatermarkLow, kvs.Get(WatermarkLow)); lowWMStr != "" {
+		cfg.WatermarkLow, err = strconv.Atoi(lowWMStr)
+		if err != nil {
+			return cfg, config.ErrInvalidCacheWatermarkLow(err)
+		}
+		// WatermarkLow should be a valid percentage.
+		if cfg.WatermarkLow < 0 || cfg.WatermarkLow > 100 {
+			err := errors.New("config min watermark value should be between 0 and 100")
+			return cfg, config.ErrInvalidCacheWatermarkLow(err)
+		}
+	}
+
+	if highWMStr := env.Get(EnvCacheWatermarkHigh, kvs.Get(WatermarkHigh)); highWMStr != "" {
+		cfg.WatermarkHigh, err = strconv.Atoi(highWMStr)
+		if err != nil {
+			return cfg, config.ErrInvalidCacheWatermarkHigh(err)
+		}
+
+		// MaxWatermark should be a valid percentage.
+		if cfg.WatermarkHigh < 0 || cfg.WatermarkHigh > 100 {
+			err := errors.New("config high watermark value should be between 0 and 100")
+			return cfg, config.ErrInvalidCacheWatermarkHigh(err)
+		}
+	}
+	if cfg.WatermarkLow > cfg.WatermarkHigh {
+		err := errors.New("config high watermark value should be greater than low watermark value")
+		return cfg, config.ErrInvalidCacheWatermarkHigh(err)
+	}
 	return cfg, nil
 }
diff --git a/cmd/config/errors.go b/cmd/config/errors.go
index dd2f01304..e1a42394c 100644
--- a/cmd/config/errors.go
+++ b/cmd/config/errors.go
@@ -72,6 +72,18 @@ var (
 		"MINIO_CACHE_AFTER: Valid cache after value must be 0 or greater",
 	)
 
+	ErrInvalidCacheWatermarkLow = newErrFn(
+		"Invalid cache low watermark value",
+		"Please check the passed value",
+		"MINIO_CACHE_WATERMARK_LOW: Valid cache low watermark value must be between 0-100",
+	)
+
+	ErrInvalidCacheWatermarkHigh = newErrFn(
+		"Invalid cache high watermark value",
+		"Please check the passed value",
+		"MINIO_CACHE_WATERMARK_HIGH: Valid cache high watermark value must be between 0-100",
+	)
+
 	ErrInvalidCacheEncryptionKey = newErrFn(
 		"Invalid cache encryption master key value",
 		"Please check the passed value",
diff --git a/cmd/disk-cache-backend.go b/cmd/disk-cache-backend.go
index b5229f72a..19891b693 100644
--- a/cmd/disk-cache-backend.go
+++ b/cmd/disk-cache-backend.go
@@ -27,6 +27,7 @@ import (
 	"log"
 	"net/http"
 	"os"
+	"strings"
 	"sync"
 	"time"
 
@@ -37,6 +38,7 @@ import (
 	"github.com/minio/minio/pkg/disk"
 	"github.com/minio/sio"
 	"github.com/ncw/directio"
+	"go.uber.org/atomic"
 )
 
 const (
@@ -44,7 +46,7 @@ const (
 	cacheMetaJSONFile = "cache.json"
 	cacheDataFile     = "part.1"
 	cacheMetaVersion  = "1.0.0"
-
+	cacheExpiryDays   = time.Duration(90 * time.Hour * 24) // defaults to 90 days
 	// SSECacheEncrypted is the metadata key indicating that the object
 	// is a cache entry encrypted with cache KMS master key in globalCacheKMS.
 	SSECacheEncrypted = "X-Minio-Internal-Encrypted-Cache"
@@ -126,15 +128,15 @@ func (m *cacheMeta) ToObjectInfo(bucket, object string) (o ObjectInfo) {
 type diskCache struct {
 	dir      string // caching directory
 	quotaPct int    // max usage in %
-	expiry   int    // cache expiry in days
 	// mark false if drive is offline
 	online bool
 	// mutex to protect updates to online variable
-	onlineMutex *sync.RWMutex
-	// purge() listens on this channel to start the cache-purge process
-	purgeChan chan struct{}
-	pool      sync.Pool
-	after     int // minimum accesses before an object is cached.
+	onlineMutex   *sync.RWMutex
+	pool          sync.Pool
+	after         int // minimum accesses before an object is cached.
+	lowWatermark  int
+	highWatermark int
+	gcCounter     atomic.Uint64
 	// nsMutex namespace lock
 	nsMutex *nsLockMap
 	// Object functions pointing to the corresponding functions of backend implementation.
@@ -142,18 +144,18 @@ type diskCache struct {
 }
 
 // Inits the disk cache dir if it is not initialized already.
-func newDiskCache(dir string, expiry int, quotaPct, after int) (*diskCache, error) {
+func newDiskCache(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,
-		expiry:      expiry,
-		quotaPct:    quotaPct,
-		after:       after,
-		purgeChan:   make(chan struct{}),
-		online:      true,
-		onlineMutex: &sync.RWMutex{},
+		dir:           dir,
+		quotaPct:      quotaPct,
+		after:         after,
+		lowWatermark:  lowWatermark,
+		highWatermark: highWatermark,
+		online:        true,
+		onlineMutex:   &sync.RWMutex{},
 		pool: sync.Pool{
 			New: func() interface{} {
 				b := directio.AlignedBlock(int(cacheBlkSize))
@@ -168,12 +170,12 @@ func newDiskCache(dir string, expiry int, quotaPct, after int) (*diskCache, erro
 	return &cache, nil
 }
 
-// Returns if the disk usage is low.
-// Disk usage is low if usage is < 80% of cacheMaxDiskUsagePct
-// Ex. for a 100GB disk, if maxUsage is configured as 70% then cacheMaxDiskUsagePct is 70G
-// hence disk usage is low if the disk usage is less than 56G (because 80% of 70G is 56G)
+// diskUsageLow() returns true if disk usage falls below the low watermark w.r.t configured cache quota.
+// Ex. for a 100GB disk, if quota is configured as 70%  and watermark_low = 80% and
+// watermark_high = 90% then garbage collection starts when 63% of disk is used and
+// stops when disk usage drops to 56%
 func (c *diskCache) diskUsageLow() bool {
-	minUsage := c.quotaPct * 80 / 100
+	gcStopPct := c.quotaPct * c.lowWatermark / 100
 	di, err := disk.GetInfo(c.dir)
 	if err != nil {
 		reqInfo := (&logger.ReqInfo{}).AppendTags("cachePath", c.dir)
@@ -182,21 +184,22 @@ func (c *diskCache) diskUsageLow() bool {
 		return false
 	}
 	usedPercent := (di.Total - di.Free) * 100 / di.Total
-	return int(usedPercent) < minUsage
+	return int(usedPercent) < gcStopPct
 }
 
-// Return if the disk usage is high.
-// Disk usage is high if disk used is > cacheMaxDiskUsagePct
+// Returns if the disk usage reaches high water mark w.r.t the configured cache quota.
+// gc starts if high water mark reached.
 func (c *diskCache) diskUsageHigh() bool {
+	gcTriggerPct := c.quotaPct * c.highWatermark / 100
 	di, err := disk.GetInfo(c.dir)
 	if err != nil {
 		reqInfo := (&logger.ReqInfo{}).AppendTags("cachePath", c.dir)
 		ctx := logger.SetReqInfo(context.Background(), reqInfo)
 		logger.LogIf(ctx, err)
-		return true
+		return false
 	}
 	usedPercent := (di.Total - di.Free) * 100 / di.Total
-	return int(usedPercent) > c.quotaPct
+	return int(usedPercent) >= gcTriggerPct
 }
 
 // Returns if size space can be allocated without exceeding
@@ -213,12 +216,42 @@ func (c *diskCache) diskAvailable(size int64) bool {
 	return int(usedPercent) < c.quotaPct
 }
 
+// toClear returns how many bytes should be cleared to reach the low watermark quota.
+// returns 0 if below quota.
+func (c *diskCache) toClear() uint64 {
+	di, err := disk.GetInfo(c.dir)
+	if err != nil {
+		reqInfo := (&logger.ReqInfo{}).AppendTags("cachePath", c.dir)
+		ctx := logger.SetReqInfo(context.Background(), reqInfo)
+		logger.LogIf(ctx, err)
+		return 0
+	}
+	return bytesToClear(int64(di.Total), int64(di.Free), uint64(c.quotaPct), uint64(c.lowWatermark))
+}
+
 // Purge cache entries that were not accessed.
-func (c *diskCache) purge() {
+func (c *diskCache) purge(ctx context.Context, doneCh <-chan struct{}) {
+	if c.diskUsageLow() {
+		return
+	}
+	toFree := c.toClear()
+	if toFree == 0 {
+		return
+	}
+	// 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(int64(toFree), time.Now().Unix(), 100)
+	if err != nil {
+		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 {
 		fm := make(map[string]os.FileInfo)
-		fname := pathJoin(pathPrefix, cacheDataFile)
+		fname := pathJoin(pathPrefix, dataFile)
 		if fi, err := os.Stat(fname); err == nil {
 			fm[fname] = fi
 		}
@@ -231,63 +264,73 @@ func (c *diskCache) purge() {
 		}
 		return fm
 	}
-	ctx := context.Background()
-	for {
-		olderThan := c.expiry * 24
-		for !c.diskUsageLow() {
-			// delete unaccessed objects older than expiry duration
-			expiry := UTCNow().Add(time.Hour * time.Duration(-1*olderThan))
-			olderThan /= 2
-			if olderThan < 1 {
-				break
-			}
-			deletedCount := 0
+	objDirs, err := ioutil.ReadDir(c.dir)
+	if err != nil {
+		log.Fatal(err)
+	}
 
-			objDirs, err := ioutil.ReadDir(c.dir)
-			if err != nil {
-				log.Fatal(err)
-			}
+	for _, obj := range objDirs {
+		if obj.Name() == minioMetaBucket {
+			continue
+		}
 
-			for _, obj := range objDirs {
-				if obj.Name() == minioMetaBucket {
-					continue
-				}
-				meta, _, _, err := c.statCachedMeta(context.Background(), pathJoin(c.dir, obj.Name()))
-				if err != nil {
-					// delete any partially filled cache entry left behind.
-					removeAll(pathJoin(c.dir, obj.Name()))
-					continue
-				}
-				// stat all cached file ranges and cacheDataFile.
-				fis := fiStatFn(meta.Ranges, cacheDataFile, pathJoin(c.dir, obj.Name()))
-				objInfo := meta.ToObjectInfo("", "")
-				cc := cacheControlOpts(objInfo)
-
-				for fname, fi := range fis {
-					if atime.Get(fi).Before(expiry) ||
-						cc.isStale(objInfo.ModTime) {
-						if err = removeAll(fname); err != nil {
-							logger.LogIf(ctx, err)
-						}
-						deletedCount++
-						// break early if sufficient disk space reclaimed.
-						if !c.diskUsageLow() {
-							break
-						}
+		cacheDir := pathJoin(c.dir, obj.Name())
+		meta, _, numHits, err := c.statCachedMeta(ctx, cacheDir)
+		if err != nil {
+			// delete any partially filled cache entry left behind.
+			removeAll(cacheDir)
+			continue
+		}
+		// stat all cached file ranges and cacheDataFile.
+		cachedFiles := fiStatFn(meta.Ranges, cacheDataFile, pathJoin(c.dir, obj.Name()))
+		objInfo := meta.ToObjectInfo("", "")
+		cc := cacheControlOpts(objInfo)
+		for fname, fi := range cachedFiles {
+			if cc != nil {
+				if cc.isStale(objInfo.ModTime) {
+					if err = removeAll(fname); err != nil {
+						logger.LogIf(ctx, err)
+					}
+					scorer.adjustSaveBytes(-fi.Size())
+					// break early if sufficient disk space reclaimed.
+					if c.diskUsageLow() {
+						return
 					}
 				}
+				continue
 			}
-			if deletedCount == 0 {
-				break
-			}
+			scorer.addFile(fname, atime.Get(fi), fi.Size(), numHits)
 		}
-		for {
-			<-c.purgeChan
-			if c.diskUsageHigh() {
-				break
-			}
+		// clean up stale cache.json files for objects that never got cached but access count was maintained in cache.json
+		fi, err := os.Stat(pathJoin(cacheDir, cacheMetaJSONFile))
+		if err != nil || (fi.ModTime().Before(expiry) && len(cachedFiles) == 0) {
+			removeAll(cacheDir)
+			scorer.adjustSaveBytes(-fi.Size())
+			continue
+		}
+		if c.diskUsageLow() {
+			return
 		}
 	}
+	for _, path := range scorer.fileNames() {
+		removeAll(path)
+		slashIdx := strings.LastIndex(path, SlashSeparator)
+		pathPrefix := path[0:slashIdx]
+		fname := path[slashIdx+1:]
+		if fname == cacheDataFile {
+			removeAll(pathPrefix)
+		}
+	}
+}
+
+func (c *diskCache) incGCCounter() {
+	c.gcCounter.Add(uint64(1))
+}
+func (c *diskCache) resetGCCounter() {
+	c.gcCounter.Store(uint64(0))
+}
+func (c *diskCache) gcCount() uint64 {
+	return c.gcCounter.Load()
 }
 
 // sets cache drive status
@@ -378,6 +421,9 @@ func (c *diskCache) statRange(ctx context.Context, bucket, object string, rs *HT
 	if !ok {
 		return oi, rngInfo, numHits, ObjectNotFound{Bucket: bucket, Object: object}
 	}
+	if _, err = os.Stat(pathJoin(cacheObjPath, rngFile)); err != nil {
+		return oi, rngInfo, numHits, ObjectNotFound{Bucket: bucket, Object: object}
+	}
 	rngInfo = RangeInfo{Range: rng, File: rngFile, Size: int64(actualRngSize)}
 
 	err = decryptCacheObjectETag(&oi)
@@ -568,10 +614,8 @@ 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() {
-		select {
-		case c.purgeChan <- struct{}{}:
-		default:
-		}
+		c.incGCCounter()
+		io.Copy(ioutil.Discard, data)
 		return errDiskFull
 	}
 	cachePath := getCacheSHADir(c.dir, bucket, object)
@@ -622,11 +666,11 @@ func (c *diskCache) Put(ctx context.Context, bucket, object string, data io.Read
 	if IsErr(err, baseErrs...) {
 		c.setOnline(false)
 	}
-
 	if err != nil {
 		removeAll(cachePath)
 		return err
 	}
+
 	if actualSize != uint64(n) {
 		removeAll(cachePath)
 		return IncompleteBody{}
diff --git a/cmd/disk-cache-utils.go b/cmd/disk-cache-utils.go
index 5c6397a5f..2f18bf630 100644
--- a/cmd/disk-cache-utils.go
+++ b/cmd/disk-cache-utils.go
@@ -17,8 +17,12 @@
 package cmd
 
 import (
+	"container/list"
 	"encoding/hex"
+	"errors"
+	"fmt"
 	"io"
+	"math"
 	"os"
 	"path"
 	"strconv"
@@ -57,13 +61,8 @@ type cacheControl struct {
 	noCache      bool
 }
 
-func (c cacheControl) isEmpty() bool {
-	return c == cacheControl{}
-
-}
-
-func (c cacheControl) isStale(modTime time.Time) bool {
-	if c.isEmpty() {
+func (c *cacheControl) isStale(modTime time.Time) bool {
+	if c == nil {
 		return false
 	}
 	// response will never be stale if only-if-cached is set
@@ -100,7 +99,8 @@ func (c cacheControl) isStale(modTime time.Time) bool {
 }
 
 // returns struct with cache-control settings from user metadata.
-func cacheControlOpts(o ObjectInfo) (c cacheControl) {
+func cacheControlOpts(o ObjectInfo) *cacheControl {
+	c := cacheControl{}
 	m := o.UserDefined
 	if o.Expires != timeSentinel {
 		c.expiry = o.Expires
@@ -114,7 +114,7 @@ func cacheControlOpts(o ObjectInfo) (c cacheControl) {
 
 	}
 	if headerVal == "" {
-		return
+		return nil
 	}
 	headerVal = strings.ToLower(headerVal)
 	headerVal = strings.TrimSpace(headerVal)
@@ -146,7 +146,7 @@ func cacheControlOpts(o ObjectInfo) (c cacheControl) {
 			p[0] == "max-stale" {
 			i, err := strconv.Atoi(p[1])
 			if err != nil {
-				return cacheControl{}
+				return nil
 			}
 			if p[0] == "max-age" {
 				c.maxAge = i
@@ -162,7 +162,7 @@ func cacheControlOpts(o ObjectInfo) (c cacheControl) {
 			}
 		}
 	}
-	return c
+	return &c
 }
 
 // backendDownError returns true if err is due to backend failure or faulty disk if in server mode
@@ -283,3 +283,161 @@ func isMetadataSame(m1, m2 map[string]string) bool {
 	}
 	return true
 }
+
+type fileScorer struct {
+	saveBytes int64
+	now       int64
+	maxHits   int
+	// 1/size for consistent score.
+	sizeMult float64
+
+	// queue is a linked list of files we want to delete.
+	// The list is kept sorted according to score, highest at top, lowest at bottom.
+	queue       list.List
+	queuedBytes int64
+}
+
+type queuedFile struct {
+	name  string
+	size  int64
+	score float64
+}
+
+// newFileScorer allows to collect files to save a specific number of bytes.
+// Each file is assigned a score based on its age, size and number of hits.
+// A list of files is maintained
+func newFileScorer(saveBytes int64, now int64, maxHits int) (*fileScorer, error) {
+	if saveBytes <= 0 {
+		return nil, errors.New("newFileScorer: saveBytes <= 0")
+	}
+	if now < 0 {
+		return nil, errors.New("newFileScorer: now < 0")
+	}
+	if maxHits <= 0 {
+		return nil, errors.New("newFileScorer: maxHits <= 0")
+	}
+	f := fileScorer{saveBytes: saveBytes, maxHits: maxHits, now: now, sizeMult: 1 / float64(saveBytes)}
+	f.queue.Init()
+	return &f, nil
+}
+
+func (f *fileScorer) addFile(name string, lastAccess time.Time, size int64, hits int) {
+	// Calculate how much we want to delete this object.
+	file := queuedFile{
+		name: name,
+		size: size,
+	}
+	score := float64(f.now - lastAccess.Unix())
+	// Size as fraction of how much we want to save, 0->1.
+	szWeight := math.Max(0, (math.Min(1, float64(size)*f.sizeMult)))
+	// 0 at f.maxHits, 1 at 0.
+	hitsWeight := (1.0 - math.Max(0, math.Min(1.0, float64(hits)/float64(f.maxHits))))
+	file.score = score * (1 + 0.25*szWeight + 0.25*hitsWeight)
+	// If we still haven't saved enough, just add the file
+	if f.queuedBytes < f.saveBytes {
+		f.insertFile(file)
+		f.trimQueue()
+		return
+	}
+	// If we score less than the worst, don't insert.
+	worstE := f.queue.Back()
+	if worstE != nil && file.score < worstE.Value.(queuedFile).score {
+		return
+	}
+	f.insertFile(file)
+	f.trimQueue()
+}
+
+// adjustSaveBytes allows to adjust the number of bytes to save.
+// This can be used to adjust the count on the fly.
+// Returns true if there still is a need to delete files (saveBytes >0),
+// false if no more bytes needs to be saved.
+func (f *fileScorer) adjustSaveBytes(n int64) bool {
+	f.saveBytes += n
+	if f.saveBytes <= 0 {
+		f.queue.Init()
+		f.saveBytes = 0
+		return false
+	}
+	if n < 0 {
+		f.trimQueue()
+	}
+	return true
+}
+
+// insertFile will insert a file into the list, sorted by its score.
+func (f *fileScorer) insertFile(file queuedFile) {
+	e := f.queue.Front()
+	for e != nil {
+		v := e.Value.(queuedFile)
+		if v.score < file.score {
+			break
+		}
+		e = e.Next()
+	}
+	f.queuedBytes += file.size
+	// We reached the end.
+	if e == nil {
+		f.queue.PushBack(file)
+		return
+	}
+	f.queue.InsertBefore(file, e)
+}
+
+// trimQueue will trim the back of queue and still keep below wantSave.
+func (f *fileScorer) trimQueue() {
+	for {
+		e := f.queue.Back()
+		if e == nil {
+			return
+		}
+		v := e.Value.(queuedFile)
+		if f.queuedBytes-v.size < f.saveBytes {
+			return
+		}
+		f.queue.Remove(e)
+		f.queuedBytes -= v.size
+	}
+}
+
+// fileNames returns all queued file names.
+func (f *fileScorer) fileNames() []string {
+	res := make([]string, 0, f.queue.Len())
+	e := f.queue.Front()
+	for e != nil {
+		res = append(res, e.Value.(queuedFile).name)
+		e = e.Next()
+	}
+	return res
+}
+
+func (f *fileScorer) reset() {
+	f.queue.Init()
+	f.queuedBytes = 0
+}
+
+func (f *fileScorer) queueString() string {
+	var res strings.Builder
+	e := f.queue.Front()
+	i := 0
+	for e != nil {
+		v := e.Value.(queuedFile)
+		if i > 0 {
+			res.WriteByte('\n')
+		}
+		res.WriteString(fmt.Sprintf("%03d: %s (score: %.3f, bytes: %d)", i, v.name, v.score, v.size))
+		i++
+		e = e.Next()
+	}
+	return res.String()
+}
+
+// bytesToClear() returns the number of bytes to clear to reach low watermark
+// w.r.t quota given disk total and free space, quota in % allocated to cache
+// and low watermark % w.r.t allowed quota.
+func bytesToClear(total, free int64, quotaPct, lowWatermark uint64) uint64 {
+	used := (total - free)
+	quotaAllowed := total * (int64)(quotaPct) / 100
+	lowWMUsage := (total * (int64)(lowWatermark*quotaPct) / (100 * 100))
+	return (uint64)(math.Min(float64(quotaAllowed), math.Max(0.0, float64(used-lowWMUsage))))
+}
diff --git a/cmd/disk-cache-utils_test.go b/cmd/disk-cache-utils_test.go
index b158fda18..8de7b4551 100644
--- a/cmd/disk-cache-utils_test.go
+++ b/cmd/disk-cache-utils_test.go
@@ -29,16 +29,16 @@ func TestGetCacheControlOpts(t *testing.T) {
 	testCases := []struct {
 		cacheControlHeaderVal string
 		expiryHeaderVal       time.Time
-		expectedCacheControl  cacheControl
+		expectedCacheControl  *cacheControl
 		expectedErr           bool
 	}{
-		{"", timeSentinel, cacheControl{}, false},
-		{"max-age=2592000, public", timeSentinel, cacheControl{maxAge: 2592000, sMaxAge: 0, minFresh: 0, expiry: time.Time{}}, false},
-		{"max-age=2592000, no-store", timeSentinel, cacheControl{maxAge: 2592000, sMaxAge: 0, noStore: true, minFresh: 0, expiry: time.Time{}}, false},
-		{"must-revalidate, max-age=600", timeSentinel, cacheControl{maxAge: 600, sMaxAge: 0, minFresh: 0, expiry: time.Time{}}, false},
-		{"s-maxAge=2500, max-age=600", timeSentinel, cacheControl{maxAge: 600, sMaxAge: 2500, minFresh: 0, expiry: time.Time{}}, false},
-		{"s-maxAge=2500, max-age=600", expiry, cacheControl{maxAge: 600, sMaxAge: 2500, minFresh: 0, expiry: time.Date(2015, time.October, 21, 07, 28, 00, 00, time.UTC)}, false},
-		{"s-maxAge=2500, max-age=600s", timeSentinel, cacheControl{maxAge: 600, sMaxAge: 2500, minFresh: 0, expiry: time.Time{}}, true},
+		{"", timeSentinel, nil, false},
+		{"max-age=2592000, public", timeSentinel, &cacheControl{maxAge: 2592000, sMaxAge: 0, minFresh: 0, expiry: time.Time{}}, false},
+		{"max-age=2592000, no-store", timeSentinel, &cacheControl{maxAge: 2592000, sMaxAge: 0, noStore: true, minFresh: 0, expiry: time.Time{}}, false},
+		{"must-revalidate, max-age=600", timeSentinel, &cacheControl{maxAge: 600, sMaxAge: 0, minFresh: 0, expiry: time.Time{}}, false},
+		{"s-maxAge=2500, max-age=600", timeSentinel, &cacheControl{maxAge: 600, sMaxAge: 2500, minFresh: 0, expiry: time.Time{}}, false},
+		{"s-maxAge=2500, max-age=600", expiry, &cacheControl{maxAge: 600, sMaxAge: 2500, minFresh: 0, expiry: time.Date(2015, time.October, 21, 07, 28, 00, 00, time.UTC)}, false},
+		{"s-maxAge=2500, max-age=600s", timeSentinel, &cacheControl{maxAge: 600, sMaxAge: 2500, minFresh: 0, expiry: time.Time{}}, true},
 	}
 
 	for _, testCase := range testCases {
@@ -49,7 +49,7 @@ func TestGetCacheControlOpts(t *testing.T) {
 				m["expires"] = testCase.expiryHeaderVal.String()
 			}
 			c := cacheControlOpts(ObjectInfo{UserDefined: m, Expires: testCase.expiryHeaderVal})
-			if testCase.expectedErr && (c != cacheControl{}) {
+			if testCase.expectedErr && (c != nil) {
 				t.Errorf("expected err, got <nil>")
 			}
 			if !testCase.expectedErr && !reflect.DeepEqual(c, testCase.expectedCacheControl) {
@@ -83,3 +83,90 @@ func TestIsMetadataSame(t *testing.T) {
 		}
 	}
 }
+
+func TestNewFileScorer(t *testing.T) {
+	fs, err := newFileScorer(1000, time.Now().Unix(), 10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(fs.fileNames()) != 0 {
+		t.Fatal("non zero files??")
+	}
+	now := time.Now()
+	fs.addFile("recent", now.Add(-time.Minute), 1000, 10)
+	fs.addFile("older", now.Add(-time.Hour), 1000, 10)
+	if !reflect.DeepEqual(fs.fileNames(), []string{"older"}) {
+		t.Fatal("unexpected file list", fs.queueString())
+	}
+	fs.reset()
+	fs.addFile("bigger", now.Add(-time.Minute), 2000, 10)
+	fs.addFile("recent", now.Add(-time.Minute), 1000, 10)
+	if !reflect.DeepEqual(fs.fileNames(), []string{"bigger"}) {
+		t.Fatal("unexpected file list", fs.queueString())
+	}
+	fs.reset()
+	fs.addFile("less", now.Add(-time.Minute), 1000, 5)
+	fs.addFile("recent", now.Add(-time.Minute), 1000, 10)
+	if !reflect.DeepEqual(fs.fileNames(), []string{"less"}) {
+		t.Fatal("unexpected file list", fs.queueString())
+	}
+	fs.reset()
+	fs.addFile("small", now.Add(-time.Minute), 200, 10)
+	fs.addFile("medium", now.Add(-time.Minute), 300, 10)
+	if !reflect.DeepEqual(fs.fileNames(), []string{"medium", "small"}) {
+		t.Fatal("unexpected file list", fs.queueString())
+	}
+	fs.addFile("large", now.Add(-time.Minute), 700, 10)
+	fs.addFile("xsmol", now.Add(-time.Minute), 7, 10)
+	if !reflect.DeepEqual(fs.fileNames(), []string{"large", "medium"}) {
+		t.Fatal("unexpected file list", fs.queueString())
+	}
+
+	fs.reset()
+	fs.addFile("less", now.Add(-time.Minute), 500, 5)
+	fs.addFile("recent", now.Add(-time.Minute), 500, 10)
+	if !fs.adjustSaveBytes(-500) {
+		t.Fatal("we should still need more bytes, got false")
+	}
+	// We should only need 500 bytes now.
+	if !reflect.DeepEqual(fs.fileNames(), []string{"less"}) {
+		t.Fatal("unexpected file list", fs.queueString())
+	}
+	if fs.adjustSaveBytes(-500) {
+		t.Fatal("we shouldn't need any more bytes, got true")
+	}
+	fs, err = newFileScorer(1000, time.Now().Unix(), 10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	fs.addFile("bigger", now.Add(-time.Minute), 50, 10)
+	// sorting should be consistent after adjusting savebytes.
+	fs.adjustSaveBytes(-800)
+	fs.addFile("smaller", now.Add(-time.Minute), 40, 10)
+	if !reflect.DeepEqual(fs.fileNames(), []string{"bigger", "smaller"}) {
+		t.Fatal("unexpected file list", fs.queueString())
+	}
+}
+func TestBytesToClear(t *testing.T) {
+	testCases := []struct {
+		total        int64
+		free         int64
+		quotaPct     uint64
+		watermarkLow uint64
+		expected     uint64
+	}{
+		{1000, 800, 40, 90, 0},
+		{1000, 200, 40, 90, 400},
+		{1000, 400, 40, 90, 240},
+		{1000, 600, 40, 90, 40},
+		{1000, 600, 40, 70, 120},
+		{1000, 1000, 90, 70, 0},
+		{1000, 0, 90, 70, 370},
+	}
+	for i, tc := range testCases {
+		toClear := bytesToClear(tc.total, tc.free, tc.quotaPct, tc.watermarkLow)
+		if tc.expected != toClear {
+			t.Errorf("test %d expected %v, got %v", i, tc.expected, toClear)
+		}
+	}
+}
diff --git a/cmd/disk-cache.go b/cmd/disk-cache.go
index 4ee9f8165..752f0c753 100644
--- a/cmd/disk-cache.go
+++ b/cmd/disk-cache.go
@@ -38,7 +38,8 @@ import (
 )
 
 const (
-	cacheBlkSize = int64(1 * 1024 * 1024)
+	cacheBlkSize    = int64(1 * 1024 * 1024)
+	cacheGCInterval = time.Minute * 30
 )
 
 // CacheStorageInfo - represents total, free capacity of
@@ -174,7 +175,7 @@ func (c *cacheObjects) GetObjectNInfo(ctx context.Context, bucket, object string
 	if c.isCacheExclude(bucket, object) || c.skipCache() {
 		return c.GetObjectNInfoFn(ctx, bucket, object, rs, h, lockType, opts)
 	}
-	var cc cacheControl
+	var cc *cacheControl
 	var cacheObjSize int64
 	// fetch diskCache if object is currently cached or nearest available cache drive
 	dcache, err := c.getCacheToLoc(ctx, bucket, object)
@@ -191,8 +192,8 @@ func (c *cacheObjects) GetObjectNInfo(ctx context.Context, bucket, object string
 			}
 		}
 		cc = cacheControlOpts(cacheReader.ObjInfo)
-		if (!cc.isEmpty() && !cc.isStale(cacheReader.ObjInfo.ModTime)) ||
-			cc.onlyIfCached {
+		if cc != nil && (!cc.isStale(cacheReader.ObjInfo.ModTime) ||
+			cc.onlyIfCached) {
 			// This is a cache hit, mark it so
 			bytesServed := cacheReader.ObjInfo.Size
 			if rs != nil {
@@ -259,11 +260,8 @@ func (c *cacheObjects) GetObjectNInfo(ctx context.Context, bucket, object string
 	c.cacheStats.incMiss()
 	// Since we got here, we are serving the request from backend,
 	// and also adding the object to the cache.
-	if !dcache.diskUsageLow() {
-		select {
-		case dcache.purgeChan <- struct{}{}:
-		default:
-		}
+	if dcache.diskUsageHigh() {
+		dcache.incGCCounter()
 	}
 
 	bkReader, bkErr := c.GetObjectNInfoFn(ctx, bucket, object, rs, h, lockType, opts)
@@ -330,12 +328,12 @@ func (c *cacheObjects) GetObjectInfo(ctx context.Context, bucket, object string,
 	if err != nil {
 		return getObjectInfoFn(ctx, bucket, object, opts)
 	}
-	var cc cacheControl
+	var cc *cacheControl
 	// if cache control setting is valid, avoid HEAD operation to backend
 	cachedObjInfo, _, cerr := dcache.Stat(ctx, bucket, object)
 	if cerr == nil {
 		cc = cacheControlOpts(cachedObjInfo)
-		if !cc.isStale(cachedObjInfo.ModTime) {
+		if cc == nil || (cc != nil && !cc.isStale(cachedObjInfo.ModTime)) {
 			// This is a cache hit, mark it so
 			c.cacheStats.incHit()
 			return cachedObjInfo, nil
@@ -522,15 +520,10 @@ func newCache(config cache.Config) ([]*diskCache, bool, error) {
 		if quota == 0 {
 			quota = config.Quota
 		}
-
-		cache, err := newDiskCache(dir, config.Expiry, quota, config.After)
+		cache, err := newDiskCache(dir, quota, config.After, config.WatermarkLow, config.WatermarkHigh)
 		if err != nil {
 			return nil, false, err
 		}
-		// Start the purging go-routine for entries that have expired if no migration in progress
-		if !migrating {
-			go cache.purge()
-		}
 		caches = append(caches, cache)
 	}
 	return caches, migrating, nil
@@ -577,13 +570,12 @@ func (c *cacheObjects) migrateCacheFromV1toV2(ctx context.Context) {
 	}
 
 	errCnt := 0
-	for index, err := range g.Wait() {
+	for _, err := range g.Wait() {
 		if err != nil {
 			errCnt++
 			logger.LogIf(ctx, err)
 			continue
 		}
-		go c.cache[index].purge()
 	}
 
 	if errCnt > 0 {
@@ -697,5 +689,38 @@ func newServerCacheObjects(ctx context.Context, config cache.Config) (CacheObjec
 	if migrateSw {
 		go c.migrateCacheFromV1toV2(ctx)
 	}
+	go c.gc(ctx, GlobalServiceDoneCh)
 	return c, nil
 }
+
+func (c *cacheObjects) gc(ctx context.Context, doneCh chan struct{}) {
+	ticker := time.NewTicker(cacheGCInterval)
+	var gcLock sync.Mutex
+
+	defer ticker.Stop()
+	for {
+		select {
+		case <-doneCh:
+			return
+		case <-ticker.C:
+			if c.migrating {
+				continue
+			}
+			var wg sync.WaitGroup
+			for _, dcache := range c.cache {
+				if dcache.gcCount() == 0 {
+					continue
+				}
+				gcLock.Lock()
+				wg.Add(1)
+				go func(d *diskCache, l *sync.Mutex) {
+					defer wg.Done()
+					d.resetGCCounter()
+					d.purge(ctx, doneCh)
+					l.Unlock()
+				}(dcache, &gcLock)
+			}
+			wg.Wait()
+		}
+	}
+}
diff --git a/cmd/disk-cache_test.go b/cmd/disk-cache_test.go
index 7f5b09b23..866205eeb 100644
--- a/cmd/disk-cache_test.go
+++ b/cmd/disk-cache_test.go
@@ -27,15 +27,15 @@ import (
 )
 
 // Initialize cache objects.
-func initCacheObjects(disk string, cacheMaxUse, cacheAfter int) (*diskCache, error) {
-	return newDiskCache(disk, 80, cacheMaxUse, cacheAfter)
+func initCacheObjects(disk string, cacheMaxUse, cacheAfter, cacheWatermarkLow, cacheWatermarkHigh int) (*diskCache, error) {
+	return newDiskCache(disk, cacheMaxUse, cacheAfter, cacheWatermarkLow, cacheWatermarkHigh)
 }
 
 // inits diskCache struct for nDisks
-func initDiskCaches(drives []string, cacheMaxUse, cacheAfter int, t *testing.T) ([]*diskCache, error) {
+func initDiskCaches(drives []string, cacheMaxUse, cacheAfter, cacheWatermarkLow, cacheWatermarkHigh int, t *testing.T) ([]*diskCache, error) {
 	var cb []*diskCache
 	for _, d := range drives {
-		obj, err := initCacheObjects(d, cacheMaxUse, cacheAfter)
+		obj, err := initCacheObjects(d, cacheMaxUse, cacheAfter, cacheWatermarkLow, cacheWatermarkHigh)
 		if err != nil {
 			return nil, err
 		}
@@ -70,7 +70,7 @@ func TestGetCachedLoc(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
-		d, err := initDiskCaches(fsDirs, 100, 1, t)
+		d, err := initDiskCaches(fsDirs, 100, 1, 80, 90, t)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -109,7 +109,7 @@ func TestGetCacheMaxUse(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
-		d, err := initDiskCaches(fsDirs, 80, 1, t)
+		d, err := initDiskCaches(fsDirs, 80, 1, 80, 90, t)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -180,7 +180,7 @@ func TestDiskCache(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	d, err := initDiskCaches(fsDirs, 100, 0, t)
+	d, err := initDiskCaches(fsDirs, 100, 0, 80, 90, t)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -258,7 +258,7 @@ func TestDiskCacheMaxUse(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	d, err := initDiskCaches(fsDirs, 80, 0, t)
+	d, err := initDiskCaches(fsDirs, 80, 0, 80, 90, t)
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/cmd/gateway/azure/gateway-azure.go b/cmd/gateway/azure/gateway-azure.go
index f47e242e7..87015be66 100644
--- a/cmd/gateway/azure/gateway-azure.go
+++ b/cmd/gateway/azure/gateway-azure.go
@@ -94,8 +94,10 @@ EXAMPLES:
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_SECRET_KEY{{.AssignmentOperator}}azureaccountkey
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_DRIVES{{.AssignmentOperator}}"/mnt/drive1,/mnt/drive2,/mnt/drive3,/mnt/drive4"
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_EXCLUDE{{.AssignmentOperator}}"bucket1/*,*.png"
-     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_EXPIRY{{.AssignmentOperator}}40
-     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_QUOTA{{.AssignmentOperator}}80
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_QUOTA{{.AssignmentOperator}}90
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_AFTER{{.AssignmentOperator}}3
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_WATERMARK_LOW{{.AssignmentOperator}}75
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_WATERMARK_HIGH{{.AssignmentOperator}}85
      {{.Prompt}} {{.HelpName}}
 `
 
diff --git a/cmd/gateway/b2/gateway-b2.go b/cmd/gateway/b2/gateway-b2.go
index 48193f6a2..fea79d673 100644
--- a/cmd/gateway/b2/gateway-b2.go
+++ b/cmd/gateway/b2/gateway-b2.go
@@ -69,8 +69,11 @@ EXAMPLES:
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_SECRET_KEY{{.AssignmentOperator}}applicationKey
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_DRIVES{{.AssignmentOperator}}"/mnt/drive1,/mnt/drive2,/mnt/drive3,/mnt/drive4"
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_EXCLUDE{{.AssignmentOperator}}"bucket1/*,*.png"
-     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_EXPIRY{{.AssignmentOperator}}40
-     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_QUOTA{{.AssignmentOperator}}80
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_QUOTA{{.AssignmentOperator}}90
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_AFTER{{.AssignmentOperator}}3
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_WATERMARK_LOW{{.AssignmentOperator}}75
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_WATERMARK_HIGH{{.AssignmentOperator}}85
+
      {{.Prompt}} {{.HelpName}}
 `
 	minio.RegisterGatewayCommand(cli.Command{
diff --git a/cmd/gateway/gcs/gateway-gcs.go b/cmd/gateway/gcs/gateway-gcs.go
index 1336b29ad..4000bc194 100644
--- a/cmd/gateway/gcs/gateway-gcs.go
+++ b/cmd/gateway/gcs/gateway-gcs.go
@@ -124,8 +124,10 @@ EXAMPLES:
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_SECRET_KEY{{.AssignmentOperator}}secretkey
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_DRIVES{{.AssignmentOperator}}"/mnt/drive1,/mnt/drive2,/mnt/drive3,/mnt/drive4"
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_EXCLUDE{{.AssignmentOperator}}"bucket1/*;*.png"
-     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_EXPIRY{{.AssignmentOperator}}40
-     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_QUOTA{{.AssignmentOperator}}80
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_AFTER{{.AssignmentOperator}}3
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_WATERMARK_LOW{{.AssignmentOperator}}75
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_WATERMARK_HIGH{{.AssignmentOperator}}85
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_QUOTA{{.AssignmentOperator}}90
      {{.Prompt}} {{.HelpName}} mygcsprojectid
 `
 
diff --git a/cmd/gateway/hdfs/gateway-hdfs.go b/cmd/gateway/hdfs/gateway-hdfs.go
index 6cab51610..5b66db243 100644
--- a/cmd/gateway/hdfs/gateway-hdfs.go
+++ b/cmd/gateway/hdfs/gateway-hdfs.go
@@ -75,8 +75,10 @@ EXAMPLES:
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_SECRET_KEY{{.AssignmentOperator}}secretkey
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_DRIVES{{.AssignmentOperator}}"/mnt/drive1,/mnt/drive2,/mnt/drive3,/mnt/drive4"
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_EXCLUDE{{.AssignmentOperator}}"bucket1/*,*.png"
-     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_EXPIRY{{.AssignmentOperator}}40
-     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_QUOTA{{.AssignmentOperator}}80
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_QUOTA{{.AssignmentOperator}}90 
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_AFTER{{.AssignmentOperator}}3
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_WATERMARK_LOW{{.AssignmentOperator}}75
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_WATERMARK_HIGH{{.AssignmentOperator}}85
      {{.Prompt}} {{.HelpName}} hdfs://namenode:8200
 `
 
diff --git a/cmd/gateway/nas/gateway-nas.go b/cmd/gateway/nas/gateway-nas.go
index b86a90edf..ad4c1b433 100644
--- a/cmd/gateway/nas/gateway-nas.go
+++ b/cmd/gateway/nas/gateway-nas.go
@@ -52,8 +52,11 @@ EXAMPLES:
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_SECRET_KEY{{.AssignmentOperator}}secretkey
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_DRIVES{{.AssignmentOperator}}"/mnt/drive1,/mnt/drive2,/mnt/drive3,/mnt/drive4"
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_EXCLUDE{{.AssignmentOperator}}"bucket1/*,*.png"
-     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_EXPIRY{{.AssignmentOperator}}40
-     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_QUOTA{{.AssignmentOperator}}80
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_QUOTA{{.AssignmentOperator}}90
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_AFTER{{.AssignmentOperator}}3
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_WATERMARK_LOW{{.AssignmentOperator}}75
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_WATERMARK_HIGH{{.AssignmentOperator}}85
+
      {{.Prompt}} {{.HelpName}} /shared/nasvol
 `
 
diff --git a/cmd/gateway/oss/gateway-oss.go b/cmd/gateway/oss/gateway-oss.go
index 7954e2f10..7283426e4 100644
--- a/cmd/gateway/oss/gateway-oss.go
+++ b/cmd/gateway/oss/gateway-oss.go
@@ -71,8 +71,10 @@ EXAMPLES:
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_SECRET_KEY{{.AssignmentOperator}}secretkey
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_DRIVES{{.AssignmentOperator}}"/mnt/drive1,/mnt/drive2,/mnt/drive3,/mnt/drive4"
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_EXCLUDE{{.AssignmentOperator}}"bucket1/*,*.png"
-     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_EXPIRY{{.AssignmentOperator}}40
-     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_QUOTA{{.AssignmentOperator}}80
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_AFTER{{.AssignmentOperator}}3
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_WATERMARK_LOW{{.AssignmentOperator}}75
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_WATERMARK_HIGH{{.AssignmentOperator}}85
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_QUOTA{{.AssignmentOperator}}90
      {{.Prompt}} {{.HelpName}}
 `
 
diff --git a/cmd/gateway/s3/gateway-s3.go b/cmd/gateway/s3/gateway-s3.go
index 6d718a2e5..86a37d04f 100644
--- a/cmd/gateway/s3/gateway-s3.go
+++ b/cmd/gateway/s3/gateway-s3.go
@@ -66,8 +66,10 @@ EXAMPLES:
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_SECRET_KEY{{.AssignmentOperator}}secretkey
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_DRIVES{{.AssignmentOperator}}"/mnt/drive1,/mnt/drive2,/mnt/drive3,/mnt/drive4"
      {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_EXCLUDE{{.AssignmentOperator}}"bucket1/*,*.png"
-     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_EXPIRY{{.AssignmentOperator}}40
-     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_QUOTA{{.AssignmentOperator}}80
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_QUOTA{{.AssignmentOperator}}90
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_AFTER{{.AssignmentOperator}}3
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_WATERMARK_LOW{{.AssignmentOperator}}75
+     {{.Prompt}} {{.EnvVarSetCommand}} MINIO_CACHE_WATERMARK_HIGH{{.AssignmentOperator}}85
      {{.Prompt}} {{.HelpName}}
 `
 
diff --git a/docs/disk-caching/DESIGN.md b/docs/disk-caching/DESIGN.md
index 11cfb32c6..f5a87e1d2 100644
--- a/docs/disk-caching/DESIGN.md
+++ b/docs/disk-caching/DESIGN.md
@@ -11,30 +11,32 @@ minio gateway <name> -h
   CACHE:
      MINIO_CACHE_DRIVES: List of mounted cache drives or directories delimited by ","
      MINIO_CACHE_EXCLUDE: List of cache exclusion patterns delimited by ","
-     MINIO_CACHE_EXPIRY: Cache expiry duration in days
      MINIO_CACHE_QUOTA: Maximum permitted usage of the cache in percentage (0-100).
      MINIO_CACHE_AFTER: Minimum number of access before caching an object.
+     MINIO_CACHE_WATERMARK_LOW: % of cache quota at which cache eviction stops
+     MINIO_CACHE_WATERMARK_HIGH: % of cache quota at which cache eviction starts
+
 
 ...
 ...
 
   7. Start MinIO gateway to s3 with edge caching enabled on '/mnt/drive1', '/mnt/drive2' and '/mnt/export1 ... /mnt/export24',
-     exclude all objects under 'mybucket', exclude all objects with '.pdf' as extension
-     with expiry up to 40 days. Cache only those objects accessed atleast 3 times.
+     exclude all objects under 'mybucket', exclude all objects with '.pdf' as extension. Cache only those objects accessed atleast 3 times. Garbage collection triggers in at high water mark (i.e. cache disk usage reaches 90% of cache quota) or at 72% and evicts oldest objects by access time until low watermark is reached ( 70% of cache quota) , i.e. 63% of disk usage.
      $ export MINIO_CACHE_DRIVES="/mnt/drive1,/mnt/drive2,/mnt/export{1..24}"
      $ export MINIO_CACHE_EXCLUDE="mybucket/*,*.pdf"
-     $ export MINIO_CACHE_EXPIRY=40
      $ export MINIO_CACHE_QUOTA=80
      $ export MINIO_CACHE_AFTER=3
+     $ export MINIO_CACHE_WATERMARK_LOW=70
+     $ export MINIO_CACHE_WATERMARK_HIGH=90
+
      $ minio gateway s3
 ```
 
 ## Assumptions
 
-- Disk cache size defaults to 80% of your drive capacity.
+- Disk cache quota defaults to 80% of your drive capacity.
 - The cache drives are required to be a filesystem mount point with [`atime`](http://kerolasa.github.io/filetimes.html) support to be enabled on the drive. Alternatively writable directories with atime support can be specified in MINIO_CACHE_DRIVES
-- Expiration of each cached entry takes user provided expiry as a hint, and defaults to 90 days if not provided.
-- Garbage collection sweep of the expired cache entries happens whenever cache usage is > 80% of drive capacity, GC continues until sufficient disk space is reclaimed.
+- Garbage collection sweep happens whenever cache disk usage reaches high watermark with respect to the configured cache quota , GC evicts least recently accessed objects until cache low watermark is reached with respect to the configured cache quota. Garbage collection runs a cache eviction sweep at 30 minute intervals.
 - An object is only cached when drive has sufficient disk space.
 
 ## Behavior
diff --git a/docs/disk-caching/README.md b/docs/disk-caching/README.md
index b56095c1a..172c26d1e 100644
--- a/docs/disk-caching/README.md
+++ b/docs/disk-caching/README.md
@@ -13,17 +13,19 @@ Install MinIO - [MinIO Quickstart Guide](https://docs.min.io/docs/minio-quicksta
 
 ### 2. Run MinIO gateway with cache
 
-Disk caching can be enabled by setting the `cache` environment variables for MinIO gateway . `cache` environment variables takes the mounted drive(s) or directory paths, cache expiry duration (in days) and any wildcard patterns to exclude from being cached.
+Disk caching can be enabled by setting the `cache` environment variables for MinIO gateway . `cache` environment variables takes the mounted drive(s) or directory paths, any wildcard patterns to exclude from being cached,low and high watermarks for garbage collection and the minimum accesses before caching an object.
 
-Following example uses `/mnt/drive1`, `/mnt/drive2` ,`/mnt/cache1` ... `/mnt/cache3` for caching, with expiry up to 90 days while excluding all objects under bucket `mybucket` and all objects with '.pdf' as extension while starting a s3 gateway setup. Objects are cached if they have been accessed three times or more.Cache max usage is restricted to 80% of disk capacity in this example.
+Following example uses `/mnt/drive1`, `/mnt/drive2` ,`/mnt/cache1` ... `/mnt/cache3` for caching, while excluding all objects under bucket `mybucket` and all objects with '.pdf' as extension on a s3 gateway setup. Objects are cached if they have been accessed three times or more.Cache max usage is restricted to 80% of disk capacity in this example. Garbage collection is triggered when high watermark is reached - i.e. at 72% of cache disk usage and clears least recently accessed entries until the disk usage drops to low watermark - i.e. cache disk usage drops to 56% (70% of 80% quota)
 
 ```bash
 export MINIO_CACHE="on"
 export MINIO_CACHE_DRIVES="/mnt/drive1,/mnt/drive2,/mnt/cache{1...3}"
-export MINIO_CACHE_EXPIRY=90
 export MINIO_CACHE_EXCLUDE="*.pdf,mybucket/*"
 export MINIO_CACHE_QUOTA=80
 export MINIO_CACHE_AFTER=3
+export MINIO_CACHE_WATERMARK_LOW=70
+export MINIO_CACHE_WATERMARK_HIGH=90
+
 minio gateway s3
 ```