Mercurial > prosody-hg

comparison net/cqueues.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 37b884d675f7
children ba409c67353b
comparison
equal deleted inserted replaced
11199:6c7c50a4de32 11200:bf8f2da84007
7 -- This module allows you to use cqueues with a net.server mainloop 7 -- This module allows you to use cqueues with a net.server mainloop
8 -- 8 --
9 9
10 local server = require "net.server"; 10 local server = require "net.server";
11 local cqueues = require "cqueues"; 11 local cqueues = require "cqueues";
12 local timer = require "util.timer";
12 assert(cqueues.VERSION >= 20150113, "cqueues newer than 20150113 required") 13 assert(cqueues.VERSION >= 20150113, "cqueues newer than 20150113 required")
13 14
14 -- Create a single top level cqueue 15 -- Create a single top level cqueue
15 local cq; 16 local cq;
16 17
17 if server.cq then -- server provides cqueues object 18 if server.cq then -- server provides cqueues object
18 cq = server.cq; 19 cq = server.cq;
19 elseif server.get_backend() == "select" and server._addtimer then -- server_select 20 elseif server.watchfd then
20 cq = cqueues.new(); 21 cq = cqueues.new();
21 local function step() 22 local timeout = timer.add_task(cq:timeout() or 0, function ()
23 -- FIXME It should be enough to reschedule this timeout instead of replacing it, but this does not work. See https://issues.prosody.im/1572
22 assert(cq:loop(0)); 24 assert(cq:loop(0));
23 end
24
25 -- Use wrapclient (as wrapconnection isn't exported) to get server_select to watch cq fd
26 local handler = server.wrapclient({
27 getfd = function() return cq:pollfd(); end;
28 settimeout = function() end; -- Method just needs to exist
29 close = function() end; -- Need close method for 'closeall'
30 }, nil, nil, {});
31
32 -- Only need to listen for readable; cqueues handles everything under the hood
33 -- readbuffer is called when `select` notes an fd as readable
34 handler.readbuffer = step;
35
36 -- Use server_select low lever timer facility,
37 -- this callback gets called *every* time there is a timeout in the main loop
38 server._addtimer(function(current_time)
39 -- This may end up in extra step()'s, but cqueues handles it for us.
40 step();
41 return cq:timeout(); 25 return cq:timeout();
42 end); 26 end);
43 elseif server.event and server.base then -- server_event 27 server.watchfd(cq:pollfd(), function ()
44 cq = cqueues.new(); 28 assert(cq:loop(0));
45 -- Only need to listen for readable; cqueues handles everything under the hood
46 local EV_READ = server.event.EV_READ;
47 -- Convert a cqueues timeout to an acceptable timeout for luaevent
48 local function luaevent_safe_timeout(cq)
49 local t = cq:timeout(); 29 local t = cq:timeout();
50 -- if you give luaevent 0 or nil, it re-uses the previous timeout. 30 if t then
51 if t == 0 then 31 timer.stop(timeout);
52 t = 0.000001; -- 1 microsecond is the smallest that works (goes into a `struct timeval`) 32 timeout = timer.add_task(cq:timeout(), function ()
53 elseif t == nil then -- pick something big if we don't have one 33 assert(cq:loop(0));
54 t = 0x7FFFFFFF; -- largest 32bit int 34 return cq:timeout();
35 end);
55 end 36 end
56 return t 37 end);
57 end
58 local event_handle;
59 event_handle = server.base:addevent(cq:pollfd(), EV_READ, function(e)
60 -- Need to reference event_handle or this callback will get collected
61 -- This creates a circular reference that can only be broken if event_handle is manually :close()'d
62 local _ = event_handle;
63 -- Run as many cqueues things as possible (with a timeout of 0)
64 -- If an error is thrown, it will break the libevent loop; but prosody resumes after logging a top level error
65 assert(cq:loop(0));
66 return EV_READ, luaevent_safe_timeout(cq);
67 end, luaevent_safe_timeout(cq));
68 else 38 else
69 error "NYI" 39 error "NYI"
70 end 40 end
71 41
72 return { 42 return {