VOL-1812: Better handle duplicate indications and concurrency
Discovery indications starts own goroutine.
This prevents sleep from stopping entire readIndications thread.
No longer return error (cant do anything with it). just log it.
Lock get onu id from resource manager otherwise duplicate onu id are
handed out under load.
Change-Id: If75a7e08bd2e2066087bc2e4907290e6c387c60c
diff --git a/adaptercore/device_handler.go b/adaptercore/device_handler.go
index 92b6445..ba4040f 100644
--- a/adaptercore/device_handler.go
+++ b/adaptercore/device_handler.go
@@ -324,7 +324,7 @@
onuDiscInd := indication.GetOnuDiscInd()
log.Infow("Received Onu discovery indication ", log.Fields{"OnuDiscInd": onuDiscInd})
sn := dh.stringifySerialNumber(onuDiscInd.SerialNumber)
- dh.onuDiscIndication(onuDiscInd, sn)
+ go dh.onuDiscIndication(onuDiscInd, sn)
case *oop.Indication_OnuInd:
onuInd := indication.GetOnuInd()
log.Infow("Received Onu indication ", log.Fields{"OnuInd": onuInd})
@@ -728,38 +728,41 @@
}
}
-func (dh *DeviceHandler) onuDiscIndication(onuDiscInd *oop.OnuDiscIndication, sn string) error {
+func (dh *DeviceHandler) onuDiscIndication(onuDiscInd *oop.OnuDiscIndication, sn string) {
channelID := onuDiscInd.GetIntfId()
parentPortNo := IntfIDToPortNo(onuDiscInd.GetIntfId(), voltha.Port_PON_OLT)
+
+ log.Debugw("new-discovery-indication", log.Fields{"sn": sn})
+ dh.lockDevice.Lock()
if _, ok := dh.discOnus[sn]; ok {
+ dh.lockDevice.Unlock()
log.Debugw("onu-sn-is-already-being-processed", log.Fields{"sn": sn})
- return nil
+ return
}
- dh.lockDevice.Lock()
dh.discOnus[sn] = true
+ log.Debugw("new-discovery-indications-list", log.Fields{"discOnus": dh.discOnus})
dh.lockDevice.Unlock()
- // evict the onu serial number from local cache
- defer func() {
- delete(dh.discOnus, sn)
- }()
kwargs := make(map[string]interface{})
if sn != "" {
kwargs["serial_number"] = sn
} else {
- log.Error("invalid onu serial number")
- return errors.New("failed to fetch onu serial number")
+ log.Errorw("invalid onu serial number", log.Fields{"sn": sn})
+ return
}
onuDevice, err := dh.coreProxy.GetChildDevice(context.TODO(), dh.device.Id, kwargs)
var onuID uint32
if onuDevice == nil || err != nil {
//This is the first time ONU discovered. Create an OnuID for it.
- onuID, err = dh.resourceMgr.GetONUID(onuDiscInd.GetIntfId())
+ ponintfid := onuDiscInd.GetIntfId()
+ dh.lockDevice.Lock()
+ onuID, err = dh.resourceMgr.GetONUID(ponintfid)
+ dh.lockDevice.Unlock()
if err != nil {
- log.Errorw("failed to fetch onuID from resource manager", log.Fields{"err": err})
- return err
+ log.Errorw("failed to fetch onuID from resource manager", log.Fields{"pon-intf-id": ponintfid, "err": err})
+ return
}
if onuDevice, err = dh.coreProxy.ChildDeviceDetected(context.TODO(), dh.device.Id, int(parentPortNo),
"brcm_openomci_onu", int(channelID),
@@ -767,8 +770,9 @@
log.Errorw("Create onu error",
log.Fields{"parent_id": dh.device.Id, "ponPort": onuDiscInd.GetIntfId(),
"onuID": onuID, "sn": sn, "error": err})
- return err
+ return
}
+ log.Debugw("onu-child-device-added", log.Fields{"onuDevice": onuDevice})
} else {
//ONU already discovered before. Use the same OnuID.
@@ -779,12 +783,16 @@
log.Debugw("ONU discovery indication key create", log.Fields{"onuID": onuID,
"intfId": onuDiscInd.GetIntfId()})
onuKey := dh.formOnuKey(onuDiscInd.GetIntfId(), onuID)
+
+ dh.lockDevice.Lock()
dh.onus[onuKey] = NewOnuDevice(onuDevice.Id, onuDevice.Type, onuDevice.SerialNumber, onuID, onuDiscInd.GetIntfId(), onuDevice.ProxyAddress.DeviceId)
+ log.Debugw("new-onu-device-discovered", log.Fields{"onu": dh.onus[onuKey]})
+ dh.lockDevice.Unlock()
err = dh.coreProxy.DeviceStateUpdate(context.TODO(), onuDevice.Id, common.ConnectStatus_REACHABLE, common.OperStatus_DISCOVERED)
if err != nil {
- log.Errorw("failed to update device state", log.Fields{"DeviceID": onuDevice.Id})
- return err
+ log.Errorw("failed to update device state", log.Fields{"DeviceID": onuDevice.Id, "err": err})
+ return
}
log.Debugw("onu-discovered-reachable", log.Fields{"deviceId": onuDevice.Id})
//TODO: We put this sleep here to prevent the race between state update and onuIndication
@@ -792,7 +800,7 @@
//then the initialisation fails.
time.Sleep(1 * time.Second)
dh.activateONU(onuDiscInd.IntfId, int64(onuID), onuDiscInd.SerialNumber, sn)
- return nil
+ return
}
func (dh *DeviceHandler) onuIndication(onuInd *oop.OnuIndication) {
@@ -831,7 +839,9 @@
}
if !foundInCache {
onuKey := dh.formOnuKey(onuInd.GetIntfId(), onuInd.GetOnuId())
+ dh.lockDevice.Lock()
dh.onus[onuKey] = NewOnuDevice(onuDevice.Id, onuDevice.Type, onuDevice.SerialNumber, onuInd.GetOnuId(), onuInd.GetIntfId(), onuDevice.ProxyAddress.DeviceId)
+ dh.lockDevice.Unlock()
}
dh.updateOnuStates(onuDevice, onuInd, foundInCache)
@@ -843,11 +853,11 @@
}
func (dh *DeviceHandler) updateOnuStates(onuDevice *voltha.Device, onuInd *oop.OnuIndication, foundInCache bool) {
+ log.Debugw("onu-indication-for-state", log.Fields{"onuIndication": onuInd, "DeviceId": onuDevice.Id, "operStatus": onuDevice.OperStatus, "adminStatus": onuDevice.AdminState})
dh.updateOnuAdminState(onuInd)
// operState
if onuInd.OperState == "down" {
- log.Debugw("inter-adapter-send-onu-ind", log.Fields{"onuIndication": onuInd})
-
+ log.Debugw("sending-interadapter-onu-indication", log.Fields{"onuIndication": onuInd, "DeviceId": onuDevice.Id, "operStatus": onuDevice.OperStatus, "adminStatus": onuDevice.AdminState})
// TODO NEW CORE do not hardcode adapter name. Handler needs Adapter reference
err := dh.AdapterProxy.SendInterAdapterMessage(context.TODO(), onuInd, ic.InterAdapterMessageType_ONU_IND_REQUEST,
"openolt", onuDevice.Type, onuDevice.Id, onuDevice.ProxyAddress.DeviceId, "")
@@ -858,9 +868,11 @@
} else if onuInd.OperState == "up" {
// Ignore operstatus if device was found in cache
if !foundInCache && onuDevice.OperStatus != common.OperStatus_DISCOVERED {
- log.Warnw("ignore onu indication", log.Fields{"intfID": onuInd.IntfId, "onuID": onuInd.OnuId, "operStatus": onuDevice.OperStatus, "msgOperStatus": onuInd.OperState})
+ log.Warnw("ignore-onu-indication", log.Fields{"intfID": onuInd.IntfId, "onuID": onuInd.OnuId, "operStatus": onuDevice.OperStatus, "msgOperStatus": onuInd.OperState})
return
}
+ log.Debugw("sending-interadapter-onu-indication", log.Fields{"onuIndication": onuInd, "DeviceId": onuDevice.Id, "operStatus": onuDevice.OperStatus, "adminStatus": onuDevice.AdminState})
+ // TODO NEW CORE do not hardcode adapter name. Handler needs Adapter reference
err := dh.AdapterProxy.SendInterAdapterMessage(context.TODO(), onuInd, ic.InterAdapterMessageType_ONU_IND_REQUEST,
"openolt", onuDevice.Type, onuDevice.Id, onuDevice.ProxyAddress.DeviceId, "")
if err != nil {