X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fxapp%2Frmr.go;h=56e0eb1583afbcc631b2b8b25ab80422923973da;hb=refs%2Fchanges%2F95%2F1395%2F1;hp=0b643d564e733bacf6ba9c762a0f5a0c6cfcc908;hpb=d969b203c9d3b7e3bbf98257ed17edaddd2453b1;p=ric-plt%2Fxapp-frame.git diff --git a/pkg/xapp/rmr.go b/pkg/xapp/rmr.go old mode 100755 new mode 100644 index 0b643d5..56e0eb1 --- a/pkg/xapp/rmr.go +++ b/pkg/xapp/rmr.go @@ -51,6 +51,17 @@ var RMRCounterOpts = []CounterOpts{ {Name: "ReceiveError", Help: "The total number of RMR receive errors"}, } +type RMRParams struct { + Mtype int + Payload []byte + PayloadLen int + Meid *RMRMeid + Xid string + SubId int + Src string + Mbuf *C.rmr_mbuf_t +} + func NewRMRClient() *RMRClient { p := C.CString(viper.GetString("rmr.protPort")) m := C.int(viper.GetInt("rmr.maxSize")) @@ -116,18 +127,35 @@ func (m *RMRClient) parseMessage(rxBuffer *C.rmr_mbuf_t) { return } - meid := &RMRMeid{} + params := &RMRParams{} + params.Mbuf = rxBuffer + params.Mtype = int(rxBuffer.mtype) + params.SubId = int(rxBuffer.sub_id) + params.Meid = &RMRMeid{} + meidBuf := make([]byte, int(C.RMR_MAX_MEID)) if meidCstr := C.rmr_get_meid(rxBuffer, (*C.uchar)(unsafe.Pointer(&meidBuf[0]))); meidCstr != nil { - meid.PlmnID = strings.TrimRight(string(meidBuf[0:16]), "\000") - meid.EnbID = strings.TrimRight(string(meidBuf[16:32]), "\000") + params.Meid.RanName = strings.TrimRight(string(meidBuf), "\000") + //params.Meid.PlmnID = strings.TrimRight(string(meidBuf[0:16]), "\000") + //params.Meid.EnbID = strings.TrimRight(string(meidBuf[16:32]), "\000") + } + + xidBuf := make([]byte, int(C.RMR_MAX_XID)) + if xidCstr := C.rmr_get_xact(rxBuffer, (*C.uchar)(unsafe.Pointer(&xidBuf[0]))); xidCstr != nil { + params.Xid = strings.TrimRight(string(xidBuf[0:32]), "\000") + } + + srcBuf := make([]byte, int(C.RMR_MAX_SRC)) + if srcStr := C.rmr_get_src(rxBuffer, (*C.uchar)(unsafe.Pointer(&srcBuf[0]))); srcStr != nil { + params.Src = strings.TrimRight(string(srcBuf[0:64]), "\000") } for _, c := range m.consumers { cptr := unsafe.Pointer(rxBuffer.payload) - payload := C.GoBytes(cptr, C.int(rxBuffer.len)) + params.Payload = C.GoBytes(cptr, C.int(rxBuffer.len)) + params.PayloadLen = int(rxBuffer.len) - err := c.Consume(int(rxBuffer.mtype), int(rxBuffer.sub_id), payload, meid) + err := c.Consume(params) if err != nil { Logger.Warn("rmrClient: Consumer returned error: %v", err) } @@ -139,53 +167,95 @@ func (m *RMRClient) Allocate() *C.rmr_mbuf_t { if buf == nil { Logger.Error("rmrClient: Allocating message buffer failed!") } - return buf } -func (m *RMRClient) Send(mtype int, sid int, payload []byte, meid *RMRMeid) bool { - buf := m.Allocate() +func (m *RMRClient) Free(mbuf *C.rmr_mbuf_t) { + if mbuf == nil { + return + } + C.rmr_free_msg(mbuf) +} + +func (m *RMRClient) SendMsg(params *RMRParams) bool { + return m.Send(params, false) +} + +func (m *RMRClient) SendRts(params *RMRParams) bool { + return m.Send(params, true) +} + +func (m *RMRClient) Send(params *RMRParams, isRts bool) bool { + txBuffer := params.Mbuf + if txBuffer == nil { + txBuffer = m.Allocate() + } - buf.mtype = C.int(mtype) - buf.sub_id = C.int(sid) - buf.len = C.int(len(payload)) - datap := C.CBytes(payload) + txBuffer.mtype = C.int(params.Mtype) + txBuffer.sub_id = C.int(params.SubId) + txBuffer.len = C.int(len(params.Payload)) + if params.PayloadLen != 0 { + txBuffer.len = C.int(params.PayloadLen) + } + datap := C.CBytes(params.Payload) defer C.free(datap) - if meid != nil { - b := make([]byte, int(C.RMR_MAX_MEID)) - copy(b, []byte(meid.PlmnID)) - copy(b[16:], []byte(meid.EnbID)) - C.rmr_bytes2meid(buf, (*C.uchar)(unsafe.Pointer(&b[0])), C.int(len(b))) + if params != nil { + if params.Meid != nil { + b := make([]byte, int(C.RMR_MAX_MEID)) + copy(b, []byte(params.Meid.RanName)) + //copy(b, []byte(params.Meid.PlmnID)) + //copy(b[16:], []byte(params.Meid.EnbID)) + C.rmr_bytes2meid(txBuffer, (*C.uchar)(unsafe.Pointer(&b[0])), C.int(len(b))) + } + xidLen := len(params.Xid) + if xidLen > 0 && xidLen <= C.RMR_MAX_XID { + b := make([]byte, int(C.RMR_MAX_XID)) + copy(b, []byte(params.Xid)) + C.rmr_bytes2xact(txBuffer, (*C.uchar)(unsafe.Pointer(&b[0])), C.int(len(b))) + } } - C.write_bytes_array(buf.payload, datap, buf.len) + C.write_bytes_array(txBuffer.payload, datap, txBuffer.len) - return m.SendBuf(buf) + return m.SendBuf(txBuffer, isRts) } -func (m *RMRClient) SendBuf(txBuffer *C.rmr_mbuf_t) bool { - for i := 0; i < 10; i++ { - txBuffer.state = 0 - txBuffer := C.rmr_send_msg(m.context, txBuffer) - if txBuffer == nil { - break - } else if txBuffer.state != C.RMR_OK { - if txBuffer.state != C.RMR_ERR_RETRY { - time.Sleep(100 * time.Microsecond) - m.UpdateStatCounter("TransmitError") - } - for j := 0; j < 100 && txBuffer.state == C.RMR_ERR_RETRY; j++ { - txBuffer = C.rmr_send_msg(m.context, txBuffer) - } - } +func (m *RMRClient) SendBuf(txBuffer *C.rmr_mbuf_t, isRts bool) bool { + var ( + currBuffer *C.rmr_mbuf_t + state bool = true + counterName string = "Transmitted" + ) + + txBuffer.state = 0 + if isRts { + currBuffer = C.rmr_rts_msg(m.context, txBuffer) + } else { + currBuffer = C.rmr_send_msg(m.context, txBuffer) + } + + if currBuffer == nil { + m.UpdateStatCounter("TransmitError") + return false + } - if txBuffer.state == C.RMR_OK { - m.UpdateStatCounter("Transmitted") - return true + // Just quick retry seems to help for K8s issue + for j := 0; j < 3 && currBuffer != nil && currBuffer.state == C.RMR_ERR_RETRY; j++ { + if isRts { + currBuffer = C.rmr_rts_msg(m.context, currBuffer) + } else { + currBuffer = C.rmr_send_msg(m.context, currBuffer) } } - m.UpdateStatCounter("TransmitError") - return false + + if currBuffer.state != C.RMR_OK { + state = false + counterName = "TransmitError" + } + + m.UpdateStatCounter(counterName) + m.Free(currBuffer) + return state } func (m *RMRClient) UpdateStatCounter(name string) {