1853890102

Committed 04 Jun 2025 08:02PM UTC coverage: 87.0% (+3.8%) from 83.158%

Build Type

push

gitlab-ci

Committed by Vladislav Grubov

Commit Message

fix: upto0

Coverage Stats

87 of 100 relevant lines covered (87.0%)

247147.42 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.0

/fun/cpu_limit.lua

local fun = require 'fun'
local fun_iter = fun.iter

local methods = debug.getmetatable(fun.ones()).__index

local rawiter = function(gen, param, state)
        return fun_iter(gen, param, state):unwrap()
end

local method1 = function(func)
    return function(self, arg1)
        return func(arg1, self.gen, self.param, self.state)
    end
end

local export1 = function(func)
    return function(arg1, gen, param, state)
        return func(arg1, rawiter(gen, param, state))
    end
end

local fiber = require 'fiber'
local clock = require 'clock'
-- local log = require 'log'
-- if log.new then log = log.new('fun.cpu_limit') end

local function ev_time_mks() return fiber.time64() end
local function realtime_mks() return clock.realtime64()/1e3 end
local function thread_time_mks() return clock.thread64()/1e3 end

local upto0 = function(v)
        if v < 0 then return 0 end
        return v
end

---@param time number
local yield_sleep = function(time)
        fiber.sleep(time)
end

local build_yield_commit = function(yield_with)
        return function(time)
                local is_committed = false
                if type(box.cfg) ~= 'function' and box.is_in_txn and box.is_in_txn() then
                        box.commit()
                        is_committed = true
                end
                yield_with(time)
                if is_committed then box.begin() end
        end
end

local build_yield_rollback = function(yield_with)
        return function(time)
                if type(box.cfg) ~= 'function' and box.is_in_txn and box.is_in_txn() then
                        box.rollback()
                end
                return yield_with(time)
        end
end

local build_yield_raise = function(time)
        return function(yield_with)
                if type(box.cfg) ~= 'function' and box.is_in_txn and box.is_in_txn() then
                        error("Transaction left openned", 3)
                end
                return yield_with(time)
        end
end

local yield_commit = build_yield_commit(yield_sleep)
local yield_rollback = build_yield_rollback(yield_sleep)
local yield_raise = build_yield_raise(yield_sleep)

local mks = 1
local sec = 1e6
local ms = 1000*mks
local max_ev_run_mks = 100*ms

local cpu_limit_gen_x = function(state, state_x, ...)
        if state_x == nil then
                return nil
        end

        state[1] = state_x
        -- state[2] = quota_mks
        -- state[3] = ev_mks
        -- state[4] = thread_mks

        return state, ...
end

local function update_quota(now_mks, max_mks, add_mks)
        local new_mks = now_mks + add_mks
        if new_mks > max_mks then
                new_mks = max_mks
        end
        -- log.info("quota: %.1fµs -> %.1fµs", now_mks, new_mks)
        return new_mks
end

local rand = math.random

local cpu_limit_gen = function(param, state)
        local max_quota_mks, sleep, gen_x, param_x = param[1], param[2], param[3], param[4]
        local state_x, quota_mks, prev_ev_mks, prev_thread_mks = state[1], state[2], state[3], state[4]

        if rand() > 0.5 then
                return cpu_limit_gen_x(state, gen_x(param_x, state_x))
        end

        local ev_mks = ev_time_mks()
        local thread_mks = thread_time_mks()
        local wall_mks = realtime_mks()

        if prev_ev_mks == ev_mks then
                -- the caller spent all this time on cpu
                -- decrease quota
                quota_mks = update_quota(quota_mks, max_quota_mks, -tonumber(thread_mks - prev_thread_mks))
                -- quota_mks = quota_mks - tonumber(thread_mks - prev_thread_mks)
        end

        local ev_loop_too_long = (ev_mks + max_ev_run_mks) < wall_mks
        if ev_loop_too_long or quota_mks <= 1 then
                -- it's time to yield
                -- Fiber needs to sleep:
                local sleep_sec = (max_ev_run_mks - (max_quota_mks - quota_mks))/sec
                sleep(upto0(sleep_sec))

                ev_mks = ev_time_mks()
                if ev_mks > wall_mks then
                        -- yes, any time drift does not add time quota.
                        local gained_mks = assert(tonumber(ev_mks-wall_mks))
                        quota_mks = update_quota(quota_mks, max_quota_mks, gained_mks*max_quota_mks/max_ev_run_mks)
                end
        end

        -- state[1] = state_x
        state[2] = quota_mks
        state[3] = ev_mks
        state[4] = thread_mks

        return cpu_limit_gen_x(state, gen_x(param_x, state_x))
end

local cpu_limit = function(opts, gen_x, param_x, state_x)
        local cpu_limit
        local sleep_with = yield_sleep
        if type(opts) == 'table' then
                cpu_limit = tonumber(opts.cpu_limit)
                local yield_with = yield_sleep
                if opts.yield_with then
                        if type(opts.yield_with) ~= 'function' then
                                error("malformed cpu_limit/yield_with: must be a function when given", 2)
                        end
                        yield_with = opts.yield_with
                end

                ---sleep_builder is a builder function that can build specific fiber.sleep()
                ---default builders already exist
                ---@type (fun(yield_with: fun(time: number)):fun(time: number))?
                local sleep_builder = nil
                if opts.txn == 'commit' then
                        sleep_builder = build_yield_commit
                        sleep_with = yield_commit
                elseif opts.txn == 'rollback' then
                        sleep_builder = build_yield_rollback
                        sleep_with = yield_rollback
                elseif opts.txn == 'raise' then
                        sleep_builder = build_yield_raise
                        sleep_with = yield_raise
                elseif opts.txn then
                        error("malformed cpu_limit/txn: commit, rollback or raise are supported", 2)
                end
                if yield_with ~= yield_sleep then
                        if sleep_builder ~= nil then
                                sleep_with = sleep_builder(yield_with)
                        else
                                sleep_with = yield_with
                        end
                end
        else
                cpu_limit = tonumber(opts)
                sleep_with = yield_sleep
        end
        assert(cpu_limit, "malformed cpu_limit given: should be positive number within (0;100)")
        assert(cpu_limit > 0, "malformed cpu_limit given: should be positive number within (0;100)")
        assert(cpu_limit < 100, "malformed cpu_limit given: should be positive number within (0;100)")

        local max_quota_mks = tonumber(max_ev_run_mks * cpu_limit / 100)
        return fun.wrap(cpu_limit_gen,
                {max_quota_mks, sleep_with, gen_x, param_x},
                {state_x, max_quota_mks, ev_time_mks(), thread_time_mks()}
        )
end

--[[
        cpu limit is like dripping bucket.

        we say that we want to limit this iterator to consume ≤10% cpu.
        after yield we caught ev_mks.

        Then we call gen_x(param_x, state_x)
        and measure how much time it took.
        It may yield inside!

        Caller:
                - ev(), real()
                        - gen_x(param_x, state_x)
                - ev(), real()

        -- We gain time to work when we sleep
        -- We lose time of the work between subsequent calls

        -- Given time quota for fiber is evaluated as % of max_ev_run_time
        -- So basically, it is not possible to the fiber be on-cpu more than max_ev_run_time (10*ms for now)

        -- When fiber or ev_run exhausts time_quota (yes, noizy neighbours make fiber yield too)
        -- fiber is sent to sleep for next time slot.
        -- sleep time slot is evaluated in the following manner:
        -- for each X time of work, fiber needs to be yielded for at least X / % to regain it's time-quota.
        -- The same approach, but slightly in a different way is implemented here:
        -- fiber should sleep at most `max_ev_run_time` (basically should skip next ev loop).
        -- more precisely, fiber need to sleep (100% - X%) * max_ev_run_time where X% is cpu quota given to fiber
        -- but, in some cases, it is sended to sleep because of noizy neighbours,
        -- so we subtract from sleep-time left time_quota.
        -- After each sleep time_quota increases for X% of the sleep but never can be higher than X% * max_ev_run_time.
]]

methods.cpu_limit = method1(cpu_limit)
---@diagnostic disable-next-line: inject-field
fun.cpu_limit = export1(cpu_limit)

return fun

--[[
        fun.cpulimit(10, ....)

        fun.ones():cpu_limit(10)

        fun.ones():cpu_limit({ cpu_limit = 3, rxn='rollback' })
]]

1	local fun = require 'fun'	1✔
2	local fun_iter = fun.iter	1✔
3
4	local methods = debug.getmetatable(fun.ones()).__index	2✔
5
6	local rawiter = function(gen, param, state)
7	return fun_iter(gen, param, state):unwrap()	2✔
8	end
9
10	local method1 = function(func)
11	return function(self, arg1)
12	return func(arg1, self.gen, self.param, self.state)	15✔
13	end
14	end
15
16	local export1 = function(func)
17	return function(arg1, gen, param, state)
18	return func(arg1, rawiter(gen, param, state))	2✔
19	end
20	end
21
22	local fiber = require 'fiber'	1✔
23	local clock = require 'clock'	1✔
24	-- local log = require 'log'
25	-- if log.new then log = log.new('fun.cpu_limit') end
26
27	local function ev_time_mks() return fiber.time64() end	1,043,239✔
28	local function realtime_mks() return clock.realtime64()/1e3 end	530,639✔
29	local function thread_time_mks() return clock.thread64()/1e3 end	530,655✔
30
31	local upto0 = function(v)
32	if v < 0 then return 0 end	512,584✔
33	return v	460✔
34	end
35
36	---@param time number
37	local yield_sleep = function(time)
38	fiber.sleep(time)	8,987✔
39	end
40
41	local build_yield_commit = function(yield_with)
42	return function(time)
43	local is_committed = false	4,142✔
44	if type(box.cfg) ~= 'function' and box.is_in_txn and box.is_in_txn() then	4,142✔
45	box.commit()	4,142✔
46	is_committed = true	4,142✔
47	end
48	yield_with(time)	4,142✔
49	if is_committed then box.begin() end	4,142✔
50	end
51	end
52
53	local build_yield_rollback = function(yield_with)
54	return function(time)
55	if type(box.cfg) ~= 'function' and box.is_in_txn and box.is_in_txn() then	×
56	box.rollback()	×
57	end
58	return yield_with(time)	×
59	end
60	end
61
62	local build_yield_raise = function(time)
63	return function(yield_with)
64	if type(box.cfg) ~= 'function' and box.is_in_txn and box.is_in_txn() then	×
65	error("Transaction left openned", 3)	×
66	end
67	return yield_with(time)	×
68	end
69	end
70
71	local yield_commit = build_yield_commit(yield_sleep)	1✔
72	local yield_rollback = build_yield_rollback(yield_sleep)	1✔
73	local yield_raise = build_yield_raise(yield_sleep)	1✔
74
75	local mks = 1	1✔
76	local sec = 1e6	1✔
77	local ms = 1000*mks	1✔
78	local max_ev_run_mks = 100*ms	1✔
79
80	local cpu_limit_gen_x = function(state, state_x, ...)
81	if state_x == nil then	1,060,374✔
82	return nil	13✔
83	end
84
85	state[1] = state_x	1,060,361✔
86	-- state[2] = quota_mks
87	-- state[3] = ev_mks
88	-- state[4] = thread_mks
89
90	return state, ...	1,060,361✔
91	end
92
93	local function update_quota(now_mks, max_mks, add_mks)
94	local new_mks = now_mks + add_mks	1,043,222✔
95	if new_mks > max_mks then	1,043,222✔
96	new_mks = max_mks	×
97	end
98	-- log.info("quota: %.1fµs -> %.1fµs", now_mks, new_mks)
99	return new_mks	1,043,222✔
100	end
101
102	local rand = math.random	1✔
103
104	local cpu_limit_gen = function(param, state)
105	local max_quota_mks, sleep, gen_x, param_x = param[1], param[2], param[3], param[4]	1,060,374✔
106	local state_x, quota_mks, prev_ev_mks, prev_thread_mks = state[1], state[2], state[3], state[4]	1,060,374✔
107
108	if rand() > 0.5 then	1,060,374✔
109	return cpu_limit_gen_x(state, gen_x(param_x, state_x))	1,059,472✔
110	end
111
112	local ev_mks = ev_time_mks()	530,638✔
113	local thread_mks = thread_time_mks()	530,638✔
114	local wall_mks = realtime_mks()	530,638✔
115
116	if prev_ev_mks == ev_mks then	530,638✔
117	-- the caller spent all this time on cpu
118	-- decrease quota
119	quota_mks = update_quota(quota_mks, max_quota_mks, -tonumber(thread_mks - prev_thread_mks))	1,061,276✔
120	-- quota_mks = quota_mks - tonumber(thread_mks - prev_thread_mks)
121	end
122
123	local ev_loop_too_long = (ev_mks + max_ev_run_mks) < wall_mks	530,638✔
124	if ev_loop_too_long or quota_mks <= 1 then	530,638✔
125	-- it's time to yield
126	-- Fiber needs to sleep:
127	local sleep_sec = (max_ev_run_mks - (max_quota_mks - quota_mks))/sec	512,584✔
128	sleep(upto0(sleep_sec))	1,025,168✔
129
130	ev_mks = ev_time_mks()	1,025,168✔
131	if ev_mks > wall_mks then	512,584✔
132	-- yes, any time drift does not add time quota.
133	local gained_mks = assert(tonumber(ev_mks-wall_mks))	512,584✔
134	quota_mks = update_quota(quota_mks, max_quota_mks, gained_mks*max_quota_mks/max_ev_run_mks)	1,025,168✔
135	end
136	end
137
138	-- state[1] = state_x
139	state[2] = quota_mks	530,638✔
140	state[3] = ev_mks	530,638✔
141	state[4] = thread_mks	530,638✔
142
143	return cpu_limit_gen_x(state, gen_x(param_x, state_x))	1,061,276✔
144	end
145
146	local cpu_limit = function(opts, gen_x, param_x, state_x)
147	local cpu_limit
148	local sleep_with = yield_sleep	16✔
149	if type(opts) == 'table' then	16✔
150	cpu_limit = tonumber(opts.cpu_limit)	14✔
151	local yield_with = yield_sleep	14✔
152	if opts.yield_with then	14✔
153	if type(opts.yield_with) ~= 'function' then	12✔
154	error("malformed cpu_limit/yield_with: must be a function when given", 2)	×
155	end
156	yield_with = opts.yield_with	12✔
157	end
158
159	---sleep_builder is a builder function that can build specific fiber.sleep()
160	---default builders already exist
161	---@type (fun(yield_with: fun(time: number)):fun(time: number))?
162	local sleep_builder = nil	14✔
163	if opts.txn == 'commit' then	14✔
164	sleep_builder = build_yield_commit	2✔
165	sleep_with = yield_commit	2✔
166	elseif opts.txn == 'rollback' then	12✔
167	sleep_builder = build_yield_rollback	×
168	sleep_with = yield_rollback	×
169	elseif opts.txn == 'raise' then	12✔
170	sleep_builder = build_yield_raise	×
171	sleep_with = yield_raise	×
172	elseif opts.txn then	12✔
173	error("malformed cpu_limit/txn: commit, rollback or raise are supported", 2)	×
174	end
175	if yield_with ~= yield_sleep then	14✔
176	if sleep_builder ~= nil then	12✔
177	sleep_with = sleep_builder(yield_with)	2✔
178	else
179	sleep_with = yield_with	11✔
180	end
181	end
182	else
183	cpu_limit = tonumber(opts)	2✔
184	sleep_with = yield_sleep	2✔
185	end
186	assert(cpu_limit, "malformed cpu_limit given: should be positive number within (0;100)")	16✔
187	assert(cpu_limit > 0, "malformed cpu_limit given: should be positive number within (0;100)")	16✔
188	assert(cpu_limit < 100, "malformed cpu_limit given: should be positive number within (0;100)")	16✔
189
190	local max_quota_mks = tonumber(max_ev_run_mks * cpu_limit / 100)	16✔
191	return fun.wrap(cpu_limit_gen,	16✔
192	{max_quota_mks, sleep_with, gen_x, param_x},	16✔
193	{state_x, max_quota_mks, ev_time_mks(), thread_time_mks()}	48✔
194	)	16✔
195	end
196
197	--[[
198	cpu limit is like dripping bucket.
199
200	we say that we want to limit this iterator to consume ≤10% cpu.
201	after yield we caught ev_mks.
202
203	Then we call gen_x(param_x, state_x)
204	and measure how much time it took.
205	It may yield inside!
206
207	Caller:
208	- ev(), real()
209	- gen_x(param_x, state_x)
210	- ev(), real()
211
212	-- We gain time to work when we sleep
213	-- We lose time of the work between subsequent calls
214
215	-- Given time quota for fiber is evaluated as % of max_ev_run_time
216	-- So basically, it is not possible to the fiber be on-cpu more than max_ev_run_time (10*ms for now)
217
218	-- When fiber or ev_run exhausts time_quota (yes, noizy neighbours make fiber yield too)
219	-- fiber is sent to sleep for next time slot.
220	-- sleep time slot is evaluated in the following manner:
221	-- for each X time of work, fiber needs to be yielded for at least X / % to regain it's time-quota.
222	-- The same approach, but slightly in a different way is implemented here:
223	-- fiber should sleep at most `max_ev_run_time` (basically should skip next ev loop).
224	-- more precisely, fiber need to sleep (100% - X%) * max_ev_run_time where X% is cpu quota given to fiber
225	-- but, in some cases, it is sended to sleep because of noizy neighbours,
226	-- so we subtract from sleep-time left time_quota.
227	-- After each sleep time_quota increases for X% of the sleep but never can be higher than X% * max_ev_run_time.
228	]]
229
230	methods.cpu_limit = method1(cpu_limit)	2✔
231	---@diagnostic disable-next-line: inject-field
232	fun.cpu_limit = export1(cpu_limit)	2✔
233
234	return fun	1✔
235
236	--[[
237	fun.cpulimit(10, ....)
238
239	fun.ones():cpu_limit(10)
240
241	fun.ones():cpu_limit({ cpu_limit = 3, rxn='rollback' })
242	]]

ochaton / fun-cpu-limit / 1853890102

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous