xl/ListFiles: return as many objects as requested. (#1383)

* xl/ListFiles: return as many objects as requested and take care of eof (#1361)

* xl/ListFiles: fix review comments.

* xl/ListFiles: Add windows filepath translation.

* xl/ListFiles: Use slashSeparator instead of "/". Remove filepath.FromSlash() as golang-windows takes care of it automatically.

fs-dir-common.go

@@ -18,7 +18,7 @@ package main
 
 import (
 	"os"
-	"path/filepath"
+	"path"
 	"sort"
 	"strings"
 	"time"
@@ -88,11 +88,11 @@ func treeWalk(bucketDir, prefixDir, entryPrefixMatch, marker string, recursive b
 	direntToFileInfo := func(dirent fsDirent) (FileInfo, error) {
 		fileInfo := FileInfo{}
 		// Convert to full object name.
-		fileInfo.Name = filepath.Join(prefixDir, dirent.name)
+		fileInfo.Name = path.Join(prefixDir, dirent.name)
 		if dirent.modTime.IsZero() && dirent.size == 0 {
 			// ModifiedTime and Size are zero, Stat() and figure out
 			// the actual values that need to be set.
-			fi, err := os.Stat(filepath.Join(bucketDir, prefixDir, dirent.name))
+			fi, err := os.Stat(path.Join(bucketDir, prefixDir, dirent.name))
 			if err != nil {
 				return FileInfo{}, err
 			}
@@ -119,10 +119,10 @@ func treeWalk(bucketDir, prefixDir, entryPrefixMatch, marker string, recursive b
 	var markerBase, markerDir string
 	if marker != "" {
 		// Ex: if marker="four/five.txt", markerDir="four/" markerBase="five.txt"
-		markerSplit := strings.SplitN(marker, string(os.PathSeparator), 2)
+		markerSplit := strings.SplitN(marker, slashSeparator, 2)
 		markerDir = markerSplit[0]
 		if len(markerSplit) == 2 {
-			markerDir += string(os.PathSeparator)
+			markerDir += slashSeparator
 			markerBase = markerSplit[1]
 		}
 	}
@@ -140,7 +140,7 @@ func treeWalk(bucketDir, prefixDir, entryPrefixMatch, marker string, recursive b
 	}
 
 	// scandir returns entries that begins with entryPrefixMatch
-	dirents, err := scandir(filepath.Join(bucketDir, prefixDir), prefixMatchFn, true)
+	dirents, err := scandir(path.Join(bucketDir, prefixDir), prefixMatchFn, true)
 	if err != nil {
 		send(treeWalkResult{err: err})
 		return false
@@ -167,7 +167,7 @@ func treeWalk(bucketDir, prefixDir, entryPrefixMatch, marker string, recursive b
 				markerArg = markerBase
 			}
 			*count--
-			if !treeWalk(bucketDir, filepath.Join(prefixDir, dirent.name), "", markerArg, recursive, send, count) {
+			if !treeWalk(bucketDir, path.Join(prefixDir, dirent.name), "", markerArg, recursive, send, count) {
 				return false
 			}
 			continue
@@ -200,7 +200,7 @@ func startTreeWalk(fsPath, bucket, prefix, marker string, recursive bool) *treeW
 	walkNotify := treeWalker{ch: ch}
 	entryPrefixMatch := prefix
 	prefixDir := ""
-	lastIndex := strings.LastIndex(prefix, string(os.PathSeparator))
+	lastIndex := strings.LastIndex(prefix, slashSeparator)
 	if lastIndex != -1 {
 		entryPrefixMatch = prefix[lastIndex+1:]
 		prefixDir = prefix[:lastIndex+1]
@@ -222,7 +222,7 @@ func startTreeWalk(fsPath, bucket, prefix, marker string, recursive bool) *treeW
 				return false
 			}
 		}
-		bucketDir := filepath.Join(fsPath, bucket)
+		bucketDir := path.Join(fsPath, bucket)
 		treeWalk(bucketDir, prefixDir, entryPrefixMatch, marker, recursive, send, &count)
 	}()
 	return &walkNotify

fs-dir-nix.go

@@ -20,7 +20,7 @@ package main
 
 import (
 	"os"
-	"path/filepath"
+	"path"
 	"runtime"
 	"sort"
 	"syscall"
@@ -88,7 +88,7 @@ func parseDirents(dirPath string, buf []byte) []fsDirent {
 		case syscall.DT_UNKNOWN:
 			// On Linux XFS does not implement d_type for on disk
 			// format << v5. Fall back to Stat().
-			if fi, err := os.Stat(filepath.Join(dirPath, name)); err == nil {
+			if fi, err := os.Stat(path.Join(dirPath, name)); err == nil {
 				mode = fi.Mode()
 			} else {
 				// Caller listing would fail, if Stat failed but we
@@ -129,7 +129,7 @@ func scandir(dirPath string, filter func(fsDirent) bool, namesOnly bool) ([]fsDi
 		}
 		for _, dirent := range parseDirents(dirPath, buf[:nbuf]) {
 			if !namesOnly {
-				dirent.name = filepath.Join(dirPath, dirent.name)
+				dirent.name = path.Join(dirPath, dirent.name)
 			}
 			if dirent.IsDir() {
 				dirent.name += string(os.PathSeparator)

fs-dir-others.go

@@ -21,7 +21,7 @@ package main
 import (
 	"io"
 	"os"
-	"path/filepath"
+	"path"
 	"sort"
 )
 
@@ -53,10 +53,11 @@ func scandir(dirPath string, filter func(fsDirent) bool, namesOnly bool) ([]fsDi
 				mode:    fi.Mode(),
 			}
 			if !namesOnly {
-				dirent.name = filepath.Join(dirPath, dirent.name)
+				dirent.name = path.Join(dirPath, dirent.name)
 			}
 			if dirent.IsDir() {
-				dirent.name += string(os.PathSeparator)
+				// append "/" instead of "\" so that sorting is done as expected.
+				dirent.name += slashSeparator
 			}
 			if filter == nil || filter(dirent) {
 				dirents = append(dirents, dirent)

fs.go

@@ -19,6 +19,7 @@ package main
 import (
 	"io"
 	"os"
+	slashpath "path"
 	"path/filepath"
 	"strings"
 	"sync"
@@ -460,10 +461,9 @@ func (s fsStorage) ListFiles(volume, prefix, marker string, recursive bool, coun
 	}
 
 	// Verify if prefix exists.
-	prefixDir := filepath.Dir(filepath.FromSlash(prefix))
-	prefixRootDir := filepath.Join(volumeDir, prefixDir)
-	var status bool
-	if status, err = isDirExist(prefixRootDir); !status {
+	prefixDir := slashpath.Dir(prefix)
+	prefixRootDir := slashpath.Join(volumeDir, prefixDir)
+	if status, err := isDirExist(prefixRootDir); !status {
 		if err == nil {
 			// Prefix does not exist, not an error just respond empty list response.
 			return nil, true, nil
@@ -487,7 +487,7 @@ func (s fsStorage) ListFiles(volume, prefix, marker string, recursive bool, coun
 	// popTreeWalker returns the channel from which rest of the objects can be retrieved.
 	walker := s.lookupTreeWalk(listParams{volume, recursive, marker, prefix})
 	if walker == nil {
-		walker = startTreeWalk(s.diskPath, volume, filepath.FromSlash(prefix), filepath.FromSlash(marker), recursive)
+		walker = startTreeWalk(filepath.ToSlash(s.diskPath), volume, prefix, marker, recursive)
 	}
 	nextMarker := ""
 	log.Debugf("Reading from the tree walk channel has begun.")

object_api_suite_test.go

@@ -163,24 +163,21 @@ func testPaging(c *check.C, create func() objectAPI) {
 		key := "obj" + strconv.Itoa(i)
 		_, err = obj.PutObject("bucket", key, int64(len("The specified multipart upload does not exist. The upload ID might be invalid, or the multipart upload might have been aborted or completed.")), bytes.NewBufferString("The specified multipart upload does not exist. The upload ID might be invalid, or the multipart upload might have been aborted or completed."), nil)
 		c.Assert(err, check.IsNil)
-		/*
-			result, err = obj.ListObjects("bucket", "", "", "", 5)
-			c.Assert(err, check.IsNil)
-			c.Assert(len(result.Objects), check.Equals, i+1)
-			c.Assert(result.IsTruncated, check.Equals, false)
-		*/
+
+		result, err = obj.ListObjects("bucket", "", "", "", 5)
+		c.Assert(err, check.IsNil)
+		c.Assert(len(result.Objects), check.Equals, i+1)
+		c.Assert(result.IsTruncated, check.Equals, false)
 	}
 	// check after paging occurs pages work
 	for i := 6; i <= 10; i++ {
 		key := "obj" + strconv.Itoa(i)
 		_, err = obj.PutObject("bucket", key, int64(len("The specified multipart upload does not exist. The upload ID might be invalid, or the multipart upload might have been aborted or completed.")), bytes.NewBufferString("The specified multipart upload does not exist. The upload ID might be invalid, or the multipart upload might have been aborted or completed."), nil)
 		c.Assert(err, check.IsNil)
-		/*
-			result, err = obj.ListObjects("bucket", "obj", "", "", 5)
-			c.Assert(err, check.IsNil)
-			c.Assert(len(result.Objects), check.Equals, 5)
-			c.Assert(result.IsTruncated, check.Equals, true)
-		*/
+		result, err = obj.ListObjects("bucket", "obj", "", "", 5)
+		c.Assert(err, check.IsNil)
+		c.Assert(len(result.Objects), check.Equals, 5)
+		c.Assert(result.IsTruncated, check.Equals, true)
 	}
 	// check paging with prefix at end returns less objects
 	{
@@ -188,11 +185,9 @@ func testPaging(c *check.C, create func() objectAPI) {
 		c.Assert(err, check.IsNil)
 		_, err = obj.PutObject("bucket", "newPrefix2", int64(len("The specified multipart upload does not exist. The upload ID might be invalid, or the multipart upload might have been aborted or completed.")), bytes.NewBufferString("The specified multipart upload does not exist. The upload ID might be invalid, or the multipart upload might have been aborted or completed."), nil)
 		c.Assert(err, check.IsNil)
-		/*
-			result, err = obj.ListObjects("bucket", "new", "", "", 5)
-			c.Assert(err, check.IsNil)
-			c.Assert(len(result.Objects), check.Equals, 2)
-		*/
+		result, err = obj.ListObjects("bucket", "new", "", "", 5)
+		c.Assert(err, check.IsNil)
+		c.Assert(len(result.Objects), check.Equals, 2)
 	}
 
 	// check ordering of pages
@@ -232,13 +227,13 @@ func testPaging(c *check.C, create func() objectAPI) {
 
 	// check results with Marker
 	{
-		/*
-			result, err = obj.ListObjects("bucket", "", "newPrefix", "", 3)
-			c.Assert(err, check.IsNil)
-			c.Assert(result.Objects[0].Name, check.Equals, "newPrefix2")
-			c.Assert(result.Objects[1].Name, check.Equals, "obj0")
-			c.Assert(result.Objects[2].Name, check.Equals, "obj1")
-		*/
+
+		result, err = obj.ListObjects("bucket", "", "newPrefix", "", 3)
+		c.Assert(err, check.IsNil)
+		c.Assert(result.Objects[0].Name, check.Equals, "newPrefix2")
+		c.Assert(result.Objects[1].Name, check.Equals, "obj0")
+		c.Assert(result.Objects[2].Name, check.Equals, "obj1")
+
 	}
 	// check ordering of results with prefix
 	{

xl-v1-errors.go

@@ -29,3 +29,6 @@ var errNumDisks = errors.New("Invalid number of disks provided, should be always
 
 // errModTime - returned for missing file modtime.
 var errModTime = errors.New("Missing 'file.modTime' in metadata")
+
+// errUnexpected - returned for any unexpected error.
+var errUnexpected = errors.New("Unexpected error - please report at https://github.com/minio/minio/issues")

xl-v1.go

@@ -20,7 +20,6 @@ import (
 	"fmt"
 	"os"
 	slashpath "path"
-	"sort"
 	"strings"
 	"sync"
 
@@ -416,58 +415,99 @@ func (xl XL) ListFiles(volume, prefix, marker string, recursive bool, count int)
 		if isLeaf {
 			// For leaf for now we just point to the first block, make it
 			// dynamic in future based on the availability of storage disks.
-			markerPath = slashpath.Join(marker, "part.0")
+			markerPath = slashpath.Join(marker, metadataFile)
 		}
 	}
 
-	// List files.
-	fsFilesInfo, eof, err = disk.ListFiles(volume, prefix, markerPath, recursive, count)
-	if err != nil {
-		log.WithFields(logrus.Fields{
-			"volume":    volume,
-			"prefix":    prefix,
-			"marker":    markerPath,
-			"recursive": recursive,
-			"count":     count,
-		}).Debugf("ListFiles failed with %s", err)
-		return nil, true, err
-	}
-
-	for _, fsFileInfo := range fsFilesInfo {
-		// Skip metadata files.
-		if strings.HasSuffix(fsFileInfo.Name, metadataFile) {
-			continue
+	for {
+		fsFilesInfo, eof, err = disk.ListFiles(volume, prefix, markerPath, recursive, count)
+		if err != nil {
+			log.WithFields(logrus.Fields{
+				"volume":    volume,
+				"prefix":    prefix,
+				"marker":    markerPath,
+				"recursive": recursive,
+				"count":     count,
+			}).Debugf("ListFiles failed with %s", err)
+			return nil, true, err
+		}
+		for _, fsFileInfo := range fsFilesInfo {
+			// Skip metadata files.
+			if strings.HasSuffix(fsFileInfo.Name, metadataFile) {
+				continue
+			}
+			var fileInfo FileInfo
+			var isLeaf bool
+			if fsFileInfo.Mode.IsDir() {
+				isLeaf = xl.isLeafDirectory(volume, fsFileInfo.Name)
+			}
+			if isLeaf || !fsFileInfo.Mode.IsDir() {
+				// Extract the parent of leaf directory or file to get the
+				// actual name.
+				path := slashpath.Dir(fsFileInfo.Name)
+				fileInfo, err = xl.extractFileInfo(volume, path)
+				if err != nil {
+					log.WithFields(logrus.Fields{
+						"volume": volume,
+						"path":   path,
+					}).Debugf("extractFileInfo failed with %s", err)
+					// For a leaf directory, if err is FileNotFound then
+					// perhaps has a missing metadata. Ignore it and let
+					// healing finish its job it will become available soon.
+					if err == errFileNotFound {
+						continue
+					}
+					// For any other errors return to the caller.
+					return nil, true, err
+				}
+			} else {
+				fileInfo = fsFileInfo
+			}
+			filesInfo = append(filesInfo, fileInfo)
+			count--
+			if count == 0 {
+				break
+			}
 		}
-		var fileInfo FileInfo
-		var isLeaf bool
-		if fsFileInfo.Mode.IsDir() {
-			isLeaf = xl.isLeafDirectory(volume, fsFileInfo.Name)
+		lenFsFilesInfo := len(fsFilesInfo)
+		if lenFsFilesInfo != 0 {
+			// markerPath for the next disk.ListFiles() iteration.
+			markerPath = fsFilesInfo[lenFsFilesInfo-1].Name
 		}
-		if isLeaf || !fsFileInfo.Mode.IsDir() {
-			// Extract the parent of leaf directory or file to get the
-			// actual name.
-			path := slashpath.Dir(fsFileInfo.Name)
-			fileInfo, err = xl.extractFileInfo(volume, path)
+		if count == 0 && recursive && !strings.HasSuffix(markerPath, metadataFile) {
+			// If last entry is not part.json then loop once more to check if we
+			// have reached eof.
+			fsFilesInfo, eof, err = disk.ListFiles(volume, prefix, markerPath, recursive, 1)
 			if err != nil {
 				log.WithFields(logrus.Fields{
-					"volume": volume,
-					"path":   path,
-				}).Debugf("extractFileInfo failed with %s", err)
-				// For a leaf directory, if err is FileNotFound then
-				// perhaps has a missing metadata. Ignore it and let
-				// healing finish its job it will become available soon.
-				if err == errFileNotFound {
-					continue
-				}
-				// For any other errors return to the caller.
+					"volume":    volume,
+					"prefix":    prefix,
+					"marker":    markerPath,
+					"recursive": recursive,
+					"count":     1,
+				}).Debugf("ListFiles failed with %s", err)
 				return nil, true, err
 			}
-		} else {
-			fileInfo = fsFileInfo
+			if !eof {
+				// part.N and part.json are always in pairs and hence this
+				// entry has to be part.json. If not better to manually investigate
+				// and fix it.
+				// For the next ListFiles() call we can safely assume that the
+				// marker is "object/part.json"
+				if !strings.HasSuffix(fsFilesInfo[0].Name, metadataFile) {
+					log.WithFields(logrus.Fields{
+						"volume":          volume,
+						"prefix":          prefix,
+						"fsFileInfo.Name": fsFilesInfo[0].Name,
+					}).Debugf("ListFiles failed with %s, expected %s to be a part.json file.", err, fsFilesInfo[0].Name)
+					return nil, true, errUnexpected
+				}
+			}
+		}
+		if count == 0 || eof {
+			break
 		}
-		filesInfo = append(filesInfo, fileInfo)
 	}
-	sort.Sort(byFileInfoName(filesInfo))
 	return filesInfo, eof, nil
 }