mirrors/ceph-csi
1
0
Fork 0
mirror of https://github.com/ceph/ceph-csi.git synced 2024-11-09 16:00:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

cephfs: safeguard subVolumeGroupCreated map from race condition

Browse Source 
Multiple go-routines may simultaneously create the
subVolumeGroupCreated map or  write into it
for a particular group.

This commit safeguards subVolumeGroupCreated map
from concurrent creation/writes while allowing for multiple
readers.

Signed-off-by: Rakshith R <rar@redhat.com>
This commit is contained in:
Rakshith R 2023-10-10 20:32:17 +05:30 committed by mergify[bot]
parent 82f1323af4
commit d516a1d66d
2 changed files with 32 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
internal/cephfs/core/metadata.go
View File

@ -51,6 +51,32 @@ func (s *subVolumeClient) isUnsupportedSubVolMetadata(err error) bool {
return true
}
// isSubVolumeGroupCreated returns true if subvolume group is created.
func (s *subVolumeClient) isSubVolumeGroupCreated() bool {
newLocalClusterState(s.clusterID)
clusterAdditionalInfo[s.clusterID].subVolumeGroupsRWMutex.RLock()
defer clusterAdditionalInfo[s.clusterID].subVolumeGroupsRWMutex.RUnlock()
if clusterAdditionalInfo[s.clusterID].subVolumeGroupsCreated == nil {
return false
}
return clusterAdditionalInfo[s.clusterID].subVolumeGroupsCreated[s.SubvolumeGroup]
}
// updateSubVolumeGroupCreated updates subvolume group created map.
// If the map is nil, it creates a new map and updates it.
func (s *subVolumeClient) updateSubVolumeGroupCreated(state bool) {
clusterAdditionalInfo[s.clusterID].subVolumeGroupsRWMutex.Lock()
defer clusterAdditionalInfo[s.clusterID].subVolumeGroupsRWMutex.Unlock()
if clusterAdditionalInfo[s.clusterID].subVolumeGroupsCreated == nil {
clusterAdditionalInfo[s.clusterID].subVolumeGroupsCreated = make(map[string]bool)
}
clusterAdditionalInfo[s.clusterID].subVolumeGroupsCreated[s.SubvolumeGroup] = state
}
// setMetadata sets custom metadata on the subvolume in a volume as a
// key-value pair.
func (s *subVolumeClient) setMetadata(key, value string) error {

10
internal/cephfs/core/volume.go
View File

@ -215,6 +215,9 @@ type localClusterState struct {
// set true once a subvolumegroup is created
// for corresponding filesystem in a cluster.
subVolumeGroupsCreated map[string]bool
// subVolumeGroupsRWMutex is used to protect subVolumeGroupsCreated map
// against concurrent writes while allowing multiple readers.
subVolumeGroupsRWMutex sync.RWMutex
}
func newLocalClusterState(clusterID string) {
@ -224,7 +227,6 @@ func newLocalClusterState(clusterID string) {
defer clusterAdditionalInfoMutex.Unlock()
if _, keyPresent := clusterAdditionalInfo[clusterID]; !keyPresent {
clusterAdditionalInfo[clusterID] = &localClusterState{}
clusterAdditionalInfo[clusterID].subVolumeGroupsCreated = make(map[string]bool)
}
}
@ -240,7 +242,7 @@ func (s *subVolumeClient) CreateVolume(ctx context.Context) error {
}
// create subvolumegroup if not already created for the cluster.
if !clusterAdditionalInfo[s.clusterID].subVolumeGroupsCreated[s.FsName] {
if !s.isSubVolumeGroupCreated() {
opts := fsAdmin.SubVolumeGroupOptions{}
err = ca.CreateSubVolumeGroup(s.FsName, s.SubvolumeGroup, &opts)
if err != nil {
@ -254,7 +256,7 @@ func (s *subVolumeClient) CreateVolume(ctx context.Context) error {
return err
}
log.DebugLog(ctx, "cephfs: created subvolume group %s", s.SubvolumeGroup)
clusterAdditionalInfo[s.clusterID].subVolumeGroupsCreated[s.FsName] = true
s.updateSubVolumeGroupCreated(true)
}
opts := fsAdmin.SubVolumeOptions{
@ -272,7 +274,7 @@ func (s *subVolumeClient) CreateVolume(ctx context.Context) error {
if errors.Is(err, rados.ErrNotFound) {
// Reset the subVolumeGroupsCreated so that we can try again to create the
// subvolumegroup in next request if the error is Not Found.
clusterAdditionalInfo[s.clusterID].subVolumeGroupsCreated[s.FsName] = false
s.updateSubVolumeGroupCreated(false)
}
return err