diff --git a/internal/pkg/core/device_handler.go b/internal/pkg/core/device_handler.go
index 504526f..7c7dc06 100644
--- a/internal/pkg/core/device_handler.go
+++ b/internal/pkg/core/device_handler.go
@@ -115,7 +115,6 @@
 	onuID         uint32
 	intfID        uint32
 	proxyDeviceID string
-	uniPorts      map[uint32]struct{}
 	losRaised     bool
 	rdiRaised     bool
 }
@@ -140,7 +139,6 @@
 	device.onuID = onuID
 	device.intfID = intfID
 	device.proxyDeviceID = proxyDevID
-	device.uniPorts = make(map[uint32]struct{})
 	device.losRaised = losRaised
 	return &device
 }
@@ -1409,19 +1407,6 @@
 	return nil
 }
 
-// AddUniPortToOnu adds the uni port to the onu device
-func (dh *DeviceHandler) AddUniPortToOnu(ctx context.Context, intfID, onuID, uniPort uint32) {
-	onuKey := dh.formOnuKey(intfID, onuID)
-
-	if onuDevice, ok := dh.onus.Load(onuKey); ok {
-		// add it to the uniPort map for the onu device
-		if _, ok = onuDevice.(*OnuDevice).uniPorts[uniPort]; !ok {
-			onuDevice.(*OnuDevice).uniPorts[uniPort] = struct{}{}
-			logger.Debugw(ctx, "adding-uni-port", log.Fields{"port": uniPort, "intf-id": intfID, "onuId": onuID})
-		}
-	}
-}
-
 // UpdatePmConfig updates the pm metrics.
 func (dh *DeviceHandler) UpdatePmConfig(ctx context.Context, pmConfigs *voltha.PmConfigs) {
 	logger.Infow(ctx, "update-pm-configs", log.Fields{"device-id": dh.device.Id, "pm-configs": pmConfigs})
diff --git a/internal/pkg/core/openolt_flowmgr.go b/internal/pkg/core/openolt_flowmgr.go
index 3d928c4..709735a 100644
--- a/internal/pkg/core/openolt_flowmgr.go
+++ b/internal/pkg/core/openolt_flowmgr.go
@@ -173,6 +173,13 @@
 	BinaryStringPrefix = "0b"
 	// BinaryBit1 is binary bit 1 expressed as a character
 	BinaryBit1 = '1'
+
+	// MapMutex
+	maxRetry  = 300
+	maxDelay  = 100000000
+	baseDelay = 10000000
+	factor    = 1.1
+	jitter    = 0.2
 )
 
 type gemPortKey struct {
@@ -215,6 +222,7 @@
 	deviceHandler      *DeviceHandler
 	resourceMgr        *rsrcMgr.OpenOltResourceMgr
 	onuIdsLock         sync.RWMutex
+	perGemPortLock     *mapmutex.Mutex                    // lock to be used to access the flowsUsedByGemPort map
 	flowsUsedByGemPort map[gemPortKey][]uint32            //gem port id to flow ids
 	packetInGemPort    map[rsrcMgr.PacketInInfoKey]uint32 //packet in gem port local cache
 	// TODO create a type rsrcMgr.OnuGemInfos to be used instead of []rsrcMgr.OnuGemInfo
@@ -256,7 +264,8 @@
 	}
 	flowMgr.onuGemInfoLock = sync.RWMutex{}
 	flowMgr.pendingFlowDelete = sync.Map{}
-	flowMgr.perUserFlowHandleLock = mapmutex.NewCustomizedMapMutex(300, 100000000, 10000000, 1.1, 0.2)
+	flowMgr.perUserFlowHandleLock = mapmutex.NewCustomizedMapMutex(maxRetry, maxDelay, baseDelay, factor, jitter)
+	flowMgr.perGemPortLock = mapmutex.NewCustomizedMapMutex(maxRetry, maxDelay, baseDelay, factor, jitter)
 	flowMgr.interfaceToMcastQueueMap = make(map[uint32]*queueInfoBrief)
 	//load interface to multicast queue map from kv store
 	flowMgr.loadInterfaceToMulticastQueueMap(ctx)
@@ -265,19 +274,35 @@
 }
 
 func (f *OpenOltFlowMgr) registerFlow(ctx context.Context, flowFromCore *ofp.OfpFlowStats, deviceFlow *openoltpb2.Flow) error {
-	logger.Debugw(ctx, "registering-flow-for-device ",
-		log.Fields{
-			"flow":      flowFromCore,
-			"device-id": f.deviceHandler.device.Id})
 	gemPK := gemPortKey{uint32(deviceFlow.AccessIntfId), uint32(deviceFlow.GemportId)}
-	flowIDList, ok := f.flowsUsedByGemPort[gemPK]
-	if !ok {
-		flowIDList = []uint32{deviceFlow.FlowId}
+	if f.perGemPortLock.TryLock(gemPK) {
+		logger.Debugw(ctx, "registering-flow-for-device ",
+			log.Fields{
+				"flow":      flowFromCore,
+				"device-id": f.deviceHandler.device.Id})
+		flowIDList, ok := f.flowsUsedByGemPort[gemPK]
+		if !ok {
+			flowIDList = []uint32{deviceFlow.FlowId}
+		}
+		flowIDList = appendUnique(flowIDList, deviceFlow.FlowId)
+		f.flowsUsedByGemPort[gemPK] = flowIDList
+
+		f.perGemPortLock.Unlock(gemPK)
+
+		// update the flowids for a gem to the KVstore
+		return f.resourceMgr.UpdateFlowIDsForGem(ctx, uint32(deviceFlow.AccessIntfId), uint32(deviceFlow.GemportId), flowIDList)
 	}
-	flowIDList = appendUnique(flowIDList, deviceFlow.FlowId)
-	f.flowsUsedByGemPort[gemPK] = flowIDList
-	// update the flowids for a gem to the KVstore
-	return f.resourceMgr.UpdateFlowIDsForGem(ctx, uint32(deviceFlow.AccessIntfId), uint32(deviceFlow.GemportId), flowIDList)
+	logger.Error(ctx, "failed-to-acquire-per-gem-port-lock",
+		log.Fields{
+			"flow-from-core": flowFromCore,
+			"device-id":      f.deviceHandler.device.Id,
+			"key":            gemPK,
+		})
+	return olterrors.NewErrAdapter("failed-to-acquire-per-gem-port-lock", log.Fields{
+		"flow-from-core": flowFromCore,
+		"device-id":      f.deviceHandler.device.Id,
+		"key":            gemPK,
+	}, nil)
 }
 
 func (f *OpenOltFlowMgr) divideAndAddFlow(ctx context.Context, intfID uint32, onuID uint32, uniID uint32, portNo uint32,
@@ -2049,26 +2074,45 @@
 			}
 
 			gemPK := gemPortKey{Intf, uint32(gemPortID)}
-			if f.isGemPortUsedByAnotherFlow(gemPK) {
-				flowIDs := f.flowsUsedByGemPort[gemPK]
-				for i, flowIDinMap := range flowIDs {
-					if flowIDinMap == flowID {
-						flowIDs = append(flowIDs[:i], flowIDs[i+1:]...)
-						// everytime flowsUsedByGemPort cache is updated the same should be updated
-						// in kv store by calling UpdateFlowIDsForGem
-						f.flowsUsedByGemPort[gemPK] = flowIDs
-						if err := f.resourceMgr.UpdateFlowIDsForGem(ctx, Intf, uint32(gemPortID), flowIDs); err != nil {
-							return err
+			used, err := f.isGemPortUsedByAnotherFlow(ctx, gemPK)
+			if err != nil {
+				return err
+			}
+			if used {
+				if f.perGemPortLock.TryLock(gemPK) {
+					flowIDs := f.flowsUsedByGemPort[gemPK]
+					for i, flowIDinMap := range flowIDs {
+						if flowIDinMap == flowID {
+							flowIDs = append(flowIDs[:i], flowIDs[i+1:]...)
+							// everytime flowsUsedByGemPort cache is updated the same should be updated
+							// in kv store by calling UpdateFlowIDsForGem
+							f.flowsUsedByGemPort[gemPK] = flowIDs
+							if err := f.resourceMgr.UpdateFlowIDsForGem(ctx, Intf, uint32(gemPortID), flowIDs); err != nil {
+								return err
+							}
+							break
 						}
-						break
 					}
+					logger.Debugw(ctx, "gem-port-id-is-still-used-by-other-flows",
+						log.Fields{
+							"gemport-id":  gemPortID,
+							"usedByFlows": flowIDs,
+							"device-id":   f.deviceHandler.device.Id})
+					f.perGemPortLock.Unlock(gemPK)
+					return nil
 				}
-				logger.Debugw(ctx, "gem-port-id-is-still-used-by-other-flows",
+
+				logger.Error(ctx, "failed-to-acquire-per-gem-port-lock",
 					log.Fields{
-						"gemport-id":  gemPortID,
-						"usedByFlows": flowIDs,
-						"device-id":   f.deviceHandler.device.Id})
-				return nil
+						"gemport-id": gemPortID,
+						"device-id":  f.deviceHandler.device.Id,
+						"key":        gemPK,
+					})
+				return olterrors.NewErrAdapter("failed-to-acquire-per-gem-port-lock", log.Fields{
+					"gemport-id": gemPortID,
+					"device-id":  f.deviceHandler.device.Id,
+					"key":        gemPK,
+				}, nil)
 			}
 			logger.Debugf(ctx, "gem-port-id %d is-not-used-by-another-flow--releasing-the-gem-port", gemPortID)
 			f.resourceMgr.RemoveGemPortIDForOnu(ctx, Intf, uint32(onuID), uint32(uniID), uint32(gemPortID))
@@ -2079,7 +2123,16 @@
 			f.onuIdsLock.Lock()
 			//everytime an entry is deleted from flowsUsedByGemPort cache, the same should be updated in kv as well
 			// by calling DeleteFlowIDsForGem
-			delete(f.flowsUsedByGemPort, gemPK)
+			if f.perGemPortLock.TryLock(gemPK) {
+				delete(f.flowsUsedByGemPort, gemPK)
+				f.perGemPortLock.Unlock(gemPK)
+			} else {
+				logger.Error(ctx, "failed-to-acquire-per-gem-port-lock",
+					log.Fields{
+						"device-id": f.deviceHandler.device.Id,
+						"key":       gemPK,
+					})
+			}
 			f.resourceMgr.DeleteFlowIDsForGem(ctx, Intf, uint32(gemPortID))
 			f.resourceMgr.FreeGemPortID(ctx, Intf, uint32(onuID), uint32(uniID), uint32(gemPortID))
 			f.onuIdsLock.Unlock()
@@ -2459,7 +2512,6 @@
 		return f.addIgmpTrapFlowOnNNI(ctx, flow, classifierInfo, portNo)
 	}
 
-	f.deviceHandler.AddUniPortToOnu(ctx, intfID, onuID, portNo)
 	f.resourceMgr.AddUniPortToOnuInfo(ctx, intfID, onuID, portNo)
 
 	TpID, err := getTpIDFromFlow(ctx, flow)
@@ -3566,9 +3618,21 @@
 	}()
 }
 
-func (f *OpenOltFlowMgr) isGemPortUsedByAnotherFlow(gemPK gemPortKey) bool {
-	flowIDList := f.flowsUsedByGemPort[gemPK]
-	return len(flowIDList) > 1
+func (f *OpenOltFlowMgr) isGemPortUsedByAnotherFlow(ctx context.Context, gemPK gemPortKey) (bool, error) {
+	if f.perGemPortLock.TryLock(gemPK) {
+		flowIDList := f.flowsUsedByGemPort[gemPK]
+		f.perGemPortLock.Unlock(gemPK)
+		return len(flowIDList) > 1, nil
+	}
+	logger.Error(ctx, "failed-to-acquire-per-gem-port-lock",
+		log.Fields{
+			"device-id": f.deviceHandler.device.Id,
+			"key":       gemPK,
+		})
+	return false, olterrors.NewErrAdapter("failed-to-acquire-per-gem-port-lock", log.Fields{
+		"device-id": f.deviceHandler.device.Id,
+		"key":       gemPK,
+	}, nil)
 }
 
 func (f *OpenOltFlowMgr) isTechProfileUsedByAnotherGem(ctx context.Context, ponIntf uint32, onuID uint32, uniID uint32, tpID uint32, tpInst *tp.TechProfile, gemPortID uint32) (bool, uint32) {
@@ -3929,7 +3993,17 @@
 	}
 	for gem, FlowIDs := range flowIDsList {
 		gemPK := gemPortKey{intf, uint32(gem)}
-		f.flowsUsedByGemPort[gemPK] = FlowIDs
+		if f.perGemPortLock.TryLock(gemPK) {
+			f.flowsUsedByGemPort[gemPK] = FlowIDs
+			f.perGemPortLock.Unlock(gemPK)
+		} else {
+			logger.Error(ctx, "failed-to-acquire-per-gem-port-lock",
+				log.Fields{
+					"intf-id":   intf,
+					"device-id": f.deviceHandler.device.Id,
+					"key":       gemPK,
+				})
+		}
 	}
 }