use the volume locks for node operations too

Author: Hantao (Will) Wang

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

@@ -137,6 +137,7 @@ func NewNodeServer(gceDriver *GCEDriver, mounter *mount.SafeFormatAndMount, devi
 		Mounter:         mounter,
 		DeviceUtils:     deviceUtils,
 		MetadataService: meta,
+		volumes:         common.NewVolumeLocks(),
 	}
 }
 

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

@@ -18,7 +18,6 @@ import (
 	"fmt"
 	"os"
 	"strings"
-	"sync"
 
 	"context"
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
@@ -39,7 +38,7 @@ type GCENodeServer struct {
 
 	// A map storing all volumes with ongoing operations so that additional operations
 	// for that same volume (as defined by VolumeID) return an Aborted error
-	volumes sync.Map
+	volumes *common.VolumeLocks
 }
 
 var _ csi.NodeServer = &GCENodeServer{}
@@ -74,10 +73,10 @@ func (ns *GCENodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePub
 		return nil, status.Error(codes.InvalidArgument, "NodePublishVolume Volume Capability must be provided")
 	}
 
-	if _, alreadyExists := ns.volumes.LoadOrStore(volumeID, true); alreadyExists {
+	if acquired := ns.volumes.TryAcquire(volumeID); !acquired {
 		return nil, status.Error(codes.Aborted, fmt.Sprintf("An operation with the given Volume ID %s already exists", volumeID))
 	}
-	defer ns.volumes.Delete(volumeID)
+	defer ns.volumes.Release(volumeID)
 
 	if err := validateVolumeCapability(volumeCapability); err != nil {
 		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("VolumeCapability is invalid: %v", err))
@@ -189,18 +188,18 @@ func (ns *GCENodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeU
 
 	// Validate Arguments
 	targetPath := req.GetTargetPath()
-	volID := req.GetVolumeId()
-	if len(volID) == 0 {
+	volumeID := req.GetVolumeId()
+	if len(volumeID) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "NodeUnpublishVolume Volume ID must be provided")
 	}
 	if len(targetPath) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "NodeUnpublishVolume Target Path must be provided")
 	}
 
-	if _, alreadyExists := ns.volumes.LoadOrStore(volID, true); alreadyExists {
-		return nil, status.Error(codes.Aborted, fmt.Sprintf("An operation with the given Volume ID %s already exists", volID))
+	if acquired := ns.volumes.TryAcquire(volumeID); !acquired {
+		return nil, status.Error(codes.Aborted, fmt.Sprintf("An operation with the given Volume ID %s already exists", volumeID))
 	}
-	defer ns.volumes.Delete(volID)
+	defer ns.volumes.Release(volumeID)
 
 	err := mount.CleanupMountPoint(targetPath, ns.Mounter.Interface, false /* bind mount */)
 	if err != nil {
@@ -227,10 +226,10 @@ func (ns *GCENodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStage
 		return nil, status.Error(codes.InvalidArgument, "NodeStageVolume Volume Capability must be provided")
 	}
 
-	if _, alreadyExists := ns.volumes.LoadOrStore(volumeID, true); alreadyExists {
+	if acquired := ns.volumes.TryAcquire(volumeID); !acquired {
 		return nil, status.Error(codes.Aborted, fmt.Sprintf("An operation with the given Volume ID %s already exists", volumeID))
 	}
-	defer ns.volumes.Delete(volumeID)
+	defer ns.volumes.Release(volumeID)
 
 	if err := validateVolumeCapability(volumeCapability); err != nil {
 		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("VolumeCapability is invalid: %v", err))
@@ -321,10 +320,10 @@ func (ns *GCENodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUns
 		return nil, status.Error(codes.InvalidArgument, "NodeUnstageVolume Staging Target Path must be provided")
 	}
 
-	if _, alreadyExists := ns.volumes.LoadOrStore(volumeID, true); alreadyExists {
+	if acquired := ns.volumes.TryAcquire(volumeID); !acquired {
 		return nil, status.Error(codes.Aborted, fmt.Sprintf("An operation with the given Volume ID %s already exists", volumeID))
 	}
-	defer ns.volumes.Delete(volumeID)
+	defer ns.volumes.Release(volumeID)
 
 	err := mount.CleanupMountPoint(stagingTargetPath, ns.Mounter.Interface, false /* bind mount */)
 	if err != nil {