ns-lock: lock namespace during FS object operations. (#2896)

cmd/fs-v1.go

@@ -223,6 +223,13 @@ func (fs fsObjects) GetObject(bucket, object string, offset int64, length int64,
 		return traceError(InvalidRange{offset, length, fi.Size})
 	}
 
+	// get a random ID for lock instrumentation.
+	opsID := getOpsID()
+
+	// Lock the object before reading.
+	nsMutex.RLock(bucket, object, opsID)
+	defer nsMutex.RUnlock(bucket, object, opsID)
+
 	var totalLeft = length
 	bufSize := int64(readSizeV1)
 	if length > 0 && bufSize > length {
@@ -426,6 +433,13 @@ func (fs fsObjects) PutObject(bucket string, object string, size int64, data io.
 		}
 	}
 
+	// get a random ID for lock instrumentation.
+	opsID := getOpsID()
+
+	// Lock the object before comitting the object.
+	nsMutex.RLock(bucket, object, opsID)
+	defer nsMutex.RUnlock(bucket, object, opsID)
+
 	// Entire object was written to the temp location, now it's safe to rename it to the actual location.
 	err = fs.storage.RenameFile(minioMetaBucket, tempObj, bucket, object)
 	if err != nil {
@@ -461,6 +475,14 @@ func (fs fsObjects) DeleteObject(bucket, object string) error {
 	if !IsValidObjectName(object) {
 		return traceError(ObjectNameInvalid{Bucket: bucket, Object: object})
 	}
+	// get a random ID for lock instrumentation.
+	opsID := getOpsID()
+
+	// Lock the object before deleting so that an in progress GetObject does not return
+	// corrupt data or there is no race with a PutObject.
+	nsMutex.RLock(bucket, object, opsID)
+	defer nsMutex.RUnlock(bucket, object, opsID)
+
 	err := fs.storage.DeleteFile(minioMetaBucket, path.Join(bucketMetaPrefix, bucket, object, fsMetaJSONFile))
 	if err != nil && err != errFileNotFound {
 		return toObjectErr(traceError(err), bucket, object)