net/cqueues.lua

   1 -- Prosody IM
   2 -- Copyright (C) 2014 Daurnimator
   3 --
   4 -- This project is MIT/X11 licensed. Please see the
   5 -- COPYING file in the source package for more information.
   6 --
   7 -- This module allows you to use cqueues with a net.server mainloop
   8 --
   9
  10 local server = require "net.server";
  11 local cqueues = require "cqueues";
  12 assert(cqueues.VERSION >= 20150113, "cqueues newer than 20150113 required")
  13
  14 -- Create a single top level cqueue
  15 local cq;
  16
  17 if server.cq then -- server provides cqueues object
  18         cq = server.cq;
  19 elseif server.get_backend() == "select" and server._addtimer then -- server_select
  20         cq = cqueues.new();
  21         local function step()
  22                 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();
  42         end);
  43 elseif server.event and server.base then -- server_event
  44         cq = cqueues.new();
  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();
  50                 -- if you give luaevent 0 or nil, it re-uses the previous timeout.
  51                 if t == 0 then
  52                         t = 0.000001; -- 1 microsecond is the smallest that works (goes into a `struct timeval`)
  53                 elseif t == nil then -- pick something big if we don't have one
  54                         t = 0x7FFFFFFF; -- largest 32bit int
  55                 end
  56                 return t
  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
  69         error "NYI"
  70 end
  71
  72 return {
  73         cq = cq;
  74 }