use the volume lock for controller parallelization

Author: Hantao (Will) Wang

pkg/common/volume_lock.go

@@ -0,0 +1,47 @@
+/*
+Copyright 2019 The Kubernetes Authors.
+
+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 common
+
+import "sync"
+
+type VolumeLocks struct {
+	locks	map[string]interface{}
+	mux	sync.Mutex
+}
+
+func NewVolumeLocks() *VolumeLocks {
+	return &VolumeLocks{
+		locks: make(map[string]interface{}),
+	}
+}
+
+func (vl *VolumeLocks) TryAcquire(lock string) bool {
+	vl.mux.Lock()
+	defer vl.mux.Unlock()
+	if _, exists := vl.locks[lock]; exists {
+		return false
+	}
+	vl.locks[lock] = nil
+	return true
+}
+
+func (vl *VolumeLocks) Release(lock string) {
+	vl.mux.Lock()
+	defer vl.mux.Unlock()
+	delete(vl.locks, lock)
+}
+

pkg/gce-pd-csi-driver/controller.go

@@ -19,7 +19,6 @@ import (
 	"math/rand"
 	"sort"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/golang/protobuf/ptypes"
@@ -46,7 +45,7 @@ type GCEControllerServer struct {
 
 	// A map storing all volumes with ongoing operations so that additional operations
 	// for that same volume (as defined by Volume Key) return an Aborted error
-	volumes sync.Map
+	volumes *common.VolumeLocks
 }
 
 var _ csi.ControllerServer = &GCEControllerServer{}
@@ -144,10 +143,10 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("CreateVolume replication type '%s' is not supported", replicationType))
 	}
 
-	if _, alreadyExists := gceCS.volumes.LoadOrStore(volKey.String(), true); alreadyExists {
+	if acquired := gceCS.volumes.TryAcquire(volKey.String()); !acquired {
 		return nil, status.Error(codes.Aborted, fmt.Sprintf("An operation with the given Volume Key %s already exists", volKey.String()))
 	}
-	defer gceCS.volumes.Delete(volKey.String())
+	defer gceCS.volumes.Release(volKey.String())
 
 	// Validate if disk already exists
 	existingDisk, err := gceCS.CloudProvider.GetDisk(ctx, volKey)
@@ -232,10 +231,10 @@ func (gceCS *GCEControllerServer) DeleteVolume(ctx context.Context, req *csi.Del
 		return nil, status.Error(codes.NotFound, fmt.Sprintf("Could not find volume with ID %v: %v", volumeID, err))
 	}
 
-	if _, alreadyExists := gceCS.volumes.LoadOrStore(volKey.String(), true); alreadyExists {
+	if acquired := gceCS.volumes.TryAcquire(volKey.String()); !acquired {
 		return nil, status.Error(codes.Aborted, fmt.Sprintf("An operation with the given Volume Key %s already exists", volKey.String()))
 	}
-	defer gceCS.volumes.Delete(volKey.String())
+	defer gceCS.volumes.Release(volKey.String())
 
 	err = gceCS.CloudProvider.DeleteDisk(ctx, volKey)
 	if err != nil {
@@ -273,10 +272,11 @@ func (gceCS *GCEControllerServer) ControllerPublishVolume(ctx context.Context, r
 		return nil, status.Error(codes.NotFound, fmt.Sprintf("Could not find volume with ID %v: %v", volumeID, err))
 	}
 
-	if _, alreadyExists := gceCS.volumes.LoadOrStore(volKey.String(), true); alreadyExists {
-		return nil, status.Error(codes.Aborted, fmt.Sprintf("An operation with the given Volume Key %s already exists", volKey.String()))
+	volKeyStr := volKey.String()
+	if acquired := gceCS.volumes.TryAcquire(nodeID + "/" + volKeyStr); !acquired {
+		return nil, status.Error(codes.Aborted, fmt.Sprintf("An operation with the given Volume Key %s or Node ID %s already exists", volKeyStr, nodeID))
 	}
-	defer gceCS.volumes.Delete(volKey.String())
+	defer gceCS.volumes.Release(nodeID + "/" + volKeyStr)
 
 	// TODO(#253): Check volume capability matches for ALREADY_EXISTS
 	if err = validateVolumeCapability(volumeCapability); err != nil {
@@ -363,10 +363,11 @@ func (gceCS *GCEControllerServer) ControllerUnpublishVolume(ctx context.Context,
 		return nil, err
 	}
 
-	if _, alreadyExists := gceCS.volumes.LoadOrStore(volKey.String(), true); alreadyExists {
-		return nil, status.Error(codes.Aborted, fmt.Sprintf("An operation with the given Volume Key %s already exists", volKey.String()))
+	volKeyStr := volKey.String()
+	if acquired := gceCS.volumes.TryAcquire(nodeID + "/" + volKeyStr); !acquired {
+		return nil, status.Error(codes.Aborted, fmt.Sprintf("An operation with the given Volume Key %s or Node ID %s already exists", volKeyStr, nodeID))
 	}
-	defer gceCS.volumes.Delete(volKey.String())
+	defer gceCS.volumes.Release(nodeID + "/" + volKeyStr)
 
 	instanceZone, instanceName, err := common.NodeIDToZoneAndName(nodeID)
 	if err != nil {
@@ -415,10 +416,10 @@ func (gceCS *GCEControllerServer) ValidateVolumeCapabilities(ctx context.Context
 		return nil, status.Error(codes.NotFound, fmt.Sprintf("Volume ID is of improper format, got %v", volumeID))
 	}
 
-	if _, alreadyExists := gceCS.volumes.LoadOrStore(volKey.String(), true); alreadyExists {
+	if acquired := gceCS.volumes.TryAcquire(volKey.String()); !acquired {
 		return nil, status.Error(codes.Aborted, fmt.Sprintf("An operation with the given Volume Key %s already exists", volKey.String()))
 	}
-	defer gceCS.volumes.Delete(volKey.String())
+	defer gceCS.volumes.Release(volKey.String())
 
 	_, err = gceCS.CloudProvider.GetDisk(ctx, volKey)
 	if err != nil {
@@ -527,10 +528,10 @@ func (gceCS *GCEControllerServer) CreateSnapshot(ctx context.Context, req *csi.C
 		return nil, status.Error(codes.NotFound, fmt.Sprintf("Could not find volume with ID %v: %v", volumeID, err))
 	}
 
-	if _, alreadyExists := gceCS.volumes.LoadOrStore(volKey.String(), true); alreadyExists {
+	if acquired := gceCS.volumes.TryAcquire(volKey.String()); !acquired {
 		return nil, status.Error(codes.Aborted, fmt.Sprintf("An operation with the given Volume Key %s already exists", volKey.String()))
 	}
-	defer gceCS.volumes.Delete(volKey.String())
+	defer gceCS.volumes.Release(volKey.String())
 
 	// Check if snapshot already exists
 	var snapshot *compute.Snapshot

pkg/gce-pd-csi-driver/gce-pd-driver.go

@@ -22,6 +22,7 @@ import (
 	"google.golang.org/grpc/status"
 	"k8s.io/klog"
 	"k8s.io/kubernetes/pkg/util/mount"
+	common "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
 	gce "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute"
 	metadataservice "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/metadata"
 	mountmanager "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/mount-manager"
@@ -144,6 +145,7 @@ func NewControllerServer(gceDriver *GCEDriver, cloudProvider gce.GCECompute, met
 		Driver:          gceDriver,
 		CloudProvider:   cloudProvider,
 		MetadataService: meta,
+		volumes:         common.NewVolumeLocks(),
 	}
 }