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