X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fxapp%2Frmr.go;h=028cc3c9473aa18af2948b27409775063c729ddd;hb=79f0680fd7bbf1c8a8c6e2a842cb18020e387a47;hp=e27d46e9bd4433cfe65278845ef7f59f7efbf49a;hpb=7032754482cc88c4a9cce424a4717de7165fee17;p=ric-plt%2Fxapp-frame.git

diff --git a/pkg/xapp/rmr.go b/pkg/xapp/rmr.go
index e27d46e..028cc3c 100755
--- a/pkg/xapp/rmr.go
+++ b/pkg/xapp/rmr.go
@@ -67,17 +67,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{
@@ -127,25 +135,65 @@ func (params *RMRParams) String() string {
 //-----------------------------------------------------------------------------
 //
 //-----------------------------------------------------------------------------
-func NewRMRClientWithParams(protPort string, maxSize int, threadType int, statDesc string) *RMRClient {
-	p := C.CString(protPort)
-	m := C.int(maxSize)
-	c := C.int(threadType)
+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)
+}
+
+//-----------------------------------------------------------------------------
+//
+//-----------------------------------------------------------------------------
+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,
-		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.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) {
@@ -168,12 +216,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)
@@ -182,6 +230,7 @@ func (m *RMRClient) Start(c MessageConsumer) {
 
 		defer m.wg.Done()
 		for {
+
 			if int(C.wait_epoll(efd, rfd)) == 0 {
 				continue
 			}
@@ -199,9 +248,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!")
@@ -240,19 +310,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 {
@@ -298,6 +370,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)
 		}
 	}
@@ -374,8 +447,7 @@ 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()
@@ -397,12 +469,11 @@ 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)
@@ -414,17 +485,24 @@ func (m *RMRClient) SendBuf(txBuffer *C.rmr_mbuf_t, isRts bool, whid int) int {
 			}
 		}
 		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) {
@@ -499,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() {
@@ -535,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
 }