<commit_before>local dpdk		= require "dpdk"
local memory	= require "memory"
local device	= require "device"
local ts		= require "timestamping"
local dpdkc		= require "dpdkc"
local filter	= require "filter"
local ffi		= require "ffi"

local incAndWrap = incAndWrap

-- set addresses here
local srcmac = "00:42:00:00:00:01"
local dstmac = "00:42:00:00:00:02"
local dstip = 0x0A000001
local srcip = 0x0A000002

function master(...)
	local txPort, rxPort, rate, flows, size = tonumberall(...)
	if not txPort or not rxPort then
		errorf("usage: txPort rxPort [rate [flows [size]]]")
	end
	flows = flows or 4
	rate = rate or 2000
	size = (size or 128) - 4 -- 4 bytes for CRC will be added by NIC
	local rxMempool = memory.createMemPool()
	if txPort == rxPort then
		txDev = device.config(txPort, rxMempool, 2, 2)
		rxDev = txDev
		txDev:wait()
	else
		txDev = device.config(txPort, rxMempool, 1, 2)
		rxDev = device.config(rxPort, rxMempool, 2, 1)
		device.waitForLinks()
	end
	txDev:getTxQueue(1):setRate(rate)
	dpdk.launchLua("timerSlave", txPort, rxPort, 0, 1, size, flows)
	dpdk.launchLua("loadSlave", txPort, 1, size, flows)
	dpdk.launchLua("counterSlave", rxPort, size)
	dpdk.waitForSlaves()
end

function loadSlave(port, queue, size, numFlows)
	local queue = device.get(port):getTxQueue(queue)
	local mempool = memory.createMemPool(function(buf)
		-- TODO: port this to new PTP/timestamping functions, ffi.cast in a userscript is not the point
		ts.fillPacket(buf, 1234, size)
		local data = ffi.cast("uint8_t*", buf.pkt.data)
		data[43] = 0x00 -- PTP version, set to 0 to disable timestamping for load packets
		local pkt = buf:getUdpPacket()
		pkt.eth.src:setString(srcmac)
		pkt.eth.dst:setString(dstmac)
		pkt.ip4.dst:set(dstip)
	end)
	local lastPrint = dpdk.getTime()
	local totalSent = 0
	local lastTotal = 0
	local lastSent = 0
	local bufs = mempool:bufArray(128)
	local baseIP = srcip
	local counter = 0
	while dpdk.running() do
		bufs:alloc(size)
		for i, buf in ipairs(bufs) do
			local pkt = buf:getUdpPacket()
			pkt.ip4.src:set(baseIP + counter)
			counter = incAndWrap(counter, numFlows)
		end
		-- UDP checksums are optional, so using just IPv4 checksums would be sufficient here
		bufs:offloadUdpChecksums()
		totalSent = totalSent + queue:send(bufs)
		local time = dpdk.getTime()
		if time - lastPrint > 1 then
			local mpps = (totalSent - lastTotal) / (time - lastPrint) / 10^6
			printf("Sent %d packets, current rate %.2f Mpps, %.2f MBit/s, %.2f MBit/s wire rate", totalSent, mpps, mpps * (size + 4) * 8, mpps * (size + 24) * 8)
			lastTotal = totalSent
			lastPrint = time
		end
	end
	printf("Sent %d packets", totalSent)
end

function counterSlave(port)
	local dev = device.get(port)
	local total = 0
	while dpdk.running() do
		local time = dpdk.getTime()
		dpdk.sleepMillis(1000)
		local elapsed = dpdk.getTime() - time
		local pkts = dev:getRxStats(port)
		total = total + pkts
		printf("Received %d packets, current rate %.2f Mpps", total, pkts / elapsed / 10^6)
	end
	printf("Received %d packets", total)
end

function timerSlave(txPort, rxPort, txQueue, rxQueue, size, numFlows)
	local txDev = device.get(txPort)
	local rxDev = device.get(rxPort)
	local txQueue = txDev:getTxQueue(txQueue)
	local rxQueue = rxDev:getRxQueue(rxQueue)
	local mem = memory.createMemPool()
	local bufs = mem:bufArray(1)
	local rxBufs = mem:bufArray(128)
	txQueue:enableTimestamps()
	rxQueue:enableTimestamps(1234)
	rxDev:filterTimestamps(rxQueue)
	local hist = {}
	-- wait one second, otherwise we might start timestamping before the load is applied
	dpdk.sleepMillis(1000)
	local counter = 0
	local baseIP = dstip
	while dpdk.running() do
		bufs:alloc(size + 4)
		local pkt = bufs[1]:getUdpPacket()
		ts.fillPacket(bufs[1], 1234, size)
		pkt.eth.src:setString(srcmac)
		pkt.eth.dst:setString(dstmac)
		pkt.ip4.src:set(baseIP + counter)
		pkt.ip4.dst:set(dstip)
		counter = (counter + 1) % numFlows
		bufs:offloadUdpChecksums()
		-- sync clocks and send
		ts.syncClocks(txDev, rxDev)
		txQueue:send(bufs)
		-- increment the wait time when using large packets or slower links
		local tx = txQueue:getTimestamp(100)
		if tx then
			dpdk.sleepMicros(500) -- minimum latency to limit the packet rate
			-- sent was successful, try to get the packet back (max. 10 ms wait time before we assume the packet is lost)
			local rx = rxQueue:tryRecv(rxBufs, 10000)
			if rx > 0 then
				local numPkts = 0
				for i = 1, rx do
					if bit.bor(rxBufs[i].ol_flags, dpdk.PKT_RX_IEEE1588_TMST) ~= 0 then
						numPkts = numPkts + 1
					end
				end
				local delay = (rxQueue:getTimestamp() - tx) * 6.4
				if numPkts == 1 then
					if delay > 0 and delay < 100000000 then
						hist[delay] = (hist[delay] or 0) + 1
					end
				end -- else: got more than one packet, so we got a problem
				-- TODO: use sequence numbers in the packets to avoid bugs here
				rxBufs:freeAll()
			end
		end
	end
	local sortedHist = {}
	for k, v in pairs(hist) do 
		table.insert(sortedHist,  { k = k, v = v })
	end
	local sum = 0
	local samples = 0
	table.sort(sortedHist, function(e1, e2) return e1.k < e2.k end)
	print("Histogram:")
	for _, v in ipairs(sortedHist) do
		sum = sum + v.k * v.v
		samples = samples + v.v
		print(v.k, v.v)
	end
	print()
	print("Average: " .. (sum / samples) .. " ns, " .. samples .. " samples")
	print("----------------------------------------------")
end
<commit_msg>l3-load-latency: refactor and modernize, now with ARP<commit_after>local dpdk		= require "dpdk"
local memory	= require "memory"
local device	= require "device"
local ts		= require "timestamping"
local filter	= require "filter"
local hist		= require "histogram"
local stats		= require "stats"
local timer		= require "timer"
local arp		= require "proto.arp"

-- set addresses here
local DST_MAC	= nil -- resolved via ARP on GW_IP or DST_IP, can be overriden with a string here
local SRC_IP	= "10.0.0.10"
local DST_IP	= "10.1.0.10"
local SRC_PORT	= 1234
local DST_PORT	= 1234

-- answer ARP requests for this IP on the rx port
-- change this if benchmarking something like a NAT device
local RX_IP		= DST_IP
-- used to resolve DST_MAC
local GW_IP		= DST_IP

function master(...)
	local txPort, rxPort, rate, flows, size = tonumberall(...)
	if not txPort or not rxPort then
		errorf("usage: txPort rxPort [rate [flows [pktSize]]]")
	end
	flows = flows or 4
	rate = rate or 2000
	size = (size or 124)
	txDev = device.config(txPort, 3, 3)
	rxDev = device.config(rxPort, 3, 3)
	device.waitForLinks()
	-- max 1mbit timestamping traffic timestamping
	-- rate will be somewhat off for high-latency links at low rates
	txDev:getTxQueue(0):setRate(rate - 1)
	dpdk.launchLua("loadSlave", txDev:getTxQueue(0), rxDev, size, flows)
	dpdk.launchLua("timerSlave", txDev:getTxQueue(1), rxDev:getRxQueue(1), size, flows)
	dpdk.launchLua(arp.arpTask, {
		-- run ARP on both ports
		{ rxQueue = rxDev:getRxQueue(2), txQueue = rxDev:getTxQueue(2), ips = RX_IP },
		{ rxQueue = txDev:getRxQueue(2), txQueue = txDev:getTxQueue(2), ips = SRC_IP }
	})
	dpdk.waitForSlaves()
end

local function fillUdpPacket(buf)
	buf:getUdpPacket():fill{
		ethSrc = queue,
		ethDst = DST_MAC,
		ip4Src = SRC_IP,
		ip4Dst = DST_IP,
		udpSrc = SRC_PORT,
		udpDst = DST_PORT
	}
end

local function doArp()
	if not DST_MAC then
		printf("Performing ARP lookup on %s", GW_IP)
		DST_MAC = arp.blockingLookup(GW_IP, 5)
		if not DST_MAC then
			printf("ARP lookup failed, using default destination mac address")
			return
		end
	end
	printf("Destination mac: %s", DST_MAC)
end

function loadSlave(queue, rxDev, size, flows)
	doArp()
	local mempool = memory.createMemPool(fillUdpPacket)
	local bufs = mempool:bufArray()
	local counter = 0
	local txCtr = stats:newDevTxCounter(queue, "plain")
	local rxCtr = stats:newDevRxCounter(rxDev, "plain")
	while dpdk.running() do
		bufs:alloc(size)
		for i, buf in ipairs(bufs) do
			local pkt = buf:getUdpPacket()
			pkt.udp:setSrcPort(SRC_PORT + counter)
			counter = incAndWrap(counter, flows)
		end
		-- UDP checksums are optional, so using just IPv4 checksums would be sufficient here
		bufs:offloadUdpChecksums()
		queue:send(bufs)
		txCtr:update()
		rxCtr:update()
	end
	txCtr:finalize()
	rxCtr:finalize()
end

function timerSlave(txQueue, rxQueue, size, flows)
	doArp()
	rxQueue.dev:filterTimestamps(rxQueue)
	local timestamper = ts:newUdpTimestamper(txQueue, rxQueue)
	local hist = hist:new()
	dpdk.sleepMillis(1000) -- ensure that the load task is running
	local counter = 0
	local rateLimit = timer:new(0.001)
	while dpdk.running() do
		hist:update(timestamper:measureLatency(size, function(buf)
			fillUdpPacket(buf)
			local pkt = buf:getUdpPacket()
			pkt.udp:setSrcPort(SRC_PORT + counter)
			counter = incAndWrap(counter, flows)
		end))
		rateLimit:wait()
		rateLimit:reset()
	end
	-- print the latency stats after all the other stuff
	dpdk.sleepMillis(300)
	hist:print()
	hist:save("histogram.csv")
end

{'fraction_non_alphanumeric': 0.11776626885836523, 'fraction_numerical': 0.025669894167980184, 'mean_word_length': 3.9295227524972254, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 1, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6691206', 'n_tokens_mistral': 3644, 'n_tokens_neox': 3291, 'n_words': 962}