X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fxapp%2Frmr.go;h=90bc64b6430d19f906572086aef59c4e4d5dd4f7;hb=refs%2Fchanges%2F63%2F3863%2F5;hp=c835408fa303f19c83d1c5b1fcaa6c8fe47212f6;hpb=192518d79ebcffe17bf569bf9037321a6fd26d44;p=ric-plt%2Fxapp-frame.git diff --git a/pkg/xapp/rmr.go b/pkg/xapp/rmr.go index c835408..90bc64b 100755 --- a/pkg/xapp/rmr.go +++ b/pkg/xapp/rmr.go @@ -24,6 +24,8 @@ package xapp #include #include #include +#include +#include #include #include @@ -31,15 +33,40 @@ void write_bytes_array(unsigned char *dst, void *data, int len) { memcpy((void *)dst, (void *)data, len); } +int init_epoll(int rcv_fd) { + struct epoll_event epe; + int epoll_fd = epoll_create1( 0 ); + epe.events = EPOLLIN; + epe.data.fd = rcv_fd; + epoll_ctl( epoll_fd, EPOLL_CTL_ADD, rcv_fd, &epe ); + return epoll_fd; +} + +void close_epoll(int epoll_fd) { + if(epoll_fd >= 0) { + close(epoll_fd); + } +} + +int wait_epoll(int epoll_fd,int rcv_fd) { + struct epoll_event events[1]; + if( epoll_wait( epoll_fd, events, 1, -1 ) > 0 ) { + if( events[0].data.fd == rcv_fd ) { + return 1; + } + } + return 0; +} + #cgo CFLAGS: -I../ -#cgo LDFLAGS: -lrmr_nng -lnng +#cgo LDFLAGS: -lrmr_si */ import "C" import ( + "fmt" "github.com/spf13/viper" - "strconv" - "sync" + "strings" "time" "unsafe" ) @@ -51,82 +78,114 @@ var RMRCounterOpts = []CounterOpts{ {Name: "ReceiveError", Help: "The total number of RMR receive errors"}, } -// To be removed ... -type RMRStatistics struct{} - -type RMRClient struct { - context unsafe.Pointer - ready int - wg sync.WaitGroup - mux sync.Mutex - stat map[string]Counter - consumers []MessageConsumer - readyCb ReadyCB - readyCbParams interface{} +var RMRErrors = map[int]string{ + C.RMR_OK: "state is good", + C.RMR_ERR_BADARG: "argument passed to function was unusable", + C.RMR_ERR_NOENDPT: "send/call could not find an endpoint based on msg type", + C.RMR_ERR_EMPTY: "msg received had no payload; attempt to send an empty message", + C.RMR_ERR_NOHDR: "message didn't contain a valid header", + C.RMR_ERR_SENDFAILED: "send failed; errno has nano reason", + C.RMR_ERR_CALLFAILED: "unable to send call() message", + C.RMR_ERR_NOWHOPEN: "no wormholes are open", + C.RMR_ERR_WHID: "wormhole id was invalid", + C.RMR_ERR_OVERFLOW: "operation would have busted through a buffer/field size", + C.RMR_ERR_RETRY: "request (send/call/rts) failed, but caller should retry (EAGAIN for wrappers)", + C.RMR_ERR_RCVFAILED: "receive failed (hard error)", + C.RMR_ERR_TIMEOUT: "message processing call timed out", + C.RMR_ERR_UNSET: "the message hasn't been populated with a transport buffer", + C.RMR_ERR_TRUNC: "received message likely truncated", + C.RMR_ERR_INITFAILED: "initialization of something (probably message) failed", + C.RMR_ERR_NOTSUPP: "the request is not supported, or RMr was not initialized for the request", } -type MessageConsumer interface { - Consume(mtype int, sid int, len int, payload []byte) error +type RMRParams struct { + Mtype int + Payload []byte + PayloadLen int + Meid *RMRMeid + Xid string + SubId int + Src string + Mbuf *C.rmr_mbuf_t + Whid int + Callid int + Timeout int + status int } -func NewRMRClient() *RMRClient { - p := C.CString(viper.GetString("rmr.protPort")) - m := C.int(viper.GetInt("rmr.maxSize")) +func NewRMRClientWithParams(protPort string, maxSize int, threadType int, statDesc string) *RMRClient { + p := C.CString(protPort) + m := C.int(maxSize) + c := C.int(threadType) defer C.free(unsafe.Pointer(p)) - - ctx := C.rmr_init(p, m, C.int(0)) + ctx := C.rmr_init(p, m, c) if ctx == nil { Logger.Error("rmrClient: Initializing RMR context failed, bailing out!") } - return &RMRClient{ + protPort: protPort, context: ctx, consumers: make([]MessageConsumer, 0), - stat: Metric.RegisterCounterGroup(RMRCounterOpts, "RMR"), + stat: Metric.RegisterCounterGroup(RMRCounterOpts, statDesc), } } -func (m *RMRClient) Start(c MessageConsumer) { - for { - Logger.Info("rmrClient: Waiting for RMR to be ready ...") - - if m.ready = int(C.rmr_ready(m.context)); m.ready == 1 { - break - } - time.Sleep(10 * time.Second) - } - m.wg.Add(viper.GetInt("rmr.numWorkers")) +func NewRMRClient() *RMRClient { + return NewRMRClientWithParams(viper.GetString("rmr.protPort"), viper.GetInt("rmr.maxSize"), viper.GetInt("rmr.threadType"), "RMR") +} +func (m *RMRClient) Start(c MessageConsumer) { if c != nil { m.consumers = append(m.consumers, c) } - for w := 0; w < viper.GetInt("rmr.numWorkers"); w++ { - go m.Worker("worker-"+strconv.Itoa(w), 0) + var counter int = 0 + for { + m.contextMux.Lock() + m.ready = int(C.rmr_ready(m.context)) + m.contextMux.Unlock() + if m.ready == 1 { + Logger.Info("rmrClient: RMR is ready after %d seconds waiting...", counter) + break + } + if counter%10 == 0 { + Logger.Info("rmrClient: Waiting for RMR to be ready ...") + } + time.Sleep(1 * time.Second) + counter++ } + m.wg.Add(1) if m.readyCb != nil { - m.readyCb(m.readyCbParams) + go m.readyCb(m.readyCbParams) } - m.Wait() -} + go func() { + m.contextMux.Lock() + rfd := C.rmr_get_rcvfd(m.context) + m.contextMux.Unlock() + efd := C.init_epoll(rfd) -func (m *RMRClient) Worker(taskName string, msgSize int) { - p := viper.GetString("rmr.protPort") - Logger.Info("rmrClient: '%s': receiving messages on [%s]", taskName, p) - - defer m.wg.Done() - for { - rxBuffer := C.rmr_rcv_msg(m.context, nil) - if rxBuffer == nil { - m.UpdateStatCounter("ReceiveError") - continue + defer m.wg.Done() + for { + if int(C.wait_epoll(efd, rfd)) == 0 { + continue + } + m.contextMux.Lock() + rxBuffer := C.rmr_rcv_msg(m.context, nil) + m.contextMux.Unlock() + + if rxBuffer == nil { + m.LogMBufError("RecvMsg failed", rxBuffer) + m.UpdateStatCounter("ReceiveError") + continue + } + m.UpdateStatCounter("Received") + m.parseMessage(rxBuffer) } - m.UpdateStatCounter("Received") + }() - go m.parseMessage(rxBuffer) - } + m.wg.Wait() } func (m *RMRClient) parseMessage(rxBuffer *C.rmr_mbuf_t) { @@ -135,62 +194,254 @@ func (m *RMRClient) parseMessage(rxBuffer *C.rmr_mbuf_t) { return } + 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 { + params.Meid.RanName = strings.TrimRight(string(meidBuf), "\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") + } + + // Default case: a single consumer + if len(m.consumers) == 1 && m.consumers[0] != nil { + params.PayloadLen = int(rxBuffer.len) + params.Payload = (*[1 << 30]byte)(unsafe.Pointer(rxBuffer.payload))[:params.PayloadLen:params.PayloadLen] + err := m.consumers[0].Consume(params) + if err != nil { + Logger.Warn("rmrClient: Consumer returned error: %v", err) + } + return + } + + // Special case for multiple consumers 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) + params.Mtype = int(rxBuffer.mtype) + params.SubId = int(rxBuffer.sub_id) - err := c.Consume(int(rxBuffer.mtype), int(rxBuffer.sub_id), int(rxBuffer.len), payload) + err := c.Consume(params) if err != nil { Logger.Warn("rmrClient: Consumer returned error: %v", err) } } } -func (m *RMRClient) Allocate() *C.rmr_mbuf_t { - buf := C.rmr_alloc_msg(m.context, 0) +func (m *RMRClient) Allocate(size int) *C.rmr_mbuf_t { + m.contextMux.Lock() + defer m.contextMux.Unlock() + buf := C.rmr_alloc_msg(m.context, C.int(size)) if buf == nil { Logger.Error("rmrClient: Allocating message buffer failed!") } - return buf } -func (m *RMRClient) Send(mtype int, sid int, len int, payload []byte) bool { - buf := m.Allocate() +func (m *RMRClient) Free(mbuf *C.rmr_mbuf_t) { + if mbuf == nil { + return + } + m.contextMux.Lock() + defer m.contextMux.Unlock() + 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) CopyBuffer(params *RMRParams) *C.rmr_mbuf_t { + if params.Mbuf != nil { + m.Free(params.Mbuf) + params.Mbuf = nil + } + + payLen := len(params.Payload) + if params.PayloadLen != 0 { + payLen = params.PayloadLen + } + + txBuffer := m.Allocate(payLen) + if txBuffer == nil { + return nil + } + txBuffer.mtype = C.int(params.Mtype) + txBuffer.sub_id = C.int(params.SubId) + txBuffer.len = C.int(payLen) - buf.mtype = C.int(mtype) - buf.sub_id = C.int(sid) - buf.len = C.int(len) - datap := C.CBytes(payload) + datap := C.CBytes(params.Payload) defer C.free(datap) - C.write_bytes_array(buf.payload, datap, C.int(len)) + if params != nil { + if params.Meid != nil { + b := make([]byte, int(C.RMR_MAX_MEID)) + copy(b, []byte(params.Meid.RanName)) + 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(txBuffer.payload, datap, txBuffer.len) + return txBuffer +} - return m.SendBuf(buf) +func (m *RMRClient) Send(params *RMRParams, isRts bool) bool { + + txBuffer := m.CopyBuffer(params) + if txBuffer == nil { + return false + } + params.status = m.SendBuf(txBuffer, isRts, params.Whid) + if params.status == int(C.RMR_OK) { + return true + } + return false } -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, whid int) int { + var ( + currBuffer *C.rmr_mbuf_t + counterName string = "Transmitted" + ) + + m.contextMux.Lock() + txBuffer.state = 0 + if whid != 0 { + currBuffer = C.rmr_wh_send_msg(m.context, C.rmr_whid_t(whid), txBuffer) + } else { + if isRts { + currBuffer = C.rmr_rts_msg(m.context, txBuffer) + } else { + currBuffer = C.rmr_send_msg(m.context, txBuffer) } + } + m.contextMux.Unlock() + + if currBuffer == nil { + m.UpdateStatCounter("TransmitError") + return m.LogMBufError("SendBuf failed", txBuffer) + } + + // Just quick retry seems to help for K8s issue + maxRetryOnFailure := viper.GetInt("rmr.maxRetryOnFailure") + if maxRetryOnFailure == 0 { + maxRetryOnFailure = 5 + } - if txBuffer.state == C.RMR_OK { - m.UpdateStatCounter("Transmitted") - return true + for j := 0; j < maxRetryOnFailure && currBuffer != nil && currBuffer.state == C.RMR_ERR_RETRY; j++ { + m.contextMux.Lock() + if whid != 0 { + currBuffer = C.rmr_wh_send_msg(m.context, C.rmr_whid_t(whid), txBuffer) + } else { + if isRts { + currBuffer = C.rmr_rts_msg(m.context, txBuffer) + } else { + currBuffer = C.rmr_send_msg(m.context, txBuffer) + } } + m.contextMux.Unlock() + } + + if currBuffer.state != C.RMR_OK { + counterName = "TransmitError" + m.LogMBufError("SendBuf failed", currBuffer) + } + + m.UpdateStatCounter(counterName) + defer m.Free(currBuffer) + + return int(currBuffer.state) +} + +func (m *RMRClient) SendCallMsg(params *RMRParams) (int, string) { + var ( + currBuffer *C.rmr_mbuf_t + counterName string = "Transmitted" + ) + txBuffer := m.CopyBuffer(params) + if txBuffer == nil { + return C.RMR_ERR_INITFAILED, "" + } + + txBuffer.state = 0 + + m.contextMux.Lock() + currBuffer = C.rmr_wh_call(m.context, C.int(params.Whid), txBuffer, C.int(params.Callid), C.int(params.Timeout)) + m.contextMux.Unlock() + + if currBuffer == nil { + m.UpdateStatCounter("TransmitError") + return m.LogMBufError("SendBuf failed", txBuffer), "" + } + + if currBuffer.state != C.RMR_OK { + counterName = "TransmitError" + m.LogMBufError("SendBuf failed", currBuffer) + } + + m.UpdateStatCounter(counterName) + defer m.Free(currBuffer) + + cptr := unsafe.Pointer(currBuffer.payload) + payload := C.GoBytes(cptr, C.int(currBuffer.len)) + + return int(currBuffer.state), string(payload) +} + +func (m *RMRClient) Openwh(target string) C.rmr_whid_t { + return m.Wh_open(target) +} + +func (m *RMRClient) Wh_open(target string) C.rmr_whid_t { + m.contextMux.Lock() + defer m.contextMux.Unlock() + endpoint := C.CString(target) + return C.rmr_wh_open(m.context, endpoint) +} + +func (m *RMRClient) Closewh(whid int) { + m.Wh_close(C.rmr_whid_t(whid)) +} + +func (m *RMRClient) Wh_close(whid C.rmr_whid_t) { + m.contextMux.Lock() + defer m.contextMux.Unlock() + C.rmr_wh_close(m.context, whid) +} + +func (m *RMRClient) IsRetryError(params *RMRParams) bool { + if params.status == int(C.RMR_ERR_RETRY) { + return true + } + return false +} + +func (m *RMRClient) IsNoEndPointError(params *RMRParams) bool { + if params.status == int(C.RMR_ERR_NOENDPT) { + return true } - m.UpdateStatCounter("TransmitError") return false } @@ -204,11 +455,6 @@ func (m *RMRClient) RegisterMetrics() { m.stat = Metric.RegisterCounterGroup(RMRCounterOpts, "RMR") } -// To be removed ... -func (m *RMRClient) GetStat() (r RMRStatistics) { - return -} - func (m *RMRClient) Wait() { m.wg.Wait() } @@ -235,3 +481,13 @@ func (m *RMRClient) GetRicMessageName(id int) (s string) { } return } + +func (m *RMRClient) LogMBufError(text string, mbuf *C.rmr_mbuf_t) int { + Logger.Debug(fmt.Sprintf("rmrClient: %s -> [tp=%v] %v - %s", text, mbuf.tp_state, mbuf.state, RMRErrors[int(mbuf.state)])) + return int(mbuf.state) +} + +// To be removed ... +func (m *RMRClient) GetStat() (r RMRStatistics) { + return +}