3797080599

Build Type

push

github

Committed by Thijs Schreijer

Commit Message

feat(timeouts) allow math.huge to wait forever

Run Details

4 of 4 new or added lines in 2 files covered. (100.0%)

1266 of 1488 relevant lines covered (85.08%)

14936.03 hits per line

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

92.93

/src/copas/semaphore.lua

local copas = require("copas")

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:give(math.huge)
    self.destroyed = true
    setmetatable(self, destroyed_semaphore_mt)
    return true
  end
end


-- Gives resources.
-- @param given (optional, default 1) number of resources to return. 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
  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
    else
      if 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
  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
-- @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 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
    wait = wait + ((self.queue[i] or {}).requested or 0)
  end
  return wait - self.count
end


return semaphore

1	local copas = require("copas")	2✔
2
3	local DEFAULT_TIMEOUT = 10	2✔
4
5	local semaphore = {}	2✔
6	semaphore.__index = semaphore	2✔
7
8
9	-- registry, semaphore indexed by the coroutines using them.
10	local registry = setmetatable({}, { __mode="kv" })	2✔
11
12
13	-- create a new semaphore
14	-- @param max maximum number of resources the semaphore can hold (this maximum does NOT include resources that have been given but not yet returned).
15	-- @param start (optional, default 0) the initial resources available
16	-- @param seconds (optional, default 10) default semaphore timeout in seconds, or `math.huge` to have no timeout.
17	function semaphore.new(max, start, seconds)	2✔
18	local timeout = tonumber(seconds or DEFAULT_TIMEOUT) or -1	6✔
19	if timeout < 0 then	6✔
20	error("expected timeout (2nd argument) to be a number greater than or equal to 0, got: " .. tostring(seconds), 2)	×
21	end
22	if type(max) ~= "number" or max < 1 then	6✔
23	error("expected max resources (1st argument) to be a number greater than 0, got: " .. tostring(max), 2)	×
24	end
25
26	local self = setmetatable({	12✔
27	count = start or 0,	6✔
28	max = max,	6✔
29	timeout = timeout,	6✔
30	q_tip = 1, -- position of next entry waiting
31	q_tail = 1, -- position where next one will be inserted
32	queue = {},	6✔
33	to_flags = setmetatable({}, { __mode = "k" }), -- timeout flags indexed by coroutine	6✔
34	}, semaphore)	6✔
35
36	return self	6✔
37	end
38
39
40	do
41	local destroyed_func = function()
42	return nil, "destroyed"	4✔
43	end
44
45	local destroyed_semaphore_mt = {	2✔
46	__index = function()
47	return destroyed_func	4✔
48	end
49	}
50
51	-- destroy a semaphore.
52	-- Releases all waiting threads with `nil+"destroyed"`
53	function semaphore:destroy()	2✔
54	self:give(math.huge)	6✔
55	self.destroyed = true	6✔
56	setmetatable(self, destroyed_semaphore_mt)	6✔
57	return true	6✔
58	end
59	end
60
61
62	-- Gives resources.
63	-- @param given (optional, default 1) number of resources to return. If more
64	-- than the maximum are returned then it will be capped at the maximum and
65	-- error "too many" will be returned.
66	function semaphore:give(given)	2✔
67	local err
68	given = given or 1	24✔
69	local count = self.count + given	24✔
70	--print("now at",count, ", after +"..given)
71	if count > self.max then	24✔
72	count = self.max	7✔
73	err = "too many"	7✔
74	end
75
76	while self.q_tip < self.q_tail do	34✔
77	local i = self.q_tip	11✔
78	local nxt = self.queue[i] -- there can be holes, so nxt might be nil	11✔
79	if not nxt then	11✔
80	self.q_tip = i + 1	2✔
81	else
82	if count >= nxt.requested then	9✔
83	-- release it
84	self.queue[i] = nil	8✔
85	self.to_flags[nxt.co] = nil	8✔
86	count = count - nxt.requested	8✔
87	self.q_tip = i + 1	8✔
88	copas.wakeup(nxt.co)	8✔
89	nxt.co = nil	8✔
90	else
91	break -- we ran out of resources
92	end
93	end
94	end
95
96	if self.q_tip == self.q_tail then -- reset queue	24✔
97	self.queue = {}	23✔
98	self.q_tip = 1	23✔
99	self.q_tail = 1	23✔
100	end
101
102	self.count = count	24✔
103	if err then	24✔
104	return nil, err	7✔
105	end
106	return true	17✔
107	end
108
109
110
111	local function timeout_handler(co)
112	local self = registry[co]	2✔
113	--print("checking timeout ", co)
114	if not self then	2✔
115	return	×
116	end
117
118	for i = self.q_tip, self.q_tail do	2✔
119	local item = self.queue[i]	2✔
120	if item and co == item.co then	2✔
121	self.queue[i] = nil	2✔
122	self.to_flags[co] = true	2✔
123	--print("marked timeout ", co)
124	copas.wakeup(co)	2✔
125	return	2✔
126	end
127	end
128	-- nothing to do here...
129	end
130
131
132	-- Requests resources from the semaphore.
133	-- Waits if there are not enough resources available before returning.
134	-- @param requested (optional, default 1) the number of resources requested
135	-- @param timeout (optional, defaults to semaphore timeout) timeout in
136	-- seconds. If 0 it will either succeed or return immediately with error "timeout".
137	-- If `math.huge` it will wait forever.
138	-- @return true, or nil+"destroyed"
139	function semaphore:take(requested, timeout)	2✔
140	requested = requested or 1	28✔
141	if self.q_tail == 1 and self.count >= requested then	28✔
142	-- nobody is waiting before us, and there is enough in store
143	self.count = self.count - requested	15✔
144	return true	15✔
145	end
146
147	if requested > self.max then	13✔
148	return nil, "too many"	1✔
149	end
150
151	local to = timeout or self.timeout	12✔
152	if to == 0 then	12✔
153	return nil, "timeout"	2✔
154	end
155
156	-- get in line
157	local co = coroutine.running()	10✔
158	self.to_flags[co] = nil	10✔
159	registry[co] = self	10✔
160	copas.timeout(to, timeout_handler)	10✔
161
162	self.queue[self.q_tail] = {	10✔
163	co = co,	10✔
164	requested = requested,	10✔
165	--timeout = nil, -- flag indicating timeout
166	}	10✔
167	self.q_tail = self.q_tail + 1	10✔
168
169	copas.pauseforever() -- block until woken	10✔
170	registry[co] = nil	10✔
171
172	if self.to_flags[co] then	10✔
173	-- a timeout happened
174	self.to_flags[co] = nil	2✔
175	return nil, "timeout"	2✔
176	end
177
178	copas.timeout(0)	8✔
179
180	if self.destroyed then	8✔
181	return nil, "destroyed"	5✔
182	end
183
184	return true	3✔
185	end
186
187	-- returns current available resources
188	function semaphore:get_count()	2✔
189	return self.count	9✔
190	end
191
192	-- returns total shortage for requested resources
193	function semaphore:get_wait()	2✔
194	local wait = 0	×
195	for i = self.q_tip, self.q_tail - 1 do	×
196	wait = wait + ((self.queue[i] or {}).requested or 0)	×
197	end
198	return wait - self.count	×
199	end
200
201
202	return semaphore	2✔

lunarmodules / copas / 3797080599

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