blob: 93eaeeb7523628aaa0caf69e6e8ef4d4dea9f189 [file] [log] [blame]
# ==================================================================================
# Copyright (c) 2020 Nokia
# Copyright (c) 2020 AT&T Intellectual Property.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==================================================================================
"""
Contains RMR functionality specific to the xapp.
The general rmr API is via "rmr"
"""
import time
import queue
from threading import Thread
from mdclogpy import Logger
from ricxappframe.rmr import rmr, helpers
mdc_logger = Logger(name=__name__)
class RmrLoop:
"""
Class represents an RMR loop that constantly reads from RMR.
Note, we use a queue here, and a thread, rather than the xapp frame just looping
and calling consume, so that a possibly slow running consume function does not
block the reading of new messages
"""
def __init__(self, port, wait_for_ready=True):
"""
sets up RMR, then launches a thread that reads and injects messages into a queue.
Parameters
----------
port: int
port to listen on
wait_for_ready: bool (optional)
If True, then this function hangs until RMR is ready to send, which includes
having a valid routing file. This can be set to False if the client only wants
to *receive only*.
"""
# Public
# thread safe queue https://docs.python.org/3/library/queue.html
# We use a thread and a queue so that a long running consume callback function can
# never block reads. IE a consume implementation could take a long time and the ring
# size for rmr blows up here and messages are lost.
self.rcv_queue = queue.Queue()
# RMR context; RMRFL_MTCALL puts RMR into a multithreaded mode, where a thread
# populates a ring of messages that receive calls read from
self.mrc = rmr.rmr_init(str(port).encode(), rmr.RMR_MAX_RCV_BYTES, rmr.RMRFL_MTCALL)
if wait_for_ready:
mdc_logger.debug("Waiting for rmr to init on port {}..".format(port))
while rmr.rmr_ready(self.mrc) == 0:
time.sleep(0.1)
# Private
self._keep_going = True # used to tell this thread to stop
self._last_ran = time.time() # used for healthcheck
self._loop_is_running = False # used in stop to know when it's safe to kill the mrc
def loop():
mdc_logger.debug("Work loop starts")
self._loop_is_running = True
while self._keep_going:
# read our mailbox
# TODO: take a flag as to whether RAW is needed or not
# RAW allows for RTS however the caller must free, and the caller may not need RTS.
# Currently after consuming, callers should do rmr.rmr_free_msg(sbuf)
for (msg, sbuf) in helpers.rmr_rcvall_msgs_raw(self.mrc, timeout=1000):
self.rcv_queue.put((msg, sbuf))
self._last_ran = time.time()
self._loop_is_running = False
mdc_logger.debug("Work loop ends")
# start the work loop
mdc_logger.debug("Starting loop thread")
self._thread = Thread(target=loop)
self._thread.start()
def stop(self):
"""
sets a flag that will cleanly stop the thread
"""
mdc_logger.debug("Stopping RMR thread. Waiting for last iteration to finish..")
self._keep_going = False
# wait until the current batch of messages is done, then kill the rmr connection
# note; I debated putting this in "loop" however if the while loop was still going
# setting mrc to close here would blow up any processing still currently happening
# probably more polite to at least finish the current batch and then close. So here
# we wait until the current batch is done, then we kill the mrc
while self._loop_is_running:
time.sleep(0.5)
mdc_logger.debug("Closing rmr connection")
rmr.rmr_close(self.mrc)
def healthcheck(self, seconds=30):
"""
returns a boolean representing whether the rmr loop is healthy, by checking two attributes:
1. is it running?,
2. is it stuck in a long (> seconds) loop?
Parameters
----------
seconds: int (optional)
the rmr loop is determined healthy if it has completed in the last (seconds)
"""
return self._thread.is_alive() and ((time.time() - self._last_ran) < seconds)