X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fxapp%2Frmr.go;h=9f988dcf1d396acc714a7d13890dd2c6729f0aa7;hb=f22b458846a20a4a9fcafb49e3195ab44a16840e;hp=0aeced219009dd3d54e3aa8d2477bd94c0ef6f12;hpb=e0948fe8b755b4d752ef9da4a46246cf3951c9dc;p=ric-plt%2Fxapp-frame.git diff --git a/pkg/xapp/rmr.go b/pkg/xapp/rmr.go index 0aeced2..9f988dc 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,41 @@ 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" @@ -72,6 +100,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 +118,34 @@ type RMRParams struct { status int } -func NewRMRClientWithParams(protPort string, maxSize int, numWorkers int, threadType int, statDesc string) *RMRClient { +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() +} + +//----------------------------------------------------------------------------- +// +//----------------------------------------------------------------------------- +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.RMRFL_NOTHREAD) ctx := C.rmr_init(p, m, c) if ctx == nil { Logger.Error("rmrClient: Initializing RMR context failed, bailing out!") } - return &RMRClient{ - protPort: protPort, - numWorkers: numWorkers, - context: ctx, - consumers: make([]MessageConsumer, 0), - stat: Metric.RegisterCounterGroup(RMRCounterOpts, statDesc), + protPort: protPort, + context: ctx, + consumers: make([]MessageConsumer, 0), + stat: Metric.RegisterCounterGroup(RMRCounterOpts, statDesc), } } func NewRMRClient() *RMRClient { - return NewRMRClientWithParams(viper.GetString("rmr.protPort"), viper.GetInt("rmr.maxSize"), viper.GetInt("rmr.numWorkers"), viper.GetInt("rmr.threadType"), "RMR") + return NewRMRClientWithParams(viper.GetString("rmr.protPort"), viper.GetInt("rmr.maxSize"), viper.GetInt("rmr.threadType"), "RMR") } func (m *RMRClient) Start(c MessageConsumer) { @@ -120,7 +155,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 +168,38 @@ func (m *RMRClient) Start(c MessageConsumer) { time.Sleep(1 * time.Second) counter++ } - m.wg.Add(m.numWorkers) + m.wg.Add(1) 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() -} - -func (m *RMRClient) Worker(taskName string, msgSize int) { - Logger.Info("rmrClient: '%s': receiving messages on [%s]", taskName, m.protPort) + go func() { + m.contextMux.Lock() + rfd := C.rmr_get_rcvfd(m.context) + m.contextMux.Unlock() + efd := C.init_epoll(rfd) - 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 + 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) { @@ -212,18 +255,32 @@ func (m *RMRClient) parseMessage(rxBuffer *C.rmr_mbuf_t) { } } -func (m *RMRClient) Allocate() *C.rmr_mbuf_t { - buf := C.rmr_alloc_msg(m.context, 0) - if buf == nil { +func (m *RMRClient) Allocate(size int) *C.rmr_mbuf_t { + 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 buf + 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 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,21 +292,48 @@ 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 { + 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 { + + payLen := len(params.Payload) + if params.PayloadLen != 0 { + payLen = params.PayloadLen + } + txBuffer := params.Mbuf - if txBuffer == nil { - txBuffer = m.Allocate() - if txBuffer == nil { - return nil - } + params.Mbuf = nil + + if txBuffer != nil { + txBuffer = m.ReAllocate(txBuffer, payLen) + } else { + 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(len(params.Payload)) - if params.PayloadLen != 0 { - txBuffer.len = C.int(params.PayloadLen) - } + txBuffer.len = C.int(payLen) + datap := C.CBytes(params.Payload) defer C.free(datap) @@ -289,6 +373,7 @@ func (m *RMRClient) SendBuf(txBuffer *C.rmr_mbuf_t, isRts bool, whid int) int { 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) @@ -299,6 +384,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") @@ -312,6 +398,7 @@ func (m *RMRClient) SendBuf(txBuffer *C.rmr_mbuf_t, isRts bool, whid int) int { } 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 { @@ -321,6 +408,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.state != C.RMR_OK { @@ -346,7 +434,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") @@ -372,6 +462,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) } @@ -381,6 +473,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) }