[VOL-5397]-removed go routines for DB operations
Signed-off-by: Akash Soni <akash.soni@radisys.com>
Change-Id: I8504004cc8e3a97a7d4de48d617471c8050ae55c
diff --git a/internal/pkg/avcfg/omci_vlan_config.go b/internal/pkg/avcfg/omci_vlan_config.go
index bfe1bc9..9eb4c10 100755
--- a/internal/pkg/avcfg/omci_vlan_config.go
+++ b/internal/pkg/avcfg/omci_vlan_config.go
@@ -43,7 +43,7 @@
// internal predefined values
cDefaultDownstreamMode = 0
cDefaultTpid = 0x8100
- cVtfdTableSize = 12 //as per G.988
+ cVtfdTableSize = 64 //as per G.988
cMaxAllowedFlows = cVtfdTableSize //which might be under discussion, for the moment connected to limit of VLAN's within VTFD
)
diff --git a/internal/pkg/core/device_handler.go b/internal/pkg/core/device_handler.go
index 02d996f..e5cc1e0 100755
--- a/internal/pkg/core/device_handler.go
+++ b/internal/pkg/core/device_handler.go
@@ -461,13 +461,11 @@
}
deadline = time.Now().Add(dh.pOpenOnuAc.maxTimeoutInterAdapterComm) //allowed run time to finish before execution
dctx2, cancel2 := context.WithDeadline(context.Background(), deadline)
- pDevEntry.ResetKvProcessingErrorIndication()
- wg.Add(1) // for the 1 go routine to finish
- go pDevEntry.UpdateOnuKvStore(log.WithSpanFromContext(dctx2, ctx), &wg)
- dh.waitForCompletion(ctx, cancel2, &wg, "TechProfDwld") //wait for background process to finish
- if kvErr := pDevEntry.GetKvProcessingErrorIndication(); kvErr != nil {
- logger.Errorw(ctx, "error-updating-KV", log.Fields{"device-id": dh.DeviceID, "err": kvErr, "tp-path": techProfMsg.TpInstancePath})
- return kvErr
+ defer cancel2()
+ err1 := pDevEntry.UpdateOnuKvStore(log.WithSpanFromContext(dctx2, ctx))
+ if err1 != nil {
+ logger.Errorf(ctx, "UpdateOnuKvStore-failed", log.Fields{"device-id": dh.DeviceID, "error": err1})
+ return err
}
return nil
default:
@@ -551,16 +549,14 @@
}
pDevEntry.FreeTcont(ctx, uint16(delTcontMsg.AllocId))
- var wg sync.WaitGroup
deadline := time.Now().Add(dh.pOpenOnuAc.maxTimeoutInterAdapterComm) //allowed run time to finish before execution
dctx, cancel := context.WithDeadline(context.Background(), deadline)
- wg.Add(1)
+ defer cancel()
logger.Debugw(ctx, "remove-tcont-in-kv", log.Fields{"device-id": dh.DeviceID, "uni-id": uniID, "tpID": tpID, "tcont": delTcontMsg.AllocId})
- go pDevEntry.UpdateOnuKvStore(log.WithSpanFromContext(dctx, ctx), &wg)
- dh.waitForCompletion(ctx, cancel, &wg, "DeleteTcont") //wait for background process to finish
- if err := pDevEntry.GetKvProcessingErrorIndication(); err != nil {
- logger.Errorw(ctx, err.Error(), log.Fields{"device-id": dh.DeviceID})
- return err
+ err1 := pDevEntry.UpdateOnuKvStore(log.WithSpanFromContext(dctx, ctx))
+ if err1 != nil {
+ logger.Errorw(ctx, "UpdateOnuKvStore-failed", log.Fields{"device-id": dh.DeviceID, "err": err1})
+ return err1
}
return dh.deleteTechProfileResource(ctx, uniID, tpID, delTcontMsg.TpInstancePath,
@@ -602,15 +598,13 @@
logger.Debugw(ctx, "techProfile-config-cleared", log.Fields{"device-id": dh.DeviceID, "uni-id": uniID, "tpID": tpID})
if bTpModify := pDevEntry.UpdateOnuUniTpPath(ctx, uniID, tpID, ""); bTpModify {
pDevEntry.ResetKvProcessingErrorIndication()
- var wg2 sync.WaitGroup
dctx2, cancel2 := context.WithDeadline(context.Background(), deadline)
- wg2.Add(1)
+ defer cancel2()
// Removal of the gem id mapping represents the removal of the tech profile
logger.Debugw(ctx, "remove-techProfile-indication-in-kv", log.Fields{"device-id": dh.DeviceID, "uni-id": uniID, "tpID": tpID})
- go pDevEntry.UpdateOnuKvStore(log.WithSpanFromContext(dctx2, ctx), &wg2)
- dh.waitForCompletion(ctx, cancel2, &wg2, "TechProfileDeleteOn"+resourceName) //wait for background process to finish
- if err := pDevEntry.GetKvProcessingErrorIndication(); err != nil {
- logger.Errorw(ctx, err.Error(), log.Fields{"device-id": dh.DeviceID})
+ err := pDevEntry.UpdateOnuKvStore(log.WithSpanFromContext(dctx2, ctx))
+ if err != nil {
+ logger.Errorw(ctx, "UpdateOnuKvStore-failed", log.Fields{"device-id": dh.DeviceID, "err": err})
return err
}
}
@@ -1189,8 +1183,7 @@
waitGroup.Done()
}
} else {
- logger.Errorw(ctx, "received unexpected UNI flowConfig done indication - is ignored",
- log.Fields{"device-id": dh.DeviceID, "uni-id": uniIndication})
+ logger.Errorw(ctx, "received unexpected UNI flowConfig done indication - is ignored", log.Fields{"device-id": dh.DeviceID, "uni-id": uniIndication})
}
} //switch uniIndication
@@ -1251,16 +1244,13 @@
//20200721: 10s proved to be less in 8*8 ONU test on local vbox machine with debug, might be further adapted
deadline := time.Now().Add(dh.pOpenOnuAc.maxTimeoutInterAdapterComm) //allowed run time to finish before execution
dctx, cancel := context.WithDeadline(ctx, deadline)
-
- pDevEntry.ResetKvProcessingErrorIndication()
-
- var wg sync.WaitGroup
- wg.Add(1) // for the 1 go routine to finish
- go pDevEntry.DeleteDataFromOnuKvStore(log.WithSpanFromContext(dctx, ctx), &wg)
- dh.waitForCompletion(ctx, cancel, &wg, "DeleteDevice") //wait for background process to finish
-
- // TODO: further actions - stop metrics and FSMs, remove device ...
- return pDevEntry.GetKvProcessingErrorIndication()
+ defer cancel()
+ err := pDevEntry.DeleteDataFromOnuKvStore(log.WithSpanFromContext(dctx, ctx))
+ if err != nil {
+ logger.Errorw(ctx, "delete data from onu kv store failed", log.Fields{"device-id": dh.DeviceID, "err": err})
+ return err
+ }
+ return nil
}
// func (dh *deviceHandler) rebootDevice(ctx context.Context, device *voltha.Device) error {
@@ -3604,15 +3594,14 @@
//20200721: 10s proved to be less in 8*8 ONU test on local vbox machine with debug, might be further adapted
deadline := time.Now().Add(dh.pOpenOnuAc.maxTimeoutInterAdapterComm) //allowed run time to finish before execution
dctx, cancel := context.WithDeadline(context.Background(), deadline)
+ defer cancel() // Ensure cancel is called to release resources
- aPDevEntry.ResetKvProcessingErrorIndication()
- var wg sync.WaitGroup
- wg.Add(1) // for the 1 go routine to finish
-
- go aPDevEntry.UpdateOnuKvStore(log.WithSpanFromContext(dctx, ctx), &wg)
- dh.waitForCompletion(ctx, cancel, &wg, "UpdateKvStore") //wait for background process to finish
-
- return aPDevEntry.GetKvProcessingErrorIndication()
+ err := aPDevEntry.UpdateOnuKvStore(log.WithSpanFromContext(dctx, ctx))
+ if err != nil {
+ logger.Errorw(ctx, "UpdateOnuKvStore-failed", log.Fields{"device-id": dh.DeviceID})
+ return err
+ }
+ return nil
}
// StorePersUniFlowConfig updates local storage of OnuUniFlowConfig and writes it into kv-store afterwards to have it
diff --git a/internal/pkg/mib/onu_device_entry.go b/internal/pkg/mib/onu_device_entry.go
index 1b7303a..6398e0a 100755
--- a/internal/pkg/mib/onu_device_entry.go
+++ b/internal/pkg/mib/onu_device_entry.go
@@ -189,7 +189,6 @@
onuKVStorePath string
mutexOnuKVStoreProcResult sync.RWMutex
onuKVStoreProcResult error //error indication of processing
- chOnuKvProcessingStep chan uint8
mutexOnuSwImageIndications sync.RWMutex
onuSwImageIndications cmn.SswImageIndications
MutexOnuImageStatus sync.RWMutex
@@ -237,7 +236,6 @@
onuDeviceEntry.ReconciledTpInstances = make(map[uint8]map[uint8]inter_adapter.TechProfileDownloadMessage)
onuDeviceEntry.chReconcilingFlowsFinished = make(chan bool)
onuDeviceEntry.reconcilingFlows = false
- onuDeviceEntry.chOnuKvProcessingStep = make(chan uint8)
onuDeviceEntry.omciRebootMessageReceivedChannel = make(chan cmn.Message, 2)
//openomciagent.lockDeviceHandlersMap = sync.RWMutex{}
//OMCI related databases are on a per-agent basis. State machines and tasks
@@ -573,28 +571,23 @@
}
// DeleteDataFromOnuKvStore - TODO: add comment
-func (oo *OnuDeviceEntry) DeleteDataFromOnuKvStore(ctx context.Context, wg *sync.WaitGroup) {
- defer wg.Done()
+func (oo *OnuDeviceEntry) DeleteDataFromOnuKvStore(ctx context.Context) error {
if oo.onuKVStore == nil {
logger.Debugw(ctx, "onuKVStore not set - abort", log.Fields{"device-id": oo.deviceID})
- oo.setKvProcessingErrorIndication(errors.New("onu-data delete aborted: onuKVStore not set"))
- return
+ return errors.New("onu-data delete aborted: onuKVStore not set")
}
- var processingStep uint8 = 1 // used to synchronize the different processing steps with chOnuKvProcessingStep
- go oo.deletePersistentData(ctx, processingStep)
- if !oo.waitForTimeoutOrCompletion(ctx, oo.chOnuKvProcessingStep, processingStep) {
- //timeout or error detected
- logger.Debugw(ctx, "ONU-data not deleted - abort", log.Fields{"device-id": oo.deviceID})
- oo.setKvProcessingErrorIndication(errors.New("onu-data delete aborted: during kv-access"))
- return
+ err := oo.deletePersistentData(ctx)
+ if err != nil {
+ logger.Errorf(ctx, "onu-data delete aborted: during kv-access", log.Fields{"device-id": oo.deviceID, "err": err})
+ return err
}
+ return nil
}
-func (oo *OnuDeviceEntry) deletePersistentData(ctx context.Context, aProcessingStep uint8) {
+func (oo *OnuDeviceEntry) deletePersistentData(ctx context.Context) error {
logger.Debugw(ctx, "delete and clear internal persistency data", log.Fields{"device-id": oo.deviceID})
-
oo.MutexPersOnuConfig.Lock()
defer oo.MutexPersOnuConfig.Unlock()
@@ -608,32 +601,27 @@
oo.mutexOnuKVStore.Unlock()
if err != nil {
logger.Errorw(ctx, "unable to delete in KVstore", log.Fields{"device-id": oo.deviceID, "err": err})
- oo.chOnuKvProcessingStep <- 0 //error indication
- return
+ return err
}
- oo.chOnuKvProcessingStep <- aProcessingStep //done
+ return nil
}
// UpdateOnuKvStore - TODO: add comment
-func (oo *OnuDeviceEntry) UpdateOnuKvStore(ctx context.Context, wg *sync.WaitGroup) {
- defer wg.Done()
+func (oo *OnuDeviceEntry) UpdateOnuKvStore(ctx context.Context) error {
if oo.onuKVStore == nil {
logger.Debugw(ctx, "onuKVStore not set - abort", log.Fields{"device-id": oo.deviceID})
- oo.setKvProcessingErrorIndication(errors.New("onu-data update aborted: onuKVStore not set"))
- return
+ return errors.New("onu-data update aborted: onuKVStore not set")
}
- var processingStep uint8 = 1 // used to synchronize the different processing steps with chOnuKvProcessingStep
- go oo.storeDataInOnuKvStore(ctx, processingStep)
- if !oo.waitForTimeoutOrCompletion(ctx, oo.chOnuKvProcessingStep, processingStep) {
- //timeout or error detected
- logger.Debugw(ctx, "ONU-data not written - abort", log.Fields{"device-id": oo.deviceID})
- oo.setKvProcessingErrorIndication(errors.New("onu-data update aborted: during writing process"))
- return
+ err := oo.storeDataInOnuKvStore(ctx)
+ if err != nil {
+ logger.Errorf(ctx, "onu-data update aborted: during writing process", log.Fields{"device-id": oo.deviceID, "err": err})
+ return err
}
+ return nil
}
-func (oo *OnuDeviceEntry) storeDataInOnuKvStore(ctx context.Context, aProcessingStep uint8) {
+func (oo *OnuDeviceEntry) storeDataInOnuKvStore(ctx context.Context) error {
oo.MutexPersOnuConfig.Lock()
defer oo.MutexPersOnuConfig.Unlock()
@@ -641,17 +629,15 @@
oo.pOpenOnuAc.RLockMutexDeviceHandlersMap()
if _, exist := oo.pOpenOnuAc.GetDeviceHandler(oo.deviceID); !exist {
logger.Debugw(ctx, "delete_device in progress - skip write request", log.Fields{"device-id": oo.deviceID})
- oo.chOnuKvProcessingStep <- aProcessingStep
oo.pOpenOnuAc.RUnlockMutexDeviceHandlersMap()
- return
+ return nil
}
oo.baseDeviceHandler.RLockMutexDeletionInProgressFlag()
if oo.baseDeviceHandler.GetDeletionInProgress() {
logger.Debugw(ctx, "delete_device in progress - skip write request", log.Fields{"device-id": oo.deviceID})
- oo.chOnuKvProcessingStep <- aProcessingStep
oo.pOpenOnuAc.RUnlockMutexDeviceHandlersMap()
oo.baseDeviceHandler.RUnlockMutexDeletionInProgressFlag()
- return
+ return nil
}
oo.pOpenOnuAc.RUnlockMutexDeviceHandlersMap()
oo.baseDeviceHandler.RUnlockMutexDeletionInProgressFlag()
@@ -666,8 +652,7 @@
oo.SOnuPersistentData.PersOperState = onuIndication.OperState
} else {
logger.Errorw(ctx, "onuIndication not set, unable to load ONU-data", log.Fields{"device-id": oo.deviceID})
- oo.chOnuKvProcessingStep <- 0 //error indication
- return
+ return errors.New("onuIndication not set, unable to load ONU-data")
}
logger.Debugw(ctx, "Update ONU-data in KVStore", log.Fields{"device-id": oo.deviceID, "SOnuPersistentData": oo.SOnuPersistentData})
@@ -676,19 +661,17 @@
if err != nil {
logger.Errorw(ctx, "unable to marshal ONU-data", log.Fields{"SOnuPersistentData": oo.SOnuPersistentData,
"device-id": oo.deviceID, "err": err})
- oo.chOnuKvProcessingStep <- 0 //error indication
- return
+ return err
}
oo.mutexOnuKVStore.Lock()
err = oo.onuKVStore.Put(ctx, oo.onuKVStorePath, Value)
oo.mutexOnuKVStore.Unlock()
if err != nil {
- logger.Errorw(ctx, "unable to write ONU-data into KVstore", log.Fields{"device-id": oo.deviceID, "err": err})
- oo.chOnuKvProcessingStep <- 0 //error indication
- return
+ logger.Errorf(ctx, "unable to write ONU-data into KVstore", log.Fields{"device-id": oo.deviceID, "err": err})
+ return err
}
- oo.chOnuKvProcessingStep <- aProcessingStep //done
+ return nil
}
// UpdateOnuUniTpPath - TODO: add comment
@@ -785,25 +768,6 @@
oo.SOnuPersistentData.PersUniConfig = append(oo.SOnuPersistentData.PersUniConfig, tmpConfig)
}
-func (oo *OnuDeviceEntry) waitForTimeoutOrCompletion(
- ctx context.Context, aChOnuProcessingStep <-chan uint8, aProcessingStep uint8) bool {
- select {
- case <-ctx.Done():
- logger.Warnw(ctx, "processing not completed in-time!",
- log.Fields{"device-id": oo.deviceID, "error": ctx.Err()})
- return false
- case rxStep := <-aChOnuProcessingStep:
- if rxStep == aProcessingStep {
- return true
- }
- //all other values are not accepted - including 0 for error indication
- logger.Warnw(ctx, "Invalid processing step received: abort!",
- log.Fields{"device-id": oo.deviceID,
- "wantedStep": aProcessingStep, "haveStep": rxStep})
- return false
- }
-}
-
// ResetKvProcessingErrorIndication - TODO: add comment
func (oo *OnuDeviceEntry) ResetKvProcessingErrorIndication() {
oo.mutexOnuKVStoreProcResult.Lock()
@@ -819,11 +783,11 @@
return value
}
-func (oo *OnuDeviceEntry) setKvProcessingErrorIndication(value error) {
- oo.mutexOnuKVStoreProcResult.Lock()
- oo.onuKVStoreProcResult = value
- oo.mutexOnuKVStoreProcResult.Unlock()
-}
+// func (oo *OnuDeviceEntry) setKvProcessingErrorIndication(value error) {
+// oo.mutexOnuKVStoreProcResult.Lock()
+// oo.onuKVStoreProcResult = value
+// oo.mutexOnuKVStoreProcResult.Unlock()
+// }
// IncrementMibDataSync - TODO: add comment
func (oo *OnuDeviceEntry) IncrementMibDataSync(ctx context.Context) {