Gitiles
Code Review Sign In
gerrit.opencord.org / voltha-openonu-adapter / 5de33fe01609df99d86a97f47b4a9e36f801834b / . / python / adapters / brcm_openomci_onu / omci / brcm_mcast_task.py
blob: 55466b15fbad25bccf0c65b967fb1d90972158dc [file] [log] [blame]
ozgecanetsiab7f73092020-02-16 23:36:25 +0300[diff] [blame]1#
2# Copyright 2018 the original author or authors.
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15#
16
17from __future__ import absolute_import
18import structlog
19from pyvoltha.adapters.extensions.omci.omci_entities import OntG, PriorityQueueG, Tcont
20from pyvoltha.adapters.extensions.omci.omci_me import MacBridgePortConfigurationDataFrame, AccessControlRow0, \
21 MulticastOperationsProfileFrame, MulticastSubscriberConfigInfoFrame, VlanTaggingFilterDataFrame, \
22 ExtendedVlanTaggingOperationConfigurationDataFrame, VlanTaggingOperation
23from twisted.internet import reactor
24from pyvoltha.adapters.extensions.omci.omci_defs import ReasonCodes, EntityOperations
25from pyvoltha.adapters.extensions.omci.tasks.task import Task
26from twisted.internet.defer import inlineCallbacks, failure
27
28
29RC = ReasonCodes
30OP = EntityOperations
31IGMP_VERSION = 2
32RESERVED_VLAN = 4095
33
34
35class BrcmMcastTask(Task):
36 task_priority = 150
37 name = "Broadcom Multicast Task"
38
39 def __init__(self, omci_agent, handler, device_id, uni_id, tp_id, vlan_id, dynamic_acl, static_acl, multicast_gem_id ):
40
41 self.log = structlog.get_logger ( device_id=device_id )
42 super ( BrcmMcastTask , self ).__init__ ( BrcmMcastTask.name ,
43 omci_agent ,
44 device_id,
45 self.task_priority)
46 self._device = omci_agent.get_device ( device_id )
47 self._results = None
48 self._local_deferred = None
Mahir Gunyel5de33fe2020-03-03 22:38:44 -0800[diff] [blame^]49 self._uni_port = handler.uni_ports[uni_id]
50 assert self._uni_port.uni_id == uni_id
ozgecanetsiab7f73092020-02-16 23:36:25 +0300[diff] [blame]51 #mcast entities IDs
52 self._mcast_operation_profile_id = handler.pon_port.mac_bridge_port_ani_entity_id + tp_id
53 self._mcast_sub_conf_info_id = self._mcast_operation_profile_id + tp_id
54 self._mac_bridge_service_profile_entity_id = handler.mac_bridge_service_profile_entity_id
55 self._mac_bridge_configuration_data_id =self._mac_bridge_service_profile_entity_id +\
56 self._uni_port.mac_bridge_port_num
57 self._ieee_mapper_service_profile_entity_id = \
58 handler.pon_port.ieee_mapper_service_profile_entity_id
59 self._gal_enet_profile_entity_id = \
60 handler.gal_enet_profile_entity_id
61 # using for multicast operations profile and multicast subscriber config info
62 self._set_igmp_function = 0
63 self._set_immediate_leave = False
64 self._set_vlan_id = vlan_id
65 # dynamic_access_control_list , static_access_control_list and multicast_gem_id
66 self._dynamic_acl= dynamic_acl
67 self._static_acl= static_acl
68 self._mcast_gem_id= multicast_gem_id
69 self._uni_port = handler.uni_ports[uni_id]
70 assert self._uni_port.uni_id == uni_id
71 # gem_port list
72 self._gem_ports = []
73 for gem_port in list(handler.pon_port.gem_ports.values()):
74 if gem_port.uni_id is not None and gem_port.uni_id != self._uni_port.uni_id: continue
75 self._gem_ports.append(gem_port)
76
77 self._tcont_list=[]
78 for gem_port in self._gem_ports:
79 self._tcont_list.append(gem_port.tcont)
80 self.tcont_me_to_queue_map = dict()
81 self.uni_port_to_queue_map = dict()
82
83 self._set_me_type = 0
84 self._set_interworking_option = 0
85 self._store_tcont_list()
86
87 def cancel_deferred(self):
88 super ( BrcmMcastTask , self ).cancel_deferred ()
89 d , self._local_deferred = self._local_deferred , None
90 try:
91 if d is not None and not d.called:
92 d.cancel ()
93 except:
94 pass
95
96 def start(self):
97 super ( BrcmMcastTask , self ).start ()
98 self._local_deferred = reactor.callLater ( 0 , self.perform_multicast )
99
100 @inlineCallbacks
101 def perform_multicast(self):
102 self.log.debug('performing-multicast', device_id=self._device._device_id, uni_id=self._uni_port.uni_id,
103 vlan_id=self._set_vlan_id, multicast_gem_id=self._mcast_gem_id)
104
105 try:
106 # create gem port and use tcont
107 yield self._create_gem_ports()
108
109 # create multicast operation profile
110 yield self._create_mcast_operation_profile()
111
112 # set multicast operation profile
113 yield self._set_mcast_operation_profile()
114
115 # create multicast subscriber config data
116 yield self._create_mcast_subscriber_conf_info()
117
118 # create mac bridge port configuration data
119 yield self._create_mac_bridge_configuration_data()
120
121 # create vlan filtering entity
122 yield self._create_vlan_tagging_filter_data()
123
124 # set vlan filtering entity
125 yield self._set_vlan_tagging_filter_data()
126
127 self.deferred.callback ( self )
128 except Exception as e:
129 self.log.exception ( 'multicast exception' , e=e )
130 self.deferred.errback ( failure.Failure ( e ) )
131
132 @inlineCallbacks
133 def _create_gem_ports(self):
134 omci_cc = self._device.omci_cc
135 try:
136 tcont = None
137 self.log.debug("tcont-list", _tcont_list=self._tcont_list)
138 for gem_tcont in self._tcont_list:
139 if gem_tcont.entity_id is not None:
140 tcont=gem_tcont
141 for gem_port in self._gem_ports:
142 gem_port_tcont= gem_port.tcont
143 if gem_port_tcont is not None and gem_port_tcont.entity_id is not None:
144 tcont = gem_port.tcont
145 ul_prior_q_entity_id = \
146 self.tcont_me_to_queue_map[tcont.entity_id][gem_port.priority_q]
147 self.log.debug("ul_prior_q_entity_id", ul_priority=ul_prior_q_entity_id)
148
149 dl_prior_q_entity_id = \
150 self.uni_port_to_queue_map[self._uni_port.entity_id][gem_port.priority_q]
151 if ul_prior_q_entity_id is not None and dl_prior_q_entity_id is not None:
152 if gem_port.direction == 'downstream' and gem_port.gem_id == self._mcast_gem_id:
153 results = yield gem_port.add_to_hardware ( omci=omci_cc ,
154 tcont_entity_id=tcont.entity_id ,
155 ieee_mapper_service_profile_entity_id=self._ieee_mapper_service_profile_entity_id +
156 self._uni_port.mac_bridge_port_num ,
157 gal_enet_profile_entity_id=self._gal_enet_profile_entity_id ,
158 ul_prior_q_entity_id=ul_prior_q_entity_id ,
159 dl_prior_q_entity_id=dl_prior_q_entity_id)
160 self.check_status_and_state ( results , 'assign-gem-port' )
161 pass
162 except Exception as e:
163 self.log.debug("failed-create-gem-ports-method", e=e)
164
165 def _store_tcont_list(self):
166 pq_to_related_port = dict()
167 onu_g = self._device.query_mib ( OntG.class_id )
168 traffic_mgmt_opt = \
169 onu_g.get ( 'attributes' , {} ).get ( 'traffic_management_options' , 0 )
170 self.log.debug ( "traffic-mgmt-option" , traffic_mgmt_opt=traffic_mgmt_opt )
171
172 prior_q = self._device.query_mib ( PriorityQueueG.class_id )
173 for k , v in prior_q.items():
174 self.log.debug("prior-q", k=k, v=v)
175 try:
176 _ = iter(v)
177 except TypeError:
178 continue
179
180 if 'instance_id' in v:
181 related_port = v['attributes']['related_port']
182 pq_to_related_port[k] = related_port
183
184 if v['instance_id'] & 0b1000000000000000:
185 tcont_me = (related_port & 0xffff0000) >> 16
186 if tcont_me not in self.tcont_me_to_queue_map:
187 self.log.debug ( "prior-q-related-port-and-tcont-me" ,
188 related_port=related_port ,
189 tcont_me=tcont_me )
190 self.tcont_me_to_queue_map[tcont_me] = list ()
191
192 self.tcont_me_to_queue_map[tcont_me].append ( k )
193 else:
194 uni_port = (related_port & 0xffff0000) >> 16
195 if uni_port == self._uni_port.entity_id:
196 if uni_port not in self.uni_port_to_queue_map:
197 self.log.debug ( "prior-q-related-port-and-uni-port-me" ,
198 related_port=related_port ,
199 uni_port_me=uni_port )
200 self.uni_port_to_queue_map[uni_port] = list ()
201
202 self.uni_port_to_queue_map[uni_port].append ( k )
203
204
205 @inlineCallbacks
206 def _create_mac_bridge_configuration_data(self):
207 self.log.debug ( 'starting-create-mac-bridge-conf-data' )
208 attributes = dict ( bridge_id_pointer=self._mac_bridge_configuration_data_id,
209 port_num=0xf0,
210 tp_type=6,
211 tp_pointer=self._mcast_gem_id)
212 msg = MacBridgePortConfigurationDataFrame ( entity_id=self._mac_bridge_configuration_data_id, attributes=attributes )
213 yield self._send_msg ( msg , 'create' , 'create-mac-bridge-port-conf-data' )
214
215
216 @inlineCallbacks
217 def _create_mcast_operation_profile(self):
218 self.log.debug ( 'starting-create-mcast-operation-profile' )
219 attributes = dict ( igmp_version=IGMP_VERSION , igmp_function=self._set_igmp_function,
220 immediate_leave=self._set_immediate_leave, robustness=0)
221 msg = MulticastOperationsProfileFrame ( entity_id=self._mcast_operation_profile_id,
222 querier_ip_address=0,
223 query_interval=125,
224 query_max_response=100,
225 last_member_query_interval=10,
226 unauthorized_join_request_behavior=False,
227 attributes=attributes)
228 yield self._send_msg ( msg , 'create' , 'create-multicast-operation-profile')
229
230 @inlineCallbacks
231 def _set_mcast_operation_profile(self):
232 self.log.debug('starting-set-mcast-operation-profile')
233 dynamic_access_control_list_table= AccessControlRow0 (
234 set_ctrl=1 ,
235 row_part_id=0 ,
236 test=0 ,
237 row_key=0 ,
238 gem_port_id=self._mcast_gem_id,
239 vlan_id=self._set_vlan_id,
240 src_ip="0.0.0.0",
241 dst_ip_start=self._dynamic_acl[0],
242 dst_ip_end=self._dynamic_acl[1],
243 ipm_group_bw=0
244
245 )
246 msg = MulticastOperationsProfileFrame(entity_id=self._mcast_operation_profile_id,
247 dynamic_access_control_list_table=dynamic_access_control_list_table
248 )
249 yield self._send_msg(msg, 'set', 'set-multicast-operations-profile')
250
251 @inlineCallbacks
252 def _create_mcast_subscriber_conf_info(self):
253 self.log.debug ( 'creating-multicast-subscriber-config-info' )
254 attributes = dict ( me_type=self._set_me_type,
255 multicast_operations_profile_pointer=self._mcast_operation_profile_id )
256 msg = MulticastSubscriberConfigInfoFrame ( entity_id=self._uni_port.entity_id, attributes=attributes )
257 yield self._send_msg ( msg , 'create' , 'create-multicast-subscriber-config-info' )
258
259
260 @inlineCallbacks
261 def _create_vlan_tagging_filter_data(self):
262 self.log.debug('creating-vlan-tagging-filter-data')
263 forward_operation = 0x10 # VID investigation
264 # When the PUSH VLAN is RESERVED_VLAN (4095), let ONU be transparent
265 if self._set_vlan_id == RESERVED_VLAN:
266 forward_operation = 0x00 # no investigation, ONU transparent
267
268 # Create bridge ani side vlan filter
269 msg = VlanTaggingFilterDataFrame(
270 self._mac_bridge_configuration_data_id, # Entity ID
271 vlan_tcis=[self._set_vlan_id], # VLAN IDs
272 forward_operation=forward_operation
273 )
274
275 yield self._send_msg(msg, 'create', 'flow-create-vlan-tagging-filter-data')
276
277 @inlineCallbacks
278 def _set_vlan_tagging_filter_data(self):
279 self.log.debug('setting-vlan-tagging-filter-data')
280 forward_operation = 0x10 # VID investigation
281 # When the PUSH VLAN is RESERVED_VLAN (4095), let ONU be transparent
282 if self._set_vlan_id == RESERVED_VLAN:
283 forward_operation = 0x00 # no investigation, ONU transparent
284
285 # Create bridge ani side vlan filter
286 msg = VlanTaggingFilterDataFrame(
287 self._mac_bridge_configuration_data_id, # Entity ID
288 vlan_tcis=[self._set_vlan_id], # VLAN IDs
289 forward_operation=forward_operation
290 )
291
292 yield self._send_msg(msg, 'set', 'flow-set-vlan-tagging-filter-data')
293
294
295 @inlineCallbacks
296 def _send_msg(self , msg , operation , log_comment):
297 if operation == 'create':
298 frame = msg.create ()
299 elif operation == 'set':
300 frame = msg.set ()
301 else:
302 frame = msg.delete ()
303 self.log.debug ( 'openomci-msg' , omci_msg=msg )
304 self.strobe_watchdog ()
305 results = yield self._device.omci_cc.send ( frame )
306 self.check_status_and_state ( results , log_comment )
307
308 def check_status_and_state(self , results , operation=''):
309 """
310 Check the results of an OMCI response. An exception is thrown
311 if the task was cancelled or an error was detected.
312
313 :param results: (OmciFrame) OMCI Response frame
314 :param operation: (str) what operation was being performed
315 :return: True if successful, False if the entity existed (already created)
316 """
317
318 omci_msg = results.fields['omci_message'].fields
319 status = omci_msg['success_code']
320 error_mask = omci_msg.get ( 'parameter_error_attributes_mask' , 'n/a' )
321 failed_mask = omci_msg.get ( 'failed_attributes_mask' , 'n/a' )
322 unsupported_mask = omci_msg.get ( 'unsupported_attributes_mask' , 'n/a' )
323
324 self.log.debug ( "OMCI Result: %s" , operation , omci_msg=omci_msg ,
325 status=status , error_mask=error_mask ,
326 failed_mask=failed_mask , unsupported_mask=unsupported_mask )
327
328 if status == RC.Success:
329 self.strobe_watchdog ()
330 return True
331
332 elif status == RC.InstanceExists:
333 return False