Merge PR #1478 (Release 2022-06: Parmigiano) into master

.version

@@ -1 +1 @@
-2022.01
+2022.07

default.nix

@@ -19,11 +19,7 @@ stdenv.mkDerivation rec {
 
   patchPhase = ''
     patchShebangs .
-
-    # some hardcodeism
-    for f in $(find src/program/snabbnfv/ -type f); do
-      substituteInPlace $f --replace "/bin/bash" "${bash}/bin/bash"
-    done
+    
   '' + lib.optionalString supportOpenstack ''
     # We need a way to pass $PATH to the scripts
     sed -i '2iexport PATH=${git}/bin:${mariadb}/bin:${which}/bin:${procps}/bin:${coreutils}/bin' src/program/snabbnfv/neutron_sync_master/neutron_sync_master.sh.inc

lib/ljsyscall/syscall/linux/syscalls.lua

@@ -814,17 +814,22 @@ function S.setegid(egid) return S.setresgid(-1, egid, -1) end
 -- note currently all returned as strings, may want to list which should be numbers
 function S.sysctl(name, new)
   name = "/proc/sys/" .. name:gsub("%.", "/")
-  local flag = c.O.RDONLY
-  if new then flag = c.O.RDWR end
-  local fd, err = S.open(name, flag)
+  local fd, err = S.open(name, c.O.RDONLY)
   if not fd then return nil, err end
   local len = 1024
   local old, err = S.read(fd, nil, len)
   if not old then return nil, err end
   old = old:sub(1, #old - 1) -- remove trailing newline
+  local ok, err = S.close(fd)
+  if not ok then return nil, err end
   if not new then return old end
+  -- Reopen fd because we want to write at pos 0
+  local fd, err = S.open(name, c.O.WRONLY)
+  if not fd then return nil, err end
   local ok, err = S.write(fd, new)
   if not ok then return nil, err end
+  local ok, err = S.close(fd)
+  if not ok then return nil, err end
   return old
 end
 

lib/luajit/.gitignore

@@ -9,3 +9,12 @@
 *.dmp
 *.swp
 .tags
+*.dwo
+/src/lj_bcdef.h
+/src/lj_ffdef.h
+/src/lj_folddef.h
+/src/lj_libdef.h
+/src/lj_recdef.h
+/src/lj_vm.S
+/src/raptorjit
+/src/host/buildvm_arch.h

src/README.md

@@ -55,7 +55,7 @@ these in a desired way using *links* and finally pass the resulting app
 network on to the Snabb engine. The engine's job is to:
 
  * Pump traffic through the app network
- * Keep the app network running (e.g. restart failed apps)
+ * Apply, and inform apps of configuration and link changes
  * Report on the network status
 
 
@@ -117,11 +117,24 @@ will be used to validate the app’s arg when it is configured using
 `config.app`.
 
 
-— Method **myapp:link**
+— Method **myapp:link** *dir* *name*
 
-*Optional*. Called any time the app’s links may have been changed (including on
-start-up). Guaranteed to be called before `pull` and `push` are called with new
-links.
+*Optional*. Called during `engine.configure()` when a link of the app is
+added. Unless `unlink` is specified this method is also called when a link
+is removed.
+Guaranteed to be called before `pull` and `push` are called with new links. 
+
+*Dir* is either `'input'` or `'output'`, and *name* is the string name
+of the link. I.e., the added link can be accessed at `self[dir][name]`.
+
+
+— Method **myapp:unlink** *dir* *name*
+
+*Optional*. Called during `engine.configure()` when a link of the app is
+removed.
+
+*Dir* is either `'input'` or `'output'`, and *name* is the string name
+of the link.
 
 
 — Method **myapp:pull**
@@ -139,6 +152,53 @@ transmitting them to output ports.
 For example: Move packets from input ports to output ports or to a
 network adapter.
 
+— Field **myapp.push_link**
+
+*Optional*. When specified must be a table of per-link `push()` methods
+that take an input link as an argument. For example an app could specify
+a **push_link** method for its input link *foo*:
+
+```
+Myapp = { push_link={} }
+function Myapp.push_link:foo (input)
+  while not link.empty(input) do something() end
+end
+```
+
+**Push_link** methods are copied to a fresh table when the app is started,
+and it is valid to create **push_link** methods dynamically during `link()`,
+for example like so:
+
+```
+Myapp = { push_link={} }
+function Myapp:link (dir, name)
+  -- NB: Myapp.push_link ~= self.push_link
+  if dir == 'input' then
+    self.push_link[name] = function (self, input)
+      while not link.empty(input) do something() end
+    end
+  end
+end
+function Myapp:unlink (dir, name)
+  if dir == 'input' then
+    self.push_link[name] = nil
+  end
+end
+```
+
+**Push** is not called when an app has **push_link** methods
+for *all* of its input links. If, however, an app at least one input link
+without an associated **push_link** method then **push** is called
+in addition to the **push_link** methods.
+
+
+— Method **myapp:tick**
+
+*Optional*. Called periodically at **engine.tick_Hz** frequency.
+
+For example: Move packets from input ports to output ports or to a
+network adapter.
+
 
 — Method **myapp:reconfig** *arg*
 
@@ -285,6 +345,13 @@ how many times per second to poll.
 
 This setting is not used when engine.busywait is true.
 
+— Variable **engine.tick_Hz**
+
+Frequency at which to call **app:tick** methods. The default value is
+1000 (call `tick()`s every millisecond).
+
+A value of 0 effectively disables `tick()` methods.
+
 ## Link (core.link)
 
 A *link* is a [ring buffer](http://en.wikipedia.org/wiki/Circular_buffer)

src/apps/interlink/freelist_instrument.lua

@@ -0,0 +1,40 @@
+-- Use of this source code is governed by the Apache 2.0 license; see COPYING.
+
+module(...,package.seeall)
+
+local histogram = require("core.histogram")
+local tsc = require("lib.tsc")
+
+function instrument_freelist ()
+   local ts = tsc.new()
+   local rebalance_latency = histogram.create('engine/rebalance_latency.histogram', 1, 100e6)
+   local reclaim_latency = histogram.create('engine/reclaim_latency.histogram', 1, 100e6)
+
+   local rebalance_step, reclaim_step = packet.rebalance_step, packet.reclaim_step
+   packet.rebalance_step = function ()
+      local start = ts:stamp()
+      rebalance_step()
+      rebalance_latency:add(tonumber(ts:to_ns(ts:stamp()-start)))
+   end
+   packet.reclaim_step = function ()
+      local start = ts:stamp()
+      reclaim_step()
+      reclaim_latency:add(tonumber(ts:to_ns(ts:stamp()-start)))
+   end
+
+   return rebalance_latency, reclaim_latency
+end
+
+function histogram_csv_header (out)
+   out = out or io.stdout
+   out:write("histogram,lo,hi,count\n")
+end
+
+function histogram_csv (histogram, name, out)
+   out = out or io.stdout
+   name = name or 'untitled'
+   for count, lo, hi in histogram:iterate() do
+      out:write(("%s,%f,%f,%d\n"):format(name, lo, hi, tonumber(count)))
+      out:flush()
+   end
+end

src/apps/interlink/test_sink.lua

@@ -0,0 +1,38 @@
+-- Use of this source code is governed by the Apache 2.0 license; see COPYING.
+
+module(...,package.seeall)
+
+local Receiver = require("apps.interlink.receiver")
+local Sink = require("apps.basic.basic_apps").Sink
+local lib = require("core.lib")
+local numa = require("lib.numa")
+
+function configure (c, name)
+   config.app(c, name, Receiver)
+   config.app(c, "sink", Sink)
+   config.link(c, name..".output -> sink.input")
+end
+
+function start (name, duration)
+   local c = config.new()
+   configure(c, name)
+   engine.configure(c)
+   engine.main{duration=duration}
+end
+
+local instr = require("apps.interlink.freelist_instrument")
+
+function start_instrument (name, duration, core)
+   numa.bind_to_cpu(core, 'skip')
+   local rebalance_latency = instr.instrument_freelist()
+   start(name, duration)
+   instr.histogram_csv(rebalance_latency, "rebalance")
+   local min, avg, max = rebalance_latency:summarize()
+   io.stderr:write(("(%d) rebalance latency (ns)    min:%16s    avg:%16s    max:%16s\n")
+      :format(core,
+              lib.comma_value(math.floor(min)),
+              lib.comma_value(math.floor(avg)),
+              lib.comma_value(math.floor(max))))
+   io.stderr:flush()
+end
+

src/apps/interlink/test_source.lua

@@ -4,12 +4,56 @@ module(...,package.seeall)
 
 local Transmitter = require("apps.interlink.transmitter")
 local Source = require("apps.basic.basic_apps").Source
+local lib = require("core.lib")
+local numa = require("lib.numa")
 
-function start (name)
-   local c = config.new()
+function configure (c, name)
    config.app(c, name, Transmitter)
    config.app(c, "source", Source)
-   config.link(c, "source.output -> "..name..".input")
+   config.link(c, "source."..name.." -> "..name..".input")
+end
+
+function start (name, duration)
+   local c = config.new()
+   configure(c, name)
    engine.configure(c)
-   engine.main()
+   engine.main{duration=duration}
 end
+
+function startn (name, duration, n)
+   local c = config.new()
+   for i=1,n do
+      configure(c, name..i)
+   end
+   engine.configure(c)
+   engine.main{duration=duration}
+end
+
+function txpackets ()
+   local txpackets = 0
+   for _, output in ipairs(engine.app_table["source"].output) do
+      txpackets = txpackets + link.stats(output).rxpackets
+   end
+   return txpackets
+end
+
+local instr = require("apps.interlink.freelist_instrument")
+
+function startn_instrument (name, duration, n, core)
+   numa.bind_to_cpu(core, 'skip')
+   local _, reclaim_latency = instr.instrument_freelist()
+   startn(name, duration, n)
+   local txpackets = txpackets()
+   instr.histogram_csv(reclaim_latency, "reclaim")
+   local min, avg, max = reclaim_latency:summarize()
+   engine.main{duration=1, no_report=true}
+   io.stderr:write(("(%d) reclaim latency (ns)     min:%16s    avg:%16s    max:%16s\n")
+      :format(core,
+              lib.comma_value(math.floor(min)),
+              lib.comma_value(math.floor(avg)),
+              lib.comma_value(math.floor(max))))
+   io.stderr:write(("%.3f Mpps\n"):format(txpackets / 1e6 / duration))
+   io.stderr:flush()
+
+   --engine.report_links()
+end

src/apps/interlink/wait_test.snabb

@@ -0,0 +1,36 @@
+#!snabb snsh
+
+-- Use of this source code is governed by the Apache 2.0 license; see COPYING.
+
+local worker = require("core.worker")
+local numa = require("lib.numa")
+
+-- Test wait times caused by group freelist rebalancing
+-- Synopsis: wait_test.snabb [duration] [nconsumers]
+local DURATION = tonumber(main.parameters[1]) or 10
+local NCONSUMERS = tonumber(main.parameters[2]) or 10
+local CPUS = numa.parse_cpuset(main.parameters[3] or "")
+
+local cores = {}
+for core in pairs(CPUS) do
+   table.insert(cores, core)
+   table.sort(cores)
+end
+
+require("apps.interlink.freelist_instrument").histogram_csv_header()
+io.stdout:flush()
+
+for i=1,NCONSUMERS do
+   worker.start("sink"..i, ([[require("apps.interlink.test_sink").start_instrument(%q, %d, %s)]])
+      :format("test"..i, DURATION, cores[1+i]))
+end
+
+worker.start("source", ([[require("apps.interlink.test_source").startn_instrument(%q, %d, %d, %s)]])
+   :format("test", DURATION, NCONSUMERS, assert(cores[1])))
+
+engine.main{done = function ()
+   for w, s in pairs(worker.status()) do
+      if s.alive then return false end
+   end
+   return true
+end}

src/apps/ipfix/README.md

@@ -33,11 +33,31 @@ idle and available for expiry.  The default is 300 seconds.
 *Optional*.  Period at which an active, non-idle flow should produce
 export records.  The default is 120 seconds.
 
+— Key **flush_timeout**
+
+*Optional*.  Maximum number of seconds after which queued data records
+are exported.  If set to a positive value, data records are queued
+until a flow export packet of maximum size according to the configured
+**mtu** can be generated or **flush_timeout** seconds have passed
+since the last export packet was generated, whichever occurs first.
+If set to zero, data records are exported immediately after each scan
+of the flow cache.  The default is 10 seconds.
+
 — Key **cache_size**
 
 *Optional*.  Initial size of flow tables, in terms of number of flows.
 The default is 20000.
 
+— Key **scan_time**
+
+*Optional*.  The flow cache for every configured template is scanned
+continously to check for entries eligible for export based on the
+**idle_timeout** and **active_timeout** parameters.  The **scan_time**
+determines the interval in seconds that a scan of the entire flow
+cache will take.  The implementation uses a token bucket mechanism by
+which access to the tables is distributed evenly over the time
+interval.  The default is 10 seconds.
+
 — Key **template_refresh_interval**
 
 *Optional*.  Period at which to send template records over UDP.  The