XL/fs: Optimize calling isBucketExist() (#1656)

* posix: Avoid using getAllVolumeInfo() in getVolumeDir()

This is necessary compromise to avoid significant slowness this
causes under load. The compromise is also substantial in a way
so that to avoid penalizing common cases v/s special cases.

For buckets with Caps on Unixes, we filter buckets based on the
latest anyways, so this is completely acceptable.

* XL/fs: Change the usage of verification of existance of buckets.

Optimize calling isBucketExists, it is not needed for all call
paths. isBucketExist should be called only for calls which use
temporary volume location for operations, for the rest rely on
the errors returned on their original call path.

Remove usage of filtering as well across all volume names.

posix.go

@@ -128,28 +128,8 @@ func checkDiskFree(diskPath string, minFreeDisk int64) (err error) {
 	return nil
 }
 
-func removeDuplicateVols(volsInfo []VolInfo) []VolInfo {
-	// Use map to record duplicates as we find them.
-	result := []VolInfo{}
-
-	m := make(map[string]VolInfo)
-	for _, v := range volsInfo {
-		if _, found := m[v.Name]; !found {
-			m[v.Name] = v
-		}
-	}
-
-	result = make([]VolInfo, 0, len(m))
-	for _, v := range m {
-		result = append(result, v)
-	}
-
-	// Return the new slice.
-	return result
-}
-
-// gets all the unique directories from diskPath.
-func getAllUniqueVols(dirPath string) ([]VolInfo, error) {
+// List all the volumes from diskPath.
+func listVols(dirPath string) ([]VolInfo, error) {
 	if err := checkPathLength(dirPath); err != nil {
 		return nil, err
 	}
@@ -179,14 +159,14 @@ func getAllUniqueVols(dirPath string) ([]VolInfo, error) {
 			Created: fi.ModTime(),
 		})
 	}
-	return removeDuplicateVols(volsInfo), nil
+	return volsInfo, nil
 }
 
-// getVolumeDir - will convert incoming volume names to
+// getVolDir - will convert incoming volume names to
 // corresponding valid volume names on the backend in a platform
 // compatible way for all operating systems. If volume is not found
 // an error is generated.
-func (s fsStorage) getVolumeDir(volume string) (string, error) {
+func (s fsStorage) getVolDir(volume string) (string, error) {
 	if !isValidVolname(volume) {
 		return "", errInvalidArgument
 	}
@@ -194,30 +174,7 @@ func (s fsStorage) getVolumeDir(volume string) (string, error) {
 		return "", err
 	}
 	volumeDir := pathJoin(s.diskPath, volume)
-	_, err := os.Stat(volumeDir)
-	if err == nil {
-		return volumeDir, nil
-	}
-	if os.IsNotExist(err) {
-		var volsInfo []VolInfo
-		volsInfo, err = getAllUniqueVols(s.diskPath)
-		if err != nil {
-			// For any errors, treat it as disk not found.
-			return volumeDir, err
-		}
-		for _, vol := range volsInfo {
-			// Verify if lowercase version of the volume is equal to
-			// the incoming volume, then use the proper  name.
-			if strings.ToLower(vol.Name) == volume {
-				volumeDir = pathJoin(s.diskPath, vol.Name)
-				return volumeDir, nil
-			}
-		}
-		return volumeDir, errVolumeNotFound
-	} else if os.IsPermission(err) {
-		return volumeDir, errVolumeAccessDenied
-	}
-	return volumeDir, err
+	return volumeDir, nil
 }
 
 // Make a volume entry.
@@ -227,20 +184,17 @@ func (s fsStorage) MakeVol(volume string) (err error) {
 		return err
 	}
 
-	volumeDir, err := s.getVolumeDir(volume)
-	if err == nil {
-		// Volume already exists, return error.
+	volumeDir, err := s.getVolDir(volume)
+	if err != nil {
+		return err
+	}
+	// Make a volume entry.
+	err = os.Mkdir(volumeDir, 0700)
+	if err != nil && os.IsExist(err) {
 		return errVolumeExists
 	}
-
-	// If volume not found create it.
-	if err == errVolumeNotFound {
-		// Make a volume entry.
-		return os.Mkdir(volumeDir, 0700)
-	}
-
-	// For all other errors return here.
-	return err
+	// Success
+	return nil
 }
 
 // ListVols - list volumes.
@@ -251,7 +205,7 @@ func (s fsStorage) ListVols() (volsInfo []VolInfo, err error) {
 	if err != nil {
 		return nil, err
 	}
-	volsInfo, err = getAllUniqueVols(s.diskPath)
+	volsInfo, err = listVols(s.diskPath)
 	if err != nil {
 		return nil, err
 	}
@@ -275,7 +229,7 @@ func (s fsStorage) ListVols() (volsInfo []VolInfo, err error) {
 // StatVol - get volume info.
 func (s fsStorage) StatVol(volume string) (volInfo VolInfo, err error) {
 	// Verify if volume is valid and it exists.
-	volumeDir, err := s.getVolumeDir(volume)
+	volumeDir, err := s.getVolDir(volume)
 	if err != nil {
 		return VolInfo{}, err
 	}
@@ -309,7 +263,7 @@ func (s fsStorage) StatVol(volume string) (volInfo VolInfo, err error) {
 // DeleteVol - delete a volume.
 func (s fsStorage) DeleteVol(volume string) error {
 	// Verify if volume is valid and it exists.
-	volumeDir, err := s.getVolumeDir(volume)
+	volumeDir, err := s.getVolDir(volume)
 	if err != nil {
 		return err
 	}
@@ -321,8 +275,7 @@ func (s fsStorage) DeleteVol(volume string) error {
 			// On windows the string is slightly different, handle it here.
 			return errVolumeNotEmpty
 		} else if strings.Contains(err.Error(), "directory not empty") {
-			// Hopefully for all other
-			// operating systems, this is
+			// Hopefully for all other operating systems, this is
 			// assumed to be consistent.
 			return errVolumeNotEmpty
 		}
@@ -335,7 +288,7 @@ func (s fsStorage) DeleteVol(volume string) error {
 // If an entry is a directory it will be returned with a trailing "/".
 func (s fsStorage) ListDir(volume, dirPath string) ([]string, error) {
 	// Verify if volume is valid and it exists.
-	volumeDir, err := s.getVolumeDir(volume)
+	volumeDir, err := s.getVolDir(volume)
 	if err != nil {
 		return nil, err
 	}
@@ -352,10 +305,18 @@ func (s fsStorage) ListDir(volume, dirPath string) ([]string, error) {
 
 // ReadFile - read a file at a given offset.
 func (s fsStorage) ReadFile(volume string, path string, offset int64) (readCloser io.ReadCloser, err error) {
-	volumeDir, err := s.getVolumeDir(volume)
+	volumeDir, err := s.getVolDir(volume)
 	if err != nil {
 		return nil, err
 	}
+	// Stat a volume entry.
+	_, err = os.Stat(volumeDir)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil, errVolumeNotFound
+		}
+		return nil, err
+	}
 
 	filePath := pathJoin(volumeDir, path)
 	if err = checkPathLength(filePath); err != nil {
@@ -388,10 +349,18 @@ func (s fsStorage) ReadFile(volume string, path string, offset int64) (readClose
 
 // CreateFile - create a file at path.
 func (s fsStorage) CreateFile(volume, path string) (writeCloser io.WriteCloser, err error) {
-	volumeDir, err := s.getVolumeDir(volume)
+	volumeDir, err := s.getVolDir(volume)
 	if err != nil {
 		return nil, err
 	}
+	// Stat a volume entry.
+	_, err = os.Stat(volumeDir)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil, errVolumeNotFound
+		}
+		return nil, err
+	}
 	if err = checkDiskFree(s.diskPath, s.minFreeDisk); err != nil {
 		return nil, err
 	}
@@ -419,10 +388,18 @@ func (s fsStorage) CreateFile(volume, path string) (writeCloser io.WriteCloser,
 
 // StatFile - get file info.
 func (s fsStorage) StatFile(volume, path string) (file FileInfo, err error) {
-	volumeDir, err := s.getVolumeDir(volume)
+	volumeDir, err := s.getVolDir(volume)
 	if err != nil {
 		return FileInfo{}, err
 	}
+	// Stat a volume entry.
+	_, err = os.Stat(volumeDir)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return FileInfo{}, errVolumeNotFound
+		}
+		return FileInfo{}, err
+	}
 
 	filePath := slashpath.Join(volumeDir, path)
 	if err = checkPathLength(filePath); err != nil {
@@ -489,10 +466,18 @@ func deleteFile(basePath, deletePath string) error {
 
 // DeleteFile - delete a file at path.
 func (s fsStorage) DeleteFile(volume, path string) error {
-	volumeDir, err := s.getVolumeDir(volume)
+	volumeDir, err := s.getVolDir(volume)
 	if err != nil {
 		return err
 	}
+	// Stat a volume entry.
+	_, err = os.Stat(volumeDir)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return errVolumeNotFound
+		}
+		return err
+	}
 
 	// Following code is needed so that we retain "/" suffix if any in
 	// path argument.
@@ -507,14 +492,29 @@ func (s fsStorage) DeleteFile(volume, path string) error {
 
 // RenameFile - rename file.
 func (s fsStorage) RenameFile(srcVolume, srcPath, dstVolume, dstPath string) error {
-	srcVolumeDir, err := s.getVolumeDir(srcVolume)
+	srcVolumeDir, err := s.getVolDir(srcVolume)
 	if err != nil {
 		return err
 	}
-	dstVolumeDir, err := s.getVolumeDir(dstVolume)
+	dstVolumeDir, err := s.getVolDir(dstVolume)
 	if err != nil {
 		return err
 	}
+	// Stat a volume entry.
+	_, err = os.Stat(srcVolumeDir)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return errVolumeNotFound
+		}
+		return err
+	}
+	_, err = os.Stat(dstVolumeDir)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return errVolumeNotFound
+		}
+	}
+
 	srcIsDir := strings.HasSuffix(srcPath, slashSeparator)
 	dstIsDir := strings.HasSuffix(dstPath, slashSeparator)
 	// Either src and dst have to be directories or files, else return error.