How to organize interrupts in single-threaded lua?

Question

Lua

In general, the situation is this:
There is lua in 1 thread, you can't create more threads. I'm trying to implement something similar to a JavaScript event loop, namely: reading from a socket, deserializing a string into an event with or without a payload, passing it to a queue, and finally processing the queue. It turned out 3 processes (coroutines), the "parallelism" of which needs to be emitted. the first 2 are very simple, since they have only 1 general case and there is no point in optimizing them, but with 3, where you need to process the queue, there is a problem, because anything can happen there up to infinite loops. You need to "invent" an interrupt mechanism to transfer control from the execution coroutine. That is, in my understanding, it is necessary to insert a check into each called function or cycle whether it is time to return control. While in ideas a preprocessor or some kind of runtime library, which will be able to generate already wrapped functions and loops. Share your experience or thoughts?
Code example:

event = {
  handlers = {
    message = { function(...) print(...) end },
    exec = { function(code) load(code, '=exec')() end }
  },
  listeners = {'message','exec' },
  on = function(self, name, cb, times, debounce)
    if not self['handlers'][name] then
      self['handlers'][name] = { cb }
      self['listeners'][#self.listeners + 1] = { name,#self['handlers'][name] }
    else
      table.insert(self['handlers'][name], cb)
    end
  end,
  fire = function(self, event, ...)
    if self['handlers'][event] then
      for _, v in ipairs(self['handlers'][event]) do
        v(event, ...)
      end
    end
  end,
  off = function(self, listener)
    for k, v in pairs(self['listeners']) do
      if v == listener then
        table.remove(self, k)
      end
    end
    self['handlers'][listener] = nil
  end
}
local queue = coroutine.wrap(
  function()
    while true do
      local buf = socket.read()
      if #buf > 0 then
        for _, v in pairs(split(buf, '\n')) do
          table.insert(messages, v)
        end
        coroutine.yield()
      end
    end
  end)

local dispatcher = coroutine.wrap(
  function()
    while true do
      while #messages > 0 do
        local msg = json.decode(messages:next())
        local t = {}
        local __event
        for k, v in pairs(msg) do
          if k == 'event' then
            __event = v
          else
            table.insert(t, v)
          end
        end
        phs(__event, json.encode(t))
      end
      coroutine.yield()
    end
  end)

local executor = coroutine.wrap(
  function()
    while true do
      event:fire(pls(0.01))
      coroutine.yield()
    end
  end)

while true do
  queue()
  dispatcher()
  executor()
end