31212597912

Committed 07 Aug 2026 07:41PM UTC coverage: 85.551% (-0.1%) from 85.665%

Build # 31212597912

Build Type

push

github

Committed by

web-flow

Commit Message

Merge 42b2ad564 into e4bbc7b69

Coverage Stats

5 of 7 new or added lines in 1 file covered. (71.43%)

14 existing lines in 4 files now uncovered.

1498 of 1751 relevant lines covered (85.55%)

15090.93 hits per line

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

89.91

/src/copas/queue.lua

local copas = require "copas"
local gettime = copas.gettime
local Sema = copas.semaphore
local Lock = copas.lock


local Queue = {}
Queue.__index = Queue


local new_name do
  local count = 0

  function new_name()
    count = count + 1
    return "copas_queue_" .. count
  end
end


-- Creates a new Queue instance
function Queue.new(opts)
  opts = opts or {}
  local self = {}
  setmetatable(self, Queue)
  self.name = opts.name or new_name()
  self.sema = Sema.new(10^9)
  self.head = 1
  self.tail = 1
  self.list = {}
  self.workers = setmetatable({}, { __mode = "k" })
  self.stopping = false
  self.worker_id = 0
  self.exit_semaphore = Sema.new(10^9)
  return self
end


-- Pushes an item in the queue (can be 'nil')
-- returns true, or nil+err ("stopping", or "destroyed")
function Queue:push(item)
  if self.stopping then
    return nil, "stopping"
  end
  self.list[self.head] = item
  self.head = self.head + 1
  self.sema:give()
  return true
end


-- Pops and item from the queue. If there are no items in the queue it will yield
-- until there are or a timeout happens (exception is when `timeout == 0`, then it will
-- not yield but return immediately). If the timeout is `math.huge` it will wait forever.
-- Returns item, or nil+err ("timeout", or "destroyed")
function Queue:pop(timeout)
  local ok, err = self.sema:take(1, timeout)
  if not ok then
    return ok, err
  end

  local item = self.list[self.tail]
  self.list[self.tail] = nil
  self.tail = self.tail + 1

  if self.tail == self.head then
    -- reset queue
    self.list = {}
    self.tail = 1
    self.head = 1
    if self.stopping then
      -- we're stopping and last item being returned, so we're done
      self:destroy()
    end
  end
  return item
end


-- return the number of items left in the queue
function Queue:get_size()
  return self.head - self.tail
end


-- instructs the queue to stop. Will not accept any more 'push' calls.
-- will autocall 'destroy' when the queue is empty.
-- returns immediately. See `finish`
function Queue:stop()
  if not self.stopping then
    self.stopping = true
    self.lock = Lock.new(nil, true)
    self.lock:get() -- close the lock
    if self:get_size() == 0 then
      -- queue is already empty, so "pop" function cannot call destroy on next
      -- pop, so destroy now.
      self:destroy()
    end
  end
  return true
end


-- Finishes a queue. Calls stop and then waits for the queue to run empty (and be
-- destroyed) before returning. returns true or nil+err ("timeout", or "destroyed")
-- Parameter no_destroy_on_timeout indicates if the queue is not to be forcefully
-- destroyed on a timeout.
function Queue:finish(timeout, no_destroy_on_timeout)
  self:stop()
  timeout = timeout or self.lock.timeout
  local endtime = gettime() + timeout
  local _, err = self.lock:get(timeout)
  -- the lock never gets released, only destroyed, so we have to check the error string
  if err == "timeout" then
    if not no_destroy_on_timeout then
      self:destroy()
    end
    return nil, err
  end

  -- if we get here, the lock was destroyed, so the queue is empty, now wait for all workers to exit
  if not next(self.workers) then
    -- all workers already exited, we're done
    return true
  end

  -- multiple threads can call this "finish" method, so we must check exiting workers
  -- one by one.
  while true do
    local _, err = self.exit_semaphore:take(1, math.max(0, endtime - gettime()))
    if err == "destroyed" then
      return true  -- someone else destroyed/finished it, so we're done
    end
    if err == "timeout" then
      if not no_destroy_on_timeout then
        self:destroy()
      end
      return nil, "timeout"
    end
    if not next(self.workers) then
      self.exit_semaphore:destroy()
      return true  -- all workers exited, we're done
    end
  end
end


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

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

  -- destroys a queue immediately. Abandons what is left in the queue.
  -- Releases all waiting threads with `nil+"destroyed"`
  function Queue:destroy()
    if self.lock then
      self.lock:destroy()
    end
    self.sema:destroy()
    setmetatable(self, destroyed_queue_mt)

    -- clear anything left in the queue
    for key in pairs(self.list) do
      self.list[key] = nil
    end

    return true
  end
end


-- adds a worker that will handle whatever is passed into the queue. Can be called
-- multiple times to add more workers.
-- The threads automatically exit when the queue is destroyed.
-- worker function signature: `function(item)` (Note: worker functions run
-- unprotected, so wrap code in an (x)pcall if errors are expected, otherwise the
-- worker will exit on an error, and queue handling will stop)
-- Returns the coroutine added.
function Queue:add_worker(worker)
  assert(type(worker) == "function", "expected worker to be a function")
  local coro

  self.worker_id = self.worker_id + 1
  local worker_name = self.name .. ":worker_" .. self.worker_id

  coro = copas.addnamedthread(worker_name, function()
    while true do
      local item, err = self:pop(math.huge) -- wait forever
      if err then
        break -- queue destroyed, exit
      end
      worker(item) -- TODO: wrap in errorhandling
    end
    self.workers[coro] = nil
    if self.exit_semaphore then
      self.exit_semaphore:give(1)
    end
  end)

  self.workers[coro] = true
  return coro
end

-- returns a list/array of current workers (coroutines) handling the queue.
-- (only the workers added by `add_worker`, and still active, will be in this list)
function Queue:get_workers()
  local lst = {}
  for coro in pairs(self.workers) do
    if coroutine.status(coro) ~= "dead" then
      lst[#lst+1] = coro
    end
  end
  return lst
end

return Queue

1	local copas = require "copas"	2✔
2	local gettime = copas.gettime	2✔
3	local Sema = copas.semaphore	2✔
4	local Lock = copas.lock	2✔
5
6
7	local Queue = {}	2✔
8	Queue.__index = Queue	2✔
9
10
11	local new_name do	2✔
12	local count = 0	2✔
13
UNCOV 14	function new_name()	1✔
15	count = count + 1	12✔
16	return "copas_queue_" .. count	12✔
17	end
18	end
19
20
21	-- Creates a new Queue instance
22	function Queue.new(opts)	2✔
23	opts = opts or {}	12✔
24	local self = {}	12✔
25	setmetatable(self, Queue)	12✔
26	self.name = opts.name or new_name()	12✔
27	self.sema = Sema.new(10^9)	12✔
28	self.head = 1	12✔
29	self.tail = 1	12✔
30	self.list = {}	12✔
31	self.workers = setmetatable({}, { __mode = "k" })	12✔
32	self.stopping = false	12✔
33	self.worker_id = 0	12✔
34	self.exit_semaphore = Sema.new(10^9)	12✔
35	return self	12✔
36	end
37
38
39	-- Pushes an item in the queue (can be 'nil')
40	-- returns true, or nil+err ("stopping", or "destroyed")
41	function Queue:push(item)	2✔
42	if self.stopping then	36✔
43	return nil, "stopping"	×
44	end
45	self.list[self.head] = item	36✔
46	self.head = self.head + 1	36✔
47	self.sema:give()	36✔
48	return true	36✔
49	end
50
51
52	-- Pops and item from the queue. If there are no items in the queue it will yield
53	-- until there are or a timeout happens (exception is when `timeout == 0`, then it will
54	-- not yield but return immediately). If the timeout is `math.huge` it will wait forever.
55	-- Returns item, or nil+err ("timeout", or "destroyed")
56	function Queue:pop(timeout)	2✔
57	local ok, err = self.sema:take(1, timeout)	44✔
58	if not ok then	44✔
59	return ok, err	10✔
60	end
61
62	local item = self.list[self.tail]	34✔
63	self.list[self.tail] = nil	34✔
64	self.tail = self.tail + 1	34✔
65
66	if self.tail == self.head then	34✔
67	-- reset queue
68	self.list = {}	14✔
69	self.tail = 1	14✔
70	self.head = 1	14✔
71	if self.stopping then	14✔
72	-- we're stopping and last item being returned, so we're done
73	self:destroy()	2✔
74	end
75	end
76	return item	34✔
77	end
78
79
80	-- return the number of items left in the queue
81	function Queue:get_size()	2✔
82	return self.head - self.tail	20✔
83	end
84
85
86	-- instructs the queue to stop. Will not accept any more 'push' calls.
87	-- will autocall 'destroy' when the queue is empty.
88	-- returns immediately. See `finish`
89	function Queue:stop()	2✔
90	if not self.stopping then	12✔
91	self.stopping = true	12✔
92	self.lock = Lock.new(nil, true)	12✔
93	self.lock:get() -- close the lock	12✔
94	if self:get_size() == 0 then	12✔
95	-- queue is already empty, so "pop" function cannot call destroy on next
96	-- pop, so destroy now.
97	self:destroy()	8✔
98	end
99	end
100	return true	12✔
101	end
102
103
104	-- Finishes a queue. Calls stop and then waits for the queue to run empty (and be
105	-- destroyed) before returning. returns true or nil+err ("timeout", or "destroyed")
106	-- Parameter no_destroy_on_timeout indicates if the queue is not to be forcefully
107	-- destroyed on a timeout.
108	function Queue:finish(timeout, no_destroy_on_timeout)	2✔
109	self:stop()	8✔
110	timeout = timeout or self.lock.timeout	8✔
111	local endtime = gettime() + timeout	8✔
112	local _, err = self.lock:get(timeout)	8✔
113	-- the lock never gets released, only destroyed, so we have to check the error string
114	if err == "timeout" then	8✔
115	if not no_destroy_on_timeout then	2✔
116	self:destroy()	2✔
117	end
118	return nil, err	2✔
119	end
120
121	-- if we get here, the lock was destroyed, so the queue is empty, now wait for all workers to exit
122	if not next(self.workers) then	6✔
123	-- all workers already exited, we're done
124	return true	×
125	end
126
127	-- multiple threads can call this "finish" method, so we must check exiting workers
128	-- one by one.
129	while true do
130	local _, err = self.exit_semaphore:take(1, math.max(0, endtime - gettime()))	8✔
131	if err == "destroyed" then	8✔
132	return true -- someone else destroyed/finished it, so we're done	×
133	end
134	if err == "timeout" then	8✔
135	if not no_destroy_on_timeout then	×
136	self:destroy()	×
137	end
138	return nil, "timeout"	×
139	end
140	if not next(self.workers) then	8✔
141	self.exit_semaphore:destroy()	6✔
142	return true -- all workers exited, we're done	6✔
143	end
144	end
145	end
146
147
148	do
149	local destroyed_func = function()
150	return nil, "destroyed"	8✔
151	end
152
153	local destroyed_queue_mt = {	2✔
154	__index = function()
155	return destroyed_func	8✔
156	end
157	}
158
159	-- destroys a queue immediately. Abandons what is left in the queue.
160	-- Releases all waiting threads with `nil+"destroyed"`
161	function Queue:destroy()	2✔
162	if self.lock then	12✔
163	self.lock:destroy()	12✔
164	end
165	self.sema:destroy()	12✔
166	setmetatable(self, destroyed_queue_mt)	12✔
167
168	-- clear anything left in the queue
169	for key in pairs(self.list) do	14✔
170	self.list[key] = nil	2✔
171	end
172
173	return true	12✔
174	end
175	end
176
177
178	-- adds a worker that will handle whatever is passed into the queue. Can be called
179	-- multiple times to add more workers.
180	-- The threads automatically exit when the queue is destroyed.
181	-- worker function signature: `function(item)` (Note: worker functions run
182	-- unprotected, so wrap code in an (x)pcall if errors are expected, otherwise the
183	-- worker will exit on an error, and queue handling will stop)
184	-- Returns the coroutine added.
185	function Queue:add_worker(worker)	2✔
186	assert(type(worker) == "function", "expected worker to be a function")	12✔
187	local coro
188
189	self.worker_id = self.worker_id + 1	12✔
190	local worker_name = self.name .. ":worker_" .. self.worker_id	12✔
191
192	coro = copas.addnamedthread(worker_name, function()	24✔
193	while true do
194	local item, err = self:pop(math.huge) -- wait forever	30✔
195	if err then	30✔
196	break -- queue destroyed, exit	6✔
197	end
198	worker(item) -- TODO: wrap in errorhandling	18✔
199	end
200	self.workers[coro] = nil	12✔
201	if self.exit_semaphore then	12✔
202	self.exit_semaphore:give(1)	12✔
203	end
204	end)
205
206	self.workers[coro] = true	12✔
207	return coro	12✔
208	end
209
210	-- returns a list/array of current workers (coroutines) handling the queue.
211	-- (only the workers added by `add_worker`, and still active, will be in this list)
212	function Queue:get_workers()	2✔
213	local lst = {}	×
214	for coro in pairs(self.workers) do	×
215	if coroutine.status(coro) ~= "dead" then	×
216	lst[#lst+1] = coro	×
217	end
218	end
219	return lst	×
220	end
221
222	return Queue	2✔

lunarmodules / copas / 31212597912

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