Merge pull request #14 from mbilal92/main

FIX: concurrent map read and map write for `globalChunkTbl`, added mutex for globalChunkTbl read/write
omec-project · Nov 8, 2023 · 1354cc3 · 1354cc3
2 parents 2a78831 + 2c2ef2a
commit 1354cc3
Show file tree

Hide file tree

Showing 5 changed files with 29 additions and 18 deletions.
diff --git a/drsm/api.go b/drsm/api.go
@@ -110,8 +110,8 @@ func (d *Drsm) ReleaseInt32ID(id int32) error {
 }
 
 func (d *Drsm) FindOwnerInt32ID(id int32) (*PodId, error) {
-	mutex.Lock()
-	defer mutex.Unlock()
+	d.globalChunkTblMutex.Lock()
+	defer d.globalChunkTblMutex.Unlock()
 	chunkId := id >> 10
 	chunk, found := d.globalChunkTbl[chunkId]
 	if found == true {

diff --git a/drsm/chunk.go b/drsm/chunk.go
@@ -31,7 +31,9 @@ func (d *Drsm) GetNewChunk() (*chunk, error) {
 	for {
 		for {
 			cn = rand.Int31n(d.chunkIdRange)
+			d.globalChunkTblMutex.Lock()
 			_, found := d.globalChunkTbl[cn]
+			d.globalChunkTblMutex.Unlock()
 			if found == true {
 				continue
 			}

diff --git a/drsm/claim.go b/drsm/claim.go
@@ -19,7 +19,9 @@ func (d *Drsm) podDownDetected() {
 			// Given Pod find out current Chunks owned by this POD
 			pd := d.podMap[p]
 			for k, _ := range pd.podChunks {
+				d.globalChunkTblMutex.Lock()
 				c, found := d.globalChunkTbl[k]
+				d.globalChunkTblMutex.Unlock()
 				logger.AppLog.Debugf("Found : %v chunk : %v ", found, c)
 				go c.claimChunk(d)
 			}

diff --git a/drsm/drsm.go b/drsm/drsm.go
@@ -45,22 +45,23 @@ type podData struct {
 }
 
 type Drsm struct {
-	mu              sync.Mutex
-	sharedPoolName  string
-	clientId        PodId
-	db              DbInfo
-	mode            DrsmMode
-	resIdSize       int32
-	localChunkTbl   map[int32]*chunk    // chunkid to chunk
-	globalChunkTbl  map[int32]*chunk    // chunkid to chunk
-	podMap          map[string]*podData // podId to podData
-	podDown         chan string
-	scanChunks      map[int32]*chunk
-	chunkIdRange    int32
-	resourceValidCb func(int32) bool
-	ipModule        ipam.Ipamer
-	prefix          map[string]*ipam.Prefix
-	mongo           *MongoDBLibrary.MongoClient
+	mu                  sync.Mutex
+	sharedPoolName      string
+	clientId            PodId
+	db                  DbInfo
+	mode                DrsmMode
+	resIdSize           int32
+	localChunkTbl       map[int32]*chunk    // chunkid to chunk
+	globalChunkTbl      map[int32]*chunk    // chunkid to chunk
+	podMap              map[string]*podData // podId to podData
+	podDown             chan string
+	scanChunks          map[int32]*chunk
+	chunkIdRange        int32
+	resourceValidCb     func(int32) bool
+	ipModule            ipam.Ipamer
+	prefix              map[string]*ipam.Prefix
+	mongo               *MongoDBLibrary.MongoClient
+	globalChunkTblMutex sync.Mutex
 }
 
 func (d *Drsm) DeletePod(podInstance string) {
@@ -87,6 +88,7 @@ func (d *Drsm) ConstuctDrsm(opt *Options) {
 	d.podMap = make(map[string]*podData)
 	d.podDown = make(chan string, 10)
 	d.scanChunks = make(map[int32]*chunk)
+	d.globalChunkTblMutex = sync.Mutex{}
 	t := time.Now().UnixNano()
 	rand.Seed(t)
 	d.initIpam(opt)

diff --git a/drsm/updates.go b/drsm/updates.go
@@ -157,7 +157,9 @@ func iterateChangeStream(d *Drsm, routineCtx context.Context, stream *mongo.Chan
 				// looks like chunk owner getting change
 				owner := s.Update.UpdFields.PodId
 				c := getChunIdFromDocId(s.DId.Id)
+				d.globalChunkTblMutex.Lock()
 				cp := d.globalChunkTbl[c]
+				d.globalChunkTblMutex.Unlock()
 				// TODO update IP address as well.
 				cp.Owner.PodName = owner
 				cp.Owner.PodIp = s.Update.UpdFields.PodIp
@@ -255,7 +257,10 @@ func (d *Drsm) addChunk(full *FullStream) {
 	c.resourceValidCb = d.resourceValidCb
 
 	pod.podChunks[cid] = c
+
+	d.globalChunkTblMutex.Lock()
 	d.globalChunkTbl[cid] = c
+	d.globalChunkTblMutex.Unlock()
 
 	logger.AppLog.Infof("Chunk id %v, podChunks %v ", cid, pod.podChunks)
 }