Merge pull request #1229 from brendanashworth/improve-fs-getobject

pkg/fs: optimize GetObject, add benchmark

pkg/fs/fs-object.go

@@ -54,50 +54,53 @@ func (fs Filesystem) GetObject(w io.Writer, bucket, object string, start, length
 
 	// normalize buckets.
 	bucket = fs.denormalizeBucket(bucket)
-	bucketPath := filepath.Join(fs.path, bucket)
-	if _, e := os.Stat(bucketPath); e != nil {
+	objectPath := filepath.Join(fs.path, bucket, object)
+
+	file, e := os.Open(objectPath)
+	if e != nil {
+		// If the object doesn't exist, the bucket might not exist either. Stat for
+		// the bucket and give a better error message if that is true.
+		if os.IsNotExist(e) {
+			_, e = os.Stat(filepath.Join(fs.path, bucket))
 			if os.IsNotExist(e) {
 				return 0, probe.NewError(BucketNotFound{Bucket: bucket})
 			}
-		return 0, probe.NewError(e)
-	}
 
-	objectPath := filepath.Join(bucketPath, object)
-	filestat, err := os.Stat(objectPath)
-	switch err := err.(type) {
-	case nil:
-		if filestat.IsDir() {
 			return 0, probe.NewError(ObjectNotFound{Bucket: bucket, Object: object})
 		}
-	default:
-		if os.IsNotExist(err) {
-			return 0, probe.NewError(ObjectNotFound{Bucket: bucket, Object: object})
-		}
-		return 0, probe.NewError(err)
-	}
-	file, err := os.Open(objectPath)
-	if err != nil {
-		return 0, probe.NewError(err)
+		return 0, probe.NewError(e)
 	}
 	defer file.Close()
 
-	_, err = file.Seek(start, os.SEEK_SET)
-	if err != nil {
-		return 0, probe.NewError(err)
+	_, e = file.Seek(start, os.SEEK_SET)
+	if e != nil {
+		// When the "handle is invalid", the file might be a directory on Windows.
+		if runtime.GOOS == "windows" && strings.Contains(e.Error(), "handle is invalid") {
+			return 0, probe.NewError(ObjectNotFound{Bucket: bucket, Object: object})
+		}
+
+		return 0, probe.NewError(e)
 	}
 
 	var count int64
+	// Copy over the whole file if the length is non-positive.
 	if length > 0 {
-		count, err = io.CopyN(w, file, length)
-		if err != nil {
-			return count, probe.NewError(err)
-		}
+		count, e = io.CopyN(w, file, length)
 	} else {
-		count, err = io.Copy(w, file)
-		if err != nil {
-			return count, probe.NewError(err)
+		count, e = io.Copy(w, file)
 	}
+
+	if e != nil {
+		// This call will fail if the object is a directory. Stat the file to see if
+		// this is true, if so, return an ObjectNotFound error.
+		stat, e := os.Stat(objectPath)
+		if e == nil && stat.IsDir() {
+			return count, probe.NewError(ObjectNotFound{Bucket: bucket, Object: object})
 		}
+
+		return count, probe.NewError(e)
+	}
+
 	return count, nil
 }
 

pkg/fs/fs-object_test.go

@@ -0,0 +1,73 @@
+/*
+ * Minio Cloud Storage, (C) 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"
+	"crypto/md5"
+	"encoding/base64"
+	"io/ioutil"
+	"os"
+	"strconv"
+	"testing"
+)
+
+func BenchmarkGetObject(b *testing.B) {
+	// Make a temporary directory to use as the filesystem.
+	directory, e := ioutil.TempDir("", "minio-benchmark-getobject")
+	if e != nil {
+		b.Fatal(e)
+	}
+	defer os.RemoveAll(directory)
+
+	// Create the filesystem.
+	filesystem, err := New(directory, 0)
+	if err != nil {
+		b.Fatal(err)
+	}
+
+	// Make a bucket and put in a few objects.
+	err = filesystem.MakeBucket("bucket")
+	if err != nil {
+		b.Fatal(err)
+	}
+
+	text := "Jack and Jill went up the hill / To fetch a pail of water."
+	hasher := md5.New()
+	hasher.Write([]byte(text))
+	sum := base64.StdEncoding.EncodeToString(hasher.Sum(nil))
+	for i := 0; i < 10; i++ {
+		_, err = filesystem.CreateObject("bucket", "object"+strconv.Itoa(i), sum, int64(len(text)), bytes.NewBufferString(text), nil)
+		if err != nil {
+			b.Fatal(err)
+		}
+	}
+
+	var w bytes.Buffer
+
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		n, err := filesystem.GetObject(&w, "bucket", "object"+strconv.Itoa(i%10), 0, 0)
+		if err != nil {
+			b.Error(err)
+		}
+		if n != int64(len(text)) {
+			b.Errorf("GetObject returned incorrect length %d (should be %d)\n", n, int64(len(text)))
+		}
+	}
+}