#!/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