X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fcontrol%2Fregistry.go;h=25709605a29c1a56c8d5d333abc19af599eba112;hb=ad41feaeb9dde8ccd6b8b508eb3df8df5d207134;hp=6abdcdb63afc46b223e5206e371f0e97706e3034;hpb=3944a22bb267f649fff113682a6ba4253007392d;p=ric-plt%2Fsubmgr.git diff --git a/pkg/control/registry.go b/pkg/control/registry.go index 6abdcdb..2570960 100644 --- a/pkg/control/registry.go +++ b/pkg/control/registry.go @@ -22,6 +22,8 @@ package control import ( "fmt" "gerrit.o-ran-sc.org/r/ric-plt/e2ap/pkg/e2ap" + "gerrit.o-ran-sc.org/r/ric-plt/submgr/pkg/xapptweaks" + "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/models" "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp" "sync" "time" @@ -30,22 +32,36 @@ import ( //----------------------------------------------------------------------------- // //----------------------------------------------------------------------------- + type Registry struct { mutex sync.Mutex - register map[uint16]*Subscription - subIds []uint16 + register map[uint32]*Subscription + subIds []uint32 rtmgrClient *RtmgrClient } func (r *Registry) Initialize() { - r.register = make(map[uint16]*Subscription) - var i uint16 + r.register = make(map[uint32]*Subscription) + var i uint32 for i = 0; i < 65535; i++ { r.subIds = append(r.subIds, i+1) } } -func (r *Registry) allocateSubs(trans *Transaction, subReqMsg *e2ap.E2APSubscriptionRequest) (*Subscription, error) { +func (r *Registry) QueryHandler() (models.SubscriptionList, error) { + r.mutex.Lock() + defer r.mutex.Unlock() + + resp := models.SubscriptionList{} + for _, subs := range r.register { + subs.mutex.Lock() + resp = append(resp, &models.SubscriptionData{SubscriptionID: int64(subs.ReqId.Seq), Meid: subs.Meid.RanName, Endpoint: subs.EpList.StringList()}) + subs.mutex.Unlock() + } + return resp, nil +} + +func (r *Registry) allocateSubs(trans *TransactionXapp, subReqMsg *e2ap.E2APSubscriptionRequest) (*Subscription, error) { if len(r.subIds) > 0 { sequenceNumber := r.subIds[0] r.subIds = r.subIds[1:] @@ -55,14 +71,15 @@ func (r *Registry) allocateSubs(trans *Transaction, subReqMsg *e2ap.E2APSubscrip } subs := &Subscription{ registry: r, - Seq: sequenceNumber, Meid: trans.Meid, SubReqMsg: subReqMsg, valid: true, } + subs.ReqId.Id = 123 + subs.ReqId.Seq = sequenceNumber if subs.EpList.AddEndpoint(trans.GetEndpoint()) == false { - r.subIds = append(r.subIds, subs.Seq) + r.subIds = append(r.subIds, subs.ReqId.Seq) return nil, fmt.Errorf("Registry: Endpoint existing already in subscription") } @@ -71,9 +88,10 @@ func (r *Registry) allocateSubs(trans *Transaction, subReqMsg *e2ap.E2APSubscrip return nil, fmt.Errorf("Registry: Failed to reserve subscription no free ids") } -func (r *Registry) findExistingSubs(trans *Transaction, subReqMsg *e2ap.E2APSubscriptionRequest) *Subscription { +func (r *Registry) findExistingSubs(trans *TransactionXapp, subReqMsg *e2ap.E2APSubscriptionRequest) *Subscription { + for _, subs := range r.register { - if subs.IsSame(trans, subReqMsg) { + if subs.IsMergeable(trans, subReqMsg) { // // check if there has been race conditions @@ -84,6 +102,11 @@ func (r *Registry) findExistingSubs(trans *Transaction, subReqMsg *e2ap.E2APSubs subs.mutex.Unlock() continue } + // If size is zero, entry is to be deleted + if subs.EpList.Size() == 0 { + subs.mutex.Unlock() + continue + } // try to add to endpointlist. if subs.EpList.AddEndpoint(trans.GetEndpoint()) == false { subs.mutex.Unlock() @@ -91,22 +114,40 @@ func (r *Registry) findExistingSubs(trans *Transaction, subReqMsg *e2ap.E2APSubs } subs.mutex.Unlock() - //Race collision during parallel incoming and deleted - xapp.Logger.Debug("Registry: Identical subs found %s for %s", subs.String(), trans.String()) + xapp.Logger.Debug("Registry: Mergeable subs found %s for %s", subs.String(), trans.String()) return subs } } return nil } -func (r *Registry) AssignToSubscription(trans *Transaction, subReqMsg *e2ap.E2APSubscriptionRequest) (*Subscription, error) { +func (r *Registry) AssignToSubscription(trans *TransactionXapp, subReqMsg *e2ap.E2APSubscriptionRequest) (*Subscription, error) { var err error var newAlloc bool r.mutex.Lock() defer r.mutex.Unlock() - subs := r.findExistingSubs(trans, subReqMsg) + // + // Check validity of subscription action types + // + actionType, err := r.CheckActionTypes(subReqMsg) + if err != nil { + xapp.Logger.Debug("CREATE %s", err) + return nil, err + } + + // + // Find possible existing Policy subscription + // + if actionType == e2ap.E2AP_ActionTypePolicy { + if subs, ok := r.register[trans.GetSubId()]; ok { + xapp.Logger.Debug("CREATE %s. Existing subscription for Policy found", subs.String()) + subs.SetCachedResponse(nil, true) + return subs, nil + } + } + subs := r.findExistingSubs(trans, subReqMsg) if subs == nil { subs, err = r.allocateSubs(trans, subReqMsg) if err != nil { @@ -128,31 +169,55 @@ func (r *Registry) AssignToSubscription(trans *Transaction, subReqMsg *e2ap.E2AP // Subscription route updates // if epamount == 1 { - subRouteAction := SubRouteInfo{CREATE, subs.EpList, subs.Seq} - err = r.rtmgrClient.SubscriptionRequestUpdate(subRouteAction) + subRouteAction := SubRouteInfo{subs.EpList, uint16(subs.ReqId.Seq)} + err = r.rtmgrClient.SubscriptionRequestCreate(subRouteAction) } else { - subRouteAction := SubRouteInfo{UPDATE, subs.EpList, subs.Seq} + subRouteAction := SubRouteInfo{subs.EpList, uint16(subs.ReqId.Seq)} err = r.rtmgrClient.SubscriptionRequestUpdate(subRouteAction) } r.mutex.Lock() if err != nil { if newAlloc { - r.subIds = append(r.subIds, subs.Seq) + r.subIds = append(r.subIds, subs.ReqId.Seq) } return nil, err } if newAlloc { - r.register[subs.Seq] = subs + r.register[subs.ReqId.Seq] = subs } - xapp.Logger.Debug("Registry: Create %s", subs.String()) + xapp.Logger.Debug("CREATE %s", subs.String()) xapp.Logger.Debug("Registry: substable=%v", r.register) return subs, nil } -// TODO: Needs better logic when there is concurrent calls -func (r *Registry) RemoveFromSubscription(subs *Subscription, trans *Transaction, waitRouteClean time.Duration) error { +func (r *Registry) CheckActionTypes(subReqMsg *e2ap.E2APSubscriptionRequest) (uint64, error) { + var reportFound bool = false + var policyFound bool = false + + for _, acts := range subReqMsg.ActionSetups { + if acts.ActionType == e2ap.E2AP_ActionTypeReport { + reportFound = true + } + if acts.ActionType == e2ap.E2AP_ActionTypePolicy { + policyFound = true + } + } + if reportFound == true && policyFound == true { + return e2ap.E2AP_ActionTypeInvalid, fmt.Errorf("Report and Policy in same RICactions-ToBeSetup-List") + } + if reportFound == true { + return e2ap.E2AP_ActionTypeReport, nil + } + if policyFound == true { + return e2ap.E2AP_ActionTypePolicy, nil + } + return e2ap.E2AP_ActionTypeInvalid, fmt.Errorf("Invalid action type in RICactions-ToBeSetup-List") +} + +// TODO: Works with concurrent calls, but check if can be improved +func (r *Registry) RemoveFromSubscription(subs *Subscription, trans *TransactionXapp, waitRouteClean time.Duration) error { r.mutex.Lock() defer r.mutex.Unlock() @@ -161,57 +226,57 @@ func (r *Registry) RemoveFromSubscription(subs *Subscription, trans *Transaction delStatus := subs.EpList.DelEndpoint(trans.GetEndpoint()) epamount := subs.EpList.Size() + seqId := subs.ReqId.Seq - // - // If last endpoint remove from register map - // - if epamount == 0 { - if _, ok := r.register[subs.Seq]; ok { - xapp.Logger.Debug("Registry: Delete %s", subs.String()) - delete(r.register, subs.Seq) - xapp.Logger.Debug("Registry: substable=%v", r.register) - } + if delStatus == false { + return nil } - r.mutex.Unlock() - // - // Wait some time before really do route updates - // - if waitRouteClean > 0 { - subs.mutex.Unlock() - time.Sleep(waitRouteClean) - subs.mutex.Lock() - } + go func() { + if waitRouteClean > 0 { + time.Sleep(waitRouteClean) + } - xapp.Logger.Info("Registry: Cleaning %s", subs.String()) + subs.mutex.Lock() + defer subs.mutex.Unlock() + xapp.Logger.Info("CLEAN %s", subs.String()) - // - // Subscription route updates - // - if delStatus { if epamount == 0 { - tmpList := RmrEndpointList{} + // + // Subscription route delete + // + tmpList := xapptweaks.RmrEndpointList{} tmpList.AddEndpoint(trans.GetEndpoint()) - subRouteAction := SubRouteInfo{DELETE, tmpList, subs.Seq} - r.rtmgrClient.SubscriptionRequestUpdate(subRouteAction) - } else { - subRouteAction := SubRouteInfo{UPDATE, subs.EpList, subs.Seq} + subRouteAction := SubRouteInfo{tmpList, uint16(seqId)} + r.rtmgrClient.SubscriptionRequestDelete(subRouteAction) + + // + // Subscription release + // + r.mutex.Lock() + defer r.mutex.Unlock() + + if _, ok := r.register[seqId]; ok { + xapp.Logger.Debug("RELEASE %s", subs.String()) + delete(r.register, seqId) + xapp.Logger.Debug("Registry: substable=%v", r.register) + } + r.subIds = append(r.subIds, seqId) + + } else if subs.EpList.Size() > 0 { + // + // Subscription route updates + // + subRouteAction := SubRouteInfo{subs.EpList, uint16(seqId)} r.rtmgrClient.SubscriptionRequestUpdate(subRouteAction) } - } - r.mutex.Lock() - // - // If last endpoint free seq nro - // - if epamount == 0 { - r.subIds = append(r.subIds, subs.Seq) - } + }() return nil } -func (r *Registry) GetSubscription(sn uint16) *Subscription { +func (r *Registry) GetSubscription(sn uint32) *Subscription { r.mutex.Lock() defer r.mutex.Unlock() if _, ok := r.register[sn]; ok { @@ -220,7 +285,7 @@ func (r *Registry) GetSubscription(sn uint16) *Subscription { return nil } -func (r *Registry) GetSubscriptionFirstMatch(ids []uint16) (*Subscription, error) { +func (r *Registry) GetSubscriptionFirstMatch(ids []uint32) (*Subscription, error) { r.mutex.Lock() defer r.mutex.Unlock() for _, id := range ids {