3719531297

Build # 3719531297

Build Type

push

github

Committed by Thijs Schreijer

Commit Message

fix(queue) finishing would not wait

Run Details

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

1224 of 1487 relevant lines covered (82.31%)

3463.29 hits per line

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

93.55

/src/copas/lock.lua

local copas = require("copas")
local gettime = require("socket").gettime

local DEFAULT_TIMEOUT = 10

local lock = {}
lock.__index = lock


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



--- Creates a new lock.
-- @param seconds (optional) default timeout in seconds when acquiring the lock (defaults to 10)
-- @param not_reentrant (optional) if truthy the lock will not allow a coroutine to grab the same lock multiple times
-- @return the lock object
function lock.new(seconds, not_reentrant)
  local timeout = tonumber(seconds or DEFAULT_TIMEOUT) or -1
  if timeout < 0 then
    error("expected timeout (1st argument) to be a number greater than or equal to 0, got: " .. tostring(seconds), 2)
  end
  return setmetatable({
            timeout = timeout,
            not_reentrant = not_reentrant,
            queue = {},
            q_tip = 0,  -- index of the first in line waiting
            q_tail = 0, -- index where the next one will be inserted
            owner = nil, -- coroutine holding lock currently
            call_count = nil, -- recursion call count
            errors = setmetatable({}, { __mode = "k" }), -- error indexed by coroutine
          }, lock)
end



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

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

  --- destroy a lock.
  -- Releases all waiting threads with `nil+"destroyed"`
  function lock:destroy()
    --print("destroying ",self)
    for i = self.q_tip, self.q_tail do
      local co = self.queue[i]
      self.queue[i] = nil

      if co then
        self.errors[co] = "destroyed"
        --print("marked destroyed ", co)
        copas.wakeup(co)
      end
    end

    if self.owner then
      self.errors[self.owner] = "destroyed"
      --print("marked destroyed ", co)
    end
    self.queue = {}
    self.q_tip = 0
    self.q_tail = 0
    self.destroyed = true

    setmetatable(self, destroyed_lock_mt)
    return true
  end
end


local function timeout_handler(co)
  local self = registry[co]
  if not self then
    return
  end

  for i = self.q_tip, self.q_tail do
    if co == self.queue[i] then
      self.queue[i] = nil
      self.errors[co] = "timeout"
      --print("marked timeout ", co)
      copas.wakeup(co)
      return
    end
  end
  -- if we get here, we own it currently, or we finished it by now, or
  -- the lock was destroyed. Anyway, nothing to do here...
end


--- Acquires the lock.
-- If the lock is owned by another thread, this will yield control, until the
-- lock becomes available, or it times out.
-- If `timeout == 0` then it will immediately return (without yielding).
-- @param timeout (optional) timeout in seconds, if given overrides the timeout passed to `new`.
-- @return wait-time on success, or nil+error+wait_time on failure. Errors can be "timeout", "destroyed", or "lock is not re-entrant"
function lock:get(timeout)
  local co = coroutine.running()
  local start_time

  -- is the lock already taken?
  if self.owner then
    -- are we re-entering?
    if co == self.owner and not self.not_reentrant then
      self.call_count = self.call_count + 1
      return 0
    end

    self.queue[self.q_tail] = co
    self.q_tail = self.q_tail + 1
    timeout = timeout or self.timeout
    if timeout == 0 then
      return nil, "timeout", 0
    end

    -- set up timeout
    registry[co] = self
    copas.timeout(timeout, timeout_handler)

    start_time = gettime()
    copas.pauseforever()

    local err = self.errors[co]
    self.errors[co] = nil
    registry[co] = nil

    --print("released ", co, err)
    if err ~= "timeout" then
      copas.timeout(0)
    end
    if err then
      return nil, err, gettime() - start_time
    end
  end

  -- it's ours to have
  self.owner = co
  self.call_count = 1
  return start_time and (gettime() - start_time) or 0
end


--- Releases the lock currently held.
-- Releasing a lock that is not owned by the current co-routine will return
-- an error.
-- returns true, or nil+err on an error
function lock:release()
  local co = coroutine.running()

  if co ~= self.owner then
    return nil, "cannot release a lock not owned"
  end

  self.call_count = self.call_count - 1
  if self.call_count > 0 then
    -- same coro is still holding it
    return true
  end

  -- need a loop, since individual coroutines might have been removed
  -- so there might be holes
  while self.q_tip < self.q_tail do
    local next_up = self.queue[self.q_tip]
    if next_up then
      self.owner = next_up
      self.queue[self.q_tip] = nil
      self.q_tip = self.q_tip + 1
      copas.wakeup(next_up)
      return true
    end
    self.q_tip = self.q_tip + 1
  end
  -- queue is empty, reset pointers
  self.owner = nil
  self.q_tip = 0
  self.q_tail = 0
  return true
end



return lock

1	local copas = require("copas")	2✔
2	local gettime = require("socket").gettime	2✔
3
4	local DEFAULT_TIMEOUT = 10	2✔
5
6	local lock = {}	2✔
7	lock.__index = lock	2✔
8
9
10	-- registry, locks indexed by the coroutines using them.
11	local registry = setmetatable({}, { __mode="kv" })	2✔
12
13
14
15	--- Creates a new lock.
16	-- @param seconds (optional) default timeout in seconds when acquiring the lock (defaults to 10)
17	-- @param not_reentrant (optional) if truthy the lock will not allow a coroutine to grab the same lock multiple times
18	-- @return the lock object
19	function lock.new(seconds, not_reentrant)	2✔
20	local timeout = tonumber(seconds or DEFAULT_TIMEOUT) or -1	6✔
21	if timeout < 0 then	6✔
22	error("expected timeout (1st argument) to be a number greater than or equal to 0, got: " .. tostring(seconds), 2)	×
23	end
24	return setmetatable({	12✔
25	timeout = timeout,	6✔
26	not_reentrant = not_reentrant,	6✔
27	queue = {},	6✔
28	q_tip = 0, -- index of the first in line waiting	6✔
29	q_tail = 0, -- index where the next one will be inserted	6✔
30	owner = nil, -- coroutine holding lock currently	6✔
31	call_count = nil, -- recursion call count	6✔
32	errors = setmetatable({}, { __mode = "k" }), -- error indexed by coroutine	6✔
33	}, lock)	12✔
34	end
35
36
37
38	do
39	local destroyed_func = function()
40	return nil, "destroyed"	2✔
41	end
42
43	local destroyed_lock_mt = {	2✔
44	__index = function()
45	return destroyed_func	2✔
46	end
47	}
48
49	--- destroy a lock.
50	-- Releases all waiting threads with `nil+"destroyed"`
51	function lock:destroy()	2✔
52	--print("destroying ",self)
53	for i = self.q_tip, self.q_tail do	262✔
54	local co = self.queue[i]	256✔
55	self.queue[i] = nil	256✔
56
57	if co then	256✔
58	self.errors[co] = "destroyed"	249✔
59	--print("marked destroyed ", co)
60	copas.wakeup(co)	249✔
61	end
62	end
63
64	if self.owner then	6✔
65	self.errors[self.owner] = "destroyed"	6✔
66	--print("marked destroyed ", co)
67	end
68	self.queue = {}	6✔
69	self.q_tip = 0	6✔
70	self.q_tail = 0	6✔
71	self.destroyed = true	6✔
72
73	setmetatable(self, destroyed_lock_mt)	6✔
74	return true	6✔
75	end
76	end
77
78
79	local function timeout_handler(co)
80	local self = registry[co]	252✔
81	if not self then	252✔
82	return	×
83	end
84
85	for i = self.q_tip, self.q_tail do	31,377✔
86	if co == self.queue[i] then	31,377✔
87	self.queue[i] = nil	252✔
88	self.errors[co] = "timeout"	252✔
89	--print("marked timeout ", co)
90	copas.wakeup(co)	252✔
91	return	252✔
92	end
93	end
94	-- if we get here, we own it currently, or we finished it by now, or
95	-- the lock was destroyed. Anyway, nothing to do here...
96	end
97
98
99	--- Acquires the lock.
100	-- If the lock is owned by another thread, this will yield control, until the
101	-- lock becomes available, or it times out.
102	-- If `timeout == 0` then it will immediately return (without yielding).
103	-- @param timeout (optional) timeout in seconds, if given overrides the timeout passed to `new`.
104	-- @return wait-time on success, or nil+error+wait_time on failure. Errors can be "timeout", "destroyed", or "lock is not re-entrant"
105	function lock:get(timeout)	2✔
106	local co = coroutine.running()	759✔
107	local start_time
108
109	-- is the lock already taken?
110	if self.owner then	759✔
111	-- are we re-entering?
112	if co == self.owner and not self.not_reentrant then	502✔
113	self.call_count = self.call_count + 1	×
114	return 0	×
115	end
116
117	self.queue[self.q_tail] = co	502✔
118	self.q_tail = self.q_tail + 1	502✔
119	timeout = timeout or self.timeout	502✔
120	if timeout == 0 then	502✔
121	return nil, "timeout", 0	×
122	end
123
124	-- set up timeout
125	registry[co] = self	502✔
126	copas.timeout(timeout, timeout_handler)	502✔
127
128	start_time = gettime()	502✔
129	copas.pauseforever()	502✔
130
131	local err = self.errors[co]	502✔
132	self.errors[co] = nil	502✔
133	registry[co] = nil	502✔
134
135	--print("released ", co, err)
136	if err ~= "timeout" then	502✔
137	copas.timeout(0)	250✔
138	end
139	if err then	502✔
140	return nil, err, gettime() - start_time	502✔
141	end
142	end
143
144	-- it's ours to have
145	self.owner = co	257✔
146	self.call_count = 1	257✔
147	return start_time and (gettime() - start_time) or 0	257✔
148	end
149
150
151	--- Releases the lock currently held.
152	-- Releasing a lock that is not owned by the current co-routine will return
153	-- an error.
154	-- returns true, or nil+err on an error
155	function lock:release()	2✔
156	local co = coroutine.running()	253✔
157
158	if co ~= self.owner then	253✔
159	return nil, "cannot release a lock not owned"	1✔
160	end
161
162	self.call_count = self.call_count - 1	252✔
163	if self.call_count > 0 then	252✔
164	-- same coro is still holding it
165	return true	×
166	end
167
168	-- need a loop, since individual coroutines might have been removed
169	-- so there might be holes
170	while self.q_tip < self.q_tail do	503✔
171	local next_up = self.queue[self.q_tip]	252✔
172	if next_up then	252✔
173	self.owner = next_up	1✔
174	self.queue[self.q_tip] = nil	1✔
175	self.q_tip = self.q_tip + 1	1✔
176	copas.wakeup(next_up)	1✔
177	return true	1✔
178	end
179	self.q_tip = self.q_tip + 1	251✔
180	end
181	-- queue is empty, reset pointers
182	self.owner = nil	251✔
183	self.q_tip = 0	251✔
184	self.q_tail = 0	251✔
185	return true	251✔
186	end
187
188
189
190	return lock	2✔

lunarmodules / copas / 3719531297

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