fiber-mon-repro.lua

-- Tarantool 3.0.2-0-gf9e82c5 Darwin-arm64-RelWithDebInfo
local fiber = require 'fiber'
local clock = require 'clock'
local log   = require 'log'

box.cfg{checkpoint_count = 1,listen=3301}
box.ctl.promote()

require 'jit'.on()
fiber.top_enable()

local ns = 1
local mks = 1000*ns
local ms = 1000*mks

local do_stress = function()
	fiber.yield()
	fiber.name("stress")
	local N = 1000
	while N > 0 do
		for _ in box.space._space:pairs() do
			N = N - 1
			if N % 100 == 0 then fiber.yield() end
			if N <= 0 then break end
		end
	end
end

rawset(_G, 'do_stress', do_stress)

local pool = {
	chan = fiber.channel(),
	worker = function(self)
		fiber.name("pool/worker")

		while true do
			fiber.testcancel()
			local pkt = self.chan:get()
			if not pkt then
				break
			end
			local func = table.remove(pkt, 1)
			func(unpack(pkt))
		end

		print("leaving")
	end,
}

rawset(_G, 'pool', pool)

for _ = 1, 8 do fiber.create(pool.worker, pool) end

-- fiber.create(function()
-- 	fiber.name("stress-commander")
-- 	fiber.yield()

-- 	while true do
-- 		pool.chan:put("do-stress")
-- 		fiber.sleep(0.001)
-- 	end
-- end)

local COLLECT_INTERVAL = 1
fiber.create(function()
	fiber.name("monitor")
	fiber.yield()

	local prev = {}
	local tx_time = clock.thread()
	local tx_time_ns = clock.thread64()

	while true do
		local cpu = fiber.top().cpu
		local info = fiber.info({ bt = false })

		local fids = {}
		for fid in pairs(info) do
			table.insert(fids, fid)
		end
		table.sort(fids)

		print("------------------------------------------------------")

		local run_diff = 0

		-- collect sched
		if prev[1] then
			local diff = cpu['1/sched'].time - prev[1]
			print(1, "sched", 1000*diff)
			run_diff = run_diff + diff
			prev[1] = cpu['1/sched'].time
		end
		prev[1] = cpu['1/sched'].time

		for _, fid in ipairs(fids) do
			local finfo = info[fid]
			local cpu_time = cpu[fid..'/'..finfo.name]
			cpu_time = cpu_time and cpu_time.time or finfo.time
			if prev[fid] then
				local diff = cpu_time - prev[fid]
				run_diff = run_diff + diff
				print(fid, finfo.name, 1000*diff)
			end
			prev[fid] = cpu_time
		end

		local current_tx = clock.thread()
		print("tx ", "tx", 1000*(current_tx-tx_time))
		print("diff", "diff", 1000*run_diff)
		tx_time = current_tx

		fiber.sleep(COLLECT_INTERVAL)
	end
end)


-- client fibers:
local fibs = {}

local netbox = require 'net.box'
local tnt = netbox.connect('127.0.0.1:3301')
assert(tnt:ping())

local N = 1e6

local do_replace = function(n)
	local fib = require 'fiber'
	fib.yield()
	local limit = 100
	box.begin()
		for i = 1, 1000 do
			limit = limit -1
			box.space.test:replace({(n + i)%1e6, fib.id(), fib.time()})
			if limit <= 0 then
				box.commit()
				fib.yield()
				box.begin()
				limit = 100
			end
		end
	box.commit()
end

rawset(_G, 'do_replace', do_replace)

for _ = 1, 8 do
	local fib = fiber.create(function(conn)
		fiber.self():set_joinable(true)
		fiber.name('client')

		local func = function(n)
			-- local fib = require 'fiber'
			-- -- fib.yield()
			-- fib.name('iproto-stress')
			-- _G.pool.chan:put({_G.do_replace, n})
			-- -- fib.yield()

			local fib = require 'fiber'
			fib.yield()
			local limit = 100
			box.begin()
				for i = 1, 1000 do
					limit = limit -1
					box.space.test:replace({(n + i)%1e6, fib.id(), fib.time()})
					if limit <= 0 then
						box.commit()
						fib.yield()
						box.begin()
						limit = 100
					end
				end
			box.commit()
		end

		local fstr = string.dump(func)

		while N > 0 do
			N = N - 1
			conn:call('dostring', { [[local f, n = ... return loadstring(f)(n)]], fstr, N }, { timeout = 1 })
		end
	end, tnt)

	table.insert(fibs, fib)
end

for _, fib in ipairs(fibs) do
	print(fib:join())
end

log.info("exiting")
os.exit(0)