30211738911

Committed 26 Jul 2026 05:02PM UTC coverage: 85.343% (+0.1%) from 85.229%

Build # 30211738911

Build Type

push

github

Committed by

Tieske

Commit Message

fix(semaphore): take() must reject requests below 1 or NaN

A negative `requested` on the fast path (`count = count - requested`)
turned a take into a give, inflating `count` past `max` with no check
at all (eg. take(-100) on a max=5 semaphore left count at 100). A NaN
request would queue forever, since `count >= NaN` is always false and
can never be satisfied, deadlocking that entry and everyone behind it
in the FIFO queue. Reject both instead.

Coverage Stats

2 of 2 new or added lines in 1 file covered. (100.0%)

9 existing lines in 5 files now uncovered.

1444 of 1692 relevant lines covered (85.34%)

55080.61 hits per line

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

97.54

/src/copas/semaphore.lua

local copas = require("copas")

local coroutine_running = coroutine.running
if _VERSION=="Lua 5.1" and not jit then     -- obsolete: only for Lua 5.1 compatibility
  coroutine_running = require("coxpcall").running
end

local DEFAULT_TIMEOUT = 10

local semaphore = {}
semaphore.__index = semaphore


-- registry, semaphore indexed by the coroutines using them.
local registry = setmetatable({}, { __mode="kv" })


-- create a new semaphore
-- @param max maximum number of resources the semaphore can hold (this maximum does NOT include resources that have been given but not yet returned).
-- @param start (optional, default 0) the initial resources available
-- @param seconds (optional, default 10) default semaphore timeout in seconds, or `math.huge` to have no timeout.
function semaphore.new(max, start, seconds)
  local timeout = tonumber(seconds or DEFAULT_TIMEOUT) or -1
  if timeout < 0 then
    error("expected timeout (2nd argument) to be a number greater than or equal to 0, got: " .. tostring(seconds), 2)
  end
  if type(max) ~= "number" or max < 1 then
    error("expected max resources (1st argument) to be a number greater than 0, got: " .. tostring(max), 2)
  end

  local self = setmetatable({
      count = start or 0,
      max = max,
      timeout = timeout,
      q_tip = 1,    -- position of next entry waiting
      q_tail = 1,   -- position where next one will be inserted
      queue = {},
      to_flags = setmetatable({}, { __mode = "k" }), -- timeout flags indexed by coroutine
    }, semaphore)

  return self
end


do
  local destroyed_func = function()
    return nil, "destroyed"
  end

  local destroyed_semaphore_mt = {
    __index = function()
      return destroyed_func
    end
  }

  -- destroy a semaphore.
  -- Releases all waiting threads with `nil+"destroyed"`
  function semaphore:destroy()
    self:release_all()
    self.destroyed = true
    setmetatable(self, destroyed_semaphore_mt)
    return true
  end
end


-- Gives resources.
-- @param given (optional, default 1) number of resources to return. Must be
-- a finite number greater than or equal to 0. If more than the maximum are
-- returned then it will be capped at the maximum and error "too many" will
-- be returned.
function semaphore:give(given)
  local err
  given = given or 1
  if given < 0 or given == math.huge or given ~= given then
    error("expected given resources (1st argument) to be a finite number greater than or equal to 0, got: " .. tostring(given), 2)
  end
  local count = self.count + given
  --print("now at",count, ", after +"..given)
  if count > self.max then
    count = self.max
    err = "too many"
  end

  while self.q_tip < self.q_tail do
    local i = self.q_tip
    local nxt = self.queue[i] -- there can be holes, so nxt might be nil
    if not nxt then
      self.q_tip = i + 1
    elseif not copas.issleeping(nxt.co) then
      -- stale entry; 'nxt.co' was canceled externally (eg. via
      -- copas.removethread/future:cancel) and will never resume to
      -- consume or give back resources. Drop it without touching 'count',
      -- and regardless of how much it had requested, so it doesn't block
      -- waiters behind it that request less than it did.
      self.queue[i] = nil
      self.to_flags[nxt.co] = nil
      self.q_tip = i + 1
      nxt.co = nil
    elseif count >= nxt.requested then
      -- release it
      self.queue[i] = nil
      self.to_flags[nxt.co] = nil
      count = count - nxt.requested
      self.q_tip = i + 1
      copas.wakeup(nxt.co)
      nxt.co = nil
    else
      break -- we ran out of resources
    end
  end

  if self.q_tip == self.q_tail then  -- reset queue
    self.queue = {}
    self.q_tip = 1
    self.q_tail = 1
  end

  self.count = count
  if err then
    return nil, err
  end
  return true
end



local function timeout_handler(co)
  local self = registry[co]
  --print("checking timeout ", co)
  if not self then
    return
  end

  for i = self.q_tip, self.q_tail do
    local item = self.queue[i]
    if item and co == item.co then
      self.queue[i] = nil
      self.to_flags[co] = true
      --print("marked timeout ", co)
      copas.wakeup(co)
      return
    end
  end
  -- nothing to do here...
end


-- Requests resources from the semaphore.
-- Waits if there are not enough resources available before returning.
-- @param requested (optional, default 1) the number of resources requested.
-- Must be a number greater than or equal to 1, and not NaN.
-- @param timeout (optional, defaults to semaphore timeout) timeout in
-- seconds. If 0 it will either succeed or return immediately with error "timeout".
-- If `math.huge` it will wait forever.
-- @return true, or nil+"destroyed"
function semaphore:take(requested, timeout)
  requested = requested or 1
  if requested < 1 or requested ~= requested then
    error("expected requested resources (1st argument) to be a number greater than or equal to 1, got: " .. tostring(requested), 2)
  end
  if self.q_tail == 1 and self.count >= requested then
    -- nobody is waiting before us, and there is enough in store
    self.count = self.count - requested
    return true
  end

  if requested > self.max then
    return nil, "too many"
  end

  local to = timeout or self.timeout
  if to == 0 then
    return nil, "timeout"
  end

  -- get in line
  local co = coroutine_running()
  self.to_flags[co] = nil
  registry[co] = self
  copas.timeout(to, timeout_handler)

  self.queue[self.q_tail] = {
    co = co,
    requested = requested,
    --timeout = nil, -- flag indicating timeout
  }
  self.q_tail = self.q_tail + 1

  copas.pauseforever() -- block until woken
  registry[co] = nil

  if self.to_flags[co] then
    -- a timeout happened
    self.to_flags[co] = nil
    return nil, "timeout"
  end

  copas.timeout(0)

  if self.destroyed then
    return nil, "destroyed"
  end

  return true
end

-- returns current available resources
function semaphore:get_count()
  return self.count
end

-- returns total shortage for requested resources
function semaphore:get_wait()
  local wait = 0
  for i = self.q_tip, self.q_tail - 1 do
    local item = self.queue[i]
    if item and copas.issleeping(item.co) then
      wait = wait + item.requested
    end
  end
  return wait - self.count
end


-- Releases every currently queued waiter, regardless of `max`.
-- Feeds `get_wait()` into `give()` in `max`-sized chunks, since a single
-- `give()` call is capped at `max` and would otherwise strand waiters
-- beyond it.
function semaphore:release_all()
  local wait = self:get_wait()
  while wait > 0 do
    if wait > self.max then
      self:give(self.max)
    else
      self:give(wait)
    end
    wait = wait - self.max
  end
end


return semaphore

1	local copas = require("copas")	162✔
2
3	local coroutine_running = coroutine.running	162✔
4	if _VERSION=="Lua 5.1" and not jit then -- obsolete: only for Lua 5.1 compatibility	162✔
UNCOV 5	coroutine_running = require("coxpcall").running	27✔
6	end
7
8	local DEFAULT_TIMEOUT = 10	162✔
9
10	local semaphore = {}	162✔
11	semaphore.__index = semaphore	162✔
12
13
14	-- registry, semaphore indexed by the coroutines using them.
15	local registry = setmetatable({}, { __mode="kv" })	162✔
16
17
18	-- create a new semaphore
19	-- @param max maximum number of resources the semaphore can hold (this maximum does NOT include resources that have been given but not yet returned).
20	-- @param start (optional, default 0) the initial resources available
21	-- @param seconds (optional, default 10) default semaphore timeout in seconds, or `math.huge` to have no timeout.
22	function semaphore.new(max, start, seconds)	162✔
23	local timeout = tonumber(seconds or DEFAULT_TIMEOUT) or -1	584✔
24	if timeout < 0 then	584✔
25	error("expected timeout (2nd argument) to be a number greater than or equal to 0, got: " .. tostring(seconds), 2)	×
26	end
27	if type(max) ~= "number" or max < 1 then	584✔
28	error("expected max resources (1st argument) to be a number greater than 0, got: " .. tostring(max), 2)	×
29	end
30
31	local self = setmetatable({	1,168✔
32	count = start or 0,	584✔
33	max = max,	584✔
34	timeout = timeout,	584✔
35	q_tip = 1, -- position of next entry waiting	488✔
36	q_tail = 1, -- position where next one will be inserted	488✔
37	queue = {},	584✔
38	to_flags = setmetatable({}, { __mode = "k" }), -- timeout flags indexed by coroutine	584✔
39	}, semaphore)	584✔
40
41	return self	584✔
42	end
43
44
45	do
46	local destroyed_func = function()
47	return nil, "destroyed"	24✔
48	end
49
50	local destroyed_semaphore_mt = {	162✔
51	__index = function()
52	return destroyed_func	24✔
53	end
54	}
55
56	-- destroy a semaphore.
57	-- Releases all waiting threads with `nil+"destroyed"`
58	function semaphore:destroy()	162✔
59	self:release_all()	428✔
60	self.destroyed = true	428✔
61	setmetatable(self, destroyed_semaphore_mt)	428✔
62	return true	428✔
63	end
64	end
65
66
67	-- Gives resources.
68	-- @param given (optional, default 1) number of resources to return. Must be
69	-- a finite number greater than or equal to 0. If more than the maximum are
70	-- returned then it will be capped at the maximum and error "too many" will
71	-- be returned.
72	function semaphore:give(given)	162✔
73	local err
74	given = given or 1	372✔
75	if given < 0 or given == math.huge or given ~= given then	372✔
76	error("expected given resources (1st argument) to be a finite number greater than or equal to 0, got: " .. tostring(given), 2)	18✔
77	end
78	local count = self.count + given	354✔
79	--print("now at",count, ", after +"..given)
80	if count > self.max then	354✔
81	count = self.max	6✔
82	err = "too many"	6✔
83	end
84
85	while self.q_tip < self.q_tail do	660✔
86	local i = self.q_tip	342✔
87	local nxt = self.queue[i] -- there can be holes, so nxt might be nil	342✔
88	if not nxt then	342✔
89	self.q_tip = i + 1	12✔
90	elseif not copas.issleeping(nxt.co) then	385✔
91	-- stale entry; 'nxt.co' was canceled externally (eg. via
92	-- copas.removethread/future:cancel) and will never resume to
93	-- consume or give back resources. Drop it without touching 'count',
94	-- and regardless of how much it had requested, so it doesn't block
95	-- waiters behind it that request less than it did.
96	self.queue[i] = nil	24✔
97	self.to_flags[nxt.co] = nil	24✔
98	self.q_tip = i + 1	24✔
99	nxt.co = nil	24✔
100	elseif count >= nxt.requested then	306✔
101	-- release it
102	self.queue[i] = nil	270✔
103	self.to_flags[nxt.co] = nil	270✔
104	count = count - nxt.requested	270✔
105	self.q_tip = i + 1	270✔
106	copas.wakeup(nxt.co)	270✔
107	nxt.co = nil	270✔
108	else
109	break -- we ran out of resources
110	end
111	end
112
113	if self.q_tip == self.q_tail then -- reset queue	354✔
114	self.queue = {}	318✔
115	self.q_tip = 1	318✔
116	self.q_tail = 1	318✔
117	end
118
119	self.count = count	354✔
120	if err then	354✔
121	return nil, err	6✔
122	end
123	return true	348✔
124	end
125
126
127
128	local function timeout_handler(co)
129	local self = registry[co]	48✔
130	--print("checking timeout ", co)
131	if not self then	48✔
UNCOV 132	return	×
133	end
134
135	for i = self.q_tip, self.q_tail do	84✔
136	local item = self.queue[i]	66✔
137	if item and co == item.co then	66✔
138	self.queue[i] = nil	30✔
139	self.to_flags[co] = true	30✔
140	--print("marked timeout ", co)
141	copas.wakeup(co)	30✔
142	return	30✔
143	end
144	end
145	-- nothing to do here...
146	end
147
148
149	-- Requests resources from the semaphore.
150	-- Waits if there are not enough resources available before returning.
151	-- @param requested (optional, default 1) the number of resources requested.
152	-- Must be a number greater than or equal to 1, and not NaN.
153	-- @param timeout (optional, defaults to semaphore timeout) timeout in
154	-- seconds. If 0 it will either succeed or return immediately with error "timeout".
155	-- If `math.huge` it will wait forever.
156	-- @return true, or nil+"destroyed"
157	function semaphore:take(requested, timeout)	162✔
158	requested = requested or 1	474✔
159	if requested < 1 or requested ~= requested then	474✔
160	error("expected requested resources (1st argument) to be a number greater than or equal to 1, got: " .. tostring(requested), 2)	18✔
161	end
162	if self.q_tail == 1 and self.count >= requested then	456✔
163	-- nobody is waiting before us, and there is enough in store
164	self.count = self.count - requested	114✔
165	return true	114✔
166	end
167
168	if requested > self.max then	342✔
169	return nil, "too many"	6✔
170	end
171
172	local to = timeout or self.timeout	336✔
173	if to == 0 then	336✔
174	return nil, "timeout"	12✔
175	end
176
177	-- get in line
178	local co = coroutine_running()	324✔
179	self.to_flags[co] = nil	324✔
180	registry[co] = self	324✔
181	copas.timeout(to, timeout_handler)	324✔
182
183	self.queue[self.q_tail] = {	324✔
184	co = co,	324✔
185	requested = requested,	324✔
186	--timeout = nil, -- flag indicating timeout
187	}	324✔
188	self.q_tail = self.q_tail + 1	324✔
189
190	copas.pauseforever() -- block until woken	324✔
191	registry[co] = nil	294✔
192
193	if self.to_flags[co] then	294✔
194	-- a timeout happened
195	self.to_flags[co] = nil	24✔
196	return nil, "timeout"	24✔
197	end
198
199	copas.timeout(0)	270✔
200
201	if self.destroyed then	270✔
202	return nil, "destroyed"	84✔
203	end
204
205	return true	186✔
206	end
207
208	-- returns current available resources
209	function semaphore:get_count()	162✔
210	return self.count	138✔
211	end
212
213	-- returns total shortage for requested resources
214	function semaphore:get_wait()	162✔
215	local wait = 0	530✔
216	for i = self.q_tip, self.q_tail - 1 do	806✔
217	local item = self.queue[i]	276✔
218	if item and copas.issleeping(item.co) then	322✔
219	wait = wait + item.requested	258✔
220	end
221	end
222	return wait - self.count	530✔
223	end
224
225
226	-- Releases every currently queued waiter, regardless of `max`.
227	-- Feeds `get_wait()` into `give()` in `max`-sized chunks, since a single
228	-- `give()` call is capped at `max` and would otherwise strand waiters
229	-- beyond it.
230	function semaphore:release_all()	162✔
231	local wait = self:get_wait()	512✔
232	while wait > 0 do	668✔
233	if wait > self.max then	156✔
234	self:give(self.max)	35✔
235	else
236	self:give(wait)	126✔
237	end
238	wait = wait - self.max	156✔
239	end
240	end
241
242
243	return semaphore	162✔

lunarmodules / copas / 30211738911

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