testsuites/vstf/vstf_scripts/vstf/controller/spirent/common/model.py

   1 ##############################################################################
   2 # Copyright (c) 2015 Huawei Technologies Co.,Ltd and others.
   3 #
   4 # All rights reserved. This program and the accompanying materials
   5 # are made available under the terms of the Apache License, Version 2.0
   6 # which accompanies this distribution, and is available at
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8 ##############################################################################
   9
  10 import re
  11 import copy
  12 import time
  13 import ConfigParser
  14
  15 fwd = {'single': ['forward'],
  16        'double': ['forward', 'reverse']
  17        }
  18 models = ['Tnv']
  19 direction = ['single', 'double']
  20 reverse_dict = {
  21     'forward': 'reverse',
  22     'reverse': 'forward'
  23 }
  24
  25
  26 class BaseModel(object):
  27
  28     def __init__(self, config):
  29         self.config = config
  30
  31     def _check_model(self):
  32         return self.config['model'] in models
  33
  34     def _check_virtenv(self):
  35         try:
  36             num = int(self.config['virtenv'])
  37             return num in range(1, 9)
  38         except:
  39             print("[ERROR]The virtenv is not a inter number.")
  40
  41     def _check_queues(self):
  42         try:
  43             num = int(self.config['queues'])
  44             return num in range(1, 9)
  45         except:
  46             print("[ERROR]The virt queues is not a inter number.")
  47
  48     @property
  49     def _check_flows(self):
  50         try:
  51             num = int(self.config['flows'])
  52             return num in range(1, 9)
  53         except:
  54             print("[ERROR]The flow is not a inter number.")
  55
  56     def _check_direct(self):
  57         return self.config['direct'] in direction
  58
  59     def _check_vlans(self):
  60         return self.config['vlans'] in ['True', 'False']
  61
  62     def _check_bind(self):
  63         return True
  64
  65     def check_parameter_invalid(self):
  66         try:
  67             if self._check_model() and \
  68                     self._check_virtenv() and \
  69                     self._check_queues() and \
  70                     self._check_flows and \
  71                     self._check_direct() and \
  72                     self._check_vlans() and \
  73                     self._check_bind():
  74                 return True
  75             else:
  76                 print("[ERROR]Paramter check invalid")
  77                 return False
  78         except:
  79             print("[ERROR]Check parameter invalid with unknown reason.")
  80             return False
  81
  82
  83 def _get_array_values(irq_array):
  84     proc_list = []
  85     for i in range(len(irq_array)):
  86         proc_list.append(irq_array[i][1])
  87     return sorted(proc_list)
  88
  89
  90 def check_dict(thread_info, flow):
  91     if thread_info['src_recv_irq'] != flow['src_recv_irq']:
  92         print("[WARN]Flow src_irq process %s not match %s in the table."
  93               % (thread_info['src_recv_irq'],
  94                  flow['src_recv_irq']))
  95         return False
  96     if thread_info['dst_send_irq'] != flow['dst_send_irq']:
  97         print("[WARN]Flow dst_irq process %s not match %s in the table."
  98               % (thread_info['dst_send_irq'],
  99                  flow['dst_send_irq']))
 100         return False
 101     return True
 102
 103
 104 def dst_ip_update(flow):
 105     try:
 106         src_dst_ip = flow['dst_ip']
 107         ip_section = '.'.join(src_dst_ip.split('.')[0:3]) + '.'
 108         number = int(src_dst_ip.split('.')[3])
 109         new_number = number + 1
 110         new_dst_ip = ip_section + str(new_number)
 111         flow['dst_ip'] = new_dst_ip
 112     except:
 113         print("[ERROR]dst ip update failed.")
 114
 115
 116 def _tranfer_array_to_range(array):
 117     return str(array[0]) + '-' + str(array[-1])
 118
 119
 120 class TnV(BaseModel):
 121
 122     def __init__(self, config):
 123         super(TnV, self).__init__(config)
 124         self.config = config
 125         self.host_instance = None
 126         self.send_instace = None
 127         self.vms = None
 128         self.init_flows = {}
 129         handle = ConfigParser.ConfigParser()
 130         handle.read(self.config['configfile'])
 131         self.handle = handle
 132
 133     def _get_vms(self):
 134         return self.host_instance.get_libvirt_vms()
 135
 136     def flow_match(self):
 137         _queues = int(self.config['queues'])
 138         _virtenv = int(self.config['virtenv'])
 139         _flows = int(self.config['flows'])
 140         return _flows == _queues * _virtenv
 141
 142     def match_virt_env(self):
 143         try:
 144             self.vms = self._get_vms()
 145             return len(self.vms) == int(self.config['virtenv'])
 146         except:
 147             print("[ERROR]vms or containers number is equal to virtenv.")
 148             return False
 149
 150     @property
 151     def match_flows_and_nic(self):
 152         # get src_nic
 153         for section in ['send', 'recv']:
 154             nic = self._get_nic_from_file(section, 'nic')
 155             try:
 156                 irq_proc = self.host_instance.get_nic_interrupt_proc(nic)
 157                 return int(self.config['flows']) == len(irq_proc)
 158             except:
 159                 print("[ERROR]match flow with nic interrupt failed.")
 160                 return False
 161
 162     def _get_nic_irq_proc(self, nic):
 163         return self.host_instance.get_nic_interrupt_proc(nic)
 164
 165     def _get_nic_from_file(self, section, column):
 166         return self.handle.get(section, column)
 167
 168     def _get_range(self, section, column):
 169         try:
 170             info = self.handle.get(section, column)
 171             return info.split(' ')
 172         except:
 173             print("[ERROR]Get mac failed.")
 174             return False
 175
 176     def check_mac_valid(self):
 177         flag = True
 178         try:
 179             for option in ['send', 'recv']:
 180                 info = self.handle.get(option, 'macs')
 181                 macs = info.split()
 182                 if len(macs) != int(self.config['virtenv']) or macs == []:
 183                     print(
 184                         "[ERROR]The macs number is not equal to vms or containers.")
 185                     return False
 186                 for mac in macs:
 187                     # check mac valid
 188                     if re.match(r'..:..:..:..:..:..', mac):
 189                         continue
 190                     else:
 191                         print("[ERROR]mac %s invalid" % mac)
 192                         flag = False
 193                         break
 194                 if not flag:
 195                     break
 196             return flag
 197         except:
 198             print("[ERROR]parse macs failed.")
 199             return False
 200
 201     def check_vlan_valid(self):
 202         flag = True
 203         for direct in ['send', 'recv']:
 204             vlans = self.handle.get(direct, 'vlans').split()
 205             if len(vlans) != int(self.config['virtenv']):
 206                 print("[ERROR]vlan un config")
 207                 return False
 208             for vlan in vlans:
 209                 if int(vlan) <= 1 or int(vlan) >= 4095:
 210                     flag = False
 211                     break
 212         return flag
 213
 214     @property
 215     def check_logic_invalid(self):
 216         return self.flow_match() and self.match_virt_env() and \
 217             self.match_flows_and_nic and self.check_mac_valid() and \
 218             self.check_vlan_valid()
 219
 220     @property
 221     def read_flow_init(self):
 222         # The
 223         temp_flow = {}
 224         src_macs = self._get_range('send', 'macs')
 225         dst_macs = self._get_range('recv', 'macs')
 226         src_vlan = self._get_range('send', 'vlans')
 227         dst_vlan = self._get_range('recv', 'vlans')
 228         src_nic = self._get_nic_from_file('send', 'nic')
 229         dst_nic = self._get_nic_from_file('recv', 'nic')
 230         src_nic_irq = _get_array_values(self._get_nic_irq_proc(src_nic))
 231         dst_nic_irq = _get_array_values(self._get_nic_irq_proc(dst_nic))
 232         src_ip_sections = self._get_range('send', 'ip_sections')
 233         dst_ip_sections = self._get_range('recv', 'ip_sections')
 234         send_port = self._get_nic_from_file('send', 'port')
 235         recv_port = self._get_nic_from_file('recv', 'port')
 236         temp_flow['tester_ip'] = self._get_nic_from_file('common', 'tester_ip')
 237         vlan = src_vlan
 238         avg_flow = int(self.config['flows']) / int(self.config['virtenv'])
 239         # build the main dictionary
 240         for _direct in sorted(fwd[self.config['direct']]):
 241             i = 0
 242             j = 0
 243             temp_flow['direct'] = _direct
 244             temp_flow['send_port'] = send_port
 245             temp_flow['recv_port'] = recv_port
 246
 247             for _vm in sorted(self.vms):
 248                 vlan_id = {
 249                     'True': vlan[i],
 250                     'False': None}
 251                 temp_flow['virt'] = _vm
 252                 _vm_info = self.host_instance.get_vm_info(_vm)
 253                 temp_flow['qemu_proc'] = _vm_info['main_pid']
 254                 # temp_flow['qemu_thread']  = _vm_info['qemu_thread']
 255                 temp_flow['mem_numa'] = _vm_info['mem_numa']
 256                 # temp_flow['vhost_thread'] = _vm_info['vhost_thread']
 257
 258                 temp_flow['src_mac'] = src_macs[i]
 259                 temp_flow['dst_mac'] = dst_macs[i]
 260                 temp_flow['vlan'] = vlan_id[self.config['vlans']]
 261                 src_ip = src_ip_sections[i]
 262                 dst_ip = dst_ip_sections[i]
 263                 temp_flow['src_ip'] = src_ip
 264                 temp_flow['dst_ip'] = dst_ip
 265                 vm_index = sorted(self.vms).index(_vm)
 266                 for _queue in range(1, int(self.config['queues']) + 1):
 267                     # flow info
 268                     temp_flow['queue'] = _queue
 269                     # fwd thread
 270
 271                     temp_flow['qemu_thread_list'] = _vm_info['qemu_thread']
 272                     forward_core = {
 273                         "forward": _vm_info['qemu_thread'][
 274                             _queue + avg_flow * vm_index],
 275                         "reverse": _vm_info['qemu_thread'][
 276                             _queue + avg_flow * vm_index + int(
 277                                 self.config['flows'])]}
 278                     temp_flow['fwd_thread'] = forward_core[_direct]
 279
 280                     temp_flow['fwd_vhost'] = None
 281                     # nic interrupts info
 282                     temp_flow['src_recv_irq'] = src_nic_irq[j]
 283                     temp_flow['src_nic'] = src_nic
 284                     temp_flow['dst_send_irq'] = dst_nic_irq[j]
 285                     temp_flow['dst_nic'] = dst_nic
 286                     # above all
 287                     j += 1
 288                     self.init_flows[_direct + '_' + _vm + '_' +
 289                                     str(_queue)] = copy.deepcopy(temp_flow)
 290                 i += 1
 291             src_nic_irq, dst_nic_irq = dst_nic_irq, src_nic_irq
 292             vlan = dst_vlan
 293             send_port, recv_port = recv_port, send_port
 294             src_nic, dst_nic = dst_nic, src_nic
 295             src_macs, dst_macs = dst_macs, src_macs
 296             src_ip_sections, dst_ip_sections = dst_ip_sections, src_ip_sections
 297         # return sorted(self.init_flows.iteritems(), key=lambda d:d[0])
 298         return self.init_flows
 299
 300     def mac_learning(self, flowa, flowb):
 301         flowa = str(flowa)
 302         flowb = str(flowb)
 303         ret = self.send_instace.mac_learning(flowa, flowb)
 304         return ret
 305
 306     def send_packet(self, flow):
 307         flow = str(flow)
 308         # return a stream block handle
 309         return self.send_instace.send_packet(flow)
 310
 311     def stop_flow(self, streamblock, flow):
 312         flow = str(flow)
 313         return self.send_instace.stop_flow(streamblock, flow)
 314
 315     def catch_thread_info(self):
 316         return self.host_instance.catch_thread_info()
 317
 318     def set_thread2flow(self, thread_info, flow):
 319         flow['fwd_vhost'] = thread_info['fwd_vhost']
 320         return True
 321
 322     @property
 323     def flow_build(self):
 324         for _direct in fwd[self.config['direct']]:
 325             for _vm in self.vms:
 326                 for _queue in range(1, int(self.config['queues']) + 1):
 327                     i = 0
 328                     while i < 50:
 329                         try:
 330                             i += 1
 331                             thread_info = None
 332                             self.mac_learning(
 333                                 self.init_flows[
 334                                     _direct +
 335                                     '_' +
 336                                     _vm +
 337                                     '_' +
 338                                     str(_queue)],
 339                                 self.init_flows[
 340                                     reverse_dict[_direct] +
 341                                     '_' +
 342                                     _vm +
 343                                     '_' +
 344                                     str(_queue)])
 345                             streamblock = self.send_packet(
 346                                 self.init_flows[_direct + '_' + _vm + '_' + str(_queue)])
 347                             time.sleep(1)
 348                             result, thread_info = self.catch_thread_info()
 349                             thread_info = eval(thread_info)
 350                             self.stop_flow(
 351                                 streamblock, self.init_flows[
 352                                     _direct + '_' + _vm + '_' + str(_queue)])
 353                             time.sleep(1)
 354                             if not result:
 355                                 print("[ERROR]Catch the thread info failed.")
 356                                 break
 357                         except:
 358                             print(
 359                                 "[ERROR]send flow failed error or get host thread info failed.")
 360
 361                         # compare the got thread info to
 362                         if check_dict(
 363                             thread_info, self.init_flows[
 364                                 _direct + '_' + _vm + '_' + str(_queue)]):
 365                             self.set_thread2flow(
 366                                 thread_info, self.init_flows[
 367                                     _direct + '_' + _vm + '_' + str(_queue)])
 368                             print(
 369                                 "[INFO]Flow %s_%s_%s :     fwd_vhost %s    src_recv_irq %s   dst_send_irq %s" %
 370                                 (_direct,
 371                                  _vm,
 372                                  _queue,
 373                                  thread_info['fwd_vhost'],
 374                                     thread_info['src_recv_irq'],
 375                                     thread_info['dst_send_irq']))
 376                             print(
 377                                 "%s" %
 378                                 (self.init_flows[
 379                                     _direct +
 380                                     '_' +
 381                                     _vm +
 382                                     '_' +
 383                                     str(_queue)]))
 384                             break
 385                         else:
 386                             dst_ip_update(
 387                                 self.init_flows[
 388                                     _direct +
 389                                     '_' +
 390                                     _vm +
 391                                     '_' +
 392                                     str(_queue)])
 393         return self.init_flows
 394
 395     def affinity_bind(self, aff_strategy):
 396         # get the forward cores
 397         qemu_list = []
 398         qemu_other = []
 399         src_vhost = []
 400         dst_vhost = []
 401         src_irq = []
 402         dst_irq = []
 403
 404         # recognize the thread id
 405         for flowname in sorted(self.init_flows.keys()):
 406             tmp_thread = self.init_flows[flowname]['fwd_thread']
 407             qemu_other = qemu_other + \
 408                 copy.deepcopy(self.init_flows[flowname]['qemu_thread_list'])
 409             qemu_list.append(tmp_thread)
 410             if self.init_flows[flowname]['direct'] == 'forward':
 411                 dst_vhost.append(self.init_flows[flowname]['fwd_vhost'])
 412                 src_irq.append(self.init_flows[flowname]['src_recv_irq'])
 413                 dst_irq.append(self.init_flows[flowname]['dst_send_irq'])
 414             elif self.init_flows[flowname]['direct'] == 'reverse':
 415                 src_vhost.append(self.init_flows[flowname]['fwd_vhost'])
 416                 dst_irq.append(self.init_flows[flowname]['src_recv_irq'])
 417                 src_irq.append(self.init_flows[flowname]['dst_send_irq'])
 418
 419         qemu_list = sorted({}.fromkeys(qemu_list).keys())
 420         src_vhost = sorted({}.fromkeys(src_vhost).keys())
 421         dst_vhost = sorted({}.fromkeys(dst_vhost).keys())
 422         src_irq = sorted({}.fromkeys(src_irq).keys())
 423         dst_irq = sorted({}.fromkeys(dst_irq).keys())
 424
 425         # get the qemu thread except the forward core
 426         qemu_other = sorted({}.fromkeys(qemu_other).keys())
 427         for i in qemu_list:
 428             qemu_other.remove(i)
 429         # get the bind strategy
 430         handle = ConfigParser.ConfigParser()
 431         handle.read(self.config['strategyfile'])
 432         try:
 433             qemu_numa = handle.get(
 434                 'strategy' +
 435                 self.config['strategy'],
 436                 'qemu_numa')
 437             src_vhost_numa = handle.get(
 438                 'strategy' + self.config['strategy'],
 439                 'src_vhost_numa')
 440             dst_vhost_numa = handle.get(
 441                 'strategy' + self.config['strategy'],
 442                 'dst_vhost_numa')
 443             src_irq_numa = handle.get(
 444                 'strategy' +
 445                 self.config['strategy'],
 446                 'src_irq_numa')
 447             dst_irq_numa = handle.get(
 448                 'strategy' +
 449                 self.config['strategy'],
 450                 'dst_irq_numa')
 451             loan_numa = handle.get(
 452                 'strategy' +
 453                 self.config['strategy'],
 454                 'loan_numa')
 455         except:
 456             print("[ERROR]Parse the strategy file failed or get the options failed.")
 457
 458         for value in [
 459                 qemu_numa,
 460                 src_vhost_numa,
 461                 dst_vhost_numa,
 462                 src_irq_numa,
 463                 dst_irq_numa,
 464                 loan_numa]:
 465             if value is not None or value == '':
 466                 raise ValueError('some option in the strategy file is none.')
 467         # cores mapping thread
 468         numa_topo = self.host_instance.get_numa_core()
 469         numa_topo = eval(numa_topo)
 470         # first check the cores number
 471
 472         # order src_irq dst_irq src_vhost dst_vhost qemu_list
 473         for node in numa_topo.keys():
 474             numa_topo[node]['process'] = []
 475             if 'node' + src_irq_numa == node:
 476                 numa_topo[node]['process'] = numa_topo[
 477                     node]['process'] + src_irq
 478             if 'node' + dst_irq_numa == node:
 479                 numa_topo[node]['process'] = numa_topo[
 480                     node]['process'] + dst_irq
 481             if 'node' + src_vhost_numa == node:
 482                 numa_topo[node]['process'] = numa_topo[
 483                     node]['process'] + src_vhost
 484             if 'node' + dst_vhost_numa == node:
 485                 numa_topo[node]['process'] = numa_topo[
 486                     node]['process'] + dst_vhost
 487             if 'node' + qemu_numa == node:
 488                 numa_topo[node]['process'] = numa_topo[
 489                     node]['process'] + qemu_list
 490         loan_cores = ''
 491         for node in numa_topo.keys():
 492             if len(
 493                     numa_topo[node]['process']) > len(
 494                     numa_topo[node]['phy_cores']):
 495                 # length distance
 496                 diff = len(numa_topo[node]['process']) - \
 497                     len(numa_topo[node]['phy_cores'])
 498                 # first deep copy
 499                 numa_topo['node' + loan_numa]['process'] = numa_topo['node' + loan_numa][
 500                     'process'] + copy.deepcopy(numa_topo[node]['process'][-diff:])
 501                 cores_str = _tranfer_array_to_range(
 502                     numa_topo[
 503                         'node' +
 504                         loan_numa]['phy_cores'][
 505                         diff:])
 506                 loan_cores = ','.join([loan_cores, cores_str])
 507                 numa_topo[node]['process'] = numa_topo[
 508                     node]['process'][0:-diff]
 509         loan_cores = loan_cores[1:]
 510         loan_bind_list = {}
 511         for proc_loan in qemu_other:
 512             loan_bind_list[proc_loan] = loan_cores
 513
 514         bind_list = {}
 515         for node in numa_topo.keys():
 516             for i in range(len(numa_topo[node]['process'])):
 517                 bind_list[numa_topo[node]['process'][i]] = str(
 518                     numa_topo[node]['phy_cores'][i])
 519         bind_list.update(loan_bind_list)
 520         for key in bind_list.keys():
 521             self.host_instance.bind_cpu(bind_list[key], key)
 522         print bind_list
 523         return True
 524
 525     def testrun(self, suite):
 526         global forward_init_flows, reverse_init_flows
 527         try:
 528             forward_init_flows = {}
 529             reverse_init_flows = {}
 530             for key in self.init_flows.keys():
 531                 if self.init_flows[key]['direct'] == "forward":
 532                     forward_init_flows[key] = self.init_flows[key]
 533                 elif self.init_flows[key]['direct'] == "reverse":
 534                     reverse_init_flows[key] = self.init_flows[key]
 535             forward_init_flows = str(forward_init_flows)
 536             reverse_init_flows = str(reverse_init_flows)
 537         except:
 538             print("[ERROR]init the forward and reverse flow failed.")
 539
 540         if suite == "throughput":
 541             print("[INFO]!!!!!!!!!!!!!!!Now begin to throughput test")
 542             ret, result = self.send_instace.run_rfc2544_throughput(
 543                 forward_init_flows, reverse_init_flows)
 544         elif suite == "frameloss":
 545             print("[INFO]!!!!!!!!!!!1!!!Now begin to frameloss test")
 546             ret, result = self.send_instace.run_rfc2544_frameloss(
 547                 forward_init_flows, reverse_init_flows)
 548         return ret, result