X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fsbi%2Fnngpush.go;h=dbd5d5910f1c036eb7e59abb88cd11f2643d7ff3;hb=refs%2Fchanges%2F31%2F4531%2F1;hp=44512999cba04b37142b1ae9284272926d5289c9;hpb=df23f25ac56eb0076e6e8801bb6080468ccf44b6;p=ric-plt%2Frtmgr.git diff --git a/pkg/sbi/nngpush.go b/pkg/sbi/nngpush.go index 4451299..dbd5d59 100644 --- a/pkg/sbi/nngpush.go +++ b/pkg/sbi/nngpush.go @@ -21,24 +21,15 @@ ================================================================================== */ /* - Mnemonic: nngpipe.go - Abstract: mangos (NNG) Pipeline SBI implementation + Mnemonic: rmrpipe.go + Abstract: mangos (RMR) Pipeline SBI implementation Date: 12 March 2019 */ package sbi /* -#include -#include -#include -#include #include -#include - - -#cgo CFLAGS: -I../ -#cgo LDFLAGS: -lrmr_nng -lnng */ import "C" @@ -47,91 +38,64 @@ import ( "crypto/md5" "errors" "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp" - "nanomsg.org/go/mangos/v2" - "nanomsg.org/go/mangos/v2/protocol/push" - _ "nanomsg.org/go/mangos/v2/transport/all" "routing-manager/pkg/rtmgr" "strconv" - "time" + "strings" "fmt" ) -type NngPush struct { +var rmrcallid = 1 +var rmrdynamiccallid = 201 + +type RmrPush struct { Sbi - NewSocket CreateNewNngSocketHandler - rcChan chan *xapp.RMRParams + rcChan chan *xapp.RMRParams } -type RMRParams struct { - *xapp.RMRParams +type EPStatus struct { + endpoint string + status bool } +type RMRParams struct { + *xapp.RMRParams +} func (params *RMRParams) String() string { - var b bytes.Buffer - sum := md5.Sum(params.Payload) - fmt.Fprintf(&b, "params(Src=%s Mtype=%d SubId=%d Xid=%s Meid=%s Paylens=%d/%d Payhash=%x)", params.Src, params.Mtype, params.SubId, params.Xid, params.Meid.RanName, params.PayloadLen, len(params.Payload), sum) - return b.String() + var b bytes.Buffer + sum := md5.Sum(params.Payload) + fmt.Fprintf(&b, "params(Src=%s Mtype=%d SubId=%d Xid=%s Meid=%s Paylens=%d/%d Payhash=%x)", params.Src, params.Mtype, params.SubId, params.Xid, params.Meid.RanName, params.PayloadLen, len(params.Payload), sum) + return b.String() } -func NewNngPush() *NngPush { - instance := new(NngPush) - instance.NewSocket = createNewPushSocket +func NewRmrPush() *RmrPush { + instance := new(RmrPush) return instance } -func createNewPushSocket() (NngSocket, error) { - xapp.Logger.Debug("Invoked: createNewPushSocket()") - socket, err := push.NewSocket() - if err != nil { - return nil, errors.New("can't create new push socket due to:" + err.Error()) - } - socket.SetPipeEventHook(pipeEventHandler) - return socket, nil -} - -func pipeEventHandler(event mangos.PipeEvent, pipe mangos.Pipe) { - xapp.Logger.Debug("Invoked: pipeEventHandler()") - xapp.Logger.Debug("Received pipe event for " + pipe.Address() + " address") - for _, ep := range rtmgr.Eps { - uri := DefaultNngPipelineSocketPrefix + ep.Ip + ":" + strconv.Itoa(DefaultNngPipelineSocketNumber) - if uri == pipe.Address() { - switch event { - case 1: - ep.IsReady = true - xapp.Logger.Debug("Endpoint " + uri + " successfully attached") - default: - ep.IsReady = false - xapp.Logger.Debug("Endpoint " + uri + " has been detached") - } - } - } -} - -func (c *NngPush) Initialize(ip string) error { +func (c *RmrPush) Initialize(ip string) error { return nil } -func (c *NngPush) Terminate() error { +func (c *RmrPush) Terminate() error { return nil } -func (c *NngPush) AddEndpoint(ep *rtmgr.Endpoint) error { +func (c *RmrPush) AddEndpoint(ep *rtmgr.Endpoint) error { xapp.Logger.Debug("Invoked sbi.AddEndpoint") - xapp.Logger.Debug("args: %v", *ep) - endpoint := ep.Ip + ":" + strconv.Itoa(DefaultNngPipelineSocketNumber) + endpoint := ep.Ip + ":" + strconv.Itoa(DefaultRmrPipelineSocketNumber) ep.Whid = int(xapp.Rmr.Openwh(endpoint)) - if ep.Whid < 0 { + if ep.Whid < 0 { return errors.New("can't open warmhole connection for endpoint:" + ep.Uuid + " due to invalid Wormhole ID: " + string(ep.Whid)) - }else { - xapp.Logger.Debug("Wormhole ID is %v and EP is %v",ep.Whid,endpoint) + } else { + xapp.Logger.Debug("Wormhole ID is %v and EP is %v", ep.Whid, endpoint) } return nil } -func (c *NngPush) DeleteEndpoint(ep *rtmgr.Endpoint) error { +func (c *RmrPush) DeleteEndpoint(ep *rtmgr.Endpoint) error { xapp.Logger.Debug("Invoked sbi. DeleteEndpoint") xapp.Logger.Debug("args: %v", *ep) @@ -139,61 +103,145 @@ func (c *NngPush) DeleteEndpoint(ep *rtmgr.Endpoint) error { return nil } -func (c *NngPush) UpdateEndpoints(rcs *rtmgr.RicComponents) { +func (c *RmrPush) UpdateEndpoints(rcs *rtmgr.RicComponents) { c.updateEndpoints(rcs, c) } -/* -NOTE: Asynchronous dial starts a goroutine which keep maintains the connection to the given endpoint -*/ -func (c *NngPush) dial(ep *rtmgr.Endpoint) error { - xapp.Logger.Debug("Dialing to endpoint: " + ep.Uuid) - uri := DefaultNngPipelineSocketPrefix + ep.Ip + ":" + strconv.Itoa(DefaultNngPipelineSocketNumber) - options := make(map[string]interface{}) - options[mangos.OptionDialAsynch] = true - if err := ep.Socket.(NngSocket).DialOptions(uri, options); err != nil { - return errors.New("can't dial on push socket to " + uri + " due to: " + err.Error()) - } - return nil -} - -func (c *NngPush) DistributeAll(policies *[]string) error { +func (c *RmrPush) DistributeAll(policies *[]string) error { xapp.Logger.Debug("Invoked: sbi.DistributeAll") xapp.Logger.Debug("args: %v", *policies) - for _, ep := range rtmgr.Eps { + /*for _, ep := range rtmgr.Eps { go c.send(ep, policies) + }*/ + channel := make(chan EPStatus) + + if rmrcallid == 200 { + rmrcallid = 1 } + for _, ep := range rtmgr.Eps { + go c.send_sync(ep, policies, channel, rmrcallid) + } + for rEp, id := range rtmgr.RmrEp { + go c.send_rt_process(rEp,id,policies,rmrcallid) + } + + rmrcallid++ + + count := 0 + result := make([]EPStatus, len(rtmgr.Eps)) + for i, _ := range result { + result[i] = <-channel + if result[i].status == true { + count++ + } else { + xapp.Logger.Error("RMR send failed for endpoint %v", result[i].endpoint) + } + } + + if count < len(rtmgr.Eps) { + return errors.New(" RMR response count " + string(count) + " is less than half of endpoint list " + string(len(rtmgr.Eps))) + } + + return nil } -func (c *NngPush) send(ep *rtmgr.Endpoint, policies *[]string) { - xapp.Logger.Debug("Push policy to endpoint: " + ep.Uuid) +func (c *RmrPush) send_sync(ep *rtmgr.Endpoint, policies *[]string, channel chan EPStatus, call_id int) { + xapp.Logger.Debug("Push policy to endpoint: " + ep.Uuid) - for _, pe := range *policies { - params := &RMRParams{&xapp.RMRParams{}} - params.Mtype = 20 - params.PayloadLen = len([]byte(pe)) - params.Payload =[]byte(pe) - params.Mbuf = nil - params.Whid = ep.Whid - time.Sleep(1 * time.Millisecond) - xapp.Rmr.SendMsg(params.RMRParams) - } - xapp.Logger.Info("NNG PUSH to endpoint " + ep.Uuid + ": OK (# of Entries:" + strconv.Itoa(len(*policies)) + ")") -} + ret := c.send_data(ep, policies, call_id) + + channel <- EPStatus{ep.Uuid, ret} -func (c *NngPush) CreateEndpoint(payload string) (*rtmgr.Endpoint){ - return c.createEndpoint(payload, c) } -func (c *NngPush) DistributeToEp(policies *[]string, ep *rtmgr.Endpoint) error { +func (c *RmrPush) send_data(ep *rtmgr.Endpoint, policies *[]string, call_id int) bool { + xapp.Logger.Debug("Invoked send_data to endpoint: " + ep.Uuid + " call_id: " + strconv.Itoa(call_id)) + var state int + var retstr string + + var policy = []byte{} + + for _, pe := range *policies { + b:= []byte(pe) + for j:=0; j