lock: Moving locking to handler layer. (#3381)

This is implemented so that the issues like in the
following flow don't affect the behavior of operation.

```
GetObjectInfo()
.... --> Time window for mutation (no lock held)
.... --> Time window for mutation (no lock held)
GetObject()
```

This happens when two simultaneous uploads are made
to the same object the object has returned wrong
info to the client.

Another classic example is "CopyObject" API itself
which reads from a source object and copies to
destination object.

Fixes #3370
Fixes #2912

cmd/bucket-policy.go

@@ -144,6 +144,12 @@ func getOldBucketsConfigPath() (string, error) {
 // if bucket policy is not found.
 func readBucketPolicyJSON(bucket string, objAPI ObjectLayer) (bucketPolicyReader io.Reader, err error) {
 	policyPath := pathJoin(bucketConfigPrefix, bucket, policyJSON)
+
+	// Acquire a read lock on policy config before reading.
+	objLock := globalNSMutex.NewNSLock(minioMetaBucket, policyPath)
+	objLock.RLock()
+	defer objLock.RUnlock()
+
 	objInfo, err := objAPI.GetObjectInfo(minioMetaBucket, policyPath)
 	if err != nil {
 		if isErrObjectNotFound(err) || isErrIncompleteBody(err) {
@@ -188,6 +194,10 @@ func readBucketPolicy(bucket string, objAPI ObjectLayer) (*bucketPolicy, error)
 // if no policies are found.
 func removeBucketPolicy(bucket string, objAPI ObjectLayer) error {
 	policyPath := pathJoin(bucketConfigPrefix, bucket, policyJSON)
+	// Acquire a write lock on policy config before modifying.
+	objLock := globalNSMutex.NewNSLock(minioMetaBucket, policyPath)
+	objLock.Lock()
+	defer objLock.Unlock()
 	if err := objAPI.DeleteObject(minioMetaBucket, policyPath); err != nil {
 		errorIf(err, "Unable to remove bucket-policy on bucket %s.", bucket)
 		err = errorCause(err)
@@ -207,9 +217,70 @@ func writeBucketPolicy(bucket string, objAPI ObjectLayer, bpy *bucketPolicy) err
 		return err
 	}
 	policyPath := pathJoin(bucketConfigPrefix, bucket, policyJSON)
+	// Acquire a write lock on policy config before modifying.
+	objLock := globalNSMutex.NewNSLock(minioMetaBucket, policyPath)
+	objLock.Lock()
+	defer objLock.Unlock()
 	if _, err := objAPI.PutObject(minioMetaBucket, policyPath, int64(len(buf)), bytes.NewReader(buf), nil, ""); err != nil {
 		errorIf(err, "Unable to set policy for the bucket %s", bucket)
 		return errorCause(err)
 	}
 	return nil
 }
+
+func parseAndPersistBucketPolicy(bucket string, policyBytes []byte, objAPI ObjectLayer) APIErrorCode {
+	// Parse bucket policy.
+	var policy = &bucketPolicy{}
+	err := parseBucketPolicy(bytes.NewReader(policyBytes), policy)
+	if err != nil {
+		errorIf(err, "Unable to parse bucket policy.")
+		return ErrInvalidPolicyDocument
+	}
+
+	// Parse check bucket policy.
+	if s3Error := checkBucketPolicyResources(bucket, policy); s3Error != ErrNone {
+		return s3Error
+	}
+
+	// Acquire a write lock on bucket before modifying its configuration.
+	bucketLock := globalNSMutex.NewNSLock(bucket, "")
+	bucketLock.Lock()
+	// Release lock after notifying peers
+	defer bucketLock.Unlock()
+
+	// Save bucket policy.
+	if err = persistAndNotifyBucketPolicyChange(bucket, policyChange{false, policy}, objAPI); err != nil {
+		switch err.(type) {
+		case BucketNameInvalid:
+			return ErrInvalidBucketName
+		case BucketNotFound:
+			return ErrNoSuchBucket
+		default:
+			errorIf(err, "Unable to save bucket policy.")
+			return ErrInternalError
+		}
+	}
+	return ErrNone
+}
+
+// persistAndNotifyBucketPolicyChange - takes a policyChange argument,
+// persists it to storage, and notify nodes in the cluster about the
+// change. In-memory state is updated in response to the notification.
+func persistAndNotifyBucketPolicyChange(bucket string, pCh policyChange, objAPI ObjectLayer) error {
+	if pCh.IsRemove {
+		if err := removeBucketPolicy(bucket, objAPI); err != nil {
+			return err
+		}
+	} else {
+		if pCh.BktPolicy == nil {
+			return errInvalidArgument
+		}
+		if err := writeBucketPolicy(bucket, objAPI, pCh.BktPolicy); err != nil {
+			return err
+		}
+	}
+
+	// Notify all peers (including self) to update in-memory state
+	S3PeersUpdateBucketPolicy(bucket, pCh)
+	return nil
+}