--------------------------------------------------------------------------------
--- Copas - Coroutine Oriented Portable Asynchronous Services
---
--- Offers a dispatcher and socket operations based on coroutines.
--- Usage:
--- copas.addserver(server, handler, timeout)
--- copas.addthread(thread, ...) Create a new coroutine thread and run it with args
--- copas.loop(timeout) - listens infinetely
--- copas.step(timeout) - executes one listening step
--- copas.receive(pattern or number) - receives data from a socket
--- copas.settimeout(client, time) if time=0 copas.receive(bufferSize) - receives partial data from a socket were data<=bufferSize
--- copas.send - sends data through a socket
--- copas.wrap - wraps a LuaSocket socket with Copas methods
--- copas.connect - blocks only the thread until connection completes
--- copas.flush - *deprecated* do nothing
---
--- Authors: Andre Carregal and Javier Guerra
--- Contributors: Diego Nehab, Mike Pall, David Burgess, Leonardo Godinho,
--- Thomas Harning Jr. and Gary NG
---
--- Copyright 2005 - Kepler Project (www.keplerproject.org)
---
--- $Id: copas.lua,v 1.31 2008/05/19 18:57:13 carregal Exp $
--------------------------------------------------------------------------------
-local socket = require "socket"
-
-require"luci.util"
-local copcall = luci.util.copcall
-
-
-local WATCH_DOG_TIMEOUT = 120
-
--- Redefines LuaSocket functions with coroutine safe versions
--- (this allows the use of socket.http from within copas)
-local function statusHandler(status, ...)
- if status then return ... end
- return nil, ...
-end
-function socket.protect(func)
- return function (...)
- return statusHandler(copcall(func, ...))
- end
-end
-
-function socket.newtry(finalizer)
- return function (...)
- local status = (...) or false
- if (status==false)then
- copcall(finalizer, select(2, ...) )
- error((select(2, ...)), 0)
- end
- return ...
- end
-end
--- end of LuaSocket redefinitions
-
-
-module ("copas", package.seeall)
-
--- Meta information is public even if begining with an "_"
-_COPYRIGHT = "Copyright (C) 2005 Kepler Project"
-_DESCRIPTION = "Coroutine Oriented Portable Asynchronous Services"
-_VERSION = "Copas 1.1.3"
-
--------------------------------------------------------------------------------
--- Simple set implementation based on LuaSocket's tinyirc.lua example
--- adds a FIFO queue for each value in the set
--------------------------------------------------------------------------------
-local function newset()
- local reverse = {}
- local set = {}
- local q = {}
- setmetatable(set, { __index = {
- insert = function(set, value)
- if not reverse[value] then
- set[#set + 1] = value
- reverse[value] = #set
- end
- end,
-
- remove = function(set, value)
- local index = reverse[value]
- if index then
- reverse[value] = nil
- local top = set[#set]
- set[#set] = nil
- if top ~= value then
- reverse[top] = index
- set[index] = top
- end
- end
- end,
-
- push = function (set, key, itm)
- local qKey = q[key]
- if qKey == nil then
- q[key] = {itm}
- else
- qKey[#qKey + 1] = itm
- end
- end,
-
- pop = function (set, key)
- local t = q[key]
- if t ~= nil then
- local ret = table.remove (t, 1)
- if t[1] == nil then
- q[key] = nil
- end
- return ret
- end
- end
- }})
- return set
-end
-
-local _servers = newset() -- servers being handled
-local _reading_log = {}
-local _writing_log = {}
-
-local _reading = newset() -- sockets currently being read
-local _writing = newset() -- sockets currently being written
-
--------------------------------------------------------------------------------
--- Coroutine based socket I/O functions.
--------------------------------------------------------------------------------
--- reads a pattern from a client and yields to the reading set on timeouts
-function receive(client, pattern, part)
- local s, err
- pattern = pattern or "*l"
- repeat
- s, err, part = client:receive(pattern, part)
- if s or err ~= "timeout" then
- _reading_log[client] = nil
- return s, err, part
- end
- _reading_log[client] = os.time()
- coroutine.yield(client, _reading)
- until false
-end
-
--- same as above but with special treatment when reading chunks,
--- unblocks on any data received.
-function receivePartial(client, pattern)
- local s, err, part
- pattern = pattern or "*l"
- repeat
- s, err, part = client:receive(pattern)
- if s or ( (type(pattern)=="number") and part~="" and part ~=nil ) or
- err ~= "timeout" then
- _reading_log[client] = nil
- return s, err, part
- end
- _reading_log[client] = os.time()
- coroutine.yield(client, _reading)
- until false
-end
-
--- sends data to a client. The operation is buffered and
--- yields to the writing set on timeouts
-function send(client,data, from, to)
- local s, err,sent
- from = from or 1
- local lastIndex = from - 1
-
- repeat
- s, err, lastIndex = client:send(data, lastIndex + 1, to)
- -- adds extra corrotine swap
- -- garantees that high throuput dont take other threads to starvation
- if (math.random(100) > 90) then
- _writing_log[client] = os.time()
- coroutine.yield(client, _writing)
- end
- if s or err ~= "timeout" then
- _writing_log[client] = nil
- return s, err,lastIndex
- end
- _writing_log[client] = os.time()
- coroutine.yield(client, _writing)
- until false
-end
-
--- waits until connection is completed
-function connect(skt,host, port)
- skt:settimeout(0)
- local ret,err = skt:connect (host, port)
- if ret or err ~= "timeout" then
- return ret, err
- end
- _writing_log[skt] = os.time()
- coroutine.yield(skt, _writing)
- ret,err = skt:connect (host, port)
- _writing_log[skt] = nil
- if (err=="already connected") then
- return 1
- end
- return ret, err
-end
-
--- flushes a client write buffer (deprecated)
-function flush(client)
-end
-
--- wraps a socket to use Copas methods (send, receive, flush and settimeout)
-local _skt_mt = {__index = {
- send = function (self, data, from, to)
- return send (self.socket, data, from, to)
- end,
-
- receive = function (self, pattern)
- if (self.timeout==0) then
- return receivePartial(self.socket, pattern)
- end
- return receive (self.socket, pattern)
- end,
-
- flush = function (self)
- return flush (self.socket)
- end,
-
- settimeout = function (self,time)
- self.timeout=time
- return
- end,
-}}
-
-function wrap (skt)
- return setmetatable ({socket = skt}, _skt_mt)
-end
-
---------------------------------------------------
--- Error handling
---------------------------------------------------
-
-local _errhandlers = {} -- error handler per coroutine
-
-function setErrorHandler (err)
- local co = coroutine.running()
- if co then
- _errhandlers [co] = err
- end
-end
-
-local function _deferror (msg, co, skt)
- print (msg, co, skt)
-end
-
--------------------------------------------------------------------------------
--- Thread handling
--------------------------------------------------------------------------------
-
-local function _doTick (co, skt, ...)
- if not co then return end
-
- local ok, res, new_q = coroutine.resume(co, skt, ...)
-
- if ok and res and new_q then
- new_q:insert (res)
- new_q:push (res, co)
- else
- if not ok then copcall (_errhandlers [co] or _deferror, res, co, skt) end
- if skt then skt:close() end
- _errhandlers [co] = nil
- end
-end
-
--- accepts a connection on socket input
-local function _accept(input, handler)
- local client = input:accept()
- if client then
- client:settimeout(0)
- local co = coroutine.create(handler)
- _doTick (co, client)
- --_reading:insert(client)
- end
- return client
-end
-
--- handle threads on a queue
-local function _tickRead (skt)
- _doTick (_reading:pop (skt), skt)
-end
-
-local function _tickWrite (skt)
- _doTick (_writing:pop (skt), skt)
-end
-
--------------------------------------------------------------------------------
--- Adds a server/handler pair to Copas dispatcher
--------------------------------------------------------------------------------
-function addserver(server, handler, timeout)
- server:settimeout(timeout or 0.1)
- _servers[server] = handler
- _reading:insert(server)
-end
-
--------------------------------------------------------------------------------
--- Adds an new courotine thread to Copas dispatcher
--------------------------------------------------------------------------------
-function addthread(thread, ...)
- local co = coroutine.create(thread)
- _doTick (co, nil, ...)
-end
-
--------------------------------------------------------------------------------
--- tasks registering
--------------------------------------------------------------------------------
-
-local _tasks = {}
-
-local function addtaskRead (tsk)
- -- lets tasks call the default _tick()
- tsk.def_tick = _tickRead
-
- _tasks [tsk] = true
-end
-
-local function addtaskWrite (tsk)
- -- lets tasks call the default _tick()
- tsk.def_tick = _tickWrite
-
- _tasks [tsk] = true
-end
-
-local function tasks ()
- return next, _tasks
-end
-
--------------------------------------------------------------------------------
--- main tasks: manage readable and writable socket sets
--------------------------------------------------------------------------------
--- a task to check ready to read events
-local _readable_t = {
- events = function(self)
- local i = 0
- return function ()
- i = i + 1
- return self._evs [i]
- end
- end,
-
- tick = function (self, input)
- local handler = _servers[input]
- if handler then
- input = _accept(input, handler)
- else
- _reading:remove (input)
- self.def_tick (input)
- end
- end
-}
-
-addtaskRead (_readable_t)
-
-
--- a task to check ready to write events
-local _writable_t = {
- events = function (self)
- local i = 0
- return function ()
- i = i+1
- return self._evs [i]
- end
- end,
-
- tick = function (self, output)
- _writing:remove (output)
- self.def_tick (output)
- end
-}
-
-addtaskWrite (_writable_t)
-
-local last_cleansing = 0
-local function _select (timeout)
- local err
- local readable={}
- local writable={}
- local r={}
- local w={}
- local now = os.time()
- local duration = os.difftime
-
-
- _readable_t._evs, _writable_t._evs, err = socket.select(_reading, _writing, timeout)
- local r_evs, w_evs = _readable_t._evs, _writable_t._evs
-
- if duration(now, last_cleansing) > WATCH_DOG_TIMEOUT then
- last_cleansing = now
- for k,v in pairs(_reading_log) do
- if not r_evs[k] and duration(now, v) > WATCH_DOG_TIMEOUT then
- _reading_log[k] = nil
- r_evs[#r_evs + 1] = k
- r_evs[k] = #r_evs
- end
- end
-
- for k,v in pairs(_writing_log) do
- if not w_evs[k] and duration(now, v) > WATCH_DOG_TIMEOUT then
- _writing_log[k] = nil
- w_evs[#w_evs + 1] = k
- w_evs[k] = #w_evs
- end
- end
- end
-
- if err == "timeout" and #r_evs + #w_evs > 0 then return nil
- else return err end
-end
-
-
--------------------------------------------------------------------------------
--- Dispatcher loop step.
--- Listen to client requests and handles them
--------------------------------------------------------------------------------
-function step(timeout)
- local err = _select (timeout)
- if err == "timeout" then return end
-
- if err then
- error(err)
- end
-
- for tsk in tasks() do
- for ev in tsk:events () do
- tsk:tick (ev)
- end
- end
-end
-
--------------------------------------------------------------------------------
--- Dispatcher endless loop.
--- Listen to client requests and handles them forever
--------------------------------------------------------------------------------
-function loop(timeout)
- while true do
- step(timeout)
- end
-end
projects / project / luci.git / commitdiff