move lockmap to util package (#7714)

Author: MartinForReal

Diff for: pkg/provider/azure.go

@@ -77,6 +77,7 @@ import (
 	"sigs.k8s.io/cloud-provider-azure/pkg/consts"
 	azureconfig "sigs.k8s.io/cloud-provider-azure/pkg/provider/config"
 	"sigs.k8s.io/cloud-provider-azure/pkg/retry"
+	"sigs.k8s.io/cloud-provider-azure/pkg/util/lockmap"
 	utilsets "sigs.k8s.io/cloud-provider-azure/pkg/util/sets"
 	"sigs.k8s.io/cloud-provider-azure/pkg/util/taints"
 )
@@ -190,7 +191,7 @@ type Cloud struct {
 	// node-sync-loop routine and service-reconcile routine should not update LoadBalancer at the same time
 	serviceReconcileLock sync.Mutex
 
-	lockMap *LockMap
+	lockMap *lockmap.LockMap
 	// multipleStandardLoadBalancerConfigurationsSynced make sure the `reconcileMultipleStandardLoadBalancerConfigurations`
 	// runs only once every time the cloud provide restarts.
 	multipleStandardLoadBalancerConfigurationsSynced bool
@@ -224,7 +225,7 @@ func NewCloud(ctx context.Context, clientBuilder cloudprovider.ControllerClientB
 
 	az.ipv6DualStackEnabled = true
 	if az.lockMap == nil {
-		az.lockMap = newLockMap()
+		az.lockMap = lockmap.NewLockMap()
 	}
 
 	if clientBuilder != nil {
@@ -386,7 +387,7 @@ func (az *Cloud) InitializeCloudFromConfig(ctx context.Context, config *config.C
 		return err
 	}
 
-	az.lockMap = newLockMap()
+	az.lockMap = lockmap.NewLockMap()
 	az.Config = *config
 	az.Environment = *env
 	az.ResourceRequestBackoff = resourceRequestBackoff

Diff for: pkg/provider/azure_fakes.go

@@ -47,6 +47,7 @@ import (
 	"sigs.k8s.io/cloud-provider-azure/pkg/provider/routetable"
 	"sigs.k8s.io/cloud-provider-azure/pkg/provider/securitygroup"
 	"sigs.k8s.io/cloud-provider-azure/pkg/provider/subnet"
+	"sigs.k8s.io/cloud-provider-azure/pkg/util/lockmap"
 	utilsets "sigs.k8s.io/cloud-provider-azure/pkg/util/sets"
 )
 
@@ -110,7 +111,7 @@ func GetTestCloud(ctrl *gomock.Controller) (az *Cloud) {
 		nodePrivateIPs:           map[string]*utilsets.IgnoreCaseSet{},
 		routeCIDRs:               map[string]string{},
 		eventRecorder:            &record.FakeRecorder{},
-		lockMap:                  newLockMap(),
+		lockMap:                  lockmap.NewLockMap(),
 	}
 	az.DisksClient = mockdiskclient.NewMockInterface(ctrl)
 	az.InterfacesClient = mockinterfaceclient.NewMockInterface(ctrl)

Diff for: pkg/provider/azure_storageaccount_test.go

@@ -45,6 +45,7 @@ import (
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/subnetclient/mocksubnetclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/cache"
 	"sigs.k8s.io/cloud-provider-azure/pkg/retry"
+	"sigs.k8s.io/cloud-provider-azure/pkg/util/lockmap"
 )
 
 const TestLocation = "testLocation"
@@ -702,7 +703,7 @@ func TestAddStorageAccountTags(t *testing.T) {
 	defer cancel()
 
 	cloud := &Cloud{}
-	cloud.lockMap = newLockMap()
+	cloud.lockMap = lockmap.NewLockMap()
 	tests := []struct {
 		name           string
 		subsID         string
@@ -806,7 +807,7 @@ func TestRemoveStorageAccountTags(t *testing.T) {
 
 	getter := func(_ context.Context, _ string) (interface{}, error) { return nil, nil }
 	cloud.storageAccountCache, _ = cache.NewTimedCache(time.Minute, getter, false)
-	cloud.lockMap = newLockMap()
+	cloud.lockMap = lockmap.NewLockMap()
 	for _, test := range tests {
 		mockStorageAccountsClient := mockstorageaccountclient.NewMockInterface(ctrl)
 		cloud.StorageAccountClient = mockStorageAccountsClient

Diff for: pkg/provider/azure_utils.go

@@ -47,44 +47,6 @@ var strToExtendedLocationType = map[string]armnetwork.ExtendedLocationTypes{
 	"edgezone": armnetwork.ExtendedLocationTypesEdgeZone,
 }
 
-// LockMap used to lock on entries
-type LockMap struct {
-	sync.Mutex
-	mutexMap map[string]*sync.Mutex
-}
-
-// NewLockMap returns a new lock map
-func newLockMap() *LockMap {
-	return &LockMap{
-		mutexMap: make(map[string]*sync.Mutex),
-	}
-}
-
-// LockEntry acquires a lock associated with the specific entry
-func (lm *LockMap) LockEntry(entry string) {
-	lm.Lock()
-	// check if entry does not exists, then add entry
-	mutex, exists := lm.mutexMap[entry]
-	if !exists {
-		mutex = &sync.Mutex{}
-		lm.mutexMap[entry] = mutex
-	}
-	lm.Unlock()
-	mutex.Lock()
-}
-
-// UnlockEntry release the lock associated with the specific entry
-func (lm *LockMap) UnlockEntry(entry string) {
-	lm.Lock()
-	defer lm.Unlock()
-
-	mutex, exists := lm.mutexMap[entry]
-	if !exists {
-		return
-	}
-	mutex.Unlock()
-}
-
 func getContextWithCancel() (context.Context, context.CancelFunc) {
 	return context.WithCancel(context.Background())
 }

Diff for: pkg/provider/azure_utils_test.go

@@ -21,7 +21,6 @@ import (
 	"reflect"
 	"sync"
 	"testing"
-	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/compute/armcompute/v6"
 	"github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/network/armnetwork/v6"
@@ -39,67 +38,6 @@ import (
 	"sigs.k8s.io/cloud-provider-azure/pkg/consts"
 )
 
-func TestSimpleLockEntry(t *testing.T) {
-	testLockMap := newLockMap()
-
-	callbackChan1 := make(chan interface{})
-	go testLockMap.lockAndCallback(t, "entry1", callbackChan1)
-	ensureCallbackHappens(t, callbackChan1)
-}
-
-func TestSimpleLockUnlockEntry(t *testing.T) {
-	testLockMap := newLockMap()
-
-	callbackChan1 := make(chan interface{})
-	go testLockMap.lockAndCallback(t, "entry1", callbackChan1)
-	ensureCallbackHappens(t, callbackChan1)
-	testLockMap.UnlockEntry("entry1")
-}
-
-func TestConcurrentLockEntry(t *testing.T) {
-	testLockMap := newLockMap()
-
-	callbackChan1 := make(chan interface{})
-	callbackChan2 := make(chan interface{})
-
-	go testLockMap.lockAndCallback(t, "entry1", callbackChan1)
-	ensureCallbackHappens(t, callbackChan1)
-
-	go testLockMap.lockAndCallback(t, "entry1", callbackChan2)
-	ensureNoCallback(t, callbackChan2)
-
-	testLockMap.UnlockEntry("entry1")
-	ensureCallbackHappens(t, callbackChan2)
-	testLockMap.UnlockEntry("entry1")
-}
-
-func (lm *LockMap) lockAndCallback(_ *testing.T, entry string, callbackChan chan<- interface{}) {
-	lm.LockEntry(entry)
-	callbackChan <- true
-}
-
-var callbackTimeout = 2 * time.Second
-
-func ensureCallbackHappens(t *testing.T, callbackChan <-chan interface{}) bool {
-	select {
-	case <-callbackChan:
-		return true
-	case <-time.After(callbackTimeout):
-		t.Fatalf("timed out waiting for callback")
-		return false
-	}
-}
-
-func ensureNoCallback(t *testing.T, callbackChan <-chan interface{}) bool {
-	select {
-	case <-callbackChan:
-		t.Fatalf("unexpected callback")
-		return false
-	case <-time.After(callbackTimeout):
-		return true
-	}
-}
-
 func TestReconcileTags(t *testing.T) {
 	for _, testCase := range []struct {
 		description, systemTags                      string

Diff for: pkg/provider/azure_vmss.go

@@ -40,6 +40,7 @@ import (
 	"sigs.k8s.io/cloud-provider-azure/pkg/consts"
 	"sigs.k8s.io/cloud-provider-azure/pkg/metrics"
 	"sigs.k8s.io/cloud-provider-azure/pkg/provider/virtualmachine"
+	"sigs.k8s.io/cloud-provider-azure/pkg/util/lockmap"
 	vmutil "sigs.k8s.io/cloud-provider-azure/pkg/util/vm"
 )
 
@@ -102,7 +103,7 @@ type ScaleSet struct {
 	nonVmssUniformNodesCache azcache.Resource
 
 	// lockMap in cache refresh
-	lockMap *LockMap
+	lockMap *lockmap.LockMap
 }
 
 // RefreshCaches invalidates and renew all related caches.
@@ -153,14 +154,14 @@ func newScaleSet(az *Cloud) (VMSet, error) {
 		Cloud:           az,
 		availabilitySet: as,
 		flexScaleSet:    fs,
-		lockMap:         newLockMap(),
+		lockMap:         lockmap.NewLockMap(),
 	}
 
 	if err := ss.RefreshCaches(); err != nil {
 		return nil, err
 	}
 
-	ss.lockMap = newLockMap()
+	ss.lockMap = lockmap.NewLockMap()
 	return ss, nil
 }
 

Diff for: pkg/provider/azure_vmssflex.go

@@ -38,6 +38,7 @@ import (
 	azcache "sigs.k8s.io/cloud-provider-azure/pkg/cache"
 	"sigs.k8s.io/cloud-provider-azure/pkg/consts"
 	"sigs.k8s.io/cloud-provider-azure/pkg/metrics"
+	"sigs.k8s.io/cloud-provider-azure/pkg/util/lockmap"
 	vmutil "sigs.k8s.io/cloud-provider-azure/pkg/util/vm"
 )
 
@@ -54,7 +55,7 @@ type FlexScaleSet struct {
 	vmssFlexVMCache          azcache.Resource
 
 	// lockMap in cache refresh
-	lockMap *LockMap
+	lockMap *lockmap.LockMap
 }
 
 // RefreshCaches invalidates and renew all related caches.
@@ -79,7 +80,7 @@ func newFlexScaleSet(az *Cloud) (VMSet, error) {
 		Cloud:                    az,
 		vmssFlexVMNameToVmssID:   &sync.Map{},
 		vmssFlexVMNameToNodeName: &sync.Map{},
-		lockMap:                  newLockMap(),
+		lockMap:                  lockmap.NewLockMap(),
 	}
 
 	if err := fs.RefreshCaches(); err != nil {

Diff for: pkg/util/lockmap/lockmap.go

@@ -0,0 +1,57 @@
+/*
+Copyright 2021 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 lockmap
+
+import "sync"
+
+// LockMap used to lock on entries
+type LockMap struct {
+	sync.Mutex
+	mutexMap map[string]*sync.Mutex
+}
+
+// NewLockMap returns a new lock map
+func NewLockMap() *LockMap {
+	return &LockMap{
+		mutexMap: make(map[string]*sync.Mutex),
+	}
+}
+
+// LockEntry acquires a lock associated with the specific entry
+func (lm *LockMap) LockEntry(entry string) {
+	lm.Lock()
+	// check if entry does not exists, then add entry
+	mutex, exists := lm.mutexMap[entry]
+	if !exists {
+		mutex = &sync.Mutex{}
+		lm.mutexMap[entry] = mutex
+	}
+	lm.Unlock()
+	mutex.Lock()
+}
+
+// UnlockEntry release the lock associated with the specific entry
+func (lm *LockMap) UnlockEntry(entry string) {
+	lm.Lock()
+	defer lm.Unlock()
+
+	mutex, exists := lm.mutexMap[entry]
+	if !exists {
+		return
+	}
+	mutex.Unlock()
+}