X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fxapp%2Frmr.go;h=59df7436cf6c16623a3de18cbfbb56fc6695d1d5;hb=9c523a68520e552d051ca922815aa936c8611894;hp=530f2230ee756e9e0613fd8129de52affea9ef3f;hpb=20a36385164677ba84d7f4159eb65b3e9c42d952;p=ric-plt%2Fxapp-frame.git diff --git a/pkg/xapp/rmr.go b/pkg/xapp/rmr.go index 530f223..59df743 100755 --- a/pkg/xapp/rmr.go +++ b/pkg/xapp/rmr.go @@ -29,6 +29,7 @@ package xapp #include #include + void write_bytes_array(unsigned char *dst, void *data, int len) { memcpy((void *)dst, (void *)data, len); } @@ -67,17 +68,25 @@ import ( "bytes" "crypto/md5" "fmt" - "github.com/spf13/viper" "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{ @@ -100,9 +109,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 @@ -124,30 +133,27 @@ func (params *RMRParams) String() string { return b.String() } -//----------------------------------------------------------------------------- +// ----------------------------------------------------------------------------- // -//----------------------------------------------------------------------------- +// ----------------------------------------------------------------------------- type RMRClientParams struct { - protPort string - maxSize int - threadType int - statDesc string - lowLatency bool - fastAck bool + StatDesc string + RmrData PortData } func (params *RMRClientParams) String() string { - return fmt.Sprintf("protPort=%s maxSize=%d threadType=%d statDesc=%s lowLatency=%t fastAck=%t", - params.protPort, params.maxSize, params.threadType, params.statDesc, params.lowLatency, params.fastAck) + 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) } -//----------------------------------------------------------------------------- +// ----------------------------------------------------------------------------- // -//----------------------------------------------------------------------------- +// ----------------------------------------------------------------------------- func NewRMRClientWithParams(params *RMRClientParams) *RMRClient { - p := C.CString(params.protPort) - m := C.int(params.maxSize) - c := C.int(params.threadType) + 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 { @@ -156,30 +162,38 @@ func NewRMRClientWithParams(params *RMRClientParams) *RMRClient { Logger.Info("new rmrClient with parameters: %s", params.String()) - if params.lowLatency { + if params.RmrData.LowLatency { C.rmr_set_low_latency(ctx) } - if params.fastAck { + if params.RmrData.FastAck { C.rmr_set_fack(ctx) } return &RMRClient{ - protPort: params.protPort, - context: ctx, - consumers: make([]MessageConsumer, 0), - stat: Metric.RegisterCounterGroup(RMRCounterOpts, params.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 { + 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{ - protPort: viper.GetString("rmr.protPort"), - maxSize: viper.GetInt("rmr.maxSize"), - threadType: viper.GetInt("rmr.threadType"), - statDesc: "RMR", - lowLatency: viper.GetBool("rmr.lowLatency"), - fastAck: viper.GetBool("rmr.fastAck"), + RmrData: p, + StatDesc: "RMR", }) } @@ -203,12 +217,12 @@ func (m *RMRClient) Start(c MessageConsumer) { time.Sleep(1 * time.Second) counter++ } - m.wg.Add(1) if m.readyCb != nil { go m.readyCb(m.readyCbParams) } + m.wg.Add(1) go func() { m.contextMux.Lock() rfd := C.rmr_get_rcvfd(m.context) @@ -217,6 +231,7 @@ func (m *RMRClient) Start(c MessageConsumer) { defer m.wg.Done() for { + if int(C.wait_epoll(efd, rfd)) == 0 { continue } @@ -234,9 +249,30 @@ func (m *RMRClient) Start(c MessageConsumer) { } }() + m.wg.Add(1) + go func() { + defer m.wg.Done() + for { + m.UpdateRmrStats() + time.Sleep(1 * time.Second) + } + }() + 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) { if len(m.consumers) == 0 { Logger.Info("rmrClient: No message handlers defined, message discarded!") @@ -275,19 +311,21 @@ 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 { @@ -333,6 +371,7 @@ func (m *RMRClient) SendWithRetry(params *RMRParams, isRts bool, to time.Duratio for ; i < int(to)*2 && status == false; i++ { status = m.Send(params, isRts) if status == false { + m.UpdateStatCounter("SendWithRetryRetry") time.Sleep(500 * time.Millisecond) } } @@ -408,58 +447,47 @@ 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" - ) - - 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 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; j++ { m.contextMux.Lock() if whid != 0 { - currBuffer = C.rmr_wh_send_msg(m.context, C.rmr_whid_t(whid), txBuffer) + txBuffer = C.rmr_wh_send_msg(m.context, C.rmr_whid_t(whid), txBuffer) } else { if isRts { - currBuffer = C.rmr_rts_msg(m.context, txBuffer) + txBuffer = C.rmr_rts_msg(m.context, txBuffer) } else { - currBuffer = C.rmr_send_msg(m.context, txBuffer) + txBuffer = C.rmr_send_msg(m.context, txBuffer) } } m.contextMux.Unlock() + if j+1 <= m.maxRetryOnFailure && txBuffer != nil && txBuffer.state == C.RMR_ERR_RETRY { + m.UpdateStatCounter("TransmitRetry") + continue + } + break } - if currBuffer.state != C.RMR_OK { - counterName = "TransmitError" - m.LogMBufError("SendBuf failed", currBuffer) + if txBuffer == nil { + m.UpdateStatCounter("TransmitError") + m.LogMBufError("SendBuf failed", txBuffer) + return int(C.RMR_ERR_INITFAILED) } - m.UpdateStatCounter(counterName) - defer m.Free(currBuffer) + if txBuffer.state != C.RMR_OK { + m.UpdateStatCounter("TransmitError") + m.LogMBufError("SendBuf failed", txBuffer) + } else { + m.UpdateStatCounter("Transmitted") + } + defer m.Free(txBuffer) + return int(txBuffer.state) - return int(currBuffer.state) } func (m *RMRClient) SendCallMsg(params *RMRParams) (int, string) { @@ -534,12 +562,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() { @@ -570,11 +599,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 }