nfvbench/traffic_client.py

   1 # Copyright 2016 Cisco Systems, Inc.  All rights reserved.
   2 #
   3 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   4 #    not use this file except in compliance with the License. You may obtain
   5 #    a copy of the License at
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #    Unless required by applicable law or agreed to in writing, software
  10 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  11 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  12 #    License for the specific language governing permissions and limitations
  13 #    under the License.
  14
  15 from attrdict import AttrDict
  16 import bitmath
  17 from datetime import datetime
  18 from log import LOG
  19 from netaddr import IPNetwork
  20 from network import Interface
  21 import socket
  22 from specs import ChainType
  23 from stats_collector import IntervalCollector
  24 from stats_collector import IterationCollector
  25 import struct
  26 import time
  27 import traffic_gen.traffic_utils as utils
  28 from utils import cast_integer
  29
  30
  31 class TrafficClientException(Exception):
  32     pass
  33
  34
  35 class TrafficRunner(object):
  36
  37     def __init__(self, client, duration_sec, interval_sec=0):
  38         self.client = client
  39         self.start_time = None
  40         self.duration_sec = duration_sec
  41         self.interval_sec = interval_sec
  42
  43     def run(self):
  44         LOG.info('Running traffic generator')
  45         self.client.gen.clear_stats()
  46         self.client.gen.start_traffic()
  47         self.start_time = time.time()
  48         return self.poll_stats()
  49
  50     def stop(self):
  51         if self.is_running():
  52             self.start_time = None
  53             self.client.gen.stop_traffic()
  54
  55     def is_running(self):
  56         return self.start_time is not None
  57
  58     def time_elapsed(self):
  59         if self.is_running():
  60             return time.time() - self.start_time
  61         else:
  62             return self.duration_sec
  63
  64     def poll_stats(self):
  65         if not self.is_running():
  66             return None
  67         time_elapsed = self.time_elapsed()
  68         if time_elapsed > self.duration_sec:
  69             self.stop()
  70             return None
  71         time_left = self.duration_sec - time_elapsed
  72         if self.interval_sec > 0.0:
  73             if time_left <= self.interval_sec:
  74                 time.sleep(time_left)
  75                 self.stop()
  76             else:
  77                 time.sleep(self.interval_sec)
  78         else:
  79             time.sleep(self.duration_sec)
  80             self.stop()
  81         return self.client.get_stats()
  82
  83
  84 class Device(object):
  85
  86     def __init__(self, port, pci, switch_port=None, vtep_vlan=None, ip=None, tg_gateway_ip=None,
  87                  gateway_ip=None, ip_addrs_step=None, tg_gateway_ip_addrs_step=None,
  88                  gateway_ip_addrs_step=None, udp_src_port=None, udp_dst_port=None,
  89                  chain_count=1, flow_count=1, vlan_tagging=False):
  90         self.chain_count = chain_count
  91         self.flow_count = flow_count
  92         self.dst = None
  93         self.port = port
  94         self.switch_port = switch_port
  95         self.vtep_vlan = vtep_vlan
  96         self.vlan_tag = None
  97         self.vlan_tagging = vlan_tagging
  98         self.pci = pci
  99         self.mac = None
 100         self.vm_mac_list = None
 101         subnet = IPNetwork(ip)
 102         self.ip = subnet.ip.format()
 103         self.ip_prefixlen = subnet.prefixlen
 104         self.ip_addrs_step = ip_addrs_step
 105         self.tg_gateway_ip_addrs_step = tg_gateway_ip_addrs_step
 106         self.gateway_ip_addrs_step = gateway_ip_addrs_step
 107         self.ip_list = self.expand_ip(self.ip, self.ip_addrs_step, self.flow_count)
 108         self.gateway_ip = gateway_ip
 109         self.gateway_ip_list = self.expand_ip(self.gateway_ip,
 110                                               self.gateway_ip_addrs_step,
 111                                               self.chain_count)
 112         self.tg_gateway_ip = tg_gateway_ip
 113         self.tg_gateway_ip_list = self.expand_ip(self.tg_gateway_ip,
 114                                                  self.tg_gateway_ip_addrs_step,
 115                                                  self.chain_count)
 116         self.udp_src_port = udp_src_port
 117         self.udp_dst_port = udp_dst_port
 118
 119     def set_mac(self, mac):
 120         if mac is None:
 121             raise TrafficClientException('Trying to set traffic generator MAC address as None')
 122         self.mac = mac
 123
 124     def set_destination(self, dst):
 125         self.dst = dst
 126
 127     def set_vm_mac_list(self, vm_mac_list):
 128         self.vm_mac_list = map(str, vm_mac_list)
 129
 130     def set_vlan_tag(self, vlan_tag):
 131         if self.vlan_tagging and vlan_tag is None:
 132             raise TrafficClientException('Trying to set VLAN tag as None')
 133         self.vlan_tag = vlan_tag
 134
 135     def get_stream_configs(self, service_chain):
 136         configs = []
 137         flow_idx = 0
 138         for chain_idx in xrange(self.chain_count):
 139             current_flow_count = (self.flow_count - flow_idx) / (self.chain_count - chain_idx)
 140             max_idx = flow_idx + current_flow_count - 1
 141             ip_src_count = self.ip_to_int(self.ip_list[max_idx]) - \
 142                 self.ip_to_int(self.ip_list[flow_idx]) + 1
 143             ip_dst_count = self.ip_to_int(self.dst.ip_list[max_idx]) - \
 144                 self.ip_to_int(self.dst.ip_list[flow_idx]) + 1
 145
 146             configs.append({
 147                 'count': current_flow_count,
 148                 'mac_src': self.mac,
 149                 'mac_dst': self.dst.mac if service_chain == ChainType.EXT
 150                 else self.vm_mac_list[chain_idx],
 151                 'ip_src_addr': self.ip_list[flow_idx],
 152                 'ip_src_addr_max': self.ip_list[max_idx],
 153                 'ip_src_count': ip_src_count,
 154                 'ip_dst_addr': self.dst.ip_list[flow_idx],
 155                 'ip_dst_addr_max': self.dst.ip_list[max_idx],
 156                 'ip_dst_count': ip_dst_count,
 157                 'ip_addrs_step': self.ip_addrs_step,
 158                 'udp_src_port': self.udp_src_port,
 159                 'udp_dst_port': self.udp_dst_port,
 160                 'mac_discovery_gw': self.gateway_ip_list[chain_idx],
 161                 'ip_src_tg_gw': self.tg_gateway_ip_list[chain_idx],
 162                 'ip_dst_tg_gw': self.dst.tg_gateway_ip_list[chain_idx],
 163                 'vlan_tag': self.vlan_tag if self.vlan_tagging else None
 164             })
 165
 166             flow_idx += current_flow_count
 167         return configs
 168
 169     @classmethod
 170     def expand_ip(cls, ip, step_ip, count):
 171         if step_ip == 'random':
 172             # Repeatable Random will used in the stream src/dst IP pairs, but we still need
 173             # to expand the IP based on the number of chains and flows configured. So we use
 174             # "0.0.0.1" as the step to have the exact IP flow ranges for every chain.
 175             step_ip = '0.0.0.1'
 176
 177         step_ip_in_int = cls.ip_to_int(step_ip)
 178         subnet = IPNetwork(ip)
 179         ip_list = []
 180         for _ in xrange(count):
 181             ip_list.append(subnet.ip.format())
 182             subnet = subnet.next(step_ip_in_int)
 183         return ip_list
 184
 185     @staticmethod
 186     def mac_to_int(mac):
 187         return int(mac.translate(None, ":.- "), 16)
 188
 189     @staticmethod
 190     def int_to_mac(i):
 191         mac = format(i, 'x').zfill(12)
 192         blocks = [mac[x:x + 2] for x in xrange(0, len(mac), 2)]
 193         return ':'.join(blocks)
 194
 195     @staticmethod
 196     def ip_to_int(addr):
 197         return struct.unpack("!I", socket.inet_aton(addr))[0]
 198
 199
 200 class RunningTrafficProfile(object):
 201     """Represents traffic configuration for currently running traffic profile."""
 202
 203     DEFAULT_IP_STEP = '0.0.0.1'
 204     DEFAULT_SRC_DST_IP_STEP = '0.0.0.1'
 205
 206     def __init__(self, config, generator_profile):
 207         generator_config = self.__match_generator_profile(config.traffic_generator,
 208                                                           generator_profile)
 209         self.generator_config = generator_config
 210         self.service_chain = config.service_chain
 211         self.service_chain_count = config.service_chain_count
 212         self.flow_count = config.flow_count
 213         self.host_name = generator_config.host_name
 214         self.name = generator_config.name
 215         self.tool = generator_config.tool
 216         self.cores = generator_config.get('cores', 1)
 217         self.ip_addrs_step = generator_config.ip_addrs_step or self.DEFAULT_SRC_DST_IP_STEP
 218         self.tg_gateway_ip_addrs_step = \
 219             generator_config.tg_gateway_ip_addrs_step or self.DEFAULT_IP_STEP
 220         self.gateway_ip_addrs_step = generator_config.gateway_ip_addrs_step or self.DEFAULT_IP_STEP
 221         self.gateway_ips = generator_config.gateway_ip_addrs
 222         self.ip = generator_config.ip
 223         self.intf_speed = bitmath.parse_string(generator_config.intf_speed.replace('ps', '')).bits
 224         self.vlan_tagging = config.vlan_tagging
 225         self.no_arp = config.no_arp
 226         self.src_device = None
 227         self.dst_device = None
 228         self.vm_mac_list = None
 229         self.__prep_interfaces(generator_config)
 230
 231     def to_json(self):
 232         return dict(self.generator_config)
 233
 234     def set_vm_mac_list(self, vm_mac_list):
 235         self.src_device.set_vm_mac_list(vm_mac_list[0])
 236         self.dst_device.set_vm_mac_list(vm_mac_list[1])
 237
 238     @staticmethod
 239     def __match_generator_profile(traffic_generator, generator_profile):
 240         generator_config = AttrDict(traffic_generator)
 241         generator_config.pop('default_profile')
 242         generator_config.pop('generator_profile')
 243         matching_profile = filter(lambda profile: profile.name == generator_profile,
 244                                   traffic_generator.generator_profile)
 245         if len(matching_profile) != 1:
 246             raise Exception('Traffic generator profile not found: ' + generator_profile)
 247
 248         generator_config.update(matching_profile[0])
 249
 250         return generator_config
 251
 252     def __prep_interfaces(self, generator_config):
 253         src_config = {
 254             'chain_count': self.service_chain_count,
 255             'flow_count': self.flow_count / 2,
 256             'ip': generator_config.ip_addrs[0],
 257             'ip_addrs_step': self.ip_addrs_step,
 258             'gateway_ip': self.gateway_ips[0],
 259             'gateway_ip_addrs_step': self.gateway_ip_addrs_step,
 260             'tg_gateway_ip': generator_config.tg_gateway_ip_addrs[0],
 261             'tg_gateway_ip_addrs_step': self.tg_gateway_ip_addrs_step,
 262             'udp_src_port': generator_config.udp_src_port,
 263             'udp_dst_port': generator_config.udp_dst_port,
 264             'vlan_tagging': self.vlan_tagging
 265         }
 266         dst_config = {
 267             'chain_count': self.service_chain_count,
 268             'flow_count': self.flow_count / 2,
 269             'ip': generator_config.ip_addrs[1],
 270             'ip_addrs_step': self.ip_addrs_step,
 271             'gateway_ip': self.gateway_ips[1],
 272             'gateway_ip_addrs_step': self.gateway_ip_addrs_step,
 273             'tg_gateway_ip': generator_config.tg_gateway_ip_addrs[1],
 274             'tg_gateway_ip_addrs_step': self.tg_gateway_ip_addrs_step,
 275             'udp_src_port': generator_config.udp_src_port,
 276             'udp_dst_port': generator_config.udp_dst_port,
 277             'vlan_tagging': self.vlan_tagging
 278         }
 279
 280         self.src_device = Device(**dict(src_config, **generator_config.interfaces[0]))
 281         self.dst_device = Device(**dict(dst_config, **generator_config.interfaces[1]))
 282         self.src_device.set_destination(self.dst_device)
 283         self.dst_device.set_destination(self.src_device)
 284
 285         if self.service_chain == ChainType.EXT and not self.no_arp \
 286                 and not self.__are_unique(self.src_device.ip_list, self.dst_device.ip_list):
 287             raise Exception('Computed IP addresses are not unique, choose different base. '
 288                             'Start IPs: {start}. End IPs: {end}'
 289                             .format(start=self.src_device.ip_list,
 290                                     end=self.dst_device.ip_list))
 291
 292     def __are_unique(self, list1, list2):
 293         return set(list1).isdisjoint(set(list2))
 294
 295     @property
 296     def devices(self):
 297         return [self.src_device, self.dst_device]
 298
 299     @property
 300     def vlans(self):
 301         return [self.src_device.vtep_vlan, self.dst_device.vtep_vlan]
 302
 303     @property
 304     def ports(self):
 305         return [self.src_device.port, self.dst_device.port]
 306
 307     @property
 308     def switch_ports(self):
 309         return [self.src_device.switch_port, self.dst_device.switch_port]
 310
 311     @property
 312     def pcis(self):
 313         return [self.src_device.pci, self.dst_device.pci]
 314
 315
 316 class TrafficGeneratorFactory(object):
 317
 318     def __init__(self, config):
 319         self.config = config
 320
 321     def get_tool(self):
 322         return self.config.generator_config.tool
 323
 324     def get_generator_client(self):
 325         tool = self.get_tool().lower()
 326         if tool == 'trex':
 327             from traffic_gen import trex
 328             return trex.TRex(self.config)
 329         elif tool == 'dummy':
 330             from traffic_gen import dummy
 331             return dummy.DummyTG(self.config)
 332         else:
 333             return None
 334
 335     def list_generator_profile(self):
 336         return [profile.name for profile in self.config.traffic_generator.generator_profile]
 337
 338     def get_generator_config(self, generator_profile):
 339         return RunningTrafficProfile(self.config, generator_profile)
 340
 341     def get_matching_profile(self, traffic_profile_name):
 342         matching_profile = filter(lambda profile: profile.name == traffic_profile_name,
 343                                   self.config.traffic_profile)
 344
 345         if len(matching_profile) > 1:
 346             raise Exception('Multiple traffic profiles with the same name found.')
 347         elif len(matching_profile) == 0:
 348             raise Exception('No traffic profile found.')
 349
 350         return matching_profile[0]
 351
 352     def get_frame_sizes(self, traffic_profile):
 353         matching_profile = self.get_matching_profile(traffic_profile)
 354         return matching_profile.l2frame_size
 355
 356
 357 class TrafficClient(object):
 358
 359     PORTS = [0, 1]
 360
 361     def __init__(self, config, notifier=None):
 362         generator_factory = TrafficGeneratorFactory(config)
 363         self.gen = generator_factory.get_generator_client()
 364         self.tool = generator_factory.get_tool()
 365         self.config = config
 366         self.notifier = notifier
 367         self.interval_collector = None
 368         self.iteration_collector = None
 369         self.runner = TrafficRunner(self, self.config.duration_sec, self.config.interval_sec)
 370         if self.gen is None:
 371             raise TrafficClientException('%s is not a supported traffic generator' % self.tool)
 372
 373         self.run_config = {
 374             'l2frame_size': None,
 375             'duration_sec': self.config.duration_sec,
 376             'bidirectional': True,
 377             'rates': None
 378         }
 379         self.current_total_rate = {'rate_percent': '10'}
 380         if self.config.single_run:
 381             self.current_total_rate = utils.parse_rate_str(self.config.rate)
 382
 383     def set_macs(self):
 384         for mac, device in zip(self.gen.get_macs(), self.config.generator_config.devices):
 385             device.set_mac(mac)
 386
 387     def start_traffic_generator(self):
 388         self.gen.init()
 389         self.gen.connect()
 390
 391     def setup(self):
 392         self.gen.set_mode()
 393         self.gen.config_interface()
 394         self.gen.clear_stats()
 395
 396     def get_version(self):
 397         return self.gen.get_version()
 398
 399     def ensure_end_to_end(self):
 400         """
 401         Ensure traffic generator receives packets it has transmitted.
 402         This ensures end to end connectivity and also waits until VMs are ready to forward packets.
 403
 404         At this point all VMs are in active state, but forwarding does not have to work.
 405         Small amount of traffic is sent to every chain. Then total of sent and received packets
 406         is compared. If ratio between received and transmitted packets is higher than (N-1)/N,
 407         N being number of chains, traffic flows through every chain and real measurements can be
 408         performed.
 409
 410         Example:
 411             PVP chain (1 VM per chain)
 412             N = 10 (number of chains)
 413             threshold = (N-1)/N = 9/10 = 0.9 (acceptable ratio ensuring working conditions)
 414             if total_received/total_sent > 0.9, traffic is flowing to more than 9 VMs meaning
 415             all 10 VMs are in operational state.
 416         """
 417         LOG.info('Starting traffic generator to ensure end-to-end connectivity')
 418         rate_pps = {'rate_pps': str(self.config.service_chain_count * 100)}
 419         self.gen.create_traffic('64', [rate_pps, rate_pps], bidirectional=True, latency=False)
 420
 421         # ensures enough traffic is coming back
 422         threshold = (self.config.service_chain_count - 1) / float(self.config.service_chain_count)
 423         retry_count = (self.config.check_traffic_time_sec +
 424                        self.config.generic_poll_sec - 1) / self.config.generic_poll_sec
 425         for it in xrange(retry_count):
 426             self.gen.clear_stats()
 427             self.gen.start_traffic()
 428             LOG.info('Waiting for packets to be received back... ({} / {})'.format(it + 1,
 429                      retry_count))
 430             time.sleep(self.config.generic_poll_sec)
 431             self.gen.stop_traffic()
 432             stats = self.gen.get_stats()
 433
 434             # compute total sent and received traffic on both ports
 435             total_rx = 0
 436             total_tx = 0
 437             for port in self.PORTS:
 438                 total_rx += float(stats[port]['rx'].get('total_pkts', 0))
 439                 total_tx += float(stats[port]['tx'].get('total_pkts', 0))
 440
 441             # how much of traffic came back
 442             ratio = total_rx / total_tx if total_tx else 0
 443
 444             if ratio > threshold:
 445                 self.gen.clear_stats()
 446                 self.gen.clear_streamblock()
 447                 LOG.info('End-to-end connectivity ensured')
 448                 return
 449
 450             time.sleep(self.config.generic_poll_sec)
 451
 452         raise TrafficClientException('End-to-end connectivity cannot be ensured')
 453
 454     def ensure_arp_successful(self):
 455         if not self.gen.resolve_arp():
 456             raise TrafficClientException('ARP cannot be resolved')
 457
 458     def set_traffic(self, frame_size, bidirectional):
 459         self.run_config['bidirectional'] = bidirectional
 460         self.run_config['l2frame_size'] = frame_size
 461         self.run_config['rates'] = [self.get_per_direction_rate()]
 462         if bidirectional:
 463             self.run_config['rates'].append(self.get_per_direction_rate())
 464         else:
 465             unidir_reverse_pps = int(self.config.unidir_reverse_traffic_pps)
 466             if unidir_reverse_pps > 0:
 467                 self.run_config['rates'].append({'rate_pps': str(unidir_reverse_pps)})
 468
 469         self.gen.clear_streamblock()
 470         self.gen.create_traffic(frame_size, self.run_config['rates'], bidirectional, latency=True)
 471
 472     def modify_load(self, load):
 473         self.current_total_rate = {'rate_percent': str(load)}
 474         rate_per_direction = self.get_per_direction_rate()
 475
 476         self.gen.modify_rate(rate_per_direction, False)
 477         self.run_config['rates'][0] = rate_per_direction
 478         if self.run_config['bidirectional']:
 479             self.gen.modify_rate(rate_per_direction, True)
 480             self.run_config['rates'][1] = rate_per_direction
 481
 482     def get_ndr_and_pdr(self):
 483         dst = 'Bidirectional' if self.run_config['bidirectional'] else 'Unidirectional'
 484         targets = {}
 485         if self.config.ndr_run:
 486             LOG.info('*** Searching NDR for %s (%s)...', self.run_config['l2frame_size'], dst)
 487             targets['ndr'] = self.config.measurement.NDR
 488         if self.config.pdr_run:
 489             LOG.info('*** Searching PDR for %s (%s)...', self.run_config['l2frame_size'], dst)
 490             targets['pdr'] = self.config.measurement.PDR
 491
 492         self.run_config['start_time'] = time.time()
 493         self.interval_collector = IntervalCollector(self.run_config['start_time'])
 494         self.interval_collector.attach_notifier(self.notifier)
 495         self.iteration_collector = IterationCollector(self.run_config['start_time'])
 496         results = {}
 497         self.__range_search(0.0, 200.0, targets, results)
 498
 499         results['iteration_stats'] = {
 500             'ndr_pdr': self.iteration_collector.get()
 501         }
 502
 503         if self.config.ndr_run:
 504             LOG.info('NDR load: %s', results['ndr']['rate_percent'])
 505             results['ndr']['time_taken_sec'] = \
 506                 results['ndr']['timestamp_sec'] - self.run_config['start_time']
 507             if self.config.pdr_run:
 508                 LOG.info('PDR load: %s', results['pdr']['rate_percent'])
 509                 results['pdr']['time_taken_sec'] = \
 510                     results['pdr']['timestamp_sec'] - results['ndr']['timestamp_sec']
 511         else:
 512             LOG.info('PDR load: %s', results['pdr']['rate_percent'])
 513             results['pdr']['time_taken_sec'] = \
 514                 results['pdr']['timestamp_sec'] - self.run_config['start_time']
 515         return results
 516
 517     def __get_dropped_rate(self, result):
 518         dropped_pkts = result['rx']['dropped_pkts']
 519         total_pkts = result['tx']['total_pkts']
 520         if not total_pkts:
 521             return float('inf')
 522         else:
 523             return float(dropped_pkts) / total_pkts * 100
 524
 525     def get_stats(self):
 526         stats = self.gen.get_stats()
 527         retDict = {'total_tx_rate': stats['total_tx_rate']}
 528         for port in self.PORTS:
 529             retDict[port] = {'tx': {}, 'rx': {}}
 530
 531         tx_keys = ['total_pkts', 'total_pkt_bytes', 'pkt_rate', 'pkt_bit_rate']
 532         rx_keys = tx_keys + ['dropped_pkts']
 533
 534         for port in self.PORTS:
 535             for key in tx_keys:
 536                 retDict[port]['tx'][key] = int(stats[port]['tx'][key])
 537             for key in rx_keys:
 538                 try:
 539                     retDict[port]['rx'][key] = int(stats[port]['rx'][key])
 540                 except ValueError:
 541                     retDict[port]['rx'][key] = 0
 542             retDict[port]['rx']['avg_delay_usec'] = cast_integer(
 543                 stats[port]['rx']['avg_delay_usec'])
 544             retDict[port]['rx']['min_delay_usec'] = cast_integer(
 545                 stats[port]['rx']['min_delay_usec'])
 546             retDict[port]['rx']['max_delay_usec'] = cast_integer(
 547                 stats[port]['rx']['max_delay_usec'])
 548             retDict[port]['drop_rate_percent'] = self.__get_dropped_rate(retDict[port])
 549
 550         ports = sorted(retDict.keys())
 551         if self.run_config['bidirectional']:
 552             retDict['overall'] = {'tx': {}, 'rx': {}}
 553             for key in tx_keys:
 554                 retDict['overall']['tx'][key] = \
 555                     retDict[ports[0]]['tx'][key] + retDict[ports[1]]['tx'][key]
 556             for key in rx_keys:
 557                 retDict['overall']['rx'][key] = \
 558                     retDict[ports[0]]['rx'][key] + retDict[ports[1]]['rx'][key]
 559             total_pkts = [retDict[ports[0]]['rx']['total_pkts'],
 560                           retDict[ports[1]]['rx']['total_pkts']]
 561             avg_delays = [retDict[ports[0]]['rx']['avg_delay_usec'],
 562                           retDict[ports[1]]['rx']['avg_delay_usec']]
 563             max_delays = [retDict[ports[0]]['rx']['max_delay_usec'],
 564                           retDict[ports[1]]['rx']['max_delay_usec']]
 565             min_delays = [retDict[ports[0]]['rx']['min_delay_usec'],
 566                           retDict[ports[1]]['rx']['min_delay_usec']]
 567             retDict['overall']['rx']['avg_delay_usec'] = utils.weighted_avg(total_pkts, avg_delays)
 568             retDict['overall']['rx']['min_delay_usec'] = min(min_delays)
 569             retDict['overall']['rx']['max_delay_usec'] = max(max_delays)
 570             for key in ['pkt_bit_rate', 'pkt_rate']:
 571                 for dirc in ['tx', 'rx']:
 572                     retDict['overall'][dirc][key] /= 2.0
 573         else:
 574             retDict['overall'] = retDict[ports[0]]
 575         retDict['overall']['drop_rate_percent'] = self.__get_dropped_rate(retDict['overall'])
 576         return retDict
 577
 578     def __convert_rates(self, rate):
 579         return utils.convert_rates(self.run_config['l2frame_size'],
 580                                    rate,
 581                                    self.config.generator_config.intf_speed)
 582
 583     def __ndr_pdr_found(self, tag, load):
 584         rates = self.__convert_rates({'rate_percent': load})
 585         self.iteration_collector.add_ndr_pdr(tag, rates['rate_pps'])
 586         last_stats = self.iteration_collector.peek()
 587         self.interval_collector.add_ndr_pdr(tag, last_stats)
 588
 589     def __format_output_stats(self, stats):
 590         for key in (self.PORTS + ['overall']):
 591             interface = stats[key]
 592             stats[key] = {
 593                 'tx_pkts': interface['tx']['total_pkts'],
 594                 'rx_pkts': interface['rx']['total_pkts'],
 595                 'drop_percentage': interface['drop_rate_percent'],
 596                 'drop_pct': interface['rx']['dropped_pkts'],
 597                 'avg_delay_usec': interface['rx']['avg_delay_usec'],
 598                 'max_delay_usec': interface['rx']['max_delay_usec'],
 599                 'min_delay_usec': interface['rx']['min_delay_usec'],
 600             }
 601
 602         return stats
 603
 604     def __targets_found(self, rate, targets, results):
 605         for tag, target in targets.iteritems():
 606             LOG.info('Found {} ({}) load: {}'.format(tag, target, rate))
 607             self.__ndr_pdr_found(tag, rate)
 608             results[tag]['timestamp_sec'] = time.time()
 609
 610     def __range_search(self, left, right, targets, results):
 611         '''Perform a binary search for a list of targets inside a [left..right] range or rate
 612
 613         left    the left side of the range to search as a % the line rate (100 = 100% line rate)
 614                 indicating the rate to send on each interface
 615         right   the right side of the range to search as a % of line rate
 616                 indicating the rate to send on each interface
 617         targets a dict of drop rates to search (0.1 = 0.1%), indexed by the DR name or "tag"
 618                 ('ndr', 'pdr')
 619         results a dict to store results
 620         '''
 621         if len(targets) == 0:
 622             return
 623         LOG.info('Range search [{} .. {}] targets: {}'.format(left, right, targets))
 624
 625         # Terminate search when gap is less than load epsilon
 626         if right - left < self.config.measurement.load_epsilon:
 627             self.__targets_found(left, targets, results)
 628             return
 629
 630         # Obtain the average drop rate in for middle load
 631         middle = (left + right) / 2.0
 632         stats, rates = self.__run_search_iteration(middle)
 633
 634         # Split target dicts based on the avg drop rate
 635         left_targets = {}
 636         right_targets = {}
 637         for tag, target in targets.iteritems():
 638             if stats['overall']['drop_rate_percent'] <= target:
 639                 # record the best possible rate found for this target
 640                 results[tag] = rates
 641                 results[tag].update({
 642                     'load_percent_per_direction': middle,
 643                     'stats': self.__format_output_stats(dict(stats)),
 644                     'timestamp_sec': None
 645                 })
 646                 right_targets[tag] = target
 647             else:
 648                 # initialize to 0 all fields of result for
 649                 # the worst case scenario of the binary search (if ndr/pdr is not found)
 650                 if tag not in results:
 651                     results[tag] = dict.fromkeys(rates, 0)
 652                     empty_stats = self.__format_output_stats(dict(stats))
 653                     for key in empty_stats:
 654                         if isinstance(empty_stats[key], dict):
 655                             empty_stats[key] = dict.fromkeys(empty_stats[key], 0)
 656                         else:
 657                             empty_stats[key] = 0
 658                     results[tag].update({
 659                         'load_percent_per_direction': 0,
 660                         'stats': empty_stats,
 661                         'timestamp_sec': None
 662                     })
 663                 left_targets[tag] = target
 664
 665         # search lower half
 666         self.__range_search(left, middle, left_targets, results)
 667
 668         # search upper half only if the upper rate does not exceed
 669         # 100%, this only happens when the first search at 100%
 670         # yields a DR that is < target DR
 671         if middle >= 100:
 672             self.__targets_found(100, right_targets, results)
 673         else:
 674             self.__range_search(middle, right, right_targets, results)
 675
 676     def __run_search_iteration(self, rate):
 677         # set load
 678         self.modify_load(rate)
 679
 680         # poll interval stats and collect them
 681         for stats in self.run_traffic():
 682             self.interval_collector.add(stats)
 683             time_elapsed_ratio = self.runner.time_elapsed() / self.run_config['duration_sec']
 684             if time_elapsed_ratio >= 1:
 685                 self.cancel_traffic()
 686         self.interval_collector.reset()
 687
 688         # get stats from the run
 689         stats = self.runner.client.get_stats()
 690         current_traffic_config = self.get_traffic_config()
 691         warning = self.compare_tx_rates(current_traffic_config['direction-total']['rate_pps'],
 692                                         stats['total_tx_rate'])
 693         if warning is not None:
 694             stats['warning'] = warning
 695
 696         # save reliable stats from whole iteration
 697         self.iteration_collector.add(stats, current_traffic_config['direction-total']['rate_pps'])
 698         LOG.info('Average drop rate: {}'.format(stats['overall']['drop_rate_percent']))
 699
 700         return stats, current_traffic_config['direction-total']
 701
 702     @staticmethod
 703     def log_stats(stats):
 704         report = {
 705             'datetime': str(datetime.now()),
 706             'tx_packets': stats['overall']['tx']['total_pkts'],
 707             'rx_packets': stats['overall']['rx']['total_pkts'],
 708             'drop_packets': stats['overall']['rx']['dropped_pkts'],
 709             'drop_rate_percent': stats['overall']['drop_rate_percent']
 710         }
 711         LOG.info('TX: %(tx_packets)d; '
 712                  'RX: %(rx_packets)d; '
 713                  'Dropped: %(drop_packets)d; '
 714                  'Drop rate: %(drop_rate_percent).4f%%',
 715                  report)
 716
 717     def run_traffic(self):
 718         stats = self.runner.run()
 719         while self.runner.is_running:
 720             self.log_stats(stats)
 721             yield stats
 722             stats = self.runner.poll_stats()
 723             if stats is None:
 724                 return
 725         self.log_stats(stats)
 726         LOG.info('Drop rate: {}'.format(stats['overall']['drop_rate_percent']))
 727         yield stats
 728
 729     def cancel_traffic(self):
 730         self.runner.stop()
 731
 732     def get_interface(self, port_index):
 733         port = self.gen.port_handle[port_index]
 734         tx, rx = 0, 0
 735         if not self.config.no_traffic:
 736             stats = self.get_stats()
 737             if port in stats:
 738                 tx, rx = int(stats[port]['tx']['total_pkts']), int(stats[port]['rx']['total_pkts'])
 739         return Interface('traffic-generator', self.tool.lower(), tx, rx)
 740
 741     def get_traffic_config(self):
 742         config = {}
 743         load_total = 0.0
 744         bps_total = 0.0
 745         pps_total = 0.0
 746         for idx, rate in enumerate(self.run_config['rates']):
 747             key = 'direction-forward' if idx == 0 else 'direction-reverse'
 748             config[key] = {
 749                 'l2frame_size': self.run_config['l2frame_size'],
 750                 'duration_sec': self.run_config['duration_sec']
 751             }
 752             config[key].update(rate)
 753             config[key].update(self.__convert_rates(rate))
 754             load_total += float(config[key]['rate_percent'])
 755             bps_total += float(config[key]['rate_bps'])
 756             pps_total += float(config[key]['rate_pps'])
 757         config['direction-total'] = dict(config['direction-forward'])
 758         config['direction-total'].update({
 759             'rate_percent': load_total,
 760             'rate_pps': cast_integer(pps_total),
 761             'rate_bps': bps_total
 762         })
 763
 764         return config
 765
 766     def get_run_config(self, results):
 767         """Returns configuration which was used for the last run."""
 768         r = {}
 769         for idx, key in enumerate(["direction-forward", "direction-reverse"]):
 770             tx_rate = results["stats"][idx]["tx"]["total_pkts"] / self.config.duration_sec
 771             rx_rate = results["stats"][idx]["rx"]["total_pkts"] / self.config.duration_sec
 772             r[key] = {
 773                 "orig": self.__convert_rates(self.run_config['rates'][idx]),
 774                 "tx": self.__convert_rates({'rate_pps': tx_rate}),
 775                 "rx": self.__convert_rates({'rate_pps': rx_rate})
 776             }
 777
 778         total = {}
 779         for direction in ['orig', 'tx', 'rx']:
 780             total[direction] = {}
 781             for unit in ['rate_percent', 'rate_bps', 'rate_pps']:
 782                 total[direction][unit] = sum(map(lambda x: float(x[direction][unit]), r.values()))
 783
 784         r['direction-total'] = total
 785         return r
 786
 787     @staticmethod
 788     def compare_tx_rates(required, actual):
 789         threshold = 0.9
 790         are_different = False
 791         try:
 792             if float(actual) / required < threshold:
 793                 are_different = True
 794         except ZeroDivisionError:
 795             are_different = True
 796
 797         if are_different:
 798             msg = "WARNING: There is a significant difference between requested TX rate ({r}) " \
 799                   "and actual TX rate ({a}). The traffic generator may not have sufficient CPU " \
 800                   "to achieve the requested TX rate.".format(r=required, a=actual)
 801             LOG.info(msg)
 802             return msg
 803
 804         return None
 805
 806     def get_per_direction_rate(self):
 807         divisor = 2 if self.run_config['bidirectional'] else 1
 808         if 'rate_percent' in self.current_total_rate:
 809             # don't split rate if it's percentage
 810             divisor = 1
 811
 812         return utils.divide_rate(self.current_total_rate, divisor)
 813
 814     def close(self):
 815         try:
 816             self.gen.stop_traffic()
 817         except Exception:
 818             pass
 819         self.gen.clear_stats()
 820         self.gen.cleanup()