--- /dev/null
+#!/usr/bin/python
+"""
+ @author: l00190809
+ @group: Huawei Ltd
+"""
+import re
+import copy
+import time
+import ConfigParser
+
+fwd = {'single': ['forward'],
+ 'double': ['forward', 'reverse']
+ }
+models = ['Tnv']
+direction = ['single', 'double']
+reverse_dict = {
+ 'forward': 'reverse',
+ 'reverse': 'forward'
+}
+
+
+class BaseModel(object):
+ def __init__(self, config):
+ self.config = config
+
+ def _check_model(self):
+ return self.config['model'] in models
+
+ def _check_virtenv(self):
+ try:
+ num = int(self.config['virtenv'])
+ return num in range(1, 9)
+ except:
+ print("[ERROR]The virtenv is not a inter number.")
+
+ def _check_queues(self):
+ try:
+ num = int(self.config['queues'])
+ return num in range(1, 9)
+ except:
+ print("[ERROR]The virt queues is not a inter number.")
+
+ @property
+ def _check_flows(self):
+ try:
+ num = int(self.config['flows'])
+ return num in range(1, 9)
+ except:
+ print("[ERROR]The flow is not a inter number.")
+
+ def _check_direct(self):
+ return self.config['direct'] in direction
+
+ def _check_vlans(self):
+ return self.config['vlans'] in ['True', 'False']
+
+ def _check_bind(self):
+ return True
+
+ def check_parameter_invalid(self):
+ try:
+ if self._check_model() and \
+ self._check_virtenv() and \
+ self._check_queues() and \
+ self._check_flows and \
+ self._check_direct() and \
+ self._check_vlans() and \
+ self._check_bind():
+ return True
+ else:
+ print("[ERROR]Paramter check invalid")
+ return False
+ except:
+ print("[ERROR]Check parameter invalid with unknown reason.")
+ return False
+
+
+def _get_array_values(irq_array):
+ proc_list = []
+ for i in range(len(irq_array)):
+ proc_list.append(irq_array[i][1])
+ return sorted(proc_list)
+
+
+def check_dict(thread_info, flow):
+ if thread_info['src_recv_irq'] != flow['src_recv_irq']:
+ print("[WARN]Flow src_irq process %s not match %s in the table."
+ % (thread_info['src_recv_irq'],
+ flow['src_recv_irq']))
+ return False
+ if thread_info['dst_send_irq'] != flow['dst_send_irq']:
+ print("[WARN]Flow dst_irq process %s not match %s in the table."
+ % (thread_info['dst_send_irq'],
+ flow['dst_send_irq']))
+ return False
+ return True
+
+
+def dst_ip_update(flow):
+ try:
+ src_dst_ip = flow['dst_ip']
+ ip_section = '.'.join(src_dst_ip.split('.')[0:3]) + '.'
+ number = int(src_dst_ip.split('.')[3])
+ new_number = number + 1
+ new_dst_ip = ip_section + str(new_number)
+ flow['dst_ip'] = new_dst_ip
+ except:
+ print("[ERROR]dst ip update failed.")
+
+
+def _tranfer_array_to_range(array):
+ return str(array[0]) + '-' + str(array[-1])
+
+
+class TnV(BaseModel):
+ def __init__(self, config):
+ super(TnV, self).__init__(config)
+ self.config = config
+ self.host_instance = None
+ self.send_instace = None
+ self.vms = None
+ self.init_flows = {}
+ handle = ConfigParser.ConfigParser()
+ handle.read(self.config['configfile'])
+ self.handle = handle
+
+ def _get_vms(self):
+ return self.host_instance.get_libvirt_vms()
+
+ def flow_match(self):
+ _queues = int(self.config['queues'])
+ _virtenv = int(self.config['virtenv'])
+ _flows = int(self.config['flows'])
+ return _flows == _queues * _virtenv
+
+ def match_virt_env(self):
+ try:
+ self.vms = self._get_vms()
+ return len(self.vms) == int(self.config['virtenv'])
+ except:
+ print("[ERROR]vms or containers number is equal to virtenv.")
+ return False
+
+ @property
+ def match_flows_and_nic(self):
+ # get src_nic
+ for section in ['send', 'recv']:
+ nic = self._get_nic_from_file(section, 'nic')
+ try:
+ irq_proc = self.host_instance.get_nic_interrupt_proc(nic)
+ return int(self.config['flows']) == len(irq_proc)
+ except:
+ print("[ERROR]match flow with nic interrupt failed.")
+ return False
+
+ def _get_nic_irq_proc(self, nic):
+ return self.host_instance.get_nic_interrupt_proc(nic)
+
+ def _get_nic_from_file(self, section, column):
+ return self.handle.get(section, column)
+
+ def _get_range(self, section, column):
+ try:
+ info = self.handle.get(section, column)
+ return info.split(' ')
+ except:
+ print("[ERROR]Get mac failed.")
+ return False
+
+ def check_mac_valid(self):
+ flag = True
+ try:
+ for option in ['send', 'recv']:
+ info = self.handle.get(option, 'macs')
+ macs = info.split()
+ if len(macs) != int(self.config['virtenv']) or macs == []:
+ print("[ERROR]The macs number is not equal to vms or containers.")
+ return False
+ for mac in macs:
+ # check mac valid
+ if re.match(r'..:..:..:..:..:..', mac):
+ continue
+ else:
+ print("[ERROR]mac %s invalid" % mac)
+ flag = False
+ break
+ if not flag:
+ break
+ return flag
+ except:
+ print("[ERROR]parse macs failed.")
+ return False
+
+ def check_vlan_valid(self):
+ flag = True
+ for direct in ['send', 'recv']:
+ vlans = self.handle.get(direct, 'vlans').split()
+ if len(vlans) != int(self.config['virtenv']):
+ print("[ERROR]vlan un config")
+ return False
+ for vlan in vlans:
+ if int(vlan) <= 1 or int(vlan) >= 4095:
+ flag = False
+ break
+ return flag
+
+ @property
+ def check_logic_invalid(self):
+ return self.flow_match() and self.match_virt_env() and \
+ self.match_flows_and_nic and self.check_mac_valid() and \
+ self.check_vlan_valid()
+
+ @property
+ def read_flow_init(self):
+ # The
+ temp_flow = {}
+ src_macs = self._get_range('send', 'macs')
+ dst_macs = self._get_range('recv', 'macs')
+ src_vlan = self._get_range('send', 'vlans')
+ dst_vlan = self._get_range('recv', 'vlans')
+ src_nic = self._get_nic_from_file('send', 'nic')
+ dst_nic = self._get_nic_from_file('recv', 'nic')
+ src_nic_irq = _get_array_values(self._get_nic_irq_proc(src_nic))
+ dst_nic_irq = _get_array_values(self._get_nic_irq_proc(dst_nic))
+ src_ip_sections = self._get_range('send', 'ip_sections')
+ dst_ip_sections = self._get_range('recv', 'ip_sections')
+ send_port = self._get_nic_from_file('send', 'port')
+ recv_port = self._get_nic_from_file('recv', 'port')
+ temp_flow['tester_ip'] = self._get_nic_from_file('common', 'tester_ip')
+ vlan = src_vlan
+ avg_flow = int(self.config['flows']) / int(self.config['virtenv'])
+ # build the main dictionary
+ for _direct in sorted(fwd[self.config['direct']]):
+ i = 0
+ j = 0
+ temp_flow['direct'] = _direct
+ temp_flow['send_port'] = send_port
+ temp_flow['recv_port'] = recv_port
+
+ for _vm in sorted(self.vms):
+ vlan_id = {
+ 'True': vlan[i],
+ 'False': None}
+ temp_flow['virt'] = _vm
+ _vm_info = self.host_instance.get_vm_info(_vm)
+ temp_flow['qemu_proc'] = _vm_info['main_pid']
+ # temp_flow['qemu_thread'] = _vm_info['qemu_thread']
+ temp_flow['mem_numa'] = _vm_info['mem_numa']
+ # temp_flow['vhost_thread'] = _vm_info['vhost_thread']
+
+ temp_flow['src_mac'] = src_macs[i]
+ temp_flow['dst_mac'] = dst_macs[i]
+ temp_flow['vlan'] = vlan_id[self.config['vlans']]
+ src_ip = src_ip_sections[i]
+ dst_ip = dst_ip_sections[i]
+ temp_flow['src_ip'] = src_ip
+ temp_flow['dst_ip'] = dst_ip
+ vm_index = sorted(self.vms).index(_vm)
+ for _queue in range(1, int(self.config['queues']) + 1):
+ # flow info
+ temp_flow['queue'] = _queue
+ # fwd thread
+
+ temp_flow['qemu_thread_list'] = _vm_info['qemu_thread']
+ forward_core = {
+ "forward": _vm_info['qemu_thread'][_queue + avg_flow * vm_index],
+ "reverse": _vm_info['qemu_thread'][_queue + avg_flow * vm_index + int(self.config['flows'])]
+ }
+ temp_flow['fwd_thread'] = forward_core[_direct]
+
+ temp_flow['fwd_vhost'] = None
+ # nic interrupts info
+ temp_flow['src_recv_irq'] = src_nic_irq[j]
+ temp_flow['src_nic'] = src_nic
+ temp_flow['dst_send_irq'] = dst_nic_irq[j]
+ temp_flow['dst_nic'] = dst_nic
+ # above all
+ j += 1
+ self.init_flows[_direct + '_' + _vm + '_' + str(_queue)] = copy.deepcopy(temp_flow)
+ i += 1
+ src_nic_irq, dst_nic_irq = dst_nic_irq, src_nic_irq
+ vlan = dst_vlan
+ send_port, recv_port = recv_port, send_port
+ src_nic, dst_nic = dst_nic, src_nic
+ src_macs, dst_macs = dst_macs, src_macs
+ src_ip_sections, dst_ip_sections = dst_ip_sections, src_ip_sections
+ # return sorted(self.init_flows.iteritems(), key=lambda d:d[0])
+ return self.init_flows
+
+ def mac_learning(self, flowa, flowb):
+ flowa = str(flowa)
+ flowb = str(flowb)
+ ret = self.send_instace.mac_learning(flowa, flowb)
+ return ret
+
+ def send_packet(self, flow):
+ flow = str(flow)
+ # return a stream block handle
+ return self.send_instace.send_packet(flow)
+
+ def stop_flow(self, streamblock, flow):
+ flow = str(flow)
+ return self.send_instace.stop_flow(streamblock, flow)
+
+ def catch_thread_info(self):
+ return self.host_instance.catch_thread_info()
+
+ def set_thread2flow(self, thread_info, flow):
+ flow['fwd_vhost'] = thread_info['fwd_vhost']
+ return True
+
+ @property
+ def flow_build(self):
+ for _direct in fwd[self.config['direct']]:
+ for _vm in self.vms:
+ for _queue in range(1, int(self.config['queues']) + 1):
+ i = 0
+ while i < 50:
+ try:
+ i += 1
+ thread_info = None
+ self.mac_learning(self.init_flows[_direct + '_' + _vm + '_' + str(_queue)],
+ self.init_flows[reverse_dict[_direct] + '_' + _vm + '_' + str(_queue)])
+ streamblock = self.send_packet(self.init_flows[_direct + '_' + _vm + '_' + str(_queue)])
+ time.sleep(1)
+ result, thread_info = self.catch_thread_info()
+ thread_info = eval(thread_info)
+ self.stop_flow(streamblock, self.init_flows[_direct + '_' + _vm + '_' + str(_queue)])
+ time.sleep(1)
+ if not result:
+ print("[ERROR]Catch the thread info failed.")
+ break
+ except:
+ print("[ERROR]send flow failed error or get host thread info failed.")
+
+ # compare the got thread info to
+ if check_dict(thread_info, self.init_flows[_direct + '_' + _vm + '_' + str(_queue)]):
+ self.set_thread2flow(thread_info, self.init_flows[_direct + '_' + _vm + '_' + str(_queue)])
+ print("[INFO]Flow %s_%s_%s : fwd_vhost %s src_recv_irq %s dst_send_irq %s"
+ % (_direct, _vm, _queue, thread_info['fwd_vhost'], thread_info['src_recv_irq'],
+ thread_info['dst_send_irq']))
+ print("%s" % (self.init_flows[_direct + '_' + _vm + '_' + str(_queue)]))
+ break
+ else:
+ dst_ip_update(self.init_flows[_direct + '_' + _vm + '_' + str(_queue)])
+ return self.init_flows
+
+ def affinity_bind(self, aff_strategy):
+ # get the forward cores
+ qemu_list = []
+ qemu_other = []
+ src_vhost = []
+ dst_vhost = []
+ src_irq = []
+ dst_irq = []
+
+ # recognize the thread id
+ for flowname in sorted(self.init_flows.keys()):
+ tmp_thread = self.init_flows[flowname]['fwd_thread']
+ qemu_other = qemu_other + copy.deepcopy(self.init_flows[flowname]['qemu_thread_list'])
+ qemu_list.append(tmp_thread)
+ if self.init_flows[flowname]['direct'] == 'forward':
+ dst_vhost.append(self.init_flows[flowname]['fwd_vhost'])
+ src_irq.append(self.init_flows[flowname]['src_recv_irq'])
+ dst_irq.append(self.init_flows[flowname]['dst_send_irq'])
+ elif self.init_flows[flowname]['direct'] == 'reverse':
+ src_vhost.append(self.init_flows[flowname]['fwd_vhost'])
+ dst_irq.append(self.init_flows[flowname]['src_recv_irq'])
+ src_irq.append(self.init_flows[flowname]['dst_send_irq'])
+
+ qemu_list = sorted({}.fromkeys(qemu_list).keys())
+ src_vhost = sorted({}.fromkeys(src_vhost).keys())
+ dst_vhost = sorted({}.fromkeys(dst_vhost).keys())
+ src_irq = sorted({}.fromkeys(src_irq).keys())
+ dst_irq = sorted({}.fromkeys(dst_irq).keys())
+
+ # get the qemu thread except the forward core
+ qemu_other = sorted({}.fromkeys(qemu_other).keys())
+ for i in qemu_list:
+ qemu_other.remove(i)
+ # get the bind strategy
+ handle = ConfigParser.ConfigParser()
+ handle.read(self.config['strategyfile'])
+ try:
+ qemu_numa = handle.get('strategy' + self.config['strategy'], 'qemu_numa')
+ src_vhost_numa = handle.get('strategy' + self.config['strategy'], 'src_vhost_numa')
+ dst_vhost_numa = handle.get('strategy' + self.config['strategy'], 'dst_vhost_numa')
+ src_irq_numa = handle.get('strategy' + self.config['strategy'], 'src_irq_numa')
+ dst_irq_numa = handle.get('strategy' + self.config['strategy'], 'dst_irq_numa')
+ loan_numa = handle.get('strategy' + self.config['strategy'], 'loan_numa')
+ except:
+ print("[ERROR]Parse the strategy file failed or get the options failed.")
+
+ for value in [qemu_numa, src_vhost_numa, dst_vhost_numa, src_irq_numa, dst_irq_numa, loan_numa]:
+ if value is not None or value == '':
+ raise ValueError('some option in the strategy file is none.')
+ # cores mapping thread
+ numa_topo = self.host_instance.get_numa_core()
+ numa_topo = eval(numa_topo)
+ # first check the cores number
+
+ # order src_irq dst_irq src_vhost dst_vhost qemu_list
+ for node in numa_topo.keys():
+ numa_topo[node]['process'] = []
+ if 'node' + src_irq_numa == node:
+ numa_topo[node]['process'] = numa_topo[node]['process'] + src_irq
+ if 'node' + dst_irq_numa == node:
+ numa_topo[node]['process'] = numa_topo[node]['process'] + dst_irq
+ if 'node' + src_vhost_numa == node:
+ numa_topo[node]['process'] = numa_topo[node]['process'] + src_vhost
+ if 'node' + dst_vhost_numa == node:
+ numa_topo[node]['process'] = numa_topo[node]['process'] + dst_vhost
+ if 'node' + qemu_numa == node:
+ numa_topo[node]['process'] = numa_topo[node]['process'] + qemu_list
+ loan_cores = ''
+ for node in numa_topo.keys():
+ if len(numa_topo[node]['process']) > len(numa_topo[node]['phy_cores']):
+ # length distance
+ diff = len(numa_topo[node]['process']) - len(numa_topo[node]['phy_cores'])
+ # first deep copy
+ numa_topo['node' + loan_numa]['process'] = numa_topo['node' + loan_numa]['process'] + copy.deepcopy(
+ numa_topo[node]['process'][-diff:])
+ cores_str = _tranfer_array_to_range(numa_topo['node' + loan_numa]['phy_cores'][diff:])
+ loan_cores = ','.join([loan_cores, cores_str])
+ numa_topo[node]['process'] = numa_topo[node]['process'][0:-diff]
+ loan_cores = loan_cores[1:]
+ loan_bind_list = {}
+ for proc_loan in qemu_other:
+ loan_bind_list[proc_loan] = loan_cores
+
+ bind_list = {}
+ for node in numa_topo.keys():
+ for i in range(len(numa_topo[node]['process'])):
+ bind_list[numa_topo[node]['process'][i]] = str(numa_topo[node]['phy_cores'][i])
+ bind_list.update(loan_bind_list)
+ for key in bind_list.keys():
+ self.host_instance.bind_cpu(bind_list[key], key)
+ print bind_list
+ return True
+
+ def testrun(self, suite):
+ global forward_init_flows, reverse_init_flows
+ try:
+ forward_init_flows = {}
+ reverse_init_flows = {}
+ for key in self.init_flows.keys():
+ if self.init_flows[key]['direct'] == "forward":
+ forward_init_flows[key] = self.init_flows[key]
+ elif self.init_flows[key]['direct'] == "reverse":
+ reverse_init_flows[key] = self.init_flows[key]
+ forward_init_flows = str(forward_init_flows)
+ reverse_init_flows = str(reverse_init_flows)
+ except:
+ print("[ERROR]init the forward and reverse flow failed.")
+
+ if suite == "throughput":
+ print("[INFO]!!!!!!!!!!!!!!!Now begin to throughput test")
+ ret, result = self.send_instace.run_rfc2544_throughput(forward_init_flows, reverse_init_flows)
+ elif suite == "frameloss":
+ print("[INFO]!!!!!!!!!!!1!!!Now begin to frameloss test")
+ ret, result = self.send_instace.run_rfc2544_frameloss(forward_init_flows, reverse_init_flows)
+ return ret, result
Code Review / bottlenecks.git / blobdiff