Merge pull request #153 from 0chain/fscache

Add minio cache for gateway

Author: dabasov

cmd/bucket-listobjects-handlers.go

@@ -19,9 +19,13 @@ package cmd
 
 import (
 	"context"
+	"log"
 	"net/http"
+	"sort"
 	"strconv"
 	"strings"
+	"sync"
+	"time"
 
 	"github.com/gorilla/mux"
 	"github.com/minio/minio/internal/logger"
@@ -46,12 +50,91 @@ func concurrentDecryptETag(ctx context.Context, objects []ObjectInfo) {
 	g.Wait()
 }
 
+func mergeListObjects(l1, l2 []ObjectInfo) []ObjectInfo {
+	mergedMap := make(map[string]ObjectInfo)
+
+	// Helper function to add/update map entries
+	addOrUpdate := func(obj ObjectInfo) {
+		if existingObj, found := mergedMap[obj.Name]; !found || obj.ModTime.After(existingObj.ModTime) {
+			mergedMap[obj.Name] = obj
+		}
+	}
+	for _, obj := range l1 {
+		addOrUpdate(obj)
+	}
+	for _, obj := range l2 {
+		addOrUpdate(obj)
+	}
+
+	mergedList := make([]ObjectInfo, 0, len(mergedMap))
+	for _, obj := range mergedMap {
+		mergedList = append(mergedList, obj)
+	}
+
+	return mergedList
+}
+
+func mergePrefixes(l1, l2 []string) []string {
+	mergedMap := make(map[string]bool)
+
+	// Helper function to add/update map entries
+	addOrUpdate := func(pre string) {
+		if _, found := mergedMap[pre]; !found {
+			mergedMap[pre] = true
+		}
+	}
+	for _, pre := range l1 {
+		addOrUpdate(pre)
+	}
+	for _, pre := range l2 {
+		addOrUpdate(pre)
+	}
+
+	mergedList := make([]string, 0, len(mergedMap))
+	for pre, _ := range mergedMap {
+		mergedList = append(mergedList, pre)
+	}
+
+	return mergedList
+}
+
+func limitMergeObjects(mergeObjects []ObjectInfo, mergePrefixes []string, maxKeys int) ([]ObjectInfo, []string, string) {
+	objPrefixMap := map[string]ObjectInfo{}
+	for _, ob := range mergeObjects {
+		objPrefixMap[ob.Name] = ob
+	}
+	for _, pre := range mergePrefixes {
+		objPrefixMap[pre] = ObjectInfo{IsDir: true}
+	}
+
+	keys := make([]string, 0, len(objPrefixMap))
+	for key := range objPrefixMap {
+		keys = append(keys, key)
+	}
+	sort.Strings(keys)
+	limitedObjs := []ObjectInfo{}
+	limitedPrefixes := []string{}
+	nextMarker := ""
+	for i, key := range keys {
+		if objPrefixMap[key].IsDir {
+			limitedPrefixes = append(limitedPrefixes, key)
+		} else {
+			limitedObjs = append(limitedObjs, objPrefixMap[key])
+		}
+		if i >= (maxKeys - 1) {
+			nextMarker = key
+			break
+		}
+	}
+	return limitedObjs, limitedPrefixes, nextMarker
+}
+
 // Validate all the ListObjects query arguments, returns an APIErrorCode
 // if one of the args do not meet the required conditions.
 // Special conditions required by MinIO server are as below
-// - delimiter if set should be equal to '/', otherwise the request is rejected.
-// - marker if set should have a common prefix with 'prefix' param, otherwise
-//   the request is rejected.
+//   - delimiter if set should be equal to '/', otherwise the request is rejected.
+//   - marker if set should have a common prefix with 'prefix' param, otherwise
+//     the request is rejected.
 func validateListObjectsArgs(marker, delimiter, encodingType string, maxKeys int) APIErrorCode {
 	// Max keys cannot be negative.
 	if maxKeys < 0 {
@@ -200,6 +283,11 @@ func (api objectAPIHandlers) ListObjectsV2MHandler(w http.ResponseWriter, r *htt
 // NOTE: It is recommended that this API to be used for application development.
 // MinIO continues to support ListObjectsV1 for supporting legacy tools.
 func (api objectAPIHandlers) ListObjectsV2Handler(w http.ResponseWriter, r *http.Request) {
+	st := time.Now()
+	defer func() {
+		elapsed := time.Since(st).Milliseconds()
+		log.Printf("ListObjectsV2Handler took %d ms\n", elapsed)
+	}()
 	ctx := newContext(r, w, "ListObjectsV2")
 
 	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
@@ -235,9 +323,17 @@ func (api objectAPIHandlers) ListObjectsV2Handler(w http.ResponseWriter, r *http
 	}
 
 	var (
-		listObjectsV2Info ListObjectsV2Info
-		err               error
+		listObjectsV2Info      ListObjectsV2Info
+		listObjectsV2InfoCache ListObjectsV2Info
+		err                    error
+		errC                   error
 	)
+	listObjectsV2Cache := objectAPI.ListObjectsV2
+	cacheEnabled := false
+	if api.CacheAPI() != nil {
+		cacheEnabled = true
+		listObjectsV2Cache = api.CacheAPI().ListObjectsV2
+	}
 
 	if r.Header.Get(xMinIOExtract) == "true" && strings.Contains(prefix, archivePattern) {
 		// Inititate a list objects operation inside a zip file based in the input params
@@ -246,13 +342,38 @@ func (api objectAPIHandlers) ListObjectsV2Handler(w http.ResponseWriter, r *http
 		// Inititate a list objects operation based on the input params.
 		// On success would return back ListObjectsInfo object to be
 		// marshaled into S3 compatible XML header.
-		listObjectsV2Info, err = objectAPI.ListObjectsV2(ctx, bucket, prefix, token, delimiter, maxKeys, fetchOwner, startAfter)
+		var wg sync.WaitGroup
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			listObjectsV2Info, err = objectAPI.ListObjectsV2(ctx, bucket, prefix, token, delimiter, maxKeys, fetchOwner, startAfter)
+		}()
+		go func() {
+			defer wg.Done()
+			if cacheEnabled {
+				stc := time.Now()
+				listObjectsV2InfoCache, errC = listObjectsV2Cache(ctx, bucket, prefix, token, delimiter, maxKeys, fetchOwner, startAfter)
+				elap := time.Since(stc)
+				log.Println("ListV2 object cache time", elap)
+			}
+		}()
+		wg.Wait()
 	}
-	if err != nil {
+	if err != nil || errC != nil {
 		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
 		return
 	}
 
+	mergeObjects := mergeListObjects(listObjectsV2Info.Objects, listObjectsV2InfoCache.Objects)
+	mergePrefixes := mergePrefixes(listObjectsV2Info.Prefixes, listObjectsV2InfoCache.Prefixes)
+	limitedObjects, limitedPrefix, nextMarker := limitMergeObjects(mergeObjects, mergePrefixes, maxKeys)
+	listObjectsV2Info.Objects = limitedObjects
+	listObjectsV2Info.Prefixes = limitedPrefix
+	if nextMarker != "" {
+		listObjectsV2Info.NextContinuationToken = nextMarker
+		listObjectsV2Info.IsTruncated = true
+	}
+
 	concurrentDecryptETag(ctx, listObjectsV2Info.Objects)
 
 	response := generateListObjectsV2Response(bucket, prefix, token, listObjectsV2Info.NextContinuationToken, startAfter,
@@ -306,8 +427,12 @@ func proxyRequestByNodeIndex(ctx context.Context, w http.ResponseWriter, r *http
 // This implementation of the GET operation returns some or all (up to 1000)
 // of the objects in a bucket. You can use the request parameters as selection
 // criteria to return a subset of the objects in a bucket.
-//
 func (api objectAPIHandlers) ListObjectsV1Handler(w http.ResponseWriter, r *http.Request) {
+	st := time.Now()
+	defer func() {
+		elapsed := time.Since(st).Milliseconds()
+		log.Printf("ListObjectsV1Handler took %d ms\n", elapsed)
+	}()
 	ctx := newContext(r, w, "ListObjectsV1")
 
 	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
@@ -340,15 +465,52 @@ func (api objectAPIHandlers) ListObjectsV1Handler(w http.ResponseWriter, r *http
 	}
 
 	listObjects := objectAPI.ListObjects
-
+	listObjectsCache := objectAPI.ListObjects
+	cacheEnabled := false
+	if api.CacheAPI() != nil {
+		cacheEnabled = true
+		listObjectsCache = api.CacheAPI().ListObjects
+	}
 	// Inititate a list objects operation based on the input params.
 	// On success would return back ListObjectsInfo object to be
 	// marshaled into S3 compatible XML header.
-	listObjectsInfo, err := listObjects(ctx, bucket, prefix, marker, delimiter, maxKeys)
-	if err != nil {
+	var (
+		listObjectsInfo      ListObjectsInfo
+		listObjectsInfoCache ListObjectsInfo
+		err                  error
+		errC                 error
+	)
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		listObjectsInfo, err = listObjects(ctx, bucket, prefix, marker, delimiter, maxKeys)
+	}()
+	go func() {
+		defer wg.Done()
+		if cacheEnabled {
+			stc := time.Now()
+			listObjectsInfoCache, errC = listObjectsCache(ctx, bucket, prefix, marker, delimiter, maxKeys)
+			elap := time.Since(stc)
+			log.Println("ListV1 object cache time", elap)
+		}
+	}()
+
+	wg.Wait()
+
+	if err != nil || errC != nil {
 		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
 		return
 	}
+	mergeObjects := mergeListObjects(listObjectsInfo.Objects, listObjectsInfoCache.Objects)
+	mergePrefixes := mergePrefixes(listObjectsInfo.Prefixes, listObjectsInfoCache.Prefixes)
+	limitedObjects, limitedPrefix, nextMarker := limitMergeObjects(mergeObjects, mergePrefixes, maxKeys)
+	listObjectsInfo.Objects = limitedObjects
+	listObjectsInfo.Prefixes = limitedPrefix
+	if nextMarker != "" {
+		listObjectsInfo.NextMarker = nextMarker
+		listObjectsInfo.IsTruncated = true
+	}
 
 	concurrentDecryptETag(ctx, listObjectsInfo.Objects)
 

cmd/disk-cache-backend.go

@@ -27,6 +27,7 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"log"
 	"net/http"
 	"os"
 	"path"
@@ -205,7 +206,7 @@ func newDiskCache(ctx context.Context, dir string, config cache.Config) (*diskCa
 		commitWriteback:    config.CacheCommitMode == CommitWriteBack,
 		commitWritethrough: config.CacheCommitMode == CommitWriteThrough,
 
-		retryWritebackCh: make(chan ObjectInfo, 10000),
+		retryWritebackCh: make(chan ObjectInfo, 100000),
 		online:           1,
 		pool: sync.Pool{
 			New: func() interface{} {
@@ -218,7 +219,22 @@ func newDiskCache(ctx context.Context, dir string, config cache.Config) (*diskCa
 	go cache.purgeWait(ctx)
 	go cache.cleanupStaleUploads(ctx)
 	if cache.commitWriteback {
-		go cache.scanCacheWritebackFailures(ctx)
+		go func() {
+			tickInterval := time.Duration(config.WriteBackInterval) * time.Second
+			log.Println("write back time interval", tickInterval)
+			ticker := time.NewTicker(tickInterval)
+			defer ticker.Stop()
+			defer close(cache.retryWritebackCh)
+			for {
+				select {
+				case <-ticker.C:
+					cache.scanCacheWritebackFailures(ctx)
+				case <-ctx.Done():
+					return
+				}
+			}
+		}()
+		//go cache.scanCacheWritebackFailures(ctx)
 	}
 	cache.diskSpaceAvailable(0) // update if cache usage is already high.
 	cache.NewNSLockFn = func(cachePath string) RWLocker {
@@ -1225,7 +1241,8 @@ func (c *diskCache) Exists(ctx context.Context, bucket, object string) bool {
 
 // queues writeback upload failures on server startup
 func (c *diskCache) scanCacheWritebackFailures(ctx context.Context) {
-	defer close(c.retryWritebackCh)
+	log.Println("scan cache write back failures")
+	//defer close(c.retryWritebackCh) // don't close the channel
 	filterFn := func(name string, typ os.FileMode) error {
 		if name == minioMetaBucket {
 			// Proceed to next file.