X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fcontrol%2Fregistry.go;h=396daded93335e834879b7442a2f6a0f0786e5d5;hb=cc7d9e0cf982acb7448e23e8c8d16b9aaaf133fb;hp=2750b78e28d45424d6dca1acd28b443c6bd847e0;hpb=12d31af1cdfcbf5f634d9cf666e8e174c74ecb27;p=ric-plt%2Fsubmgr.git diff --git a/pkg/control/registry.go b/pkg/control/registry.go index 2750b78..396dade 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,44 +32,130 @@ 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) AssignToSubscription(trans *Transaction, subReqMsg *e2ap.E2APSubscriptionRequest) (*Subscription, error) { +func (r *Registry) QueryHandler() (models.SubscriptionList, error) { r.mutex.Lock() defer r.mutex.Unlock() - var sequenceNumber uint16 + resp := models.SubscriptionList{} + for _, subs := range r.register { + subs.mutex.Lock() + resp = append(resp, &models.SubscriptionData{SubscriptionID: int64(subs.ReqId.InstanceId), Meid: subs.Meid.RanName, Endpoint: subs.EpList.StringList()}) + subs.mutex.Unlock() + } + return resp, nil +} - // - // Allocate subscription - // +func (r *Registry) allocateSubs(trans *TransactionXapp, subReqMsg *e2ap.E2APSubscriptionRequest) (*Subscription, error) { if len(r.subIds) > 0 { - sequenceNumber = r.subIds[0] + subId := r.subIds[0] r.subIds = r.subIds[1:] - if _, ok := r.register[sequenceNumber]; ok == true { - r.subIds = append(r.subIds, sequenceNumber) - return nil, fmt.Errorf("Registry: Failed to reserves subscription") + if _, ok := r.register[subId]; ok == true { + r.subIds = append(r.subIds, subId) + return nil, fmt.Errorf("Registry: Failed to reserve subscription exists") + } + subs := &Subscription{ + registry: r, + Meid: trans.Meid, + SubReqMsg: subReqMsg, + valid: true, + } + subs.ReqId.Id = 123 + subs.ReqId.InstanceId = subId + + if subs.EpList.AddEndpoint(trans.GetEndpoint()) == false { + r.subIds = append(r.subIds, subs.ReqId.InstanceId) + return nil, fmt.Errorf("Registry: Endpoint existing already in subscription") + } + + return subs, nil + } + return nil, fmt.Errorf("Registry: Failed to reserve subscription no free ids") +} + +func (r *Registry) findExistingSubs(trans *TransactionXapp, subReqMsg *e2ap.E2APSubscriptionRequest) *Subscription { + + for _, subs := range r.register { + if subs.IsMergeable(trans, subReqMsg) { + + // + // check if there has been race conditions + // + subs.mutex.Lock() + //subs has been set to invalid + if subs.valid == false { + 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() + continue + } + subs.mutex.Unlock() + + xapp.Logger.Debug("Registry: Mergeable subs found %s for %s", subs.String(), trans.String()) + return subs } - } else { - return nil, fmt.Errorf("Registry: Failed to reserves subscription no free ids") } - subs := &Subscription{ - registry: r, - Seq: sequenceNumber, - Meid: trans.Meid, + return nil +} + +func (r *Registry) AssignToSubscription(trans *TransactionXapp, subReqMsg *e2ap.E2APSubscriptionRequest) (*Subscription, error) { + var err error + var newAlloc bool + r.mutex.Lock() + defer r.mutex.Unlock() + + // + // 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()) + // Update message data to subscription + subs.SubReqMsg = subReqMsg + subs.SetCachedResponse(nil, true) + return subs, nil + } + } + + subs := r.findExistingSubs(trans, subReqMsg) + if subs == nil { + subs, err = r.allocateSubs(trans, subReqMsg) + if err != nil { + return nil, err + } + newAlloc = true } // @@ -76,39 +164,63 @@ func (r *Registry) AssignToSubscription(trans *Transaction, subReqMsg *e2ap.E2AP subs.mutex.Lock() defer subs.mutex.Unlock() - if subs.EpList.AddEndpoint(trans.GetEndpoint()) == false { - r.subIds = append(r.subIds, sequenceNumber) - return nil, fmt.Errorf("Registry: Endpoint existing already in subscription") - } epamount := subs.EpList.Size() r.mutex.Unlock() // // Subscription route updates // - var err error if epamount == 1 { - subRouteAction := SubRouteInfo{CREATE, subs.EpList, subs.Seq} - err = r.rtmgrClient.SubscriptionRequestUpdate(subRouteAction) + subRouteAction := SubRouteInfo{subs.EpList, uint16(subs.ReqId.InstanceId)} + err = r.rtmgrClient.SubscriptionRequestCreate(subRouteAction) } else { - subRouteAction := SubRouteInfo{UPDATE, subs.EpList, subs.Seq} + subRouteAction := SubRouteInfo{subs.EpList, uint16(subs.ReqId.InstanceId)} err = r.rtmgrClient.SubscriptionRequestUpdate(subRouteAction) } r.mutex.Lock() if err != nil { - r.subIds = append(r.subIds, sequenceNumber) + if newAlloc { + r.subIds = append(r.subIds, subs.ReqId.InstanceId) + } return nil, err } - subs.SubReqMsg = subReqMsg - r.register[sequenceNumber] = subs - xapp.Logger.Debug("Registry: Create %s", subs.String()) + if newAlloc { + r.register[subs.ReqId.InstanceId] = subs + } + xapp.Logger.Debug("CREATE %s", subs.String()) xapp.Logger.Debug("Registry: substable=%v", r.register) return subs, nil } -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() subs.mutex.Lock() @@ -116,72 +228,72 @@ func (r *Registry) RemoveFromSubscription(subs *Subscription, trans *Transaction delStatus := subs.EpList.DelEndpoint(trans.GetEndpoint()) epamount := subs.EpList.Size() + subId := subs.ReqId.InstanceId - // - // 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(subId)} + r.rtmgrClient.SubscriptionRequestDelete(subRouteAction) + + // + // Subscription release + // + r.mutex.Lock() + defer r.mutex.Unlock() + + if _, ok := r.register[subId]; ok { + xapp.Logger.Debug("RELEASE %s", subs.String()) + delete(r.register, subId) + xapp.Logger.Debug("Registry: substable=%v", r.register) + } + r.subIds = append(r.subIds, subId) + + } else if subs.EpList.Size() > 0 { + // + // Subscription route updates + // + subRouteAction := SubRouteInfo{subs.EpList, uint16(subId)} 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(subId uint32) *Subscription { r.mutex.Lock() defer r.mutex.Unlock() - if _, ok := r.register[sn]; ok { - return r.register[sn] + if _, ok := r.register[subId]; ok { + return r.register[subId] } return nil } -func (r *Registry) GetSubscriptionFirstMatch(ids []uint16) (*Subscription, error) { +func (r *Registry) GetSubscriptionFirstMatch(subIds []uint32) (*Subscription, error) { r.mutex.Lock() defer r.mutex.Unlock() - for _, id := range ids { - if _, ok := r.register[id]; ok { - return r.register[id], nil + for _, subId := range subIds { + if _, ok := r.register[subId]; ok { + return r.register[subId], nil } } - return nil, fmt.Errorf("No valid subscription found with ids %v", ids) + return nil, fmt.Errorf("No valid subscription found with subIds %v", subIds) }