X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fxapp%2Frmr.go;h=028cc3c9473aa18af2948b27409775063c729ddd;hb=79f0680fd7bbf1c8a8c6e2a842cb18020e387a47;hp=9a3116d78ccc88c8a7768a7b4e5d3713b8bf7895;hpb=1307b2d8d34f48bf44267efc2d1e607823c8d06d;p=ric-plt%2Fxapp-frame.git diff --git a/pkg/xapp/rmr.go b/pkg/xapp/rmr.go index 9a3116d..028cc3c 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,25 +33,59 @@ 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_si */ import "C" import ( + "bytes" + "crypto/md5" "fmt" - "github.com/spf13/viper" - "strconv" "strings" "time" "unsafe" + + "github.com/spf13/viper" ) var RMRCounterOpts = []CounterOpts{ {Name: "Transmitted", Help: "The total number of transmited RMR messages"}, - {Name: "Received", Help: "The total number of received RMR messages"}, {Name: "TransmitError", Help: "The total number of RMR transmission errors"}, + {Name: "TransmitRetry", Help: "The total number of transmit retries on failure"}, + {Name: "Received", Help: "The total number of received RMR messages"}, {Name: "ReceiveError", Help: "The total number of RMR receive errors"}, + {Name: "SendWithRetryRetry", Help: "SendWithRetry service retries"}, +} + +var RMRGaugeOpts = []CounterOpts{ + {Name: "Enqueued", Help: "The total number of enqueued in RMR library"}, + {Name: "Dropped", Help: "The total number of dropped in RMR library"}, } var RMRErrors = map[int]string{ @@ -72,6 +108,9 @@ var RMRErrors = map[int]string{ C.RMR_ERR_NOTSUPP: "the request is not supported, or RMr was not initialized for the request", } +//----------------------------------------------------------------------------- +// +//----------------------------------------------------------------------------- type RMRParams struct { Mtype int Payload []byte @@ -87,30 +126,74 @@ type RMRParams struct { status int } -func NewRMRClientWithParams(protPort string, maxSize int, numWorkers int, threadType int, statDesc string) *RMRClient { - p := C.CString(protPort) - m := C.int(maxSize) - c := C.int(threadType) - defer C.free(unsafe.Pointer(p)) +func (params *RMRParams) String() string { + var b bytes.Buffer + fmt.Fprintf(&b, "params(Src=%s Mtype=%d SubId=%d Xid=%s Meid=%s Paylens=%d/%d Paymd5=%x)", params.Src, params.Mtype, params.SubId, params.Xid, params.Meid, params.PayloadLen, len(params.Payload), md5.Sum(params.Payload)) + return b.String() +} + +//----------------------------------------------------------------------------- +// +//----------------------------------------------------------------------------- +type RMRClientParams struct { + StatDesc string + RmrData PortData +} + +func (params *RMRClientParams) String() string { + return fmt.Sprintf("ProtPort=%d MaxSize=%d ThreadType=%d StatDesc=%s LowLatency=%t FastAck=%t Policies=%v", + params.RmrData.Port, params.RmrData.MaxSize, params.RmrData.ThreadType, params.StatDesc, + params.RmrData.LowLatency, params.RmrData.FastAck, params.RmrData.Policies) +} - //ctx := C.rmr_init(p, m, C.int(0)) - //ctx := C.rmr_init(p, m, C.RMRFL_NOTHREAD) +//----------------------------------------------------------------------------- +// +//----------------------------------------------------------------------------- +func NewRMRClientWithParams(params *RMRClientParams) *RMRClient { + p := C.CString(fmt.Sprintf("%d", params.RmrData.Port)) + m := C.int(params.RmrData.MaxSize) + c := C.int(params.RmrData.ThreadType) + defer C.free(unsafe.Pointer(p)) ctx := C.rmr_init(p, m, c) if ctx == nil { Logger.Error("rmrClient: Initializing RMR context failed, bailing out!") } + Logger.Info("new rmrClient with parameters: %s", params.String()) + + if params.RmrData.LowLatency { + C.rmr_set_low_latency(ctx) + } + if params.RmrData.FastAck { + C.rmr_set_fack(ctx) + } + return &RMRClient{ - protPort: protPort, - numWorkers: numWorkers, - context: ctx, - consumers: make([]MessageConsumer, 0), - stat: Metric.RegisterCounterGroup(RMRCounterOpts, statDesc), + context: ctx, + consumers: make([]MessageConsumer, 0), + statc: Metric.RegisterCounterGroup(RMRCounterOpts, params.StatDesc), + statg: Metric.RegisterGaugeGroup(RMRGaugeOpts, params.StatDesc), + maxRetryOnFailure: params.RmrData.MaxRetryOnFailure, } } func NewRMRClient() *RMRClient { - return NewRMRClientWithParams(viper.GetString("rmr.protPort"), viper.GetInt("rmr.maxSize"), viper.GetInt("rmr.numWorkers"), viper.GetInt("rmr.threadType"), "RMR") + p := GetPortData("rmrdata") + if p.Port == 0 || viper.IsSet("rmr.protPort") { + // Old xApp descriptor used, fallback to rmr section + fmt.Sscanf(viper.GetString("rmr.protPort"), "tcp:%d", &p.Port) + p.MaxSize = viper.GetInt("rmr.maxSize") + p.ThreadType = viper.GetInt("rmr.threadType") + p.LowLatency = viper.GetBool("rmr.lowLatency") + p.FastAck = viper.GetBool("rmr.fastAck") + p.MaxRetryOnFailure = viper.GetInt("rmr.maxRetryOnFailure") + } + + return NewRMRClientWithParams( + &RMRClientParams{ + RmrData: p, + StatDesc: "RMR", + }) } func (m *RMRClient) Start(c MessageConsumer) { @@ -120,7 +203,10 @@ func (m *RMRClient) Start(c MessageConsumer) { var counter int = 0 for { - if m.ready = int(C.rmr_ready(m.context)); m.ready == 1 { + 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 } @@ -130,33 +216,60 @@ func (m *RMRClient) Start(c MessageConsumer) { time.Sleep(1 * time.Second) counter++ } - m.wg.Add(m.numWorkers) if m.readyCb != nil { go m.readyCb(m.readyCbParams) } - for w := 0; w < m.numWorkers; w++ { - go m.Worker("worker-"+strconv.Itoa(w), 0) - } - m.Wait() -} + m.wg.Add(1) + 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) { - Logger.Info("rmrClient: '%s': receiving messages on [%s]", taskName, m.protPort) + defer m.wg.Done() + for { - defer m.wg.Done() - for { - rxBuffer := C.rmr_rcv_msg(m.context, nil) - if rxBuffer == nil { - m.LogMBufError("RecvMsg failed", rxBuffer) - m.UpdateStatCounter("ReceiveError") - continue + 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.wg.Add(1) + go func() { + defer m.wg.Done() + for { + m.UpdateRmrStats() + time.Sleep(1 * time.Second) } - m.UpdateStatCounter("Received") + }() - go m.parseMessage(rxBuffer) - } + m.wg.Wait() +} + +func (m *RMRClient) UpdateRmrStats() { + param := (*C.rmr_rx_debug_t)(C.malloc(C.size_t(unsafe.Sizeof(C.rmr_rx_debug_t{})))) + m.contextMux.Lock() + C.rmr_get_rx_debug_info(m.context, param) + m.contextMux.Unlock() + m.mux.Lock() + m.statg["Enqueued"].Set(float64(param.enqueue)) + m.statg["Dropped"].Set(float64(param.drop)) + m.mux.Unlock() + C.free(unsafe.Pointer(param)) } func (m *RMRClient) parseMessage(rxBuffer *C.rmr_mbuf_t) { @@ -197,33 +310,49 @@ func (m *RMRClient) parseMessage(rxBuffer *C.rmr_mbuf_t) { return } - // Special case for multiple consumers - for _, c := range m.consumers { - cptr := unsafe.Pointer(rxBuffer.payload) - 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) + /* + // Special case for multiple consumers + for _, c := range m.consumers { + cptr := unsafe.Pointer(rxBuffer.payload) + 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(params) - if err != nil { - Logger.Warn("rmrClient: Consumer returned error: %v", err) + err := c.Consume(params) + if err != nil { + Logger.Warn("rmrClient: Consumer returned error: %v", err) + } } - } + */ } func (m *RMRClient) Allocate(size int) *C.rmr_mbuf_t { - buf := C.rmr_alloc_msg(m.context, C.int(size)) - if buf == nil { + m.contextMux.Lock() + defer m.contextMux.Unlock() + outbuf := C.rmr_alloc_msg(m.context, C.int(size)) + if outbuf == nil { + Logger.Error("rmrClient: Allocating message buffer failed!") + } + return outbuf +} + +func (m *RMRClient) ReAllocate(inbuf *C.rmr_mbuf_t, size int) *C.rmr_mbuf_t { + m.contextMux.Lock() + defer m.contextMux.Unlock() + outbuf := C.rmr_realloc_msg(inbuf, C.int(size)) + if outbuf == nil { Logger.Error("rmrClient: Allocating message buffer failed!") } - return buf + return outbuf } 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) } @@ -235,10 +364,30 @@ func (m *RMRClient) SendRts(params *RMRParams) bool { return m.Send(params, true) } +func (m *RMRClient) SendWithRetry(params *RMRParams, isRts bool, to time.Duration) (err error) { + status := m.Send(params, isRts) + i := 0 + for ; i < int(to)*2 && status == false; i++ { + status = m.Send(params, isRts) + if status == false { + m.UpdateStatCounter("SendWithRetryRetry") + time.Sleep(500 * time.Millisecond) + } + } + if status == false { + err = fmt.Errorf("Failed with retries(%d) %s", i, params.String()) + if params.Mbuf != nil { + m.Free(params.Mbuf) + params.Mbuf = nil + } + } + return +} + func (m *RMRClient) CopyBuffer(params *RMRParams) *C.rmr_mbuf_t { - if params.Mbuf != nil { - m.Free(params.Mbuf) - params.Mbuf = nil + + if params == nil { + return nil } payLen := len(params.Payload) @@ -246,7 +395,15 @@ func (m *RMRClient) CopyBuffer(params *RMRParams) *C.rmr_mbuf_t { payLen = params.PayloadLen } - txBuffer := m.Allocate(payLen) + txBuffer := params.Mbuf + params.Mbuf = nil + + if txBuffer != nil { + txBuffer = m.ReAllocate(txBuffer, payLen) + } else { + txBuffer = m.Allocate(payLen) + } + if txBuffer == nil { return nil } @@ -257,20 +414,21 @@ func (m *RMRClient) CopyBuffer(params *RMRParams) *C.rmr_mbuf_t { datap := C.CBytes(params.Payload) defer C.free(datap) - 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))) - } + 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 } @@ -289,10 +447,10 @@ func (m *RMRClient) Send(params *RMRParams, isRts bool) bool { func (m *RMRClient) SendBuf(txBuffer *C.rmr_mbuf_t, isRts bool, whid int) int { var ( - currBuffer *C.rmr_mbuf_t - counterName string = "Transmitted" + currBuffer *C.rmr_mbuf_t ) + m.contextMux.Lock() txBuffer.state = 0 if whid != 0 { currBuffer = C.rmr_wh_send_msg(m.context, C.rmr_whid_t(whid), txBuffer) @@ -303,6 +461,7 @@ func (m *RMRClient) SendBuf(txBuffer *C.rmr_mbuf_t, isRts bool, whid int) int { currBuffer = C.rmr_send_msg(m.context, txBuffer) } } + m.contextMux.Unlock() if currBuffer == nil { m.UpdateStatCounter("TransmitError") @@ -310,12 +469,12 @@ func (m *RMRClient) SendBuf(txBuffer *C.rmr_mbuf_t, isRts bool, whid int) int { } // Just quick retry seems to help for K8s issue - maxRetryOnFailure := viper.GetInt("rmr.maxRetryOnFailure") - if maxRetryOnFailure == 0 { - maxRetryOnFailure = 5 + if m.maxRetryOnFailure == 0 { + m.maxRetryOnFailure = 5 } - for j := 0; j < maxRetryOnFailure && currBuffer != nil && currBuffer.state == C.RMR_ERR_RETRY; j++ { + for j := 0; j < m.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 { @@ -325,17 +484,25 @@ func (m *RMRClient) SendBuf(txBuffer *C.rmr_mbuf_t, isRts bool, whid int) int { currBuffer = C.rmr_send_msg(m.context, txBuffer) } } + m.contextMux.Unlock() + m.UpdateStatCounter("TransmitRetry") } - if currBuffer.state != C.RMR_OK { - counterName = "TransmitError" + if currBuffer == nil { + m.UpdateStatCounter("TransmitError") m.LogMBufError("SendBuf failed", currBuffer) + return int(C.RMR_ERR_INITFAILED) } - m.UpdateStatCounter(counterName) + if currBuffer.state != C.RMR_OK { + m.UpdateStatCounter("TransmitError") + m.LogMBufError("SendBuf failed", currBuffer) + } else { + m.UpdateStatCounter("Transmitted") + } defer m.Free(currBuffer) - return int(currBuffer.state) + } func (m *RMRClient) SendCallMsg(params *RMRParams) (int, string) { @@ -350,7 +517,9 @@ func (m *RMRClient) SendCallMsg(params *RMRParams) (int, string) { 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") @@ -376,6 +545,8 @@ func (m *RMRClient) Openwh(target string) C.rmr_whid_t { } 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) } @@ -385,6 +556,8 @@ func (m *RMRClient) Closewh(whid int) { } func (m *RMRClient) Wh_close(whid C.rmr_whid_t) { + m.contextMux.Lock() + defer m.contextMux.Unlock() C.rmr_wh_close(m.context, whid) } @@ -404,12 +577,13 @@ func (m *RMRClient) IsNoEndPointError(params *RMRParams) bool { func (m *RMRClient) UpdateStatCounter(name string) { m.mux.Lock() - m.stat[name].Inc() + m.statc[name].Inc() m.mux.Unlock() } func (m *RMRClient) RegisterMetrics() { - m.stat = Metric.RegisterCounterGroup(RMRCounterOpts, "RMR") + m.statc = Metric.RegisterCounterGroup(RMRCounterOpts, "RMR") + m.statg = Metric.RegisterGaugeGroup(RMRGaugeOpts, "RMR") } func (m *RMRClient) Wait() { @@ -440,11 +614,10 @@ func (m *RMRClient) GetRicMessageName(id int) (s string) { } 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 + if mbuf != nil { + Logger.Debug(fmt.Sprintf("rmrClient: %s -> [tp=%v] %v - %s", text, mbuf.tp_state, mbuf.state, RMRErrors[int(mbuf.state)])) + return int(mbuf.state) + } + Logger.Debug(fmt.Sprintf("rmrClient: %s -> mbuf nil", text)) + return 0 }