Mercurial > prosody-hg

comparison util/async.lua @ 11200:bf8f2da84007

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Merge 0.11->trunk
author Kim Alvefur <zash@zash.se>
date Thu, 05 Nov 2020 22:31:25 +0100
parents 558f0555ba02
children 542a9a503073
comparison
equal deleted inserted replaced
11199:6c7c50a4de32 11200:bf8f2da84007
51 if not runner then 51 if not runner then
52 log("error", "unexpected async state: unable to locate runner during error handling"); 52 log("error", "unexpected async state: unable to locate runner during error handling");
53 return false; 53 return false;
54 end 54 end
55 call_watcher(runner, "error", debug.traceback(thread, err)); 55 call_watcher(runner, "error", debug.traceback(thread, err));
56 runner.state, runner.thread = "ready", nil; 56 runner.state = "ready";
57 return runner:run(); 57 return runner:run();
58 elseif state == "ready" then 58 elseif state == "ready" then
59 -- If state is 'ready', it is our responsibility to update runner.state from 'waiting'. 59 -- If state is 'ready', it is our responsibility to update runner.state from 'waiting'.
60 -- We also have to :run(), because the queue might have further items that will not be 60 -- We also have to :run(), because the queue might have further items that will not be
61 -- processed otherwise. FIXME: It's probably best to do this in a nexttick (0 timer). 61 -- processed otherwise. FIXME: It's probably best to do this in a nexttick (0 timer).
157 return true, self.state, #self.queue; 157 return true, self.state, #self.queue;
158 end 158 end
159 159
160 local q, thread = self.queue, self.thread; 160 local q, thread = self.queue, self.thread;
161 if not thread or coroutine.status(thread) == "dead" then 161 if not thread or coroutine.status(thread) == "dead" then
162 --luacheck: ignore 143/coroutine
163 if thread and coroutine.close then
164 coroutine.close(thread);
165 end
162 self:log("debug", "creating new coroutine"); 166 self:log("debug", "creating new coroutine");
163 -- Create a new coroutine for this runner 167 -- Create a new coroutine for this runner
164 thread = runner_create_thread(self.func, self); 168 thread = runner_create_thread(self.func, self);
165 self.thread = thread; 169 self.thread = thread;
166 end 170 end
244 248
245 local function ready() 249 local function ready()
246 return pcall(checkthread); 250 return pcall(checkthread);
247 end 251 end
248 252
253 local function wait_for(promise)
254 local async_wait, async_done = waiter();
255 local ret, err = nil, nil;
256 promise:next(
257 function (r) ret = r; end,
258 function (e) err = e; end)
259 :finally(async_done);
260 async_wait();
261 if ret then
262 return ret;
263 else
264 return nil, err;
265 end
266 end
267
249 return { 268 return {
250 ready = ready; 269 ready = ready;
251 waiter = waiter; 270 waiter = waiter;
252 guarder = guarder; 271 guarder = guarder;
253 runner = runner; 272 runner = runner;
273 wait = wait_for; -- COMPAT w/trunk pre-0.12
274 wait_for = wait_for;
254 }; 275 };