14 files changed, 1774 insertions, 0 deletions
diff --git a/examples/Makefile.am b/examples/Makefile.am
new file mode 100644
index 0000000..96361b9
--- /dev/null
+++ b/examples/Makefile.am
@@ -0,0 +1,21 @@
+# Copyright (c) 2018-2021, OARC, Inc.
+# All rights reserved.
+#
+# This file is part of dnsjit.
+#
+# dnsjit is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# dnsjit is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with dnsjit.  If not, see <http://www.gnu.org/licenses/>.
+
+dist_doc_DATA = capture.lua dumpdns2pcap.lua dumpdns.lua dumpdns-qr.lua \
+  filter_rcode.lua qr-multi-pcap-state.lua readme.lua replay.lua \
+  replay_multicli.lua respdiff.lua test_pcap_read.lua test_throughput.lua
diff --git a/examples/capture.lua b/examples/capture.lua
new file mode 100755
index 0000000..a4b9e75
--- /dev/null
+++ b/examples/capture.lua
@@ -0,0 +1,47 @@
+#!/usr/bin/env dnsjit
+local interface = arg[2]
+
+if interface == nil then
+    print("usage: "..arg[1].." <interface or any/all>")
+    return
+end
+
+local object = require("dnsjit.core.objects")
+local input = require("dnsjit.input.pcap").new()
+local layer = require("dnsjit.filter.layer").new()
+local dns = require("dnsjit.core.object.dns").new()
+
+input:create(interface)
+input:activate()
+layer:producer(input)
+local producer, ctx = layer:produce()
+
+while true do
+    local obj = producer(ctx)
+    if obj == nil then break end
+    local pl = obj:cast()
+    if obj:type() == "payload" and pl.len > 0 then
+        local transport = obj.obj_prev
+        while transport ~= nil do
+            if transport.obj_type == object.IP or transport.obj_type == object.IP6 then
+                break
+            end
+            transport = transport.obj_prev
+        end
+        local protocol = obj.obj_prev
+        while protocol ~= nil do
+            if protocol.obj_type == object.UDP or protocol.obj_type == object.TCP then
+                break
+            end
+            protocol = protocol.obj_prev
+        end
+
+        dns.obj_prev = obj
+        if transport ~= nil and protocol ~= nil then
+            transport = transport:cast()
+            protocol = protocol:cast()
+            print(protocol:type().." "..transport:source()..":"..tonumber(protocol.sport).." -> "..transport:destination()..":"..tonumber(protocol.dport))
+            dns:print()
+        end
+    end
+end
diff --git a/examples/count-pkts-per-ip.lua b/examples/count-pkts-per-ip.lua
new file mode 100755
index 0000000..9a6908e
--- /dev/null
+++ b/examples/count-pkts-per-ip.lua
@@ -0,0 +1,50 @@
+#!/usr/bin/env dnsjit
+-- count-pkts-per-ip.lua: count number of packets received from each IP/IPv6 address
+
+local input = require("dnsjit.input.pcap").new()
+local layer = require("dnsjit.filter.layer").new()
+local object = require("dnsjit.core.objects")
+local ip = require("dnsjit.lib.ip")
+local trie = require("dnsjit.lib.trie").new("uint64_t", true)
+local getopt = require("dnsjit.lib.getopt").new({})
+
+local pcap = unpack(getopt:parse())
+if pcap == nil then
+    print("usage: "..arg[1].." <pcap>")
+end
+
+-- Set up input
+input:open_offline(pcap)
+layer:producer(input)
+local produce, pctx = layer:produce()
+
+-- Read input and count packets
+while true do
+	local obj = produce(pctx)
+	if obj == nil then break end
+    local pkt = obj:cast_to(object.IP) or obj:cast_to(object.IP6)
+
+    if pkt ~= nil then
+        local iplen = 4
+        if pkt:type() == "ip6" then
+            iplen = 16
+        end
+
+        local node = trie:get_ins(pkt.src, iplen)
+        node:set(node:get() + 1)
+    end
+end
+
+-- Print statistics
+local iter = trie:iter()
+local node = iter:node()
+
+while node ~= nil do
+    local npkts = tonumber(node:get())
+    local key = node:key()
+    local ipstr = ip.tostring(key, true)
+
+    print(ipstr.." sent "..npkts.." packets")
+    iter:next()
+    node = iter:node()
+end
diff --git a/examples/dumpdns-qr.lua b/examples/dumpdns-qr.lua
new file mode 100755
index 0000000..d726e96
--- /dev/null
+++ b/examples/dumpdns-qr.lua
@@ -0,0 +1,85 @@
+#!/usr/bin/env dnsjit
+local pcap = arg[2]
+
+if pcap == nil then
+    print("usage: "..arg[1].." <pcap>")
+    return
+end
+
+local object = require("dnsjit.core.objects")
+local input = require("dnsjit.input.pcap").new()
+local layer = require("dnsjit.filter.layer").new()
+local dns = require("dnsjit.core.object.dns").new()
+local label = require("dnsjit.core.object.dns.label")
+
+local ffi = require("ffi")
+local labels = require("dnsjit.core.object.dns.label").new(16)
+local q = require("dnsjit.core.object.dns.q").new()
+
+input:open_offline(pcap)
+layer:producer(input)
+local producer, ctx = layer:produce()
+
+local queries = {}
+local responses = {}
+
+while true do
+    local obj = producer(ctx)
+    if obj == nil then break end
+    local pl = obj:cast()
+    if obj:type() == "payload" and pl.len > 0 then
+        local transport = obj.obj_prev
+        while transport ~= nil do
+            if transport.obj_type == object.IP or transport.obj_type == object.IP6 then
+                break
+            end
+            transport = transport.obj_prev
+        end
+        local protocol = obj.obj_prev
+        while protocol ~= nil do
+            if protocol.obj_type == object.UDP or protocol.obj_type == object.TCP then
+                break
+            end
+            protocol = protocol.obj_prev
+        end
+
+        dns.obj_prev = obj
+        if transport ~= nil and protocol ~= nil and dns:parse_header() == 0 then
+            transport = transport:cast()
+            protocol = protocol:cast()
+
+            if dns.qr == 1 then
+                table.insert(responses, {
+                    src = transport:source(),
+                    sport = protocol.sport,
+                    dst = transport:destination(),
+                    dport = protocol.dport,
+                    id = dns.id,
+                    rcode = dns.rcode_tostring(dns.rcode),
+                })
+            else
+                if dns.qdcount > 0 and dns:parse_q(q, labels, 16) == 0 then
+                    table.insert(queries, {
+                        src = transport:source(),
+                        sport = protocol.sport,
+                        dst = transport:destination(),
+                        dport = protocol.dport,
+                        id = dns.id,
+                        qname = label.tooffstr(dns, labels, 16),
+                        qtype = dns.type_tostring(q.type)
+                    })
+                end
+            end
+        end
+    end
+end
+
+print("src", "dst", "id", "rcode", "qname", "qtype")
+local q, r
+for _, q in pairs(queries) do
+    for _, r in pairs(responses) do
+        if q.id == r.id and q.sport == r.dport and q.dport == r.sport and q.src == r.dst and q.dst == r.src then
+            print(q.src, q.dst, q.id, r.rcode, q.qname, q.qtype)
+        end
+    end
+end
diff --git a/examples/dumpdns.lua b/examples/dumpdns.lua
new file mode 100755
index 0000000..7c6fb8c
--- /dev/null
+++ b/examples/dumpdns.lua
@@ -0,0 +1,46 @@
+#!/usr/bin/env dnsjit
+local pcap = arg[2]
+
+if pcap == nil then
+    print("usage: "..arg[1].." <pcap>")
+    return
+end
+
+local object = require("dnsjit.core.objects")
+local input = require("dnsjit.input.pcap").new()
+local layer = require("dnsjit.filter.layer").new()
+local dns = require("dnsjit.core.object.dns").new()
+
+input:open_offline(pcap)
+layer:producer(input)
+local producer, ctx = layer:produce()
+
+while true do
+    local obj = producer(ctx)
+    if obj == nil then break end
+    local pl = obj:cast()
+    if obj:type() == "payload" and pl.len > 0 then
+        local transport = obj.obj_prev
+        while transport ~= nil do
+            if transport.obj_type == object.IP or transport.obj_type == object.IP6 then
+                break
+            end
+            transport = transport.obj_prev
+        end
+        local protocol = obj.obj_prev
+        while protocol ~= nil do
+            if protocol.obj_type == object.UDP or protocol.obj_type == object.TCP then
+                break
+            end
+            protocol = protocol.obj_prev
+        end
+
+        dns.obj_prev = obj
+        if transport ~= nil and protocol ~= nil then
+            transport = transport:cast()
+            protocol = protocol:cast()
+            print(protocol:type().." "..transport:source()..":"..tonumber(protocol.sport).." -> "..transport:destination()..":"..tonumber(protocol.dport))
+            dns:print()
+        end
+    end
+end
diff --git a/examples/dumpdns2pcap.lua b/examples/dumpdns2pcap.lua
new file mode 100755
index 0000000..08656cc
--- /dev/null
+++ b/examples/dumpdns2pcap.lua
@@ -0,0 +1,38 @@
+#!/usr/bin/env dnsjit
+local pcap_in = arg[2]
+local pcap_out = arg[3]
+
+if pcap_in == nil or pcap_out == nil then
+    print("usage: "..arg[1].." <pcap in> <pcap out>")
+    return
+end
+
+local object = require("dnsjit.core.objects")
+local input = require("dnsjit.input.pcap").new()
+local layer = require("dnsjit.filter.layer").new()
+local dns = require("dnsjit.core.object.dns").new()
+local output = require("dnsjit.output.pcap").new()
+
+input:open_offline(pcap_in)
+layer:producer(input)
+local producer, ctx = layer:produce()
+
+output:open(pcap_out, input:linktype(), input:snaplen())
+local receiver, rctx = output:receive()
+
+local n = 0
+while true do
+    local obj = producer(ctx)
+    if obj == nil then break end
+    local pl = obj:cast()
+    if obj:type() == "payload" and pl.len > 0 then
+        dns.obj_prev = obj
+        if dns:parse_header() == 0 then
+            receiver(rctx, obj)
+            n = n + 1
+        end
+    end
+end
+
+output:close()
+print(n, "DNS packets dumped")
diff --git a/examples/filter_rcode.lua b/examples/filter_rcode.lua
new file mode 100755
index 0000000..c3f0254
--- /dev/null
+++ b/examples/filter_rcode.lua
@@ -0,0 +1,38 @@
+#!/usr/bin/env dnsjit
+local pcap = arg[2]
+local rcode = tonumber(arg[3])
+
+if pcap == nil or rcode == nil then
+    print("usage: "..arg[1].." <pcap> <rcode>")
+    return
+end
+
+local object = require("dnsjit.core.objects")
+local input = require("dnsjit.input.pcap").new()
+local layer = require("dnsjit.filter.layer").new()
+local dns = require("dnsjit.core.object.dns").new()
+
+input:open_offline(pcap)
+layer:producer(input)
+local producer, ctx = layer:produce()
+
+while true do
+    local obj = producer(ctx)
+    if obj == nil then break end
+    local pl = obj:cast()
+    if obj:type() == "payload" and pl.len > 0 then
+        local transport = obj.obj_prev
+        while transport ~= nil do
+            if transport.obj_type == object.IP or transport.obj_type == object.IP6 then
+                break
+            end
+            transport = transport.obj_prev
+        end
+
+        dns.obj_prev = obj
+        if transport and dns and dns:parse_header() == 0 and dns.have_rcode == 1 and dns.rcode == rcode then
+            transport = transport:cast()
+            print(dns.id, transport:source().." -> "..transport:destination())
+        end
+    end
+end
diff --git a/examples/qr-multi-pcap-state.lua b/examples/qr-multi-pcap-state.lua
new file mode 100755
index 0000000..cf2cf0d
--- /dev/null
+++ b/examples/qr-multi-pcap-state.lua
@@ -0,0 +1,270 @@
+#!/usr/bin/env dnsjit
+local clock = require("dnsjit.lib.clock")
+local log = require("dnsjit.core.log")
+local getopt = require("dnsjit.lib.getopt").new({
+    { "v", "verbose", 0, "Enable and increase verbosity for each time given", "?+" },
+    { "r", "read-state", "", "File to read state from before processing", "?" },
+    { "w", "write-state", "", "File to write state to after processing", "?" },
+})
+local pcap = unpack(getopt:parse())
+if getopt:val("help") then
+    getopt:usage()
+    return
+end
+local v = getopt:val("v")
+if v > 0 then
+    log.enable("warning")
+end
+if v > 1 then
+    log.enable("notice")
+end
+if v > 2 then
+    log.enable("info")
+end
+if v > 3 then
+    log.enable("debug")
+end
+
+if pcap == nil then
+    print("usage: "..arg[1].." <pcap>")
+    return
+end
+
+local object = require("dnsjit.core.objects")
+local input = require("dnsjit.input.mmpcap").new()
+local layer = require("dnsjit.filter.layer").new()
+local dns = require("dnsjit.core.object.dns").new()
+local label = require("dnsjit.core.object.dns.label")
+local ffi = require("ffi")
+local labels = require("dnsjit.core.object.dns.label").new(16)
+local q = require("dnsjit.core.object.dns.q").new()
+local bit = require("bit")
+
+hash_u32 = ffi.new("uint32_t[2]")
+hash_u32p = ffi.cast("uint32_t*", hash_u32)
+function hashkey(dns, transport, protocol)
+    if transport.obj_type == object.IP then
+        ffi.copy(hash_u32p, transport.src, 4)
+        hash_u32[1] = hash_u32[0]
+        ffi.copy(hash_u32p, transport.dst, 4)
+        hash_u32[1] = bit.bxor(hash_u32[0], hash_u32[1])
+    else
+        local srcp = ffi.cast("uint8_t*", transport.src)
+        ffi.copy(hash_u32p, srcp, 4)
+        hash_u32[1] = hash_u32[0]
+        ffi.copy(hash_u32p, srcp+4, 4)
+        hash_u32[1] = bit.bxor(hash_u32[0], hash_u32[1])
+        srcp = ffi.cast("uint8_t*", transport.dst)
+        ffi.copy(hash_u32p, srcp, 4)
+        hash_u32[1] = bit.bxor(hash_u32[0], hash_u32[1])
+        ffi.copy(hash_u32p, srcp+4, 4)
+        hash_u32[1] = bit.bxor(hash_u32[0], hash_u32[1])
+    end
+    if dns.qr == 1 then
+        hash_u32[0] = protocol.dport + bit.lshift(protocol.sport, 16)
+    else
+        hash_u32[0] = protocol.sport + bit.lshift(protocol.dport, 16)
+    end
+    hash_u32[1] = bit.bxor(hash_u32[0], hash_u32[1])
+    hash_u32[0] = dns.id + bit.lshift(dns.id, 16)
+    return bit.bxor(hash_u32[0], hash_u32[1])
+end
+
+function dump_inflight(hkey, obj)
+    return string.format("return %d, { qsec = %d, qnsec = %d, rsec = %d, rnsec = %d, src = %q, sport = %d, dst = %q, dport = %d, id = %d, qname = %q, qtype = %q, rcode = %d }",
+        hkey,
+        tonumber(obj.qsec), tonumber(obj.qnsec),
+        tonumber(obj.rsec), tonumber(obj.rnsec),
+        obj.src, obj.sport,
+        obj.dst, obj.dport,
+        obj.id,
+        obj.qname, obj.qtype,
+        obj.rcode
+    )
+end
+
+function qrout(res)
+    local tsq = tonumber(res.qsec) + (tonumber(res.qnsec)/1000000000)
+    local tsr = tonumber(res.rsec) + (tonumber(res.rnsec)/1000000000)
+    print(tsq, tsr, math.floor(((tsr-tsq)*1000000)+0.5),
+        res.src, res.dst, res.id, res.rcode, res.qname, res.qtype)
+end
+
+input:open(pcap)
+layer:producer(input)
+local producer, ctx = layer:produce()
+
+local inflight = {}
+
+if getopt:val("read-state") > "" then
+    local f, _ = io.open(getopt:val("read-state"))
+    local inflights = 0
+    if f ~= nil then
+        for chunk in f:lines() do
+            local hkey, query = assert(loadstring(chunk))()
+            if hkey and query then
+                if not inflight[hkey] then
+                    inflight[hkey] = {
+                        queries = {},
+                        size = 0,
+                    }
+                end
+
+                table.insert(inflight[hkey].queries, query)
+                inflight[hkey].size = inflight[hkey].size + 1
+                inflights = inflights + 1
+            end
+        end
+        f:close()
+        print(string.format("== read %d inflight states from %q", inflights, getopt:val("read-state")))
+    end
+end
+
+local stat = {
+    packets = 0,
+    queries = 0,
+    responses = 0,
+    dropped = 0,
+}
+local start_sec, start_nsec = clock:monotonic()
+while true do
+    local obj = producer(ctx)
+    if obj == nil then break end
+    stat.packets = stat.packets + 1
+    local pl = obj:cast()
+    if obj:type() == "payload" and pl.len > 0 then
+        local protocol = obj.obj_prev
+        while protocol ~= nil do
+            if protocol.obj_type == object.UDP or protocol.obj_type == object.TCP then
+                break
+            end
+            protocol = protocol.obj_prev
+        end
+        local transport = protocol.obj_prev
+        while transport ~= nil do
+            if transport.obj_type == object.IP or transport.obj_type == object.IP6 then
+                break
+            end
+            transport = transport.obj_prev
+        end
+        local pcap = transport.obj_prev
+        while pcap ~= nil do
+            if pcap.obj_type == object.PCAP then
+                break
+            end
+            pcap = pcap.obj_prev
+        end
+
+        dns.obj_prev = obj
+        if pcap ~= nil and transport ~= nil and protocol ~= nil and dns:parse_header() == 0 then
+            transport = transport:cast()
+            protocol = protocol:cast()
+            pcap = pcap:cast()
+
+            local hkey = hashkey(dns, transport, protocol)
+
+            if dns.qr == 1 then
+                stat.responses = stat.responses + 1
+                if inflight[hkey] then
+                    for k, n in pairs(inflight[hkey].queries) do
+                        if n.id == dns.id
+                            and n.sport == protocol.dport
+                            and n.dport == protocol.sport
+                            and n.src == transport:destination()
+                            and n.dst == transport:source()
+                        then
+                            n.rsec = pcap.ts.sec
+                            n.rnsec = pcap.ts.nsec
+                            n.rcode = dns.rcode
+                            qrout(n)
+                            inflight[hkey].queries[k] = nil
+                            inflight[hkey].size = inflight[hkey].size - 1
+                            if inflight[hkey].size < 1 then
+                                inflight[hkey] = nil
+                            end
+                            break
+                        end
+                    end
+                else
+                    print("== dropped",
+                        tonumber(pcap.ts.sec) + (tonumber(pcap.ts.nsec) / 1000000000),
+                        transport:source(),
+                        transport:destination(),
+                        dns.id,
+                        label.tooffstr(dns, labels, 16),
+                        dns.type_tostring(q.type)
+                    )
+                    stat.dropped = stat.dropped + 1
+                end
+            else
+                stat.queries = stat.queries + 1
+                if dns.qdcount > 0 and dns:parse_q(q, labels, 16) == 0 then
+                    if not inflight[hkey] then
+                        inflight[hkey] = {
+                            queries = {},
+                            size = 0,
+                        }
+                    end
+
+                    table.insert(inflight[hkey].queries, {
+                        qsec = pcap.ts.sec,
+                        qnsec = pcap.ts.nsec,
+                        rsec = -1,
+                        rnsec = -1,
+                        src = transport:source(),
+                        sport = protocol.sport,
+                        dst = transport:destination(),
+                        dport = protocol.dport,
+                        id = dns.id,
+                        qname = label.tooffstr(dns, labels, 16),
+                        qtype = dns.type_tostring(q.type),
+                        rcode = -1,
+                    })
+                    inflight[hkey].size = inflight[hkey].size + 1
+                end
+            end
+        end
+    end
+end
+local end_sec, end_nsec = clock:monotonic()
+
+local runtime = 0
+if end_sec > start_sec then
+    runtime = ((end_sec - start_sec) - 1) + ((1000000000 - start_nsec + end_nsec)/1000000000)
+elseif end_sec == start_sec and end_nsec > start_nsec then
+    runtime = (end_nsec - start_nsec) / 1000000000
+end
+
+print("== runtime", runtime)
+print("== packets", stat.packets, stat.packets/runtime)
+print("== queries", stat.queries, stat.queries/runtime)
+print("== responses", stat.responses, stat.responses/runtime)
+print("== dropped", stat.dropped, stat.dropped/runtime)
+
+if getopt:val("write-state") > "" then
+    local f, _ = io.open(getopt:val("write-state"), "w+")
+    local inflights = 0
+    if f ~= nil then
+        for hkey, unanswered in pairs(inflight) do
+            for _, query in pairs(unanswered.queries) do
+                f:write(dump_inflight(hkey, query), "\n")
+                inflights = inflights + 1
+            end
+        end
+        f:close()
+        print(string.format("== wrote %d inflight states to %q", inflights, getopt:val("write-state")))
+    end
+else
+    inflights = 0
+    for hkey, unanswered in pairs(inflight) do
+        inflights = inflights + unanswered.size
+    end
+    if inflights > 0 then
+        print("== inflight queries (tsq, src, dst, id, qname, qtype)")
+        for hkey, unanswered in pairs(inflight) do
+            for _, query in pairs(unanswered.queries) do
+                print(tonumber(query.qsec) + (tonumber(query.qnsec)/1000000000), query.src, query.dst, query.id, query.qname, query.qtype)
+            end
+        end
+    end
+end
diff --git a/examples/readme.lua b/examples/readme.lua
new file mode 100755
index 0000000..dc6de89
--- /dev/null
+++ b/examples/readme.lua
@@ -0,0 +1,20 @@
+#!/usr/bin/env dnsjit
+require("dnsjit.core.objects")
+local input = require("dnsjit.input.pcap").new()
+local layer = require("dnsjit.filter.layer").new()
+local dns = require("dnsjit.core.object.dns").new()
+
+input:open_offline(arg[2])
+layer:producer(input)
+local producer, ctx = layer:produce()
+
+while true do
+    local object = producer(ctx)
+    if object == nil then break end
+    if object:type() == "payload" then
+        dns.obj_prev = object
+        if dns:parse_header() == 0 then
+            print(dns.id)
+        end
+    end
+end
diff --git a/examples/replay.lua b/examples/replay.lua
new file mode 100755
index 0000000..5281133
--- /dev/null
+++ b/examples/replay.lua
@@ -0,0 +1,119 @@
+#!/usr/bin/env dnsjit
+local clock = require("dnsjit.lib.clock")
+local log = require("dnsjit.core.log")
+local getopt = require("dnsjit.lib.getopt").new({
+    { "v", "verbose", 0, "Enable and increase verbosity for each time given", "?+" },
+    { "R", "responses", false, "Wait for responses to the queries and print both", "?" },
+    { "t", "tcp", false, "Use TCP instead of UDP", "?"},
+    { "T", "tls", false, "Use TLS instead of UDP/TCP", "?"},
+})
+local pcap, host, port = unpack(getopt:parse())
+if getopt:val("help") then
+    getopt:usage()
+    return
+end
+local v = getopt:val("v")
+if v > 0 then
+    log.enable("warning")
+end
+if v > 1 then
+    log.enable("notice")
+end
+if v > 2 then
+    log.enable("info")
+end
+if v > 3 then
+    log.enable("debug")
+end
+
+if pcap == nil or host == nil or port == nil then
+    print("usage: "..arg[1].." <pcap> <host> <port>")
+    return
+end
+
+local ffi = require("ffi")
+
+require("dnsjit.core.objects")
+local input = require("dnsjit.input.mmpcap").new()
+local layer = require("dnsjit.filter.layer").new()
+
+input:open(pcap)
+layer:producer(input)
+
+local query = require("dnsjit.core.object.dns").new()
+local response = require("dnsjit.core.object.dns").new()
+
+local dnscli = require("dnsjit.output.dnscli")
+local output
+if getopt:val("t") then
+    output = dnscli.new(dnscli.TCP)
+    response.includes_dnslen = 1
+elseif getopt:val("T") then
+    output = dnscli.new(dnscli.TLS)
+    response.includes_dnslen = 1
+else
+    output = dnscli.new(dnscli.UDP)
+end
+output:connect(host, port)
+
+local printdns = false
+if getopt:val("responses") then
+    printdns = true
+end
+
+local prod, pctx = layer:produce()
+local recv, rctx = output:receive()
+local oprod, opctx = output:produce()
+local start_sec, start_nsec = clock:monotonic()
+
+while true do
+    local obj = prod(pctx)
+    if obj == nil then break end
+    local pl = obj:cast()
+    if obj:type() == "payload" and pl.len > 0 then
+        query.obj_prev = obj
+
+        local trs = pl.obj_prev:cast()
+        if trs:type() == "tcp" then
+            query.includes_dnslen = 1
+        else
+            query.includes_dnslen = 0
+        end
+
+        if query:parse_header() == 0 and query.qr == 0 then
+            recv(rctx, query:uncast())
+
+            if printdns then
+                print("query:")
+                query:print()
+
+                local pobj = oprod(opctx)
+                if pobj == nil then
+                    log.fatal("producer error")
+                end
+                local rpl = pobj:cast()
+                if rpl.len == 0 then
+                    print("timed out")
+                else
+                    response.obj_prev = pobj
+                    print("response:")
+                    response:print()
+                end
+            end
+        end
+    end
+end
+
+local end_sec, end_nsec = clock:monotonic()
+
+local runtime = 0
+if end_sec > start_sec then
+    runtime = ((end_sec - start_sec) - 1) + ((1000000000 - start_nsec + end_nsec)/1000000000)
+elseif end_sec == start_sec and end_nsec > start_nsec then
+    runtime = (end_nsec - start_nsec) / 1000000000
+end
+
+print("runtime", runtime)
+print("packets", input:packets(), input:packets()/runtime, "/pps")
+print("queries", output:packets(), output:packets()/runtime, "/qps")
+print("errors", output:errors())
diff --git a/examples/replay_multicli.lua b/examples/replay_multicli.lua
new file mode 100755
index 0000000..c883aa5
--- /dev/null
+++ b/examples/replay_multicli.lua
@@ -0,0 +1,192 @@
+#!/usr/bin/env dnsjit
+local clock = require("dnsjit.lib.clock")
+local log = require("dnsjit.core.log")
+local getopt = require("dnsjit.lib.getopt").new({
+    { "c", "clients", 10, "Number of clients run", "?" },
+    { "v", "verbose", 0, "Enable and increase verbosity for each time given", "?+" },
+    { "R", "responses", false, "Wait for responses to the queries and print both", "?" },
+    { "t", "tcp", false, "Use TCP instead of UDP", "?"},
+    { "T", "tls", false, "Use TLS instead of UDP/TCP", "?"},
+})
+local pcap, host, port = unpack(getopt:parse())
+if getopt:val("help") then
+    getopt:usage()
+    return
+end
+local v = getopt:val("v")
+if v > 0 then
+    log.enable("warning")
+end
+if v > 1 then
+    log.enable("notice")
+end
+if v > 2 then
+    log.enable("info")
+end
+if v > 3 then
+    log.enable("debug")
+end
+
+if pcap == nil or host == nil or port == nil then
+    print("usage: "..arg[1].." <pcap> <host> <port>")
+    return
+end
+
+local ffi = require("ffi")
+
+require("dnsjit.core.objects")
+local input = require("dnsjit.input.mmpcap").new()
+local layer = require("dnsjit.filter.layer").new()
+
+input:open(pcap)
+layer:producer(input)
+
+local query = require("dnsjit.core.object.dns").new()
+local response = require("dnsjit.core.object.dns").new()
+
+local dnscli = require("dnsjit.output.dnscli")
+
+local clients = {}
+local last_client = nil
+local num_clients = getopt:val("c")
+for n = 1, num_clients do
+    local output
+    if getopt:val("t") then
+        output = dnscli.new(dnscli.TCP + dnscli.NONBLOCKING)
+    elseif getopt:val("T") then
+        output = dnscli.new(dnscli.TLS + dnscli.NONBLOCKING)
+    else
+        output = dnscli.new(dnscli.UDP + dnscli.NONBLOCKING)
+    end
+    output:connect(host, port)
+
+    local recv, rctx = output:receive()
+    local prod, pctx = output:produce()
+    local client = {
+        output = output,
+        last = last_client,
+        busy = false,
+        recv = recv, rctx = rctx,
+        prod = prod, pctx = pctx,
+        id = nil,
+        done = false,
+    }
+    table.insert(clients, client)
+    last_client = client
+end
+local output = clients[1]
+output.last = last_client
+
+if getopt:val("t") then
+    response.includes_dnslen = 1
+elseif getopt:val("T") then
+    response.includes_dnslen = 1
+end
+
+local printdns = false
+if getopt:val("responses") then
+    printdns = true
+end
+
+local prod, pctx = layer:produce()
+local start_sec, start_nsec = clock:monotonic()
+
+local done = false
+while true do
+    output = output.last
+    if printdns and output.busy then
+        local pobj = output.prod(output.pctx)
+        if pobj == nil then
+            log.fatal("producer error")
+        end
+        local rpl = pobj:cast()
+        if rpl.len == 0 then
+            -- print(output, "busy")
+        else
+            response.obj_prev = pobj
+            if response:parse_header() == 0 and response.qr == 1 and response.id == output.id then
+                print("response:")
+                response:print()
+                output.busy = false
+            end
+        end
+    end
+    if not output.busy then
+        while true do
+            local obj = prod(pctx)
+            if obj == nil then
+                done = true
+                break
+            end
+            local pl = obj:cast()
+            if obj:type() == "payload" and pl.len > 0 then
+                query.obj_prev = obj
+
+                local trs = pl.obj_prev:cast()
+                if trs:type() == "tcp" then
+                    query.includes_dnslen = 1
+                else
+                    query.includes_dnslen = 0
+                end
+
+                if query:parse_header() == 0 and query.qr == 0 then
+                    output.recv(output.rctx, query:uncast())
+                    if printdns then
+                        print("query:")
+                        query:print()
+                        output.busy = true
+                        output.id = query.id
+                    end
+                    break
+                end
+            end
+        end
+    end
+    if done then break end
+end
+
+local queries, timeouts, errors = 0, 0, 0
+done = 0
+while true do
+    output = output.last
+    if printdns and output.busy then
+        local pobj = output.prod(output.pctx)
+        if pobj == nil then
+            log.fatal("producer error")
+        end
+        local rpl = pobj:cast()
+        if rpl.len == 0 then
+            -- print(output, "busy")
+        else
+            response.obj_prev = pobj
+            if response:parse_header() == 0 and response.qr == 1 and response.id == output.id then
+                print("response:")
+                response:print()
+                output.busy = false
+            end
+        end
+    end
+    if not output.busy and not output.done then
+        output.done = true
+        done = done + 1
+        queries = queries + output.output:packets()
+        timeouts = timeouts + output.output:timeouts()
+        errors = errors + output.output:errors()
+    end
+    if done >= num_clients then break end
+end
+
+local end_sec, end_nsec = clock:monotonic()
+
+local runtime = 0
+if end_sec > start_sec then
+    runtime = ((end_sec - start_sec) - 1) + ((1000000000 - start_nsec + end_nsec)/1000000000)
+elseif end_sec == start_sec and end_nsec > start_nsec then
+    runtime = (end_nsec - start_nsec) / 1000000000
+end
+
+print("runtime", runtime)
+print("packets", input:packets(), input:packets()/runtime, "/pps")
+print("queries", queries, queries/runtime, "/qps")
+print("timeouts", timeouts)
+print("errors", errors)
diff --git a/examples/respdiff.lua b/examples/respdiff.lua
new file mode 100755
index 0000000..831f349
--- /dev/null
+++ b/examples/respdiff.lua
@@ -0,0 +1,160 @@
+#!/usr/bin/env dnsjit
+local ffi = require("ffi")
+local clock = require("dnsjit.lib.clock")
+local log = require("dnsjit.core.log")
+log.display_file_line(true)
+local getopt = require("dnsjit.lib.getopt").new({
+    { "v", "verbose", 0, "Enable and increase verbosity for each time given", "?+" },
+})
+local pcap, host, port, path, origname, recvname = unpack(getopt:parse())
+if getopt:val("help") then
+    getopt:usage()
+    return
+end
+local v = getopt:val("v")
+if v > 0 then
+    log.enable("warning")
+end
+if v > 1 then
+    log.enable("notice")
+end
+if v > 2 then
+    log.enable("info")
+end
+if v > 3 then
+    log.enable("debug")
+end
+
+if pcap == nil or host == nil or port == nil or path == nil or origname == nil or recvname == nil then
+    print("usage: "..arg[1].." <pcap> <host> <port> <LMDB path> <origname> <recvname>")
+    return
+end
+
+local object = require("dnsjit.core.objects")
+local dns = require("dnsjit.core.object.dns").new()
+local input = require("dnsjit.input.mmpcap").new()
+input:open(pcap)
+local layer = require("dnsjit.filter.layer").new()
+layer:producer(input)
+
+local udpcli, tcpcli
+local udprecv, udpctx, tcprecv, tcpctx
+local udpprod, tcpprod
+
+local prod, pctx = layer:produce()
+local queries = {}
+local clipayload = ffi.new("core_object_payload_t")
+clipayload.obj_type = object.PAYLOAD
+local cliobject = ffi.cast("core_object_t*", clipayload)
+
+local respdiff = require("dnsjit.output.respdiff").new(path, origname, recvname)
+local resprecv, respctx = respdiff:receive()
+local query_payload, original_payload, response_payload = ffi.new("core_object_payload_t"), ffi.new("core_object_payload_t"), ffi.new("core_object_payload_t")
+query_payload.obj_type = object.PAYLOAD
+original_payload.obj_type = object.PAYLOAD
+response_payload.obj_type = object.PAYLOAD
+local query_payload_obj = ffi.cast("core_object_t*", query_payload)
+query_payload.obj_prev = ffi.cast("core_object_t*", original_payload)
+original_payload.obj_prev = ffi.cast("core_object_t*", response_payload)
+
+local start_sec, start_nsec = clock:realtime()
+while true do
+    local obj = prod(pctx)
+    if obj == nil then break end
+    local payload = obj:cast()
+    if obj:type() == "payload" and payload.len > 0 then
+        dns.obj_prev = obj
+        if dns:parse_header() == 0 then
+            local transport = obj.obj_prev
+            while transport ~= nil do
+                if transport.obj_type == object.IP or transport.obj_type == object.IP6 then
+                    break
+                end
+                transport = transport.obj_prev
+            end
+            local protocol = obj.obj_prev
+            while protocol ~= nil do
+                if protocol.obj_type == object.UDP or protocol.obj_type == object.TCP then
+                    break
+                end
+                protocol = protocol.obj_prev
+            end
+
+            if transport ~= nil and protocol ~= nil then
+                transport = transport:cast()
+                protocol = protocol:cast()
+
+                if dns.qr == 0 then
+                    local k = string.format("%s %d %s %d", transport:source(), protocol.sport, transport:destination(), protocol.dport)
+                    local q = {
+                        id = dns.id,
+                        proto = protocol:type(),
+                        payload = ffi.new("uint8_t[?]", payload.len),
+                        len = tonumber(payload.len)
+                    }
+                    ffi.copy(q.payload, payload.payload, payload.len)
+                    queries[k] = q
+                else
+                    local k = string.format("%s %d %s %d", transport:destination(), protocol.dport, transport:source(), protocol.sport)
+                    local q = queries[k]
+                    if q then
+                        queries[k] = nil
+                        clipayload.payload = q.payload
+                        clipayload.len = q.len
+
+                        local prod, pctx
+
+                        if q.proto == "udp" then
+                            if not udpcli then
+                                udpcli = require("dnsjit.output.udpcli").new()
+                                udpcli:connect(host, port)
+                                udprecv, udpctx = udpcli:receive()
+                                udpprod, _ = udpcli:produce()
+                            end
+                            udprecv(udpctx, cliobject)
+                            prod = udpprod
+                            pctx = udpctx
+                        elseif q.proto == "tcp" then
+                            if not tcpcli then
+                                tcpcli = require("dnsjit.output.tcpcli").new()
+                                tcpcli:connect(host, port)
+                                tcprecv, tcpctx = tcpcli:receive()
+                                tcpprod, _ = tcpcli:produce()
+                            end
+                            tcprecv(tcpctx, cliobject)
+                            prod = tcpprod
+                            pctx = tcpctx
+                        end
+
+                        while true do
+                            local response = prod(pctx)
+                            if response == nil then
+                                log.fatal("producer error")
+                            end
+                            local rpl = response:cast()
+                            if rpl.len == 0 then
+                                log.info("timed out")
+                            else
+                                dns.obj_prev = response
+                                if dns:parse_header() == 0 and dns.id == q.id then
+                                    query_payload.payload = q.payload
+                                    query_payload.len = q.len
+                                    original_payload.payload = payload.payload
+                                    original_payload.len = payload.len
+                                    response_payload.payload = rpl.payload
+                                    response_payload.len = rpl.len
+
+                                    resprecv(respctx, query_payload_obj)
+                                    break
+                                end
+                            end
+                        end
+                    end
+                end
+            end
+        end
+    end
+end
+local end_sec, end_nsec = clock:realtime()
+
+respdiff:commit(start_sec, end_sec)
diff --git a/examples/test_pcap_read.lua b/examples/test_pcap_read.lua
new file mode 100755
index 0000000..7d29c7a
--- /dev/null
+++ b/examples/test_pcap_read.lua
@@ -0,0 +1,149 @@
+#!/usr/bin/env dnsjit
+local clock = require("dnsjit.lib.clock")
+local log = require("dnsjit.core.log")
+local getopt = require("dnsjit.lib.getopt").new({
+    { "v", "verbose", 0, "Enable and increase verbosity for each time given", "?+" },
+    { "l", "layer", false, "Test also with dnsjit.filter.layer", "?" },
+    { "p", "producer", false, "Test with the producer interface rather then receiver interface", "?" },
+})
+local pcap, runs = unpack(getopt:parse())
+if getopt:val("help") then
+    getopt:usage()
+    return
+end
+local v = getopt:val("v")
+if v > 0 then
+    log.enable("warning")
+end
+if v > 1 then
+    log.enable("notice")
+end
+if v > 2 then
+    log.enable("info")
+end
+if v > 3 then
+    log.enable("debug")
+end
+
+if pcap == nil then
+    print("usage: "..arg[1].." <pcap> [runs]")
+    return
+end
+
+inputs = { "fpcap", "mmpcap", "pcap" }
+result = {}
+results = {}
+highest = nil
+
+if runs == nil then
+    runs = 10
+else
+    runs = tonumber(runs)
+end
+
+if getopt:val("p") then
+    for _, name in pairs(inputs) do
+        rt = 0.0
+        p = 0
+
+        print("run", name)
+        for n = 1, runs do
+            o = require("dnsjit.output.null").new()
+            i = require("dnsjit.input."..name).new()
+
+            if name == "pcap" then
+                i:open_offline(pcap)
+            else
+                i:open(pcap)
+            end
+
+            if getopt:val("l") then
+                f = require("dnsjit.filter.layer").new()
+                f:producer(i)
+                o:producer(f)
+            else
+                o:producer(i)
+            end
+
+            ss, sns = clock:monotonic()
+            o:run()
+            es, ens = clock:monotonic()
+
+            if es > ss then
+                rt = rt + ((es - ss) - 1) + ((1000000000 - sns + ens)/1000000000)
+            elseif es == ss and ens > sns then
+                rt = rt + (ens - sns) / 1000000000
+            end
+
+            p = p + o:packets()
+        end
+
+        result[name] = {
+            rt = rt,
+            p = p
+        }
+        if highest == nil or rt > result[highest].rt then
+            highest = name
+        end
+        table.insert(results, name)
+    end
+else
+    for _, name in pairs(inputs) do
+        rt = 0.0
+        p = 0
+
+        print("run", name)
+        for n = 1, runs do
+            o = require("dnsjit.output.null").new()
+            i = require("dnsjit.input."..name).new()
+
+            if name == "pcap" then
+                i:open_offline(pcap)
+            else
+                i:open(pcap)
+            end
+
+            if getopt:val("l") then
+                f = require("dnsjit.filter.layer").new()
+                f:receiver(o)
+                i:receiver(f)
+            else
+                i:receiver(o)
+            end
+
+            ss, sns = clock:monotonic()
+            if name == "pcap" then
+                i:dispatch()
+            else
+                i:run()
+            end
+            es, ens = clock:monotonic()
+
+            if es > ss then
+                rt = rt + ((es - ss) - 1) + ((1000000000 - sns + ens)/1000000000)
+            elseif es == ss and ens > sns then
+                rt = rt + (ens - sns) / 1000000000
+            end
+
+            p = p + o:packets()
+        end
+
+        result[name] = {
+            rt = rt,
+            p = p
+        }
+        if highest == nil or rt > result[highest].rt then
+            highest = name
+        end
+        table.insert(results, name)
+    end
+end
+
+print("name", "runtime", "pps", "x", "pkts")
+print(highest, result[highest].rt, result[highest].p/result[highest].rt, 1.0, result[highest].p)
+for _, name in pairs(results) do
+    if name ~= highest then
+        local f = result[name].p / result[highest].p
+        print(name, result[name].rt, result[name].p/result[name].rt, (result[highest].rt/result[name].rt)*f, result[name].p)
+    end
+end
diff --git a/examples/test_throughput.lua b/examples/test_throughput.lua
new file mode 100755
index 0000000..b936c41
--- /dev/null
+++ b/examples/test_throughput.lua
@@ -0,0 +1,539 @@
+#!/usr/bin/env dnsjit
+local ffi = require("ffi")
+local object = require("dnsjit.core.objects")
+local clock = require("dnsjit.lib.clock")
+local log = require("dnsjit.core.log")
+local getopt = require("dnsjit.lib.getopt").new({
+    { "v", "verbose", 0, "Enable and increase verbosity for each time given", "?+" },
+    { "s", "split", false, "Test also with dnsjit.filter.split", "?" },
+    { "t", "thread", false, "Test also with dnsjit.core.thread using dnsjit.core.channel", "?" },
+})
+local num, runs = unpack(getopt:parse())
+if getopt:val("help") then
+    getopt:usage()
+    return
+end
+local v = getopt:val("v")
+if v > 0 then
+    log.enable("warning")
+end
+if v > 1 then
+    log.enable("notice")
+end
+if v > 2 then
+    log.enable("info")
+end
+if v > 3 then
+    log.enable("debug")
+end
+
+if num == nil then
+    print("usage: "..arg[1].." <num> [runs]")
+    return
+else
+    num = tonumber(num)
+end
+
+if runs == nil then
+    runs = 1
+else
+    runs = tonumber(runs)
+end
+
+print("zero:receiver() -> null:receive()")
+local run
+for run = 1, runs do
+    local i = require("dnsjit.input.zero").new()
+    local o = require("dnsjit.output.null").new()
+
+    i:receiver(o)
+    local start_sec, start_nsec = clock:monotonic()
+    i:run(num)
+    local end_sec, end_nsec = clock:monotonic()
+
+    local runtime = 0
+    if end_sec > start_sec then
+        runtime = ((end_sec - start_sec) - 1) + ((1000000000 - start_nsec + end_nsec)/1000000000)
+    elseif end_sec == start_sec and end_nsec > start_nsec then
+        runtime = (end_nsec - start_nsec) / 1000000000
+    end
+
+    print(run, "runtime", runtime, num/runtime, "/sec", o:packets())
+end
+
+print("lua -> null:receive()")
+local run
+for run = 1, runs do
+    local o = require("dnsjit.output.null").new()
+    local recv, rctx = o:receive()
+    local pkt = ffi.new("core_object_null_t")
+    pkt.obj_type = object.NULL
+    local obj = ffi.cast("core_object_t*", pkt)
+
+    local start_sec, start_nsec = clock:monotonic()
+    for n = 1, num do
+        recv(rctx, obj)
+    end
+    local end_sec, end_nsec = clock:monotonic()
+
+    local runtime = 0
+    if end_sec > start_sec then
+        runtime = ((end_sec - start_sec) - 1) + ((1000000000 - start_nsec + end_nsec)/1000000000)
+    elseif end_sec == start_sec and end_nsec > start_nsec then
+        runtime = (end_nsec - start_nsec) / 1000000000
+    end
+
+    print(run, "runtime", runtime, num/runtime, "/sec", o:packets())
+end
+
+-- TODO: use core.thread
+
+print("zero:produce() <- null:producer()")
+local run
+for run = 1, runs do
+    local i = require("dnsjit.input.zero").new()
+    local o = require("dnsjit.output.null").new()
+
+    local start_sec, start_nsec = clock:monotonic()
+    o:producer(i)
+    o:run(num)
+    local end_sec, end_nsec = clock:monotonic()
+
+    local runtime = 0
+    if end_sec > start_sec then
+        runtime = ((end_sec - start_sec) - 1) + ((1000000000 - start_nsec + end_nsec)/1000000000)
+    elseif end_sec == start_sec and end_nsec > start_nsec then
+        runtime = (end_nsec - start_nsec) / 1000000000
+    end
+
+    print(run, "runtime", runtime, num/runtime, "/sec", o:packets())
+end
+
+print("zero:produce() <- lua")
+local run
+for run = 1, runs do
+    local i = require("dnsjit.input.zero").new()
+    local prod, pctx = i:produce()
+
+    local start_sec, start_nsec = clock:monotonic()
+    for n = 1, num do
+        prod(pctx)
+    end
+    local end_sec, end_nsec = clock:monotonic()
+
+    local runtime = 0
+    if end_sec > start_sec then
+        runtime = ((end_sec - start_sec) - 1) + ((1000000000 - start_nsec + end_nsec)/1000000000)
+    elseif end_sec == start_sec and end_nsec > start_nsec then
+        runtime = (end_nsec - start_nsec) / 1000000000
+    end
+
+    print(run, "runtime", runtime, num/runtime, "/sec", num)
+end
+
+print("zero:produce() <- lua -> null:receive()")
+local run
+for run = 1, runs do
+    local i = require("dnsjit.input.zero").new()
+    local o = require("dnsjit.output.null").new()
+    local prod, pctx = i:produce()
+    local recv, rctx = o:receive()
+
+    local start_sec, start_nsec = clock:monotonic()
+    for n = 1, num do
+        recv(rctx, prod(pctx))
+    end
+    local end_sec, end_nsec = clock:monotonic()
+
+    local runtime = 0
+    if end_sec > start_sec then
+        runtime = ((end_sec - start_sec) - 1) + ((1000000000 - start_nsec + end_nsec)/1000000000)
+    elseif end_sec == start_sec and end_nsec > start_nsec then
+        runtime = (end_nsec - start_nsec) / 1000000000
+    end
+
+    print(run, "runtime", runtime, num/runtime, "/sec", num)
+end
+
+if getopt:val("s") then
+    print("zero:receiver() -> split:receiver() -> null:receive() x1")
+    local run
+    for run = 1, runs do
+        local i = require("dnsjit.input.zero").new()
+        local s = require("dnsjit.filter.split").new()
+        local o1 = require("dnsjit.output.null").new()
+
+        s:receiver(o1)
+        i:receiver(s)
+        local start_sec, start_nsec = clock:monotonic()
+        i:run(num)
+        local end_sec, end_nsec = clock:monotonic()
+
+        local runtime = 0
+        if end_sec > start_sec then
+            runtime = ((end_sec - start_sec) - 1) + ((1000000000 - start_nsec + end_nsec)/1000000000)
+        elseif end_sec == start_sec and end_nsec > start_nsec then
+            runtime = (end_nsec - start_nsec) / 1000000000
+        end
+
+        print(run, "runtime", runtime, num/runtime, "/sec", o1:packets(), o1:packets())
+    end
+
+    print("zero:receiver() -> split:receiver() -> null:receive() x2")
+    local run
+    for run = 1, runs do
+        local i = require("dnsjit.input.zero").new()
+        local s = require("dnsjit.filter.split").new()
+        local o1 = require("dnsjit.output.null").new()
+        local o2 = require("dnsjit.output.null").new()
+
+        s:receiver(o1)
+        s:receiver(o2)
+        i:receiver(s)
+        local start_sec, start_nsec = clock:monotonic()
+        i:run(num)
+        local end_sec, end_nsec = clock:monotonic()
+
+        local runtime = 0
+        if end_sec > start_sec then
+            runtime = ((end_sec - start_sec) - 1) + ((1000000000 - start_nsec + end_nsec)/1000000000)
+        elseif end_sec == start_sec and end_nsec > start_nsec then
+            runtime = (end_nsec - start_nsec) / 1000000000
+        end
+
+        print(run, "runtime", runtime, num/runtime, "/sec", o1:packets() + o2:packets(), o1:packets(), o2:packets())
+    end
+
+    print("zero:receiver() -> split:receiver() -> null:receive() x4")
+    local run
+    for run = 1, runs do
+        local i = require("dnsjit.input.zero").new()
+        local s = require("dnsjit.filter.split").new()
+        local o1 = require("dnsjit.output.null").new()
+        local o2 = require("dnsjit.output.null").new()
+        local o3 = require("dnsjit.output.null").new()
+        local o4 = require("dnsjit.output.null").new()
+
+        s:receiver(o1)
+        s:receiver(o2)
+        s:receiver(o3)
+        s:receiver(o4)
+        i:receiver(s)
+        local start_sec, start_nsec = clock:monotonic()
+        i:run(num)
+        local end_sec, end_nsec = clock:monotonic()
+
+        local runtime = 0
+        if end_sec > start_sec then
+            runtime = ((end_sec - start_sec) - 1) + ((1000000000 - start_nsec + end_nsec)/1000000000)
+        elseif end_sec == start_sec and end_nsec > start_nsec then
+            runtime = (end_nsec - start_nsec) / 1000000000
+        end
+
+        print(run, "runtime", runtime, num/runtime, "/sec", o1:packets() + o2:packets() + o3:packets() + o4:packets(), o1:packets(), o2:packets(), o3:packets(), o4:packets())
+    end
+
+    print("zero:receiver() -> lua split table -> null:receive() x4")
+    local run
+    for run = 1, runs do
+        local i = require("dnsjit.input.zero").new()
+        local o1 = require("dnsjit.output.null").new()
+        local o2 = require("dnsjit.output.null").new()
+        local o3 = require("dnsjit.output.null").new()
+        local o4 = require("dnsjit.output.null").new()
+
+        local prod, pctx = i:produce()
+        local recv, rctx = {}, {}
+
+        local f, c = o1:receive()
+        table.insert(recv, f)
+        table.insert(rctx, c)
+        f, c = o2:receive()
+        table.insert(recv, f)
+        table.insert(rctx, c)
+        f, c = o3:receive()
+        table.insert(recv, f)
+        table.insert(rctx, c)
+        f, c = o4:receive()
+        table.insert(recv, f)
+        table.insert(rctx, c)
+
+        local start_sec, start_nsec = clock:monotonic()
+        local idx = 1
+        for n = 1, num do
+            local f, c = recv[idx], rctx[idx]
+            if not f then
+                idx = 1
+                f, c = recv[1], rctx[1]
+            end
+            f(c, prod(pctx))
+            idx = idx + 1
+        end
+        local end_sec, end_nsec = clock:monotonic()
+
+        local runtime = 0
+        if end_sec > start_sec then
+            runtime = ((end_sec - start_sec) - 1) + ((1000000000 - start_nsec + end_nsec)/1000000000)
+        elseif end_sec == start_sec and end_nsec > start_nsec then
+            runtime = (end_nsec - start_nsec) / 1000000000
+        end
+
+        print(run, "runtime", runtime, num/runtime, "/sec", o1:packets() + o2:packets() + o3:packets() + o4:packets(), o1:packets(), o2:packets(), o3:packets(), o4:packets())
+    end
+
+    print("zero:receiver() -> lua split gen code -> null:receive() x4")
+    local run
+    for run = 1, runs do
+        local i = require("dnsjit.input.zero").new()
+        local o1 = require("dnsjit.output.null").new()
+        local o2 = require("dnsjit.output.null").new()
+        local o3 = require("dnsjit.output.null").new()
+        local o4 = require("dnsjit.output.null").new()
+
+        local prod, pctx = i:produce()
+        local f1, c1 = o1:receive()
+        local f2, c2 = o2:receive()
+        local f3, c3 = o3:receive()
+        local f4, c4 = o4:receive()
+
+        local code = "return function (num, prod, pctx, f1, c1, f2, c2, f3, c3, f4, c4)\nlocal n = 0\nwhile n < num do\n"
+        code = code .. "f1(c1,prod(pctx))\n"
+        code = code .. "n = n + 1\n"
+        code = code .. "f2(c2,prod(pctx))\n"
+        code = code .. "n = n + 1\n"
+        code = code .. "f3(c3,prod(pctx))\n"
+        code = code .. "n = n + 1\n"
+        code = code .. "f4(c4,prod(pctx))\n"
+        code = code .. "n = n + 1\n"
+        code = code .. "end\n"
+        code = code .. "end"
+        local f = assert(loadstring(code))()
+
+        local start_sec, start_nsec = clock:monotonic()
+        f(num, prod, pctx, f1, c1, f2, c2, f3, c3, f4, c4)
+        local end_sec, end_nsec = clock:monotonic()
+
+        local runtime = 0
+        if end_sec > start_sec then
+            runtime = ((end_sec - start_sec) - 1) + ((1000000000 - start_nsec + end_nsec)/1000000000)
+        elseif end_sec == start_sec and end_nsec > start_nsec then
+            runtime = (end_nsec - start_nsec) / 1000000000
+        end
+
+        print(run, "runtime", runtime, num/runtime, "/sec", o1:packets() + o2:packets() + o3:packets() + o4:packets(), o1:packets(), o2:packets(), o3:packets(), o4:packets())
+    end
+end
+
+if getopt:val("t") then
+    print("zero:receiver() -> thread lua x1")
+    local run
+    for run = 1, runs do
+        local i = require("dnsjit.input.zero").new()
+        local c = require("dnsjit.core.channel").new()
+        local t = require("dnsjit.core.thread").new()
+
+        t:start(function(t)
+            local c = t:pop()
+
+            while true do
+                local o = c:get()
+                if o == nil then break end
+            end
+        end)
+        t:push(c)
+
+        local prod, pctx = i:produce()
+        local start_sec, start_nsec = clock:monotonic()
+        for n = 1, num do
+            c:put(prod(pctx))
+        end
+        c:close()
+        t:stop()
+        local end_sec, end_nsec = clock:monotonic()
+
+        local runtime = 0
+        if end_sec > start_sec then
+            runtime = ((end_sec - start_sec) - 1) + ((1000000000 - start_nsec + end_nsec)/1000000000)
+        elseif end_sec == start_sec and end_nsec > start_nsec then
+            runtime = (end_nsec - start_nsec) / 1000000000
+        end
+
+        print(run, "runtime", runtime, num/runtime, "/sec")
+    end
+
+    print("zero:receiver() -> thread lua x2")
+    local run
+    for run = 1, runs do
+        local i = require("dnsjit.input.zero").new()
+        local c = require("dnsjit.core.channel").new()
+        local c2 = require("dnsjit.core.channel").new()
+        local t = require("dnsjit.core.thread").new()
+        local t2 = require("dnsjit.core.thread").new()
+
+        local f = function(t)
+            local c = t:pop()
+
+            while true do
+                local o = c:get()
+                if o == nil then break end
+            end
+        end
+
+        t:start(f)
+        t2:start(f)
+        t:push(c)
+        t2:push(c2)
+
+        local prod, pctx = i:produce()
+        local start_sec, start_nsec = clock:monotonic()
+        for n = 1, num/2 do
+            c:put(prod(pctx))
+            c2:put(prod(pctx))
+        end
+        c:close()
+        c2:close()
+        t:stop()
+        t2:stop()
+        local end_sec, end_nsec = clock:monotonic()
+
+        local runtime = 0
+        if end_sec > start_sec then
+            runtime = ((end_sec - start_sec) - 1) + ((1000000000 - start_nsec + end_nsec)/1000000000)
+        elseif end_sec == start_sec and end_nsec > start_nsec then
+            runtime = (end_nsec - start_nsec) / 1000000000
+        end
+
+        print(run, "runtime", runtime, num/runtime, "/sec")
+    end
+
+    print("zero:receiver() -> thread lua x4")
+    local run
+    for run = 1, runs do
+        local i = require("dnsjit.input.zero").new()
+        local c = require("dnsjit.core.channel").new()
+        local c2 = require("dnsjit.core.channel").new()
+        local c3 = require("dnsjit.core.channel").new()
+        local c4 = require("dnsjit.core.channel").new()
+        local t = require("dnsjit.core.thread").new()
+        local t2 = require("dnsjit.core.thread").new()
+        local t3 = require("dnsjit.core.thread").new()
+        local t4 = require("dnsjit.core.thread").new()
+
+        local f = function(t)
+            local c = t:pop()
+
+            while true do
+                local o = c:get()
+                if o == nil then break end
+            end
+        end
+
+        t:start(f)
+        t2:start(f)
+        t3:start(f)
+        t4:start(f)
+        t:push(c)
+        t2:push(c2)
+        t3:push(c3)
+        t4:push(c4)
+
+        local prod, pctx = i:produce()
+        local start_sec, start_nsec = clock:monotonic()
+        for n = 1, num/4 do
+            c:put(prod(pctx))
+            c2:put(prod(pctx))
+            c3:put(prod(pctx))
+            c4:put(prod(pctx))
+        end
+        c:close()
+        c2:close()
+        c3:close()
+        c4:close()
+        t:stop()
+        t2:stop()
+        t3:stop()
+        t4:stop()
+        local end_sec, end_nsec = clock:monotonic()
+
+        local runtime = 0
+        if end_sec > start_sec then
+            runtime = ((end_sec - start_sec) - 1) + ((1000000000 - start_nsec + end_nsec)/1000000000)
+        elseif end_sec == start_sec and end_nsec > start_nsec then
+            runtime = (end_nsec - start_nsec) / 1000000000
+        end
+
+        print(run, "runtime", runtime, num/runtime, "/sec")
+    end
+
+    print("zero:receiver() -> thread lua x1 -> null:receiver()")
+    local run
+    for run = 1, runs do
+        local i = require("dnsjit.input.zero").new()
+        local c = require("dnsjit.core.channel").new()
+        local t = require("dnsjit.core.thread").new()
+
+        t:start(function(t)
+            local c = t:pop()
+            local o = require("dnsjit.output.null").new()
+
+            local recv, rctx = o:receive()
+            while true do
+                local obj = c:get()
+                if obj == nil then break end
+                recv(rctx, obj)
+            end
+        end)
+        t:push(c)
+
+        local prod, pctx = i:produce()
+        local start_sec, start_nsec = clock:monotonic()
+        for n = 1, num do
+            c:put(prod(pctx))
+        end
+        c:close()
+        t:stop()
+        local end_sec, end_nsec = clock:monotonic()
+
+        local runtime = 0
+        if end_sec > start_sec then
+            runtime = ((end_sec - start_sec) - 1) + ((1000000000 - start_nsec + end_nsec)/1000000000)
+        elseif end_sec == start_sec and end_nsec > start_nsec then
+            runtime = (end_nsec - start_nsec) / 1000000000
+        end
+
+        print(run, "runtime", runtime, num/runtime, "/sec")
+    end
+
+    print("zero:receiver() -> thread x1 -> null:receiver()")
+    local run
+    for run = 1, runs do
+        local i = require("dnsjit.input.zero").new()
+        local c = require("dnsjit.core.channel").new()
+        local t = require("dnsjit.core.thread").new()
+
+        t:start(function(t)
+            local c = t:pop()
+            local o = require("dnsjit.output.null").new()
+
+            c:receiver(o)
+            c:run()
+        end)
+        t:push(c)
+
+        i:receiver(c)
+        local start_sec, start_nsec = clock:monotonic()
+        i:run(num)
+        c:close()
+        t:stop()
+        local end_sec, end_nsec = clock:monotonic()
+
+        local runtime = 0
+        if end_sec > start_sec then
+            runtime = ((end_sec - start_sec) - 1) + ((1000000000 - start_nsec + end_nsec)/1000000000)
+        elseif end_sec == start_sec and end_nsec > start_nsec then
+            runtime = (end_nsec - start_nsec) / 1000000000
+        end
+
+        print(run, "runtime", runtime, num/runtime, "/sec")
+    end
+end