X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fxapp%2Frmr.go;h=94070242c02ab2ab52604035f630ed1cf70e1a68;hb=refs%2Fchanges%2F16%2F5016%2F1;hp=9f988dcf1d396acc714a7d13890dd2c6729f0aa7;hpb=f22b458846a20a4a9fcafb49e3195ab44a16840e;p=ric-plt%2Fxapp-frame.git diff --git a/pkg/xapp/rmr.go b/pkg/xapp/rmr.go index 9f988dc..9407024 100755 --- a/pkg/xapp/rmr.go +++ b/pkg/xapp/rmr.go @@ -67,7 +67,6 @@ import ( "bytes" "crypto/md5" "fmt" - "github.com/spf13/viper" "strings" "time" "unsafe" @@ -127,25 +126,54 @@ 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), + stat: Metric.RegisterCounterGroup(RMRCounterOpts, 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("rmr-data") + return NewRMRClientWithParams( + &RMRClientParams{ + RmrData: p, + StatDesc: "RMR", + }) } func (m *RMRClient) Start(c MessageConsumer) { @@ -313,6 +341,10 @@ func (m *RMRClient) SendWithRetry(params *RMRParams, isRts bool, to time.Duratio func (m *RMRClient) CopyBuffer(params *RMRParams) *C.rmr_mbuf_t { + if params == nil { + return nil + } + payLen := len(params.Payload) if params.PayloadLen != 0 { payLen = params.PayloadLen @@ -337,20 +369,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 } @@ -369,8 +402,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() @@ -392,12 +424,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) @@ -411,15 +442,21 @@ func (m *RMRClient) SendBuf(txBuffer *C.rmr_mbuf_t, isRts bool, whid int) int { m.contextMux.Unlock() } - 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) { @@ -530,11 +567,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 }