Merge pull request snabbco#37 from andywingo/transient

Add transient program

src/apps/basic/basic_apps.lua

@@ -204,6 +204,10 @@ function RateLimitedRepeater:new (arg)
    return setmetatable(o, {__index=RateLimitedRepeater})
 end
 
+function RateLimitedRepeater:set_rate (byte_rate)
+   self.rate = math.max(byte_rate, 0)
+end
+
 function RateLimitedRepeater:push ()
    local i, o = self.input.input, self.output.output
    for _ = 1, link.nreadable(i) do
@@ -222,7 +226,7 @@ function RateLimitedRepeater:push ()
    end
 
    local npackets = #self.packets
-   if npackets > 0 then
+   if npackets > 0 and self.rate > 0 then
       for _ = 1, link.nwritable(o) do
          local p = self.packets[self.index]
          if p.length > self.bucket_content then break end

src/program/transient/README

@@ -0,0 +1,30 @@
+Usage: transient [OPTIONS] <PCAP-FILE> <NAME> <PCI> [<PCAP-FILE> <NAME> <PCI>]...
+
+  -b BITRATE, --bitrate BITRATE
+                             Peak at BITRATE bits/second.
+  -s STEP, --step STEP
+                             Increase bitrate in increments of STEP bits/second.
+  -D DURATION, --duration DURATION
+                             Linger on each step for DURATION seconds.
+  -p period, --period PERIOD
+                             Measure each PERIOD seconds.
+  -h, --help
+                             Print usage information.
+
+Transmit packets from the PCAP-FILE packet captures to the corresponding
+PCI network adaptors.  Start at zero bits per second, ramping up to
+BITRATE bits per second in increments of STEP bits per second, lingering
+at each step for DURATION seconds.  The default is to ramp to 10 Gbps in
+increments of 1 Gbps, lingering for 5 seconds at each step.  Once the
+peak bitrate is reached, back down the same way until 0 is reached, and
+end the test.
+
+Packets received on the network interfaces are counted and recorded, and
+the corresponding incoming and outgoing packet rates are written to
+standard output in CSV format, suitable for passing to a graphing
+program.  The NAME values are used to label the columns.  Measurements
+will be taken each PERIOD seconds, which defaults to 1.
+
+Examples:
+  transient cap1.pcap tx 01:00.0
+  transient -d 1 -b 5e9 -s 0.2e9 cap1.pcap "Name 1" 01:00.0 cap2 "Name 2" 01:00.1

src/program/transient/README.inc

@@ -0,0 +1 @@
+README

src/program/transient/transient.lua

@@ -0,0 +1,201 @@
+module(..., package.seeall)
+
+local engine = require("core.app")
+local counter = require("core.counter")
+local config = require("core.config")
+local timer = require("core.timer")
+local pci = require("lib.hardware.pci")
+local Intel82599 = require("apps.intel.intel_app").Intel82599
+local basic_apps = require("apps.basic.basic_apps")
+local main = require("core.main")
+local PcapReader = require("apps.pcap.pcap").PcapReader
+local lib = require("core.lib")
+local ffi = require("ffi")
+
+function show_usage(code)
+   print(require("program.transient.README_inc"))
+   main.exit(code)
+end
+
+function find_devices(pattern)
+   if #pci.devices == 0 then pci.scan_devices() end
+   local ret = {}
+   for _,device in ipairs(pci.devices) do
+      if (device.usable and device.driver == 'apps.intel.intel_app' and
+          device.pciaddress:match(pattern)) then
+         table.insert(ret, device.pciaddress)
+      end
+   end
+   return ret
+end
+
+function find_device(pattern)
+   local devices = find_devices(pattern)
+   if #devices == 0 then
+      error('no devices matched pattern "'..pattern..'"')
+   elseif #devices == 1 then
+      return devices[1]
+   else
+      local devices_str = table.concat(devices, ' ')
+      error('multiple devices matched pattern "'..pattern..'":'..devices_str)
+   end
+end
+
+function parse_args(args)
+   local handlers = {}
+   local opts = { bitrate = 10e9, duration = 5, period = 1 }
+   function handlers.b(arg)
+      opts.bitrate = assert(tonumber(arg), 'bitrate must be a number')
+   end
+   function handlers.s(arg)
+      opts.step = assert(tonumber(arg), 'step must be a number')
+   end
+   function handlers.D(arg)
+      opts.duration = assert(tonumber(arg), 'duration must be a number')
+   end
+   function handlers.p(arg)
+      opts.period = assert(tonumber(arg), 'period must be a number')
+   end
+   function handlers.h() show_usage(0) end
+   args = lib.dogetopt(args, handlers, "hb:s:D:p:",
+                       { bitrate="b", step="s", duration="D", period="p",
+                         help="h" })
+   if not opts.step then opts.step = opts.bitrate / 10 end
+   assert(opts.bitrate > 0, 'bitrate must be positive')
+   assert(opts.step > 0, 'step must be positive')
+   assert(opts.duration > 0, 'duration must be positive')
+   assert(opts.period > 0, 'period must be positive')
+   if #args == 0 or #args % 3 ~= 0 then show_usage(1) end
+   local streams = {}
+   for i=1,#args,3 do
+      local capture_file, name, pattern = args[i], args[i+1], args[i+2]
+      local nic = {
+         capture_file = capture_file,
+         name = name,
+         id = name:gsub('[^%w]', '_'),
+         pci_addr = find_device(pattern)
+      }
+      table.insert(streams, nic)
+   end
+   return opts, streams
+end
+
+-- This ramps the repeater up from 0 Gbps to the max bitrate, lingering
+-- at the top only for one period, then comes back down in the same way.
+-- We can add more of these for different workloads.
+function adjust_rate(opts, streams)
+   local count = math.ceil(opts.bitrate / opts.step)
+   return function()
+      local byte_rate = (opts.bitrate - math.abs(count) * opts.step) / 8
+      for _,stream in ipairs(streams) do
+         local app = engine.app_table[stream.repeater_id]
+         app:set_rate(byte_rate)
+      end
+      count = count - 1
+   end
+end
+
+local function fout(tmpl, ...)
+   io.write(tmpl:format(...))
+end
+
+local function print_headings(streams)
+   fout('Time (s)')
+   for _,stream in ipairs(streams) do
+      fout(',%s TX (MPPS),%s TX (Gbps),%s RX (MPPS),%s RX (Gbps)',
+           stream.name, stream.name, stream.name, stream.name)
+   end
+   fout('\n')
+   io.flush()
+end
+
+local function print_sample(sample)
+   fout('%f', sample.elapsed)
+   for _,stat in ipairs(sample) do
+      fout(',%f,%f,%f,%f', stat.sent_mpps, stat.sent_gbps,
+           stat.received_mpps, stat.received_gbps)
+   end
+   fout('\n')
+   io.flush()
+end
+
+local statistics = {}
+function record_stats(opts, streams)
+   local count = 0
+   local elapsed = 0
+   local prev_stats = {}
+   print_headings(streams)
+   for _,stream in ipairs(streams) do
+      prev_stats[stream.id] = {
+         sent_bytes = 0,
+         sent_packets = 0,
+         received_bytes = 0,
+         received_packets = 0
+      }
+   end
+   return function()
+      elapsed = elapsed + opts.period
+      count = count + 1
+      local sample = { elapsed = elapsed, count = count }
+      for _,stream in ipairs(streams) do
+         local app = engine.app_table[stream.nic_id]
+         local prev = prev_stats[stream.id]
+         local sent, received = app.input.rx, app.output.tx
+         local cur = {
+            sent_bytes = counter.read(sent.stats.txbytes),
+            sent_packets = counter.read(sent.stats.txpackets),
+            received_bytes = counter.read(received.stats.txbytes),
+            received_packets = counter.read(received.stats.txpackets)
+         }
+         prev_stats[stream.id] = cur
+         local diff = {
+            sent_bytes = tonumber(cur.sent_bytes - prev.sent_bytes),
+            sent_packets = tonumber(cur.sent_packets - prev.sent_packets),
+            received_bytes = tonumber(cur.received_bytes - prev.received_bytes),
+            received_packets = tonumber(cur.received_packets - prev.received_packets)
+         }
+         local stat = {
+            name = stream.name,
+            sent_mpps = diff.sent_packets / opts.period / 1e6,
+            sent_gbps = diff.sent_bytes * 8 / opts.period / 1e9,
+            received_mpps = diff.received_packets / opts.period / 1e6,
+            received_gbps = diff.received_bytes * 8 / opts.period / 1e9
+         }
+         table.insert(sample, stat)
+      end
+      table.insert(statistics, sample)
+      print_sample(sample)
+   end
+end
+
+function run(args)
+   local opts, streams = parse_args(args)
+   local c = config.new()
+   for _,stream in ipairs(streams) do
+      stream.pcap_id = 'pcap_'..stream.id
+      stream.repeater_id = 'repeater_'..stream.id
+      stream.nic_id = 'nic_'..stream.id
+      stream.rx_sink_id = 'rx_sink_'..stream.id
+
+      config.app(c, stream.pcap_id, PcapReader, stream.capture_file)
+      config.app(c, stream.repeater_id, basic_apps.RateLimitedRepeater, {})
+      config.app(c, stream.nic_id, Intel82599, { pciaddr = stream.pci_addr })
+      config.app(c, stream.rx_sink_id, basic_apps.Sink)
+
+      config.link(c, stream.pcap_id..".output -> "..stream.repeater_id..".input")
+      config.link(c, stream.repeater_id..".output -> "..stream.nic_id..".rx")
+
+      config.link(c, stream.nic_id..".tx -> "..stream.rx_sink_id..".input")
+   end
+   engine.configure(c)
+
+   local rate_adjuster = adjust_rate(opts, streams)
+   -- Initialize rates before anything happens.
+   rate_adjuster()
+   timer.activate(timer.new("adjust_rate", rate_adjuster,
+                            opts.duration * 1e9, 'repeating'))
+   timer.activate(timer.new("record_stats", record_stats(opts, streams),
+                            opts.period * 1e9, 'repeating'))
+
+   engine.main({duration=opts.duration*((opts.bitrate/opts.step)*2+1)})
+end