[VOL-3269] Using a global lock on onuGemInfoLock
Change-Id: Id6fb3ab27b3c5a6fca74d18d306d07e706a355bf
diff --git a/go.mod b/go.mod
old mode 100755
new mode 100644
diff --git a/internal/pkg/core/openolt_flowmgr.go b/internal/pkg/core/openolt_flowmgr.go
index b31c312..28d2065 100644
--- a/internal/pkg/core/openolt_flowmgr.go
+++ b/internal/pkg/core/openolt_flowmgr.go
@@ -218,8 +218,9 @@
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
- onuGemInfo [][]rsrcMgr.OnuGemInfo //onu, gem and uni info local cache, indexed by IntfId
- onuGemInfoLock []sync.RWMutex // lock by Pon Port
+ onuGemInfo map[uint32][]rsrcMgr.OnuGemInfo //onu, gem and uni info local cache, indexed by IntfId
+ // We need to have a global lock on the onuGemInfo map
+ onuGemInfoLock sync.RWMutex
pendingFlowDelete sync.Map
// The mapmutex.Mutex can be fine tuned to use mapmutex.NewCustomizedMapMutex
perUserFlowHandleLock *mapmutex.Mutex
@@ -244,7 +245,7 @@
flowMgr.flowsUsedByGemPort = make(map[gemPortKey][]uint32)
flowMgr.packetInGemPort = make(map[rsrcMgr.PacketInInfoKey]uint32)
ponPorts := rMgr.DevInfo.GetPonPorts()
- flowMgr.onuGemInfo = make([][]rsrcMgr.OnuGemInfo, ponPorts)
+ flowMgr.onuGemInfo = make(map[uint32][]rsrcMgr.OnuGemInfo, ponPorts)
//Load the onugem info cache from kv store on flowmanager start
for idx = 0; idx < ponPorts; idx++ {
if flowMgr.onuGemInfo[idx], err = rMgr.GetOnuGemInfo(ctx, idx); err != nil {
@@ -253,7 +254,7 @@
//Load flowID list per gem map per interface from the kvstore.
flowMgr.loadFlowIDlistForGem(ctx, idx)
}
- flowMgr.onuGemInfoLock = make([]sync.RWMutex, ponPorts)
+ flowMgr.onuGemInfoLock = sync.RWMutex{}
flowMgr.pendingFlowDelete = sync.Map{}
flowMgr.perUserFlowHandleLock = mapmutex.NewCustomizedMapMutex(300, 100000000, 10000000, 1.1, 0.2)
flowMgr.interfaceToMcastQueueMap = make(map[uint32]*queueInfoBrief)
@@ -1556,29 +1557,6 @@
return &flows
}
-//func (f *OpenOltFlowMgr) getUpdatedFlowInfo(flow *openolt_pb2.Flow, flowStoreCookie uint64, flowCategory string) *[]rsrcMgr.FlowInfo {
-// var flows []rsrcMgr.FlowInfo = []rsrcMgr.FlowInfo{rsrcMgr.FlowInfo{Flow: flow, FlowCategory: flowCategory, FlowStoreCookie: flowStoreCookie}}
-// var intfId uint32
-// /* For flows which trap out of the NNI, the AccessIntfId is invalid
-// (set to -1). In such cases, we need to refer to the NetworkIntfId .
-// */
-// if flow.AccessIntfId != -1 {
-// intfId = uint32(flow.AccessIntfId)
-// } else {
-// intfId = uint32(flow.NetworkIntfId)
-// }
-// // Get existing flows matching flowid for given subscriber from KV store
-// existingFlows := f.resourceMgr.GetFlowIDInfo(intfId, uint32(flow.OnuId), uint32(flow.UniId), flow.FlowId)
-// if existingFlows != nil {
-// logger.Debugw(ctx, "Flow exists for given flowID, appending it to current flow", log.Fields{"flowID": flow.FlowId})
-// for _, f := range *existingFlows {
-// flows = append(flows, f)
-// }
-// }
-// logger.Debugw(ctx, "Updated flows for given flowID and onuid", log.Fields{"updatedflow": flows, "flowid": flow.FlowId, "onu": flow.OnuId})
-// return &flows
-//}
-
func (f *OpenOltFlowMgr) updateFlowInfoToKVStore(ctx context.Context, intfID int32, onuID int32, uniID int32, flowID uint32, flows *[]rsrcMgr.FlowInfo) error {
logger.Debugw(ctx, "storing-flow(s)-into-kv-store", log.Fields{
"flow-id": flowID,
@@ -1678,26 +1656,6 @@
return nil
}
-/*func register_flow(deviceFlow *openolt_pb2.Flow, logicalFlow *ofp.OfpFlowStats){
- //update core flows_proxy : flows_proxy.update('/', flows)
-}
-
-func generateStoredId(flowId uint32, direction string)uint32{
-
- if direction == Upstream{
- logger.Debug(ctx, "Upstream flow shifting flowid")
- return ((0x1 << 15) | flowId)
- }else if direction == Downstream{
- logger.Debug(ctx, "Downstream flow not shifting flowid")
- return flowId
- }else{
- logger.Errorw(ctx, "Unrecognized direction",log.Fields{"direction": direction})
- return flowId
- }
-}
-
-*/
-
func (f *OpenOltFlowMgr) addLLDPFlow(ctx context.Context, flow *ofp.OfpFlowStats, portNo uint32) error {
classifierInfo := make(map[string]interface{})
@@ -2002,17 +1960,18 @@
// is conveyed to ONOS during packet-in OF message.
func (f *OpenOltFlowMgr) deleteGemPortFromLocalCache(ctx context.Context, intfID uint32, onuID uint32, gemPortID uint32) {
- f.onuGemInfoLock[intfID].Lock()
- defer f.onuGemInfoLock[intfID].Unlock()
+ f.onuGemInfoLock.Lock()
+ defer f.onuGemInfoLock.Unlock()
logger.Infow(ctx, "deleting-gem-from-local-cache",
log.Fields{
- "gem": gemPortID,
- "intf-id": intfID,
- "onu-id": onuID,
- "device-id": f.deviceHandler.device.Id,
- "onu-gem": f.onuGemInfo[intfID]})
+ "gem-port-id": gemPortID,
+ "intf-id": intfID,
+ "onu-id": onuID,
+ "device-id": f.deviceHandler.device.Id,
+ "onu-gem": f.onuGemInfo[intfID]})
onugem := f.onuGemInfo[intfID]
+deleteLoop:
for i, onu := range onugem {
if onu.OnuID == onuID {
for j, gem := range onu.GemPorts {
@@ -2027,10 +1986,10 @@
"deletedgemport-id": gemPortID,
"gemports": onu.GemPorts,
"device-id": f.deviceHandler.device.Id})
- break
+ break deleteLoop
}
}
- break
+ break deleteLoop
}
}
}
@@ -2937,8 +2896,8 @@
//UpdateOnuInfo function adds onu info to cache and kvstore
func (f *OpenOltFlowMgr) UpdateOnuInfo(ctx context.Context, intfID uint32, onuID uint32, serialNum string) error {
- f.onuGemInfoLock[intfID].Lock()
- defer f.onuGemInfoLock[intfID].Unlock()
+ f.onuGemInfoLock.Lock()
+ defer f.onuGemInfoLock.Unlock()
onu := rsrcMgr.OnuGemInfo{OnuID: onuID, SerialNumber: serialNum, IntfID: intfID}
f.onuGemInfo[intfID] = append(f.onuGemInfo[intfID], onu)
@@ -2958,16 +2917,16 @@
//addGemPortToOnuInfoMap function adds GEMport to ONU map
func (f *OpenOltFlowMgr) addGemPortToOnuInfoMap(ctx context.Context, intfID uint32, onuID uint32, gemPort uint32) {
- f.onuGemInfoLock[intfID].Lock()
- defer f.onuGemInfoLock[intfID].Unlock()
+ f.onuGemInfoLock.Lock()
+ defer f.onuGemInfoLock.Unlock()
logger.Infow(ctx, "adding-gem-to-onu-info-map",
log.Fields{
- "gem": gemPort,
- "intf": intfID,
- "onu": onuID,
- "device-id": f.deviceHandler.device.Id,
- "onu-gem": f.onuGemInfo[intfID]})
+ "gem-port-id": gemPort,
+ "intf-id": intfID,
+ "onu-id": onuID,
+ "device-id": f.deviceHandler.device.Id,
+ "onu-gem": f.onuGemInfo[intfID]})
onugem := f.onuGemInfo[intfID]
// update the gem to the local cache as well as to kv strore
for idx, onu := range onugem {
@@ -2998,11 +2957,11 @@
}
logger.Infow(ctx, "gem-added-to-onu-info-map",
log.Fields{
- "gem": gemPort,
- "intf": intfID,
- "onu": onuID,
- "device-id": f.deviceHandler.device.Id,
- "onu-gem": f.onuGemInfo[intfID]})
+ "gem-port-id": gemPort,
+ "intf-id": intfID,
+ "onu-id": onuID,
+ "device-id": f.deviceHandler.device.Id,
+ "onu-gem": f.onuGemInfo[intfID]})
}
// This function Lookup maps by serialNumber or (intfId, gemPort)
@@ -3010,8 +2969,8 @@
//getOnuIDfromGemPortMap Returns OnuID,nil if found or set 0,error if no onuId is found for serialNumber or (intfId, gemPort)
func (f *OpenOltFlowMgr) getOnuIDfromGemPortMap(ctx context.Context, intfID uint32, gemPortID uint32) (uint32, error) {
- f.onuGemInfoLock[intfID].Lock()
- defer f.onuGemInfoLock[intfID].Unlock()
+ f.onuGemInfoLock.RLock()
+ defer f.onuGemInfoLock.RUnlock()
logger.Infow(ctx, "getting-onu-id-from-gem-port-and-pon-port",
log.Fields{
@@ -3030,6 +2989,11 @@
}
}
}
+ logger.Errorw(ctx, "onu-id-from-gem-port-not-found", log.Fields{
+ "gem-port-id": gemPortID,
+ "interface-id": intfID,
+ "all-gems-on-port": onu,
+ })
return uint32(0), olterrors.NewErrNotFound("onu-id", log.Fields{
"interface-id": intfID,
"gem-port-id": gemPortID},
@@ -3073,18 +3037,18 @@
var gemPortID uint32
var err error
- f.onuGemInfoLock[intfID].Lock()
- defer f.onuGemInfoLock[intfID].Unlock()
-
+ f.onuGemInfoLock.RLock()
+ defer f.onuGemInfoLock.RUnlock()
pktInkey := rsrcMgr.PacketInInfoKey{IntfID: intfID, OnuID: onuID, LogicalPort: portNum}
-
- gemPortID, ok := f.packetInGemPort[pktInkey]
+ var ok bool
+ gemPortID, ok = f.packetInGemPort[pktInkey]
if ok {
logger.Debugw(ctx, "found-gemport-for-pktin-key",
log.Fields{
"pktinkey": pktInkey,
"gem": gemPortID})
- return gemPortID, err
+
+ return gemPortID, nil
}
//If gem is not found in cache try to get it from kv store, if found in kv store, update the cache and return.
gemPortID, err = f.resourceMgr.GetGemPortFromOnuPktIn(ctx, intfID, onuID, portNum)
@@ -3102,6 +3066,7 @@
log.Fields{
"pktinkey": pktInkey,
"gem": gemPortID}, err)
+
}
// nolint: gocyclo
@@ -3828,11 +3793,11 @@
// UpdateGemPortForPktIn updates gemport for packet-in in to the cache and to the kv store as well.
func (f *OpenOltFlowMgr) UpdateGemPortForPktIn(ctx context.Context, intfID uint32, onuID uint32, logicalPort uint32, gemPort uint32) {
+
+ f.onuGemInfoLock.Lock()
+ defer f.onuGemInfoLock.Unlock()
+
pktInkey := rsrcMgr.PacketInInfoKey{IntfID: intfID, OnuID: onuID, LogicalPort: logicalPort}
-
- f.onuGemInfoLock[intfID].Lock()
- defer f.onuGemInfoLock[intfID].Unlock()
-
lookupGemPort, ok := f.packetInGemPort[pktInkey]
if ok {
if lookupGemPort == gemPort {
@@ -3851,13 +3816,14 @@
"pktinkey": pktInkey,
"gem": gemPort})
return
+
}
// AddUniPortToOnuInfo adds uni port to the onugem info both in cache and kvstore.
func (f *OpenOltFlowMgr) AddUniPortToOnuInfo(ctx context.Context, intfID uint32, onuID uint32, portNum uint32) {
- f.onuGemInfoLock[intfID].Lock()
- defer f.onuGemInfoLock[intfID].Unlock()
+ f.onuGemInfoLock.Lock()
+ defer f.onuGemInfoLock.Unlock()
onugem := f.onuGemInfo[intfID]
for idx, onu := range onugem {
@@ -3873,6 +3839,7 @@
}
}
f.resourceMgr.AddUniPortToOnuInfo(ctx, intfID, onuID, portNum)
+
}
func (f *OpenOltFlowMgr) loadFlowIDlistForGem(ctx context.Context, intf uint32) {
diff --git a/internal/pkg/core/openolt_flowmgr_test.go b/internal/pkg/core/openolt_flowmgr_test.go
index 26697aa..1faa8c0 100644
--- a/internal/pkg/core/openolt_flowmgr_test.go
+++ b/internal/pkg/core/openolt_flowmgr_test.go
@@ -20,6 +20,9 @@
import (
"context"
"fmt"
+ "reflect"
+ "strconv"
+ "sync"
"testing"
"time"
@@ -46,10 +49,14 @@
}
func newMockResourceMgr() *resourcemanager.OpenOltResourceMgr {
ranges := []*openolt.DeviceInfo_DeviceResourceRanges{
- {IntfIds: []uint32{0, 1, 2}}}
+ {
+ IntfIds: []uint32{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15},
+ Technology: "Default",
+ },
+ }
deviceinfo := &openolt.DeviceInfo{Vendor: "openolt", Model: "openolt", HardwareVersion: "1.0", FirmwareVersion: "1.0",
- DeviceId: "olt", DeviceSerialNumber: "openolt", PonPorts: 3, Technology: "Default",
+ DeviceId: "olt", DeviceSerialNumber: "openolt", PonPorts: 16, Technology: "Default",
OnuIdStart: 1, OnuIdEnd: 1, AllocIdStart: 1, AllocIdEnd: 1,
GemportIdStart: 1, GemportIdEnd: 1, FlowIdStart: 1, FlowIdEnd: 1,
Ranges: ranges,
@@ -595,37 +602,97 @@
}
func TestOpenOltFlowMgr_UpdateOnuInfo(t *testing.T) {
- // flowMgr := newMockFlowmgr()
- type args struct {
- intfID uint32
- onuID uint32
- serialNum string
- }
- tests := []struct {
- name string
- args args
- }{
- // TODO: Add test cases.
- {"UpdateOnuInfo", args{1, 1, "onu1"}},
- {"UpdateOnuInfo", args{2, 3, "onu1"}},
- }
+ flwMgr := newMockFlowmgr()
+
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
- for _, tt := range tests {
- t.Run(tt.name, func(t *testing.T) {
- flowMgr.UpdateOnuInfo(ctx, tt.args.intfID, tt.args.onuID, tt.args.serialNum)
- })
+ wg := sync.WaitGroup{}
+
+ intfCount := 16
+ onuCount := 32
+
+ for i := 0; i < intfCount; i++ {
+ for j := 0; j < onuCount; j++ {
+ wg.Add(1)
+ go func(i uint32, j uint32) {
+ flwMgr.UpdateOnuInfo(ctx, i, i, fmt.Sprintf("onu-%d", i))
+ wg.Done()
+ }(uint32(i), uint32(j))
+ }
+
+ }
+
+ wg.Wait()
+}
+
+func TestOpenOltFlowMgr_addGemPortToOnuInfoMap(t *testing.T) {
+ flowMgr = newMockFlowmgr()
+ intfNum := 16
+ onuNum := 32
+
+ // clean the flowMgr
+ flowMgr.onuGemInfo = make(map[uint32][]rsrcMgr.OnuGemInfo, intfNum)
+
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+
+ // Create OnuInfo
+ for i := 0; i < intfNum; i++ {
+ for o := 0; o < onuNum; o++ {
+ flowMgr.UpdateOnuInfo(ctx, uint32(i), uint32(o), fmt.Sprintf("i%do%d", i, o))
+ }
+ }
+
+ // Add gemPorts to OnuInfo in parallel threads
+ wg := sync.WaitGroup{}
+
+ for o := 0; o < onuNum; o++ {
+ for i := 0; i < intfNum; i++ {
+ wg.Add(1)
+ go func(intfId uint32, onuId uint32) {
+ gemID, _ := strconv.Atoi(fmt.Sprintf("90%d%d", intfId, onuId))
+
+ flowMgr.addGemPortToOnuInfoMap(ctx, intfId, onuId, uint32(gemID))
+ wg.Done()
+ }(uint32(i), uint32(o))
+ }
+ }
+
+ wg.Wait()
+
+ // check that each entry of onuGemInfo has the correct number of ONUs
+ for i := 0; i < intfNum; i++ {
+ lenofOnu := len(flowMgr.onuGemInfo[uint32(i)])
+ if onuNum != lenofOnu {
+ t.Errorf("OnuGemInfo length is not as expected len = %d, want %d", lenofOnu, onuNum)
+ }
+
+ for o := 0; o < onuNum; o++ {
+ lenOfGemPorts := len(flowMgr.onuGemInfo[uint32(i)][o].GemPorts)
+ // check that each onuEntry has 1 gemPort
+ if lenOfGemPorts != 1 {
+ t.Errorf("Expected 1 GemPort per ONU, found %d", lenOfGemPorts)
+ }
+
+ // check that the value of the gemport is correct
+ gemID, _ := strconv.Atoi(fmt.Sprintf("90%d%d", i, o))
+ currentValue := flowMgr.onuGemInfo[uint32(i)][o].GemPorts[0]
+ if uint32(gemID) != currentValue {
+ t.Errorf("Expected GemPort value to be %d, found %d", gemID, currentValue)
+ }
+ }
}
}
func TestOpenOltFlowMgr_deleteGemPortFromLocalCache(t *testing.T) {
- // flowMgr := newMockFlowmgr()
+ flwMgr := newMockFlowmgr()
type args struct {
intfID uint32
onuID uint32
gemPortIDs []uint32
gemPortIDsToBeDeleted []uint32
+ gemPortIDsRemaining []uint32
serialNum string
finalLength int
}
@@ -634,38 +701,42 @@
args args
}{
// Add/Delete single gem port
- {"DeleteGemPortFromLocalCache1", args{0, 1, []uint32{1}, []uint32{1}, "onu1", 0}},
+ {"DeleteGemPortFromLocalCache1", args{0, 1, []uint32{1}, []uint32{1}, []uint32{}, "onu1", 0}},
// Delete all gemports
- {"DeleteGemPortFromLocalCache2", args{0, 1, []uint32{1, 2, 3, 4}, []uint32{1, 2, 3, 4}, "onu1", 0}},
+ {"DeleteGemPortFromLocalCache2", args{0, 1, []uint32{1, 2, 3, 4}, []uint32{1, 2, 3, 4}, []uint32{}, "onu1", 0}},
// Try to delete when there is no gem port
- {"DeleteGemPortFromLocalCache3", args{0, 1, []uint32{}, []uint32{1, 2}, "onu1", 0}},
+ {"DeleteGemPortFromLocalCache3", args{0, 1, []uint32{}, []uint32{1, 2}, []uint32{}, "onu1", 0}},
// Try to delete non-existent gem port
- {"DeleteGemPortFromLocalCache4", args{0, 1, []uint32{1}, []uint32{2}, "onu1", 1}},
+ {"DeleteGemPortFromLocalCache4", args{0, 1, []uint32{1}, []uint32{2}, []uint32{1}, "onu1", 1}},
// Try to delete two of the gem ports
- {"DeleteGemPortFromLocalCache5", args{0, 1, []uint32{1, 2, 3, 4}, []uint32{2, 4}, "onu1", 2}},
+ {"DeleteGemPortFromLocalCache5", args{0, 1, []uint32{1, 2, 3, 4}, []uint32{2, 4}, []uint32{1, 3}, "onu1", 2}},
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
- flowMgr.UpdateOnuInfo(ctx, tt.args.intfID, tt.args.onuID, tt.args.serialNum)
+ flwMgr.UpdateOnuInfo(ctx, tt.args.intfID, tt.args.onuID, tt.args.serialNum)
for _, gemPort := range tt.args.gemPortIDs {
- flowMgr.addGemPortToOnuInfoMap(ctx, tt.args.intfID, tt.args.onuID, gemPort)
+ flwMgr.addGemPortToOnuInfoMap(ctx, tt.args.intfID, tt.args.onuID, gemPort)
}
for _, gemPortDeleted := range tt.args.gemPortIDsToBeDeleted {
- flowMgr.deleteGemPortFromLocalCache(context.Background(), tt.args.intfID, tt.args.onuID, gemPortDeleted)
+ flwMgr.deleteGemPortFromLocalCache(ctx, tt.args.intfID, tt.args.onuID, gemPortDeleted)
}
- lenofGemPorts := len(flowMgr.onuGemInfo[tt.args.intfID][0].GemPorts)
+ lenofGemPorts := len(flwMgr.onuGemInfo[tt.args.intfID][0].GemPorts)
if lenofGemPorts != tt.args.finalLength {
t.Errorf("GemPorts length is not as expected len = %d, want %d", lenofGemPorts, tt.args.finalLength)
}
+ gemPorts := flwMgr.onuGemInfo[tt.args.intfID][0].GemPorts
+ if !reflect.DeepEqual(tt.args.gemPortIDsRemaining, gemPorts) {
+ t.Errorf("GemPorts are not as expected = %v, want %v", gemPorts, tt.args.gemPortIDsRemaining)
+ }
})
}
}
func TestOpenOltFlowMgr_GetLogicalPortFromPacketIn(t *testing.T) {
- // flowMgr := newMockFlowmgr()
+ flwMgr := newMockFlowmgr()
type args struct {
packetIn *openoltpb2.PacketIndication
}
@@ -687,7 +758,7 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
- got, err := flowMgr.GetLogicalPortFromPacketIn(ctx, tt.args.packetIn)
+ got, err := flwMgr.GetLogicalPortFromPacketIn(ctx, tt.args.packetIn)
if (err != nil) != tt.wantErr {
t.Errorf("OpenOltFlowMgr.GetLogicalPortFromPacketIn() error = %v, wantErr %v", err, tt.wantErr)
return