Dispatcher working for check-links. Need to get it working with forwarder.

1 files changed, 267 insertions, 0 deletions

diff --git a/etc/dispatch.lua b/etc/dispatch.lua
new file mode 100644
index 0000000..e6c14a6
--- /dev/null
+++ b/etc/dispatch.lua
@@ -0,0 +1,267 @@
+-----------------------------------------------------------------------------
+-- A hacked dispatcher module
+-- LuaSocket sample files
+-- Author: Diego Nehab
+-- RCS ID: $$
+-----------------------------------------------------------------------------
+local base = _G
+local socket = require("socket")
+local coroutine = require("coroutine")
+module("dispatch")
+
+-- if too much time goes by without any activity in one of our sockets, we
+-- just kill it
+TIMEOUT = 10
+
+-----------------------------------------------------------------------------
+-- Mega hack. Don't try to do this at home.
+-----------------------------------------------------------------------------
+-- Lua 5.1 has coroutine.running(). We need it here, so we use this terrible
+-- hack to emulate it in Lua itself
+-- This is very inefficient, but is very good for debugging.
+local running
+local resume = coroutine.resume
+function coroutine.resume(co, ...)
+    running = co
+    return resume(co, unpack(arg))
+end
+
+function coroutine.running()
+    return running
+end
+
+-----------------------------------------------------------------------------
+-- Mega hack. Don't try to do this at home.
+-----------------------------------------------------------------------------
+-- we can't yield across calls to protect, so we rewrite it with coxpcall
+-- make sure you don't require any module that uses socket.protect before
+-- loading our hack
+function socket.protect(f)
+    return f
+end
+
+function socket.protect(f)
+  return function(...)
+    local co = coroutine.create(f)
+    while true do
+      local results = {resume(co, unpack(arg))}
+      local status = table.remove(results, 1)
+      if not status then
+        if type(results[1]) == 'table' then
+          return nil, results[1][1]
+        else error(results[1]) end 
+      end
+      if coroutine.status(co) == "suspended" then
+        arg = {coroutine.yield(unpack(results))}
+      else
+        return unpack(results)
+      end
+    end
+  end
+end
+
+-----------------------------------------------------------------------------
+-- socket.tcp() replacement for non-blocking I/O
+-----------------------------------------------------------------------------
+local function newtrap(dispatcher)
+    -- try to create underlying socket
+    local tcp, error = socket.tcp()
+    if not tcp then return nil, error end
+    -- put it in non-blocking mode right away
+    tcp:settimeout(0)
+    -- metatable for trap produces new methods on demand for those that we
+    -- don't override explicitly.
+    local metat = { __index = function(table, key) 
+        table[key] = function(...)
+            return tcp[key](tcp, unpack(arg))
+        end
+    end}
+    -- does user want to do his own non-blocking I/O?
+    local zero = false
+    -- create a trap object that will behave just like a real socket object
+    local trap = {  } 
+    -- we ignore settimeout to preserve our 0 timeout, but record whether
+    -- the user wants to do his own non-blocking I/O
+    function trap:settimeout(mode, value)
+        if value == 0 then
+            zero = true
+        else
+            zero = false
+        end
+        return 1
+    end
+    -- send in non-blocking mode and yield on timeout
+    function trap:send(data, first, last) 
+        first = (first or 1) - 1
+        local result, error
+        while true do
+            -- tell dispatcher we want to keep sending before we yield 
+            dispatcher.sending:insert(tcp)                   
+            -- mark time we started waiting
+            dispatcher.context[tcp].last = socket.gettime()
+            -- return control to dispatcher
+            -- if upon return the dispatcher tells us we timed out,
+            -- return an error to whoever called us
+            if coroutine.yield() == "timeout" then 
+                return nil, "timeout" 
+            end
+            -- try sending
+            result, error, first = tcp:send(data, first+1, last)
+            -- if we are done, or there was an unexpected error, 
+            -- break away from loop
+            if error ~= "timeout" then return result, error, first end
+        end
+    end
+    -- receive in non-blocking mode and yield on timeout
+    -- or simply return partial read, if user requested timeout = 0
+    function trap:receive(pattern, partial)
+        local error = "timeout"
+        local value
+        while true do 
+            -- tell dispatcher we want to keep receiving before we yield
+            dispatcher.receiving:insert(tcp)
+            -- mark time we started waiting
+            dispatcher.context[tcp].last = socket.gettime()
+            -- return control to dispatcher
+            -- if upon return the dispatcher tells us we timed out,
+            -- return an error to whoever called us
+            if coroutine.yield() == "timeout" then 
+                return nil, "timeout" 
+            end
+            -- try receiving
+            value, error, partial = tcp:receive(pattern, partial)
+            -- if we are done, or there was an unexpected error, 
+            -- break away from loop
+            if (error ~= "timeout") or zero then 
+                return value, error, partial 
+            end
+        end
+    end
+    -- connect in non-blocking mode and yield on timeout
+    function trap:connect(host, port)
+        local result, error = tcp:connect(host, port)
+        -- mark time we started waiting
+        dispatcher.context[tcp].last = socket.gettime()
+        if error == "timeout" then
+            -- tell dispatcher we will be able to write uppon connection
+            dispatcher.sending:insert(tcp)
+            -- return control to dispatcher
+            -- if upon return the dispatcher tells us we have a
+            -- timeout, just abort
+            if coroutine.yield() == "timeout" then 
+                return nil, "timeout" 
+            end
+            -- when we come back, check if connection was successful
+            result, error = tcp:connect(host, port)
+            if result or error == "already connected" then return 1
+            else return nil, "non-blocking connect failed" end
+        else return result, error end
+    end
+    -- accept in non-blocking mode and yield on timeout
+    function trap:accept()
+        local result, error = tcp:accept()
+        while error == "timeout" do
+            -- mark time we started waiting
+            dispatcher.context[tcp].last = socket.gettime()
+            -- tell dispatcher we will be able to read uppon connection
+            dispatcher.receiving:insert(tcp)
+            -- return control to dispatcher
+            -- if upon return the dispatcher tells us we have a
+            -- timeout, just abort
+            if coroutine.yield() == "timeout" then 
+                return nil, "timeout" 
+            end
+        end 
+        return result, error
+    end
+    -- remove thread from context
+    function trap:close()
+        dispatcher.context[tcp] = nil
+        return tcp:close()
+    end
+    -- add newly created socket to context
+    dispatcher.context[tcp] = {
+        thread = coroutine.running()
+    }
+    return setmetatable(trap, metat)
+end
+
+-----------------------------------------------------------------------------
+-- Our set data structure
+-----------------------------------------------------------------------------
+local function newset()
+    local reverse = {}
+    local set = {}
+    return setmetatable(set, {__index = {
+        insert = function(set, value)
+            if not reverse[value] then
+                table.insert(set, value)
+                reverse[value] = table.getn(set)
+            end
+        end,
+        remove = function(set, value)
+            local index = reverse[value]
+            if index then
+                reverse[value] = nil
+                local top = table.remove(set)
+                if top ~= value then 
+                    reverse[top] = index
+                    set[index] = top
+                end 
+            end
+        end
+    }}) 
+end
+
+-----------------------------------------------------------------------------
+-- Our dispatcher API. 
+-----------------------------------------------------------------------------
+local metat = { __index = {} }
+
+function metat.__index:start(func) 
+    local co = coroutine.create(func)
+    assert(coroutine.resume(co))
+end
+
+function newhandler()
+    local dispatcher = { 
+        context = {},
+        sending = newset(),
+        receiving = newset()
+    }
+    function dispatcher.tcp()
+        return newtrap(dispatcher)
+    end
+    return setmetatable(dispatcher, metat)
+end
+
+-- step through all active threads
+function metat.__index:step()
+    -- check which sockets are interesting and act on them
+    local readable, writable = socket.select(self.receiving, 
+        self.sending, 1)
+    -- for all readable connections, resume their threads
+    for _, who in ipairs(readable) do
+        if self.context[who] then
+            self.receiving:remove(who)
+            assert(coroutine.resume(self.context[who].thread))
+        end
+    end
+    -- for all writable connections, do the same
+    for _, who in ipairs(writable) do
+        if self.context[who] then
+            self.sending:remove(who)
+            assert(coroutine.resume(self.context[who].thread))
+        end
+    end
+    -- politely ask replacement I/O functions in idle threads to 
+    -- return reporting a timeout
+    local now = socket.gettime()
+    for who, data in pairs(self.context) do
+        if  data.last and now - data.last > TIMEOUT then
+            self.sending:remove(who)
+            self.receiving:remove(who)
+            assert(coroutine.resume(self.context[who].thread, "timeout"))
+        end
+    end
+end