Task Manager

CMakeLists.txt

@@ -39,6 +39,9 @@ link_directories(
 	${CMAKE_CURRENT_SOURCE_DIR}/libmoon/deps/luajit/usr/local/lib
 )
 
+add_custom_target(tman ALL
+  COMMAND ${CMAKE_COMMAND} -E create_symlink MoonGen tman)
+
 add_executable(MoonGen ${files})
 target_link_libraries(MoonGen ${libraries})
 

README.md

@@ -134,6 +134,19 @@ MoonGen prints all available ports on startup, so adjust this if necessary.
 You can also check out the examples of the [libmoon](https://github.com/libmoon/libmoon) project.
 All libmoon scripts are also valid MoonGen scripts as MoonGen extends libmoon.
 
+## Using tman (task manager)
+A newer CLI interface allows to start a number of various tasks at once. Here is the synopsis:
+```
+ ./build/tman [tman options...] [-- <task1-file> [task1 options...]] [-- <task2-file> [task2 options...]] [-- ...]
+```
+Also `./build/tman -h` gives more enhanced help.
+
+Ordinary task Lua files, however, are incompatible with tman interface. A task file for `tman` has to determine the following objects:
+- `function configure(parser)` which describes task's options.
+- `function task(taskNum, txInfo, rxInfo, args)` which describes what is being run. Here `txInfo` and `rxInfo` are arrays of RX and TX queues respectively, `args` keeps the parameters of the task manager and the task itself.
+
+See examples in `examples/tman/`.
+
 # Frequently Asked Questions
 
 ### Which NICs do you support?

examples/delay-stream.lua

@@ -0,0 +1,341 @@
+local mg		= require "moongen"
+local memory	= require "memory"
+local device	= require "device"
+local stats		= require "stats"
+local log 		= require "log"
+local ip4		= require "proto.ip4"
+local libmoon	= require "libmoon"
+local dpdkc		= require "dpdkc"
+local ffi		= require "ffi"
+local syscall	= require "syscall"
+require "pepperfish"
+
+function configure(parser)
+    -- do nothing, just check parse errors
+	function convertMac_fake(str)
+		mac = parseMacAddress(str, true)
+		if not mac then
+			parser:error("failed to parse MAC "..str)
+		end
+		return str
+	end
+
+	function convertTime(str)
+		local pattern = "^(%d+)([mu]?s)$"
+		local _, _, n, unit = string.find(str, pattern)
+		if not (n and unit) then
+			parser:error("failed to parse time '"..str.."', it should match '"..pattern.."' pattern")
+		end
+		return {n=tonumber(n), unit=unit}
+	end
+
+	parser:description("Redirects stream between devices, optionally mangling and delaying data.")
+	parser:argument("srcDev", "Source device."):convert(tonumber)
+	parser:argument("dstDev", "Destination device."):convert(tonumber)
+	parser:option("-n", "Number of threads."):convert(tonumber):default(1)
+	parser:option("-d --destination", "Rewrite destination IP.")
+	parser:option("-m --ethDst", "Rewrite destination MAC."):convert(convertMac_fake)
+	parser:option("--delay", "Delay in us."):convert(tonumber):default(0)
+	parser:option("--buf", "Buffer size."):convert(tonumber):default(63)
+	parser:option("-l --max-pktlen", "Maximum packet length."):convert(tonumber):default(60)
+	parser:flag("--feed-self", "Do not drop traffic generated by ourselves.")
+	parser:flag("--fake-write", "Do not actually issue traffic.")
+	parser:flag("--debug", "Debug ring buffer.")
+	parser:flag("--profile-read", "Turn on profiling for reading.")
+	parser:flag("--profile-write", "Turn no profiling for writing.")
+end
+
+local shm_filename = "/MoonGen"
+
+local function setup(isMaster, taskNum, args)
+	if args.debug then
+		log:setLevel("DEBUG")
+	end
+
+	local cfg = {}
+	local shm_filename = shm_filename .. "-" .. tostring(taskNum)
+
+	cfg.chunkSize = args.buf + 1
+	cfg.packetSpace = args.max_pktlen + 4
+	local delay = args.delay; if delay <= 0 then delay = 1 end
+	local extra = 10.0 -- just in case
+	local _, ringLogsize = math.frexp(14.88 * delay * extra / args.buf)
+	if ringLogsize < 2 then ringLogsize = 2 end
+	cfg.ringSize = bit.lshift(1, ringLogsize)
+	cfg.ringLogsize = ringLogsize
+	if isMaster and taskNum == 1 then
+		log:info("buf size: %d, ringSize: %d, e: %d", args.buf, cfg.ringSize, ringLogsize)
+	end
+	cfg.ringDblSize = cfg.ringSize * 2
+	cfg.ringDblMask = 2 * cfg.ringDblSize - 1
+
+	local oflags
+	if isMaster then oflags = "CREAT,RDWR,EXCL" else oflags = "RDWR" end
+
+	local shm_fd, err = syscall.shm_open(shm_filename, oflags, "RUSR,WUSR")
+	if not shm_fd then
+		if err.EXIST then
+			-- try to unlink and retry to open
+			syscall.shm_unlink(shm_filename)
+			shm_fd, err = syscall.shm_open(shm_filename, oflags, "RUSR,WUSR")
+		end
+		if not shm_fd then
+			errorf("shm_open(\"%s\", ...) failed: %s", shm_filename, err)
+		end
+	end
+	local ringRawSize = cfg.ringSize * cfg.chunkSize * cfg.packetSpace
+	local shm_size = ringRawSize + 8
+	local ok, err = syscall.ftruncate(shm_fd, shm_size)
+	if not ok then
+		errorf("ftruncate(...) failed: %s", err)
+	end
+	local ptr, err = syscall.mmap(nil, shm_size, "READ,WRITE", "SHARED", shm_fd, 0)
+	if not ptr then
+		errorf("nmap(...) failed: %s", err)
+	end
+	cfg.ringBufferRaw = ffi.cast("uint8_t*", ptr)
+	cfg.ax_ptr = ffi.cast("volatile uint32_t*", cfg.ringBufferRaw+ringRawSize)
+	cfg.bx_ptr = ffi.cast("volatile uint32_t*", cfg.ringBufferRaw+ringRawSize+4)
+
+	if isMaster then
+		cfg.ax_ptr[0] = 0
+		cfg.bx_ptr[0] = 0
+		cfg.cleanupFunc = function()
+			syscall.munmap(ptr, shm_size)
+			syscall.close(shm_fd)
+			syscall.shm_unlink(shm_filename)
+		end
+	else
+		cfg.cleanupFunc = function() syscall.munmap(ptr, shm_size); syscall.close(shm_fd) end
+	end
+
+	return cfg
+end
+
+function master(args)
+	if bit.band(args.buf, args.buf + 1) ~= 0 then
+		log:warn("Recommended buffer size is 2^n-1")
+	end
+
+	local n = args.n
+	local cfgs = {}
+	for i = 1, n do
+		cfgs[i] = setup(true, i, args)
+	end
+
+	local srcDev, dstDev
+	if args.srcDev == args.dstDev then
+		srcDev = device.config{port = args.srcDev, rxQueues = n, txQueues = n, rssQueues = n}
+		dstDev = srcDev
+		srcDev:wait()
+	else
+		srcDev = device.config{port = args.srcDev, rxQueues = n, rssQueues = n}
+		dstDev = device.config{port = args.dstDev, txQueues = n}
+		srcDev:wait()
+		dstDev:wait()
+	end
+
+	for i = 0, n - 1 do
+		mg.startTask("task_read", i+1, srcDev:getRxQueue(i), dstDev:getMac(true), args)
+		mg.startTask("task_write", i+1, dstDev:getTxQueue(i), args)
+	end
+
+	mg.waitForTasks()
+	for _, cfg in ipairs(cfgs) do cfg.cleanupFunc() end
+end
+
+local function gettimeofday_n()
+	local sec, usec = gettimeofday()
+	return tonumber(sec), tonumber(usec)
+end
+
+local zero16 = hton16(0)
+
+function task_read(taskNum, rxQ, ethSelf, args)
+	local cfg = setup(false, taskNum, args)
+	local ringDblMask = cfg.ringDblMask
+	local ax_ptr = cfg.ax_ptr
+	local bx_ptr = cfg.bx_ptr
+	local ringBufferRaw = cfg.ringBufferRaw
+	local chunkSize = cfg.chunkSize
+	local packetSpace = cfg.packetSpace
+	local ringMask = cfg.ringSize - 1
+	local ringDblSize = cfg.ringDblSize
+
+	local max_pktlen, feed_self = args.max_pktlen, args.feed_self
+
+	local rxStats = stats:newDevRxCounter(rxQ, "plain")
+	local manStats = stats:newManualRxCounter("man:" .. tostring(taskNum))
+	local timesBufferFull = 0
+
+	local mem = memory.createMemPool()
+	local bufs_read = mem:bufArray(chunkSize - 1)
+
+	-- ring buffer pointers (not using "tx", "rx" due to confusion), ax is to be "greater" than bx
+	local ax, bx = 0, 0
+
+	if args.profile_read then
+		profiler = newProfiler("call")
+		profiler:start()
+	end
+
+	while mg.running() do
+		bx = bx_ptr[0]
+		if bit.band(ax - bx, ringDblSize) == 0 -- ring not full
+		then
+			local sec, usec = gettimeofday_n()
+			local rx = rxQ:recv(bufs_read)
+			manStats:update(rx, 0)
+
+			local chunkIdx = bit.band(ax, ringMask)
+			local j = 0
+			for i = 1, rx do
+				local buf = bufs_read.array[i-1]
+				local pkt = buf:getTcpPacket(ipv4)
+				local pkt_raw = ringBufferRaw + ((chunkIdx * chunkSize + j) * packetSpace)
+				local bufSize = buf:getSize()
+				if bufSize <= max_pktlen and (feed_self or pkt.eth.src:get() ~= ethSelf)
+				then
+					ffi.cast("uint32_t*", pkt_raw)[0] = bufSize
+					ffi.copy(pkt_raw + 4, buf:getRawPacket(), bufSize)
+					j = j + 1
+				end
+			end
+
+			local meta_raw = ffi.cast("volatile uint32_t*", ringBufferRaw + (((chunkIdx + 1) * chunkSize - 1) * packetSpace))
+			meta_raw[0], meta_raw[1], meta_raw[2] = rx, sec, usec
+			log:debug("READ-%d %03x, %03x", taskNum, ax, bx)
+			ax = bit.band(ax + 1, ringDblMask)
+			ax_ptr[0] = ax
+
+			rxStats:update()
+			bufs_read:freeAll()
+		else
+			if timesBufferFull == 0 then
+				log:warn("Buffer is full")
+			end
+			timesBufferFull = timesBufferFull + 1
+		end
+	end
+
+	if args.profile_read then
+		profiler:stop()
+		local outfile = io.open( "profile-read.txt", "w+" )
+		profiler:report(outfile)
+		outfile:close()
+	end
+
+	rxStats:finalize()
+	manStats:finalize()
+	if timesBufferFull ~= 0 then
+		log:warn("Buffer was full %d times", timesBufferFull)
+	end
+	cfg.cleanupFunc()
+end
+
+function task_write(taskNum, txQ, args)
+	local cfg = setup(false, taskNum, args)
+	local ringDblMask = cfg.ringDblMask
+	local ax_ptr = cfg.ax_ptr
+	local bx_ptr = cfg.bx_ptr
+	local ringBufferRaw = cfg.ringBufferRaw
+	local chunkSize = cfg.chunkSize
+	local packetSpace = cfg.packetSpace
+	local ringMask = cfg.ringSize - 1
+
+	local delay, max_pktlen, fake_write = args.delay, args.max_pktlen, args.fake_write
+
+	local ipDst = args.destination and parseIPAddress(args.destination)
+	local ethDst = args.ethDst and parseMacAddress(args.ethDst, true)
+	local ethSrc = txQ.dev:getMac(true)
+
+	local txStats = stats:newDevTxCounter(txQ, "plain")
+
+	local mem = memory.createMemPool()
+	local bufs_write = mem:bufArray(chunkSize - 1)
+
+	-- ring buffer pointers (not using "tx", "rx" due to confusion), ax is to be "greater" than bx
+	local ax, bx = 0, 0
+
+	local function do_write()
+		ax = ax_ptr[0]
+		while bit.band(ax - bx, ringDblMask) ~= 0 -- ring not empty
+		do
+			if not fake_write then
+				local chunkIdx = bit.band(bx, ringMask)
+				local meta_raw = ffi.cast("volatile uint32_t*", ringBufferRaw + (((chunkIdx + 1) * chunkSize - 1) * packetSpace))
+				local sec, usec = gettimeofday_n()
+				local delta_usec = (sec - meta_raw[1]) * 1000000 + (usec - meta_raw[2])
+
+				if delta_usec > delay then
+					log:debug("WRITE-%d %03x, %03x, delta: %d", taskNum, ax, bx, delta_usec)
+
+					local nRecvd = meta_raw[0]
+					if nRecvd > chunkSize - 1 then
+						nRecvd = chunkSize - 1
+						log:warn("Probably garbage in ring buffer (nRecvd): %d > %d", nRecvd, chunkSize-1)
+					end
+					bufs_write:resize(nRecvd)
+					bufs_write:alloc(max_pktlen)
+
+					local i = 0
+					for j = 0, nRecvd-1 do
+						local pkt_raw = ringBufferRaw + ((chunkIdx * chunkSize + j) * packetSpace)
+						local pktSize = ffi.cast("uint32_t*", pkt_raw)[0]
+						if pktSize ~= 0 then
+							i = i + 1
+							local buf = bufs_write.array[i-1]
+							buf:setSize(pktSize)
+							ffi.copy(buf:getData(), pkt_raw+4, pktSize)
+
+							local pkt = buf:getTcpPacket(ipv4)
+							if pkt.ip4:getProtocol() == ip4.PROTO_TCP then
+								if ethDst then
+									pkt.eth.dst:set(ethDst)
+									pkt.eth.src:set(ethSrc)
+								end
+								if ipDst then pkt.ip4.dst:set(ipDst) end
+								--pkt.ip4:setChecksum(0)
+								pkt.ip4.cs = zero16 -- FIXME: setChecksum() is extremely slow
+							end
+						end
+					end
+					if i ~= 0 then
+						bufs_write:resize(i)
+						bufs_write:offloadTcpChecksums(ipv4)
+						txQ:send(bufs_write)
+					end
+					txStats:update()
+					bx = bit.band(bx + 1, ringDblMask)
+					bx_ptr[0] = bx
+				else
+					if delta_usec < 0 then
+						log:warn("Oops, something wrong with time")
+					end
+					return
+				end
+			else
+				bx = bit.band(bx + 1, ringDblMask)
+				bx_ptr[0] = bx
+			end
+		end
+	end
+
+	if args.profile_write then
+		profiler = newProfiler("call")
+		profiler:start()
+	end
+
+	while mg.running() do
+		do_write()
+	end
+		if args.profile_write then
+		profiler:stop()
+		local outfile = io.open( "profile-write.txt", "w+" )
+		profiler:report(outfile)
+		outfile:close()
+	end
+
+	txStats:finalize()
+	cfg.cleanupFunc()
+end