VOL-184 ONU software upgrade based on OpenOMCI framework
** Implement OMCI download software procedure with two state machine:
1. Software OMCI downloading state machine
2. Activate software image state machine, which includes OMCI software download and OMCI software activation
3. Add OMCI message/frames for Start/End/DownloadSection/ActiveImage/CommitImage
** Change OMCI_CC.send to support:
1. Send OMCI messages without waiting for response (Download Section)
2. Add timeout handler to support retry max 3 times on timeout
3. Add a clock(reactor) in contructor to support automation test with robotframwork
RobotFramework use global reactor to loop run all test cases.
It will conflict with the reactor in VOLTHA classes.
The solution is to assign a reactor when create VOLTHA classes so that they will share
the same reactor and make RobotFramework works.
If reactor is not assigned VOLTHA classes will use default global reactor.
So it is back-compitable and won't have impact on other code
** OMCI download/activate task are exclusive True to avoid blocking by OMCI PM/Alarm STM/tasks
** Log is detailed for debug at current stage; and will improve after functions are fully verified
** Add test cases based on RobotFramework
1. Test cases for Software download
2. Mock Adapter/Mock ONU are used in the test cases
** Implement relative commands:
img_dnld_request
img_dnld_status
img_dnld_cancel
img_activate
Change-Id: I1796fd694f312378facbcd441bfc15ee7c093d91
diff --git a/voltha/extensions/omci/omci_cc.py b/voltha/extensions/omci/omci_cc.py
index baacd41..6632d28 100644
--- a/voltha/extensions/omci/omci_cc.py
+++ b/voltha/extensions/omci/omci_cc.py
@@ -23,24 +23,28 @@
from twisted.internet.defer import DeferredQueue, TimeoutError, CancelledError, failure, fail, inlineCallbacks
from common.frameio.frameio import hexify
from voltha.extensions.omci.omci import *
-from voltha.extensions.omci.omci_me import OntGFrame, OntDataFrame
+from voltha.extensions.omci.omci_me import OntGFrame, OntDataFrame, SoftwareImageFrame
from voltha.extensions.omci.me_frame import MEFrame
from voltha.extensions.omci.omci_defs import ReasonCodes
from common.event_bus import EventBusClient
from enum import IntEnum
from binascii import hexlify
+def hexify(buffer):
+ """Return a hexadecimal string encoding of input buffer"""
+ return ''.join('%02x' % ord(c) for c in buffer)
DEFAULT_OMCI_TIMEOUT = 3 # Seconds
MAX_OMCI_REQUEST_AGE = 60 # Seconds
MAX_OMCI_TX_ID = 0xFFFF # 2 Octets max
+DEFAULT_OMCI_DOWNLOAD_SECTION_SIZE = 31 # Bytes
+#DEFAULT_OMCI_DOWNLOAD_WINDOW_SIZE = 32 # sections
CONNECTED_KEY = 'connected'
TX_REQUEST_KEY = 'tx-request'
RX_RESPONSE_KEY = 'rx-response'
UNKNOWN_CLASS_ATTRIBUTE_KEY = 'voltha-unknown-blob'
-
class OmciCCRxEvents(IntEnum):
AVC_Notification = 0,
MIB_Upload = 1,
@@ -60,7 +64,6 @@
OP = EntityOperations
RxEvent = OmciCCRxEvents
-
class OMCI_CC(object):
""" Handle OMCI Communication Channel specifics for Adtran ONUs"""
@@ -75,16 +78,21 @@
OmciGetAllAlarmsNextResponse.message_id: RxEvent.Get_ALARM_Get_Next
}
- def __init__(self, adapter_agent, device_id, me_map=None):
+ def __init__(self, adapter_agent, device_id, me_map=None,
+ clock=None):
self.log = structlog.get_logger(device_id=device_id)
self._adapter_agent = adapter_agent
self._device_id = device_id
self._proxy_address = None
self._tx_tid = 1
self._enabled = False
- self._requests = dict() # Tx ID -> (timestamp, deferred, tx_frame, timeout)
+ self._requests = dict() # Tx ID -> (timestamp, deferred, tx_frame, timeout, retry, delayedCall)
self._me_map = me_map
-
+ if clock is None:
+ self.reactor = reactor
+ else:
+ self.reactor = clock
+
# Statistics
self._tx_frames = 0
self._rx_frames = 0
@@ -221,15 +229,15 @@
if msg_type == EntityOperations.AlarmNotification.value:
topic = OMCI_CC.event_bus_topic(self._device_id, RxEvent.Alarm_Notification)
- reactor.callLater(0, self.event_bus.publish, topic, msg)
+ self.reactor.callLater(0, self.event_bus.publish, topic, msg)
elif msg_type == EntityOperations.AttributeValueChange.value:
topic = OMCI_CC.event_bus_topic(self._device_id, RxEvent.AVC_Notification)
- reactor.callLater(0, self.event_bus.publish, topic, msg)
+ self.reactor.callLater(0, self.event_bus.publish, topic, msg)
elif msg_type == EntityOperations.TestResult.value:
topic = OMCI_CC.event_bus_topic(self._device_id, RxEvent.Test_Result)
- reactor.callLater(0, self.event_bus.publish, topic, msg)
+ self.reactor.callLater(0, self.event_bus.publish, topic, msg)
else:
self.log.warn('onu-unsupported-autonomous-message', type=msg_type)
@@ -256,7 +264,7 @@
omci_entities.entity_id_to_class_map = self._me_map
try:
- rx_frame = OmciFrame(msg)
+ rx_frame = msg if isinstance(msg, OmciFrame) else OmciFrame(msg)
rx_tid = rx_frame.fields['transaction_id']
if rx_tid == 0:
@@ -265,7 +273,7 @@
# Previously unreachable if this is the very first Rx or we
# have been running consecutive errors
if self._rx_frames == 0 or self._consecutive_errors != 0:
- reactor.callLater(0, self._publish_connectivity_event, True)
+ self.reactor.callLater(0, self._publish_connectivity_event, True)
self._rx_frames += 1
self._consecutive_errors = 0
@@ -285,7 +293,11 @@
omci_entities.entity_id_to_class_map = saved_me_map # Always restore it.
try:
- (ts, d, tx_frame, timeout) = self._requests.pop(rx_tid)
+ # (0: timestamp, 1: defer, 2: frame, 3: timeout, 4: retry, 5: delayedCall)
+ (ts, d, tx_frame, timeout, retry, dc) = self._requests.pop(rx_tid)
+ if dc is not None and not dc.cancelled and not dc.called:
+ self.log.debug("cancel timeout call")
+ dc.cancel()
ts_diff = now - arrow.Arrow.utcfromtimestamp(ts)
secs = ts_diff.total_seconds()
@@ -489,14 +501,17 @@
def flush(self, max_age=0):
limit = arrow.utcnow().float_timestamp - max_age
- old = [tid for tid, (ts, _, _, _) in self._requests.iteritems()
+ old = [tid for tid, (ts, _, _, _, _, _) in self._requests.iteritems()
if ts <= limit]
for tid in old:
- (_, d, _, _) = self._requests.pop(tid)
+ (_, d, _, _, _, dc) = self._requests.pop(tid)
if d is not None and not d.called:
d.cancel()
+ if dc is not None and not dc.called and not dc.cancelled:
+ dc.cancel()
+
def _get_tx_tid(self):
"""
Get the next Transaction ID for a tx. Note TID=0 is reserved
@@ -518,9 +533,9 @@
:param tx_tid: (int) Associated Tx TID
"""
if tx_tid in self._requests:
- (_, _, _, timeout) = self._requests.pop(tx_tid)
- else:
- timeout = 0
+ (_, _, _, timeout, retry, dc) = self._requests.pop(tx_tid)
+ if dc is not None and not dc.called and not dc.cancelled:
+ dc.cancel()
if isinstance(value, failure.Failure):
value.trap(CancelledError)
@@ -546,12 +561,26 @@
# Continue with Rx Success callbacks.
return rx_frame
- def send(self, frame, timeout=DEFAULT_OMCI_TIMEOUT):
+ def _request_timeout(self, tx_tid):
+ self.log.debug("_request_timeout", tx_tid=tx_tid)
+ if tx_tid in self._requests:
+ req = self._requests[tx_tid] # (0: timestamp, 1: defer, 2: frame, 3: timeout, 4: retry, 5: delayedCall)
+ frame = req[2]
+ timeout = req[3]
+ retry = req[4]
+ if retry > 0:
+ retry -= 1
+ self.send(frame, timeout, retry)
+ else:
+ d = req[1]
+ d.errback(failure.Failure(TimeoutError(timeout, "Send OMCI TID -{}".format(tx_tid))))
+
+ def send(self, frame, timeout=DEFAULT_OMCI_TIMEOUT, retry=0):
"""
Send the OMCI Frame to the ONU via the proxy_channel
:param frame: (OMCIFrame) Message to send
- :param timeout: (int) Rx Timeout. 0=Forever
+ :param timeout: (int) Rx Timeout. 0=No response needed
:return: (deferred) A deferred that fires when the response frame is received
or if an error/timeout occurs
"""
@@ -562,7 +591,7 @@
assert isinstance(frame, OmciFrame), \
"Invalid frame class '{}'".format(type(frame))
- if not self.enabled or self._proxy_address is None:
+ if not self.enabled:
# TODO custom exceptions throughout this code would be helpful
return fail(result=failure.Failure(Exception('OMCI is not enabled')))
@@ -576,7 +605,17 @@
assert tx_tid >= 0, 'Invalid Tx TID: {}'.format(tx_tid)
ts = arrow.utcnow().float_timestamp
- d = defer.Deferred()
+
+ if tx_tid in self._requests:
+ req = self._requests[tx_tid] # (0: timestamp, 1: defer, 2: frame, 3: timeout, 4: retry, 5: delayedCall)
+ d = req[1]
+ timeout = req[3]
+ dc = req[5]
+ if dc is not None and not dc.cancelled: # delayedCall returned from last send
+ dc.cancel()
+ else:
+ req = None
+ d = defer.Deferred()
# NOTE: Since we may need to do an independent ME map on a per-ONU basis
# save the current value of the entity_id_to_class_map, then
@@ -592,20 +631,21 @@
omci_entities.entity_id_to_class_map = saved_me_map
self._tx_frames += 1
- self._requests[tx_tid] = (ts, d, frame, timeout)
-
- d.addCallbacks(self._request_success, self._request_failure,
- errbackArgs=(tx_tid,))
-
+
if timeout > 0:
- d.addTimeout(timeout, reactor)
+ dc = self.reactor.callLater(timeout, self._request_timeout, tx_tid)
+ req = self._requests[tx_tid] = (ts, d, frame, timeout, retry, dc)
+ d.addCallbacks(self._request_success, self._request_failure, errbackArgs=(tx_tid,))
+ # d.addTimeout(timeout, reactor)
+ else:
+ self.reactor.callLater(0, d.callback, tx_tid) # no response needed to trigger the defer; just fire it.
except Exception as e:
self._tx_errors += 1
self._consecutive_errors += 1
if self._consecutive_errors == 1:
- reactor.callLater(0, self._publish_connectivity_event, False)
+ self.reactor.callLater(0, self._publish_connectivity_event, False)
self.log.exception('send-omci', e=e)
return fail(result=failure.Failure(e))
@@ -658,3 +698,39 @@
frame = OntDataFrame().get_all_alarm_next(seq_no)
return self.send(frame, timeout)
+
+ def send_start_software_download(self, image_inst_id, image_size, window_size, timeout=DEFAULT_OMCI_TIMEOUT):
+ frame = SoftwareImageFrame(image_inst_id).start_software_download(image_size, window_size-1)
+ return self.send(frame, timeout, 3)
+
+ def send_download_section(self, image_inst_id, section_num, data, size=DEFAULT_OMCI_DOWNLOAD_SECTION_SIZE, timeout=0):
+ """
+ # timeout=0 indicates no repons needed
+ """
+ # self.log.debug("send_download_section", instance_id=image_inst_id, section=section_num, timeout=timeout)
+ if timeout > 0:
+ frame = SoftwareImageFrame(image_inst_id).download_section(True, section_num, data)
+ else:
+ frame = SoftwareImageFrame(image_inst_id).download_section(False, section_num, data)
+ return self.send(frame, timeout)
+
+ # if timeout > 0:
+ # self.reactor.callLater(0, self.sim_receive_download_section_resp,
+ # frame.fields["transaction_id"],
+ # frame.fields["omci_message"].fields["section_number"])
+ # return d
+
+ def send_end_software_download(self, image_inst_id, crc32, image_size, timeout=DEFAULT_OMCI_TIMEOUT):
+ frame = SoftwareImageFrame(image_inst_id).end_software_download(crc32, image_size)
+ return self.send(frame, timeout)
+ # self.reactor.callLater(0, self.sim_receive_end_software_download_resp, frame.fields["transaction_id"])
+ # return d
+
+ def send_active_image(self, image_inst_id, flag=0, timeout=DEFAULT_OMCI_TIMEOUT):
+ frame = SoftwareImageFrame(image_inst_id).activate_image(flag)
+ return self.send(frame, timeout)
+
+ def send_commit_image(self, image_inst_id, timeout=DEFAULT_OMCI_TIMEOUT):
+ frame = SoftwareImageFrame(image_inst_id).commit_image()
+ return self.send(frame, timeout)
+