Merge pull request #1245 from hackintoshrao/fs-bucket-tests

api: ListObject - tests, benchmark, optimization

pkg/fs/api_suite_test.go

@@ -24,7 +24,6 @@ import (
 	"encoding/xml"
 	"math/rand"
 	"strconv"
-	"time"
 
 	"gopkg.in/check.v1"
 )
@@ -216,7 +215,6 @@ func testPaging(c *check.C, create func() Filesystem) {
 		c.Assert(len(result.Objects), check.Equals, 1)
 		c.Assert(result.Prefixes[0], check.Equals, "this/is/also/")
 	}
-	time.Sleep(time.Second)
 
 	// check delimited results with delimiter without prefix
 	{

pkg/fs/dir.go

@@ -26,11 +26,11 @@ import (
 )
 
 const (
-	// listObjectsLimit - maximum list objects limit
+	// ListObjectsLimit - maximum list objects limit.
 	listObjectsLimit = 1000
 )
 
-// isDirEmpty - returns whether given directory is empty or not
+// IsDirEmpty - returns whether given directory is empty or not.
 func isDirEmpty(dirname string) (status bool, err error) {
 	f, err := os.Open(dirname)
 	if err == nil {
@@ -44,7 +44,7 @@ func isDirEmpty(dirname string) (status bool, err error) {
 	return
 }
 
-// isDirExist - returns whether given directory is exist or not
+// IsDirExist - returns whether given directory is exist or not.
 func isDirExist(dirname string) (status bool, err error) {
 	fi, err := os.Lstat(dirname)
 	if err == nil {
@@ -54,11 +54,17 @@ func isDirExist(dirname string) (status bool, err error) {
 	return
 }
 
-// byName implements sort.Interface for sorting os.FileInfo list
+// ByName implements sort.Interface for sorting os.FileInfo list.
 type byName []os.FileInfo
 
-func (f byName) Len() int      { return len(f) }
+func (f byName) Len() int {
-func (f byName) Swap(i, j int) { f[i], f[j] = f[j], f[i] }
+	return len(f)
+}
+
+func (f byName) Swap(i, j int) {
+	f[i], f[j] = f[j], f[i]
+}
+
 func (f byName) Less(i, j int) bool {
 	n1 := f[i].Name()
 	if f[i].IsDir() {
@@ -73,7 +79,7 @@ func (f byName) Less(i, j int) bool {
 	return n1 < n2
 }
 
-// ObjectInfo - object info
+// ObjectInfo - object info.
 type ObjectInfo struct {
 	Bucket       string
 	Name         string
@@ -85,31 +91,61 @@ type ObjectInfo struct {
 	Err          error
 }
 
-// readDir - read 'scanDir' directory.  It returns list of ObjectInfo where
+// Using sort.Search() internally to jump to the file entry containing the prefix.
-//           each object name is appended with 'namePrefix'
+func searchFileInfos(fileInfos []os.FileInfo, x string) int {
-func readDir(scanDir, namePrefix string) (objInfos []ObjectInfo) {
+	processFunc := func(i int) bool {
+		return fileInfos[i].Name() >= x
+	}
+	return sort.Search(len(fileInfos), processFunc)
+}
+
+// ReadDir - read 'scanDir' directory.  It returns list of ObjectInfo.
+// Each object name is appended with 'namePrefix'.
+func readDir(scanDir, namePrefix, queryPrefix string, isFirst bool) (objInfos []ObjectInfo) {
 	f, err := os.Open(scanDir)
 	if err != nil {
 		objInfos = append(objInfos, ObjectInfo{Err: err})
 		return
 	}
-
 	fis, err := f.Readdir(-1)
 	if err != nil {
 		f.Close()
 		objInfos = append(objInfos, ObjectInfo{Err: err})
 		return
 	}
-
+	// Close the directory.
-	// Close the directory
 	f.Close()
-
 	// Sort files by Name.
 	sort.Sort(byName(fis))
 
-	// Populate []ObjectInfo from []FileInfo
+	var prefixIndex int
+	// Searching for entries with objectName containing  prefix.
+	// Binary search is used  for efficient search.
+	if queryPrefix != "" && isFirst {
+		prefixIndex = searchFileInfos(fis, queryPrefix)
+		if prefixIndex == len(fis) {
+			return
+		}
+
+		if !strings.HasPrefix(fis[prefixIndex].Name(), queryPrefix) {
+			return
+		}
+		fis = fis[prefixIndex:]
+
+	}
+
+	// Populate []ObjectInfo from []FileInfo.
 	for _, fi := range fis {
 		name := fi.Name()
+		if queryPrefix != "" && isFirst {
+			// If control is here then there is a queryPrefix, and there are objects which satisfies the prefix.
+			// Since the result is sorted, the object names which satisfies query prefix would be stored one after the other.
+			// Push the objectInfo only if its contains the prefix.
+			// This ensures that the channel containing object Info would only has objects with the given queryPrefix.
+			if !strings.HasPrefix(name, queryPrefix) {
+				return
+			}
+		}
 		size := fi.Size()
 		modTime := fi.ModTime()
 		isDir := fi.Mode().IsDir()
@@ -122,7 +158,8 @@ func readDir(scanDir, namePrefix string) (objInfos []ObjectInfo) {
 		// For directories explicitly end with '/'.
 		if isDir {
 			name += "/"
-			size = 0 // Size is set to '0' for directories explicitly.
+			// Size is set to '0' for directories explicitly.
+			size = 0
 		}
 
 		if fi.Mode()&os.ModeSymlink == os.ModeSymlink {
@@ -150,12 +187,13 @@ func readDir(scanDir, namePrefix string) (objInfos []ObjectInfo) {
 			Size:         size,
 			IsDir:        isDir,
 		})
+
 	}
 
 	return
 }
 
-// ObjectInfoChannel - object info channel
+// ObjectInfoChannel - object info channel.
 type ObjectInfoChannel struct {
 	ch        <-chan ObjectInfo
 	objInfo   *ObjectInfo
@@ -170,7 +208,7 @@ func (oic *ObjectInfoChannel) Read() (ObjectInfo, bool) {
 	}
 
 	if oic.objInfo == nil {
-		// first read
+		// First read.
 		if oi, ok := <-oic.ch; ok {
 			oic.objInfo = &oi
 		} else {
@@ -183,7 +221,7 @@ func (oic *ObjectInfoChannel) Read() (ObjectInfo, bool) {
 	status := true
 	oic.objInfo = nil
 
-	// read once more to know whether it was last read
+	// Read once more to know whether it was last read.
 	if oi, ok := <-oic.ch; ok {
 		oic.objInfo = &oi
 	} else {
@@ -193,7 +231,7 @@ func (oic *ObjectInfoChannel) Read() (ObjectInfo, bool) {
 	return retObjInfo, status
 }
 
-// IsClosed - return whether channel is closed or not
+// IsClosed - return whether channel is closed or not.
 func (oic ObjectInfoChannel) IsClosed() bool {
 	if oic.objInfo != nil {
 		return false
@@ -202,7 +240,7 @@ func (oic ObjectInfoChannel) IsClosed() bool {
 	return oic.closed
 }
 
-// IsTimedOut - return whether channel is closed due to timeout
+// IsTimedOut - return whether channel is closed due to timeout.
 func (oic ObjectInfoChannel) IsTimedOut() bool {
 	if oic.timedOut {
 		return true
@@ -220,19 +258,19 @@ func (oic ObjectInfoChannel) IsTimedOut() bool {
 	}
 }
 
-// treeWalk - walk into 'scanDir' recursively when 'recursive' is true.
+// TreeWalk - walk into 'scanDir' recursively when 'recursive' is true.
-//            It uses 'bucketDir' to get name prefix for object name.
+// It uses 'bucketDir' to get name prefix for object name.
-func treeWalk(scanDir, bucketDir string, recursive bool) ObjectInfoChannel {
+func treeWalk(scanDir, bucketDir string, recursive bool, queryPrefix string) ObjectInfoChannel {
 	objectInfoCh := make(chan ObjectInfo, listObjectsLimit)
 	timeoutCh := make(chan struct{}, 1)
 
-	// goroutine - retrieves directory entries, makes ObjectInfo and sends into the channel
+	// goroutine - retrieves directory entries, makes ObjectInfo and sends into the channel.
 	go func() {
 		defer close(objectInfoCh)
 		defer close(timeoutCh)
 
-		// send function - returns true if ObjectInfo is sent
+		// Send function - returns true if ObjectInfo is sent.
-		//                 within (time.Second * 15) else false on time-out
+		// Within (time.Second * 15) else false on time-out.
 		send := func(oi ObjectInfo) bool {
 			timer := time.After(time.Second * 15)
 			select {
@@ -246,11 +284,14 @@ func treeWalk(scanDir, bucketDir string, recursive bool) ObjectInfoChannel {
 
 		namePrefix := strings.Replace(filepath.ToSlash(scanDir), filepath.ToSlash(bucketDir), "", 1)
 		if strings.HasPrefix(namePrefix, "/") {
-			/* remove beginning "/" */
+			// Remove forward slash ("/") from beginning.
 			namePrefix = namePrefix[1:]
 		}
-
+		// The last argument (isFisrt), is set to `true` only during the first run of the function.
-		for objInfos := readDir(scanDir, namePrefix); len(objInfos) > 0; {
+		// This makes sure that the sub-directories inside the prefixDir are recursed
+		// without being asserted for prefix in the object name.
+		isFirst := true
+		for objInfos := readDir(scanDir, namePrefix, queryPrefix, isFirst); len(objInfos) > 0; {
 			var objInfo ObjectInfo
 			objInfo, objInfos = objInfos[0], objInfos[1:]
 			if !send(objInfo) {
@@ -259,14 +300,15 @@ func treeWalk(scanDir, bucketDir string, recursive bool) ObjectInfoChannel {
 
 			if objInfo.IsDir && recursive {
 				scanDir := filepath.Join(bucketDir, filepath.FromSlash(objInfo.Name))
-
 				namePrefix = strings.Replace(filepath.ToSlash(scanDir), filepath.ToSlash(bucketDir), "", 1)
+
 				if strings.HasPrefix(namePrefix, "/") {
 					/* remove beginning "/" */
 					namePrefix = namePrefix[1:]
 				}
-
+				// The last argument is set to false in the further calls to readdir.
-				objInfos = append(readDir(scanDir, namePrefix), objInfos...)
+				isFirst = false
+				objInfos = append(readDir(scanDir, namePrefix, queryPrefix, isFirst), objInfos...)
 			}
 		}
 	}()

pkg/fs/fs-bucket-listobjects.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"net/url"
 	"os"
+	"path"
 	"path/filepath"
 	"strings"
 
@@ -30,6 +31,7 @@ import (
 // maxKeys number of objects per call.
 func (fs Filesystem) ListObjects(bucket, prefix, marker, delimiter string, maxKeys int) (ListObjectsResult, *probe.Error) {
 	result := ListObjectsResult{}
+	var queryPrefix string
 
 	// Input validation.
 	if !IsValidBucketName(bucket) {
@@ -40,8 +42,10 @@ func (fs Filesystem) ListObjects(bucket, prefix, marker, delimiter string, maxKe
 
 	if status, err := isDirExist(filepath.Join(fs.path, bucket)); !status {
 		if err == nil {
+			// File exists, but its not a directory.
 			return result, probe.NewError(BucketNotFound{Bucket: bucket})
 		} else if os.IsNotExist(err) {
+			// File does not exist.
 			return result, probe.NewError(BucketNotFound{Bucket: bucket})
 		} else {
 			return result, probe.NewError(err)
@@ -60,8 +64,9 @@ func (fs Filesystem) ListObjects(bucket, prefix, marker, delimiter string, maxKe
 		}
 
 		if !strings.HasPrefix(marker, prefix) {
-			return result, probe.NewError(fmt.Errorf("marker '%s' and prefix '%s' do not match", marker, prefix))
+			return result, probe.NewError(fmt.Errorf("Invalid combination of marker '%s' and prefix '%s'", marker, prefix))
 		}
+
 	}
 
 	// Return empty response for a valid request when maxKeys is 0.
@@ -85,17 +90,37 @@ func (fs Filesystem) ListObjects(bucket, prefix, marker, delimiter string, maxKe
 	prefixDir := filepath.Dir(filepath.FromSlash(prefix))
 	rootDir := filepath.Join(bucketDir, prefixDir)
 
+	// Maximum 1000 objects returned in a single to listObjects.
+	// Further calls will set right marker value to continue reading the rest of the objectList.
+	// popListObjectCh returns nil if the call to ListObject is done for the first time.
+	// On further calls to ListObjects to retrive more objects within the timeout period,
+	// popListObjectCh returns the channel from which rest of the objects can be retrieved.
 	objectInfoCh := fs.popListObjectCh(ListObjectParams{bucket, delimiter, marker, prefix})
 	if objectInfoCh == nil {
-		ch := treeWalk(rootDir, bucketDir, recursive)
+		if prefix != "" {
+			// queryPrefix variable is set to value of the prefix to be searched.
+			// If prefix contains directory hierarchy queryPrefix is set to empty string,
+			// this ensure that all objects inside the prefixDir is listed.
+			// Otherwise the base name is extracted from path.Base and it'll be will be set to Querystring.
+			// if prefix = /Asia/India/, queryPrefix will be set to empty string(""), so that all objects in prefixDir are listed.
+			// if prefix = /Asia/India/summerpics , Querystring will be set to "summerpics",
+			// so those all objects with the prefix "summerpics" inside the /Asia/India/ prefix folder gets listed.
+			if prefix[len(prefix)-1:] == "/" {
+				queryPrefix = ""
+			} else {
+				queryPrefix = path.Base(prefix)
+			}
+		}
+		ch := treeWalk(rootDir, bucketDir, recursive, queryPrefix)
 		objectInfoCh = &ch
 	}
 
 	nextMarker := ""
 	for i := 0; i < maxKeys; {
+
 		objInfo, ok := objectInfoCh.Read()
 		if !ok {
-			// closed channel
+			// Closed channel.
 			return result, nil
 		}
 
@@ -113,18 +138,18 @@ func (fs Filesystem) ListObjects(bucket, prefix, marker, delimiter string, maxKe
 
 		// Add the bucket.
 		objInfo.Bucket = bucket
-
+		// In case its not the first call to ListObjects (before timeout),
-		if strings.HasPrefix(objInfo.Name, prefix) {
+		// The result is already inside the buffered channel.
-			if objInfo.Name > marker {
+		if objInfo.Name > marker {
-				if objInfo.IsDir {
+			if objInfo.IsDir {
-					result.Prefixes = append(result.Prefixes, objInfo.Name)
+				result.Prefixes = append(result.Prefixes, objInfo.Name)
-				} else {
+			} else {
-					result.Objects = append(result.Objects, objInfo)
+				result.Objects = append(result.Objects, objInfo)
-				}
-				nextMarker = objInfo.Name
-				i++
 			}
+			nextMarker = objInfo.Name
+			i++
 		}
+
 	}
 
 	if !objectInfoCh.IsClosed() {

pkg/fs/fs-bucket-listobjects_test.go

@@ -0,0 +1,611 @@
+/*
+ * Minio Cloud Storage, (C) 2015-2016 Minio, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package fs
+
+import (
+	"bytes"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"testing"
+)
+
+func TestListObjects(t *testing.T) {
+
+	// Make a temporary directory to use as the filesystem.
+	directory, e := ioutil.TempDir("", "minio-list-object-test")
+	if e != nil {
+		t.Fatal(e)
+	}
+	defer os.RemoveAll(directory)
+
+	// Create the filesystem.
+	fs, err := New(directory, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// This bucket is used for testing ListObject operations.
+	err = fs.MakeBucket("test-bucket-list-object")
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Will not store any objects in this bucket,
+	// Its to test ListObjects on an empty bucket.
+	err = fs.MakeBucket("empty-bucket")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	tmpfile, e := ioutil.TempFile("", "simple-file.txt")
+	if e != nil {
+		t.Fatal(e)
+	}
+	defer os.Remove(tmpfile.Name()) // clean up
+
+	_, err = fs.CreateObject("test-bucket-list-object", "Asia-maps", "", int64(len("asia-maps")), bytes.NewBufferString("asia-maps"), nil)
+	if err != nil {
+		t.Fatal(e)
+	}
+
+	_, err = fs.CreateObject("test-bucket-list-object", "Asia/India/India-summer-photos-1", "", int64(len("contentstring")), bytes.NewBufferString("contentstring"), nil)
+	if err != nil {
+		t.Fatal(e)
+	}
+
+	_, err = fs.CreateObject("test-bucket-list-object", "Asia/India/Karnataka/Bangalore/Koramangala/pics", "", int64(len("contentstring")), bytes.NewBufferString("contentstring"), nil)
+	if err != nil {
+		t.Fatal(e)
+	}
+
+	for i := 0; i < 2; i++ {
+		key := "newPrefix" + strconv.Itoa(i)
+		_, err = fs.CreateObject("test-bucket-list-object", key, "", int64(len(key)), bytes.NewBufferString(key), nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+	_, err = fs.CreateObject("test-bucket-list-object", "newzen/zen/recurse/again/again/again/pics", "", int64(len("recurse")), bytes.NewBufferString("recurse"), nil)
+	if err != nil {
+		t.Fatal(e)
+	}
+
+	for i := 0; i < 3; i++ {
+		key := "obj" + strconv.Itoa(i)
+		_, err = fs.CreateObject("test-bucket-list-object", key, "", int64(len(key)), bytes.NewBufferString(key), nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	// Formualting the result data set to be expected from ListObjects call inside the tests,
+	// This will be used in testCases and used for asserting the correctness of ListObjects output in the tests.
+
+	resultCases := []ListObjectsResult{
+		// ListObjectsResult-0.
+		// Testing for listing all objects in the bucket, (testCase 20,21,22).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "Asia-maps"},
+				ObjectInfo{Name: "Asia/India/India-summer-photos-1"},
+				ObjectInfo{Name: "Asia/India/Karnataka/Bangalore/Koramangala/pics"},
+				ObjectInfo{Name: "newPrefix0"},
+				ObjectInfo{Name: "newPrefix1"},
+				ObjectInfo{Name: "newzen/zen/recurse/again/again/again/pics"},
+				ObjectInfo{Name: "obj0"},
+				ObjectInfo{Name: "obj1"},
+				ObjectInfo{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-1.
+		// Used for asserting the truncated case, (testCase 23).
+		ListObjectsResult{
+			IsTruncated: true,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "Asia-maps"},
+				ObjectInfo{Name: "Asia/India/India-summer-photos-1"},
+				ObjectInfo{Name: "Asia/India/Karnataka/Bangalore/Koramangala/pics"},
+				ObjectInfo{Name: "newPrefix0"},
+				ObjectInfo{Name: "newPrefix1"},
+			},
+		},
+		// ListObjectsResult-2.
+		// (TestCase 24).
+		ListObjectsResult{
+			IsTruncated: true,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "Asia-maps"},
+				ObjectInfo{Name: "Asia/India/India-summer-photos-1"},
+				ObjectInfo{Name: "Asia/India/Karnataka/Bangalore/Koramangala/pics"},
+				ObjectInfo{Name: "newPrefix0"},
+			},
+		},
+		// ListObjectsResult-3.
+		// (TestCase 25).
+		ListObjectsResult{
+			IsTruncated: true,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "Asia-maps"},
+				ObjectInfo{Name: "Asia/India/India-summer-photos-1"},
+				ObjectInfo{Name: "Asia/India/Karnataka/Bangalore/Koramangala/pics"},
+			},
+		},
+		// ListObjectsResult-4.
+		// Again used for truncated case.
+		// (TestCase 26).
+		ListObjectsResult{
+			IsTruncated: true,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "Asia-maps"},
+			},
+		},
+		// ListObjectsResult-5.
+		// Used for Asserting prefixes.
+		// Used for test case with prefix "new", (testCase 27-29).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "newPrefix0"},
+				ObjectInfo{Name: "newPrefix1"},
+				ObjectInfo{Name: "newzen/zen/recurse/again/again/again/pics"},
+			},
+		},
+		// ListObjectsResult-6.
+		// Used for Asserting prefixes.
+		// Used for test case with prefix = "obj", (testCase 30).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "obj0"},
+				ObjectInfo{Name: "obj1"},
+				ObjectInfo{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-7.
+		// Used for Asserting prefixes and truncation.
+		// Used for test case with prefix = "new" and maxKeys = 1, (testCase 31).
+		ListObjectsResult{
+			IsTruncated: true,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "newPrefix0"},
+			},
+		},
+		// ListObjectsResult-8.
+		// Used for Asserting prefixes.
+		// Used for test case with prefix = "obj" and maxKeys = 2, (testCase 32).
+		ListObjectsResult{
+			IsTruncated: true,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "obj0"},
+				ObjectInfo{Name: "obj1"},
+			},
+		},
+		// ListObjectsResult-9.
+		// Used for asserting the case with marker, but without prefix.
+		//marker is set to "newPrefix0" in the testCase, (testCase 33).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "newPrefix1"},
+				ObjectInfo{Name: "newzen/zen/recurse/again/again/again/pics"},
+				ObjectInfo{Name: "obj0"},
+				ObjectInfo{Name: "obj1"},
+				ObjectInfo{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-10.
+		//marker is set to "newPrefix1" in the testCase, (testCase 34).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "newzen/zen/recurse/again/again/again/pics"},
+				ObjectInfo{Name: "obj0"},
+				ObjectInfo{Name: "obj1"},
+				ObjectInfo{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-11.
+		//marker is set to "obj0" in the testCase, (testCase 35).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "obj1"},
+				ObjectInfo{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-12.
+		// Marker is set to "obj1" in the testCase, (testCase 36).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-13.
+		// Marker is set to "man" in the testCase, (testCase37).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "newPrefix0"},
+				ObjectInfo{Name: "newPrefix1"},
+				ObjectInfo{Name: "newzen/zen/recurse/again/again/again/pics"},
+				ObjectInfo{Name: "obj0"},
+				ObjectInfo{Name: "obj1"},
+				ObjectInfo{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-14.
+		// Marker is set to "Abc" in the testCase, (testCase 39).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "Asia-maps"},
+				ObjectInfo{Name: "Asia/India/India-summer-photos-1"},
+				ObjectInfo{Name: "Asia/India/Karnataka/Bangalore/Koramangala/pics"},
+				ObjectInfo{Name: "newPrefix0"},
+				ObjectInfo{Name: "newPrefix1"},
+				ObjectInfo{Name: "newzen/zen/recurse/again/again/again/pics"},
+				ObjectInfo{Name: "obj0"},
+				ObjectInfo{Name: "obj1"},
+				ObjectInfo{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-15.
+		// Marker is set to "Asia/India/India-summer-photos-1" in the testCase, (testCase 40).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "Asia/India/Karnataka/Bangalore/Koramangala/pics"},
+				ObjectInfo{Name: "newPrefix0"},
+				ObjectInfo{Name: "newPrefix1"},
+				ObjectInfo{Name: "newzen/zen/recurse/again/again/again/pics"},
+				ObjectInfo{Name: "obj0"},
+				ObjectInfo{Name: "obj1"},
+				ObjectInfo{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-16.
+		// Marker is set to "Asia/India/Karnataka/Bangalore/Koramangala/pics" in the testCase, (testCase 41).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "newPrefix0"},
+				ObjectInfo{Name: "newPrefix1"},
+				ObjectInfo{Name: "newzen/zen/recurse/again/again/again/pics"},
+				ObjectInfo{Name: "obj0"},
+				ObjectInfo{Name: "obj1"},
+				ObjectInfo{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-17.
+		// Used for asserting the case with marker, without prefix but with truncation.
+		// Marker =  "newPrefix0" & maxKeys = 3 in the testCase, (testCase42).
+		// Output truncated to 3 values.
+		ListObjectsResult{
+			IsTruncated: true,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "newPrefix1"},
+				ObjectInfo{Name: "newzen/zen/recurse/again/again/again/pics"},
+				ObjectInfo{Name: "obj0"},
+			},
+		},
+		// ListObjectsResult-18.
+		// Marker = "newPrefix1" & maxkeys = 1 in the testCase, (testCase43).
+		// Output truncated to 1 value.
+		ListObjectsResult{
+			IsTruncated: true,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "newzen/zen/recurse/again/again/again/pics"},
+			},
+		},
+		// ListObjectsResult-19.
+		// Marker = "obj0" & maxKeys = 1 in the testCase, (testCase44).
+		// Output truncated to 1 value.
+		ListObjectsResult{
+			IsTruncated: true,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "obj1"},
+			},
+		},
+		// ListObjectsResult-20.
+		// Marker = "obj0" & prefix = "obj" in the testCase, (testCase 45).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "obj1"},
+				ObjectInfo{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-21.
+		// Marker = "obj1" & prefix = "obj" in the testCase, (testCase 46).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-22.
+		// Marker = "newPrefix0" & prefix = "new" in the testCase,, (testCase 47).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "newPrefix1"},
+				ObjectInfo{Name: "newzen/zen/recurse/again/again/again/pics"},
+			},
+		},
+		// ListObjectsResult-23.
+		// Prefix is set to "Asia/India/" in the testCase, and delimiter is not set (testCase 55).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "Asia/India/India-summer-photos-1"},
+				ObjectInfo{Name: "Asia/India/Karnataka/Bangalore/Koramangala/pics"},
+			},
+		},
+
+		// ListObjectsResult-24.
+		// Prefix is set to "Asia" in the testCase, and delimiter is not set (testCase 56).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "Asia-maps"},
+				ObjectInfo{Name: "Asia/India/India-summer-photos-1"},
+				ObjectInfo{Name: "Asia/India/Karnataka/Bangalore/Koramangala/pics"},
+			},
+		},
+
+		// ListObjectsResult-25.
+		// Prefix is set to "Asia" in the testCase, and delimiter is set (testCase 57).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "Asia-maps"},
+			},
+		},
+		// ListObjectsResult-26.
+		// prefix = "new" and delimiter is set in the testCase.(testCase 58).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "newPrefix0"},
+				ObjectInfo{Name: "newPrefix1"},
+			},
+		},
+		// ListObjectsResult-27.
+		// Prefix is set to "Asia/India/" in the testCase, and delimiter is set to forward slash '/' (testCase 59).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "Asia/India/India-summer-photos-1"},
+			},
+		},
+		// ListObjectsResult-28.
+		// Marker is set to "Asia/India/India-summer-photos-1" and delimiter set in the testCase, (testCase 60).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "newPrefix0"},
+				ObjectInfo{Name: "newPrefix1"},
+				ObjectInfo{Name: "obj0"},
+				ObjectInfo{Name: "obj1"},
+				ObjectInfo{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-29.
+		// Marker is set to "Asia/India/Karnataka/Bangalore/Koramangala/pics" in the testCase and delimeter set, (testCase 61).
+		ListObjectsResult{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				ObjectInfo{Name: "newPrefix0"},
+				ObjectInfo{Name: "newPrefix1"},
+				ObjectInfo{Name: "obj0"},
+				ObjectInfo{Name: "obj1"},
+				ObjectInfo{Name: "obj2"},
+			},
+		},
+	}
+
+	testCases := []struct {
+		// Inputs to ListObjects.
+		bucketName string
+		prefix     string
+		marker     string
+		delimeter  string
+		maxKeys    int
+		// Expected output of ListObjects.
+		result ListObjectsResult
+		err    error
+		// Flag indicating whether the test is expected to pass or not.
+		shouldPass bool
+	}{
+		// Test cases with invalid bucket names ( Test number 1-4 ).
+		{".test", "", "", "", 0, ListObjectsResult{}, BucketNameInvalid{Bucket: ".test"}, false},
+		{"Test", "", "", "", 0, ListObjectsResult{}, BucketNameInvalid{Bucket: "Test"}, false},
+		{"---", "", "", "", 0, ListObjectsResult{}, BucketNameInvalid{Bucket: "---"}, false},
+		{"ad", "", "", "", 0, ListObjectsResult{}, BucketNameInvalid{Bucket: "ad"}, false},
+		// Using an existing file for bucket name, but its not a directory (5).
+		{"simple-file.txt", "", "", "", 0, ListObjectsResult{}, BucketNotFound{Bucket: "simple-file.txt"}, false},
+		// Valid bucket names, but they donot exist (6-8).
+		{"volatile-bucket-1", "", "", "", 0, ListObjectsResult{}, BucketNotFound{Bucket: "volatile-bucket-1"}, false},
+		{"volatile-bucket-2", "", "", "", 0, ListObjectsResult{}, BucketNotFound{Bucket: "volatile-bucket-2"}, false},
+		{"volatile-bucket-3", "", "", "", 0, ListObjectsResult{}, BucketNotFound{Bucket: "volatile-bucket-3"}, false},
+		// Valid, existing bucket, but sending invalid delimeter values (9-10).
+		// Empty string < "" > and forward slash < / > are the ony two valid arguments for delimeter.
+		{"test-bucket-list-object", "", "", "*", 0, ListObjectsResult{}, fmt.Errorf("delimiter '%s' is not supported", "*"), false},
+		{"test-bucket-list-object", "", "", "-", 0, ListObjectsResult{}, fmt.Errorf("delimiter '%s' is not supported", "-"), false},
+		// Marker goes through url QueryUnescape, sending inputs for which QueryUnescape would fail (11-12).
+		// Here is how QueryUnescape behaves https://golang.org/pkg/net/url/#QueryUnescape.
+		// QueryUnescape is necessasry since marker is provided as URL query parameter.
+		{"test-bucket-list-object", "", "test%", "", 0, ListObjectsResult{}, fmt.Errorf("invalid URL escape"), false},
+		{"test-bucket-list-object", "", "test%A", "", 0, ListObjectsResult{}, fmt.Errorf("invalid URL escape"), false},
+		// Testing for failure cases with both perfix and marker (13).
+		// The prefix and marker combination to be valid it should satisy strings.HasPrefix(marker, prefix).
+		{"test-bucket-list-object", "asia", "europe-object", "", 0, ListObjectsResult{}, fmt.Errorf("Invalid combination of marker '%s' and prefix '%s'", "europe-object", "asia"), false},
+		// Setting a non-existing directory to be prefix (14-15).
+		{"empty-bucket", "europe/france/", "", "", 1, ListObjectsResult{}, fmt.Errorf("%s:", filepath.FromSlash("/empty-bucket/europe/france")), false},
+		{"empty-bucket", "europe/tunisia/", "", "", 1, ListObjectsResult{}, fmt.Errorf("%s:", filepath.FromSlash("/empty-bucket/europe/tunisia")), false},
+		// Testing on empty bucket, that is, bucket without any objects in it (16).
+		{"empty-bucket", "", "", "", 0, ListObjectsResult{}, nil, true},
+		// Setting maxKeys to negative value (17-18).
+		{"empty-bucket", "", "", "", -1, ListObjectsResult{}, nil, true},
+		{"empty-bucket", "", "", "", 1, ListObjectsResult{}, nil, true},
+		// Setting maxKeys to a very large value (19).
+		{"empty-bucket", "", "", "", 1111000000000000, ListObjectsResult{}, nil, true},
+		// Testing for all 7 objects in the bucket (20).
+		{"test-bucket-list-object", "", "", "", 9, resultCases[0], nil, true},
+		//Testing for negative value of maxKey, this should set maxKeys to listObjectsLimit (21).
+		{"test-bucket-list-object", "", "", "", -1, resultCases[0], nil, true},
+		// Testing for very large value of maxKey, this should set maxKeys to listObjectsLimit (22).
+		{"test-bucket-list-object", "", "", "", 1234567891011, resultCases[0], nil, true},
+		// Testing for trancated value (23-26).
+		{"test-bucket-list-object", "", "", "", 5, resultCases[1], nil, true},
+		{"test-bucket-list-object", "", "", "", 4, resultCases[2], nil, true},
+		{"test-bucket-list-object", "", "", "", 3, resultCases[3], nil, true},
+		{"test-bucket-list-object", "", "", "", 1, resultCases[4], nil, true},
+		// Testing with prefix (27-30).
+		{"test-bucket-list-object", "new", "", "", 3, resultCases[5], nil, true},
+		{"test-bucket-list-object", "new", "", "", 4, resultCases[5], nil, true},
+		{"test-bucket-list-object", "new", "", "", 5, resultCases[5], nil, true},
+		{"test-bucket-list-object", "obj", "", "", 3, resultCases[6], nil, true},
+		// Testing with prefix and truncation (31-32).
+		{"test-bucket-list-object", "new", "", "", 1, resultCases[7], nil, true},
+		{"test-bucket-list-object", "obj", "", "", 2, resultCases[8], nil, true},
+		// Testing with marker, but without prefix and truncation (33-37).
+		{"test-bucket-list-object", "", "newPrefix0", "", 5, resultCases[9], nil, true},
+		{"test-bucket-list-object", "", "newPrefix1", "", 4, resultCases[10], nil, true},
+		{"test-bucket-list-object", "", "obj0", "", 2, resultCases[11], nil, true},
+		{"test-bucket-list-object", "", "obj1", "", 1, resultCases[12], nil, true},
+		{"test-bucket-list-object", "", "man", "", 10, resultCases[13], nil, true},
+		// Marker being set to a value which is greater than and all object names when sorted (38).
+		// Expected to send an empty response in this case.
+		{"test-bucket-list-object", "", "zen", "", 10, ListObjectsResult{}, nil, true},
+		// Marker being set to a value which is lesser than and all object names when sorted (39).
+		// Expected to send all the objects in the bucket in this case.
+		{"test-bucket-list-object", "", "Abc", "", 10, resultCases[14], nil, true},
+		// Marker is to a hierarhical value (40-41).
+		{"test-bucket-list-object", "", "Asia/India/India-summer-photos-1", "", 10, resultCases[15], nil, true},
+		{"test-bucket-list-object", "", "Asia/India/Karnataka/Bangalore/Koramangala/pics", "", 10, resultCases[16], nil, true},
+		// Testing with marker and truncation, but no prefix (42-44).
+		{"test-bucket-list-object", "", "newPrefix0", "", 3, resultCases[17], nil, true},
+		{"test-bucket-list-object", "", "newPrefix1", "", 1, resultCases[18], nil, true},
+		{"test-bucket-list-object", "", "obj0", "", 1, resultCases[19], nil, true},
+		// Testing with both marker and prefix, but without truncation (45-47).
+		// The valid combination of marker and prefix should satisfy strings.HasPrefix(marker, prefix).
+		{"test-bucket-list-object", "obj", "obj0", "", 2, resultCases[20], nil, true},
+		{"test-bucket-list-object", "obj", "obj1", "", 1, resultCases[21], nil, true},
+		{"test-bucket-list-object", "new", "newPrefix0", "", 2, resultCases[22], nil, true},
+		// Testing with maxKeys set to 0 (48-54).
+		// The parameters have to valid.
+		{"test-bucket-list-object", "", "obj1", "", 0, ListObjectsResult{}, nil, true},
+		{"test-bucket-list-object", "", "obj0", "", 0, ListObjectsResult{}, nil, true},
+		{"test-bucket-list-object", "new", "", "", 0, ListObjectsResult{}, nil, true},
+		{"test-bucket-list-object", "obj", "", "", 0, ListObjectsResult{}, nil, true},
+		{"test-bucket-list-object", "obj", "obj0", "", 0, ListObjectsResult{}, nil, true},
+		{"test-bucket-list-object", "obj", "obj1", "", 0, ListObjectsResult{}, nil, true},
+		{"test-bucket-list-object", "new", "newPrefix0", "", 0, ListObjectsResult{}, nil, true},
+		// Tests on hierarchical key names as prefix.
+		// Without delimteter the code should recurse into the prefix Dir.
+		// Tests with prefix, but without delimiter (55-56).
+		{"test-bucket-list-object", "Asia/India/", "", "", 10, resultCases[23], nil, true},
+		{"test-bucket-list-object", "Asia", "", "", 10, resultCases[24], nil, true},
+		// Tests with prefix and delimiter (57-59).
+		// With delimeter the code shouldnot recurse into the sub-directories of prefix Dir.
+		{"test-bucket-list-object", "Asia", "", "/", 10, resultCases[25], nil, true},
+		{"test-bucket-list-object", "new", "", "/", 10, resultCases[26], nil, true},
+		{"test-bucket-list-object", "Asia/India/", "", "/", 10, resultCases[27], nil, true},
+		// Test with marker set as hierarhical value and with delimiter. (60-61)
+		{"test-bucket-list-object", "", "Asia/India/India-summer-photos-1", "/", 10, resultCases[28], nil, true},
+		{"test-bucket-list-object", "", "Asia/India/Karnataka/Bangalore/Koramangala/pics", "/", 10, resultCases[29], nil, true},
+	}
+
+	for i, testCase := range testCases {
+		result, err := fs.ListObjects(testCase.bucketName, testCase.prefix, testCase.marker, testCase.delimeter, testCase.maxKeys)
+		if err != nil && testCase.shouldPass {
+			t.Errorf("Test %d: Expected to pass, but failed with: <ERROR> %s", i+1, err.Cause.Error())
+		}
+		if err == nil && !testCase.shouldPass {
+			t.Errorf("Test %d: Expected to fail with <ERROR> \"%s\", but passed instead", i+1, testCase.err.Error())
+		}
+		// Failed as expected, but does it fail for the expected reason.
+		if err != nil && !testCase.shouldPass {
+			if !strings.Contains(err.Cause.Error(), testCase.err.Error()) {
+				t.Errorf("Test %d: Expected to fail with error \"%s\", but instead failed with error \"%s\" instead", i+1, testCase.err.Error(), err.Cause.Error())
+			}
+		}
+		// Since there are cases for which ListObjects fails, this is necessary.
+		// Test passes as expected, but the output values are verified for correctness here.
+		if err == nil && testCase.shouldPass {
+			// The length of the expected ListObjectsResult.Objects should match in both expected result from test cases and in the output.
+			// On failure calling t.Fatalf, otherwise it may lead to index out of range error in assertion following this.
+			if len(testCase.result.Objects) != len(result.Objects) {
+				t.Fatalf("Test %d: Expected number of object in the result to be '%d', but found '%d' objects instead", i+1, len(testCase.result.Objects), len(result.Objects))
+			}
+			for j := 0; j < len(testCase.result.Objects); j++ {
+				if testCase.result.Objects[j].Name != result.Objects[j].Name {
+					t.Errorf("Test %d: Expected object name to be \"%s\", but found \"%s\" instead", i+1, testCase.result.Objects[j].Name, result.Objects[j].Name)
+				}
+			}
+			if testCase.result.IsTruncated != result.IsTruncated {
+				t.Errorf("Test %d: Expected IsTruncated flag to be %v, but instead found it to be %v", i+1, testCase.result.IsTruncated, result.IsTruncated)
+			}
+
+		}
+	}
+
+}
+
+func BenchmarkListObjects(b *testing.B) {
+	// Make a temporary directory to use as the filesystem.
+	directory, e := ioutil.TempDir("", "minio-list-benchmark")
+	if e != nil {
+		b.Fatal(e)
+	}
+	defer os.RemoveAll(directory)
+
+	// Create the filesystem.
+	filesystem, err := New(directory, 0)
+	if err != nil {
+		b.Fatal(err)
+	}
+
+	// Create a bucket.
+	err = filesystem.MakeBucket("ls-benchmark-bucket")
+	if err != nil {
+		b.Fatal(err)
+	}
+
+	for i := 0; i < 20000; i++ {
+		key := "obj" + strconv.Itoa(i)
+		_, err = filesystem.CreateObject("ls-benchmark-bucket", key, "", int64(len(key)), bytes.NewBufferString(key), nil)
+		if err != nil {
+			b.Fatal(err)
+		}
+	}
+
+	b.ResetTimer()
+
+	// List the buckets over and over and over.
+	for i := 0; i < b.N; i++ {
+		_, err = filesystem.ListObjects("ls-benchmark-bucket", "", "obj9000", "", -1)
+		if err != nil {
+			b.Fatal(err)
+		}
+	}
+}

pkg/fs/fs-bucket_test.go

@@ -69,7 +69,7 @@ func TestGetBucketInfo(t *testing.T) {
 		{"meta-test-bucket.3", BucketInfo{Name: "meta-test-bucket.3"}, nil, true},
 	}
 	for i, testCase := range testCases {
-		//the err returned is of type *probe.Error
+		// The err returned is of type *probe.Error.
 		bucketInfo, err := filesystem.GetBucketInfo(testCase.bucketName)
 
 		if err != nil && testCase.shouldPass {