-- original file : copas.lua
-- for more into : see util-soc.lua
local socket = socket or require("socket")
local ssl = ssl or nil -- only loaded upon demand
local WATCH_DOG_TIMEOUT = 120
local UDP_DATAGRAM_MAX = 8192
local type, next, pcall, getmetatable, tostring = type, next, pcall, getmetatable, tostring
local min, max, random = math.min, math.max, math.random
local find = string.find
local insert, remove = table.insert, table.remove
local gettime = socket.gettime
local selectsocket = socket.select
local createcoroutine = coroutine.create
local resumecoroutine = coroutine.resume
local yieldcoroutine = coroutine.yield
local runningcoroutine = coroutine.running
-- Redefines LuaSocket functions with coroutine safe versions (this allows the use
-- of socket.http from within copas).
-- Meta information is public even if beginning with an "_"
local function report(fmt,first,...)
if logs then
report = logs and logs.reporter("copas")
report(fmt,first,...)
elseif fmt then
fmt = "copas: " .. fmt
if first then
print(format(fmt,first,...))
else
print(fmt)
end
end
end
local copas = {
_COPYRIGHT = "Copyright (C) 2005-2016 Kepler Project",
_DESCRIPTION = "Coroutine Oriented Portable Asynchronous Services",
_VERSION = "Copas 2.0.1",
autoclose = true,
running = false,
report = report,
trace = false,
}
local function statushandler(status, ...)
if status then
return ...
end
local err = (...)
if type(err) == "table" then
err = err[1]
end
if copas.trace then
report("error: %s",tostring(err))
end
return nil, err
end
function socket.protect(func)
return function(...)
return statushandler(pcall(func,...))
end
end
function socket.newtry(finalizer)
return function (...)
local status = (...)
if not status then
local detail = select(2,...)
pcall(finalizer,detail)
if copas.trace then
report("error: %s",tostring(detail))
end
return
end
return ...
end
end
-- 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 queue = { }
setmetatable(set, {
__index = {
insert =
function(set, value)
if not reverse[value] then
local n = #set +1
set[n] = value
reverse[value] = n
end
end,
remove =
function(set, value)
local index = reverse[value]
if index then
reverse[value] = nil
local n = #set
local top = set[n]
set[n] = nil
if top ~= value then
reverse[top] = index
set[index] = top
end
end
end,
push =
function (set, key, itm)
local entry = queue[key]
if entry == nil then -- hm can it be false then?
queue[key] = { itm }
else
entry[#entry + 1] = itm
end
end,
pop =
function (set, key)
local top = queue[key]
if top ~= nil then
local ret = remove(top,1)
if top[1] == nil then
queue[key] = nil
end
return ret
end
end
}
} )
return set
end
local _sleeping = {
times = { }, -- list with wake-up times
cos = { }, -- list with coroutines, index matches the 'times' list
lethargy = { }, -- list of coroutines sleeping without a wakeup time
insert =
function()
end,
remove =
function()
end,
push =
function(self, sleeptime, co)
if not co then
return
end
if sleeptime < 0 then
--sleep until explicit wakeup through copas.wakeup
self.lethargy[co] = true
return
else
sleeptime = gettime() + sleeptime
end
local t = self.times
local c = self.cos
local i = 1
local n = #t
while i <= n and t[i] <= sleeptime do
i = i + 1
end
insert(t,i,sleeptime)
insert(c,i,co)
end,
getnext =
-- returns delay until next sleep expires, or nil if there is none
function(self)
local t = self.times
local delay = t[1] and t[1] - gettime() or nil
return delay and max(delay, 0) or nil
end,
pop =
-- find the thread that should wake up to the time
function(self, time)
local t = self.times
local c = self.cos
if #t == 0 or time < t[1] then
return
end
local co = c[1]
remove(t,1)
remove(c,1)
return co
end,
wakeup =
function(self, co)
local let = self.lethargy
if let[co] then
self:push(0, co)
let[co] = nil
else
local c = self.cos
local t = self.times
for i=1,#c do
if c[i] == co then
remove(c,i)
remove(t,i)
self:push(0, co)
return
end
end
end
end
}
local _servers = newset() -- servers being handled
local _reading = newset() -- sockets currently being read
local _writing = newset() -- sockets currently being written
local _reading_log = { }
local _writing_log = { }
local _is_timeout = { -- set of errors indicating a timeout
timeout = true, -- default LuaSocket timeout
wantread = true, -- LuaSec specific timeout
wantwrite = true, -- LuaSec specific timeout
}
-- Coroutine based socket I/O functions.
local function isTCP(socket)
return not find(tostring(socket),"^udp")
end
-- Reads a pattern from a client and yields to the reading set on timeouts UDP: a
-- UDP socket expects a second argument to be a number, so it MUST be provided as
-- the 'pattern' below defaults to a string. Will throw a 'bad argument' error if
-- omitted.
local function copasreceive(client, pattern, part)
if not pattern or pattern == "" then
pattern = "*l"
end
local current_log = _reading_log
local s, err
repeat
s, err, part = client:receive(pattern, part)
if s or (not _is_timeout[err]) then
current_log[client] = nil
return s, err, part
end
if err == "wantwrite" then
current_log = _writing_log
current_log[client] = gettime()
yieldcoroutine(client, _writing)
else
current_log = _reading_log
current_log[client] = gettime()
yieldcoroutine(client, _reading)
end
until false
end
-- Receives data from a client over UDP. Not available for TCP. (this is a copy of
-- receive() method, adapted for receivefrom() use).
local function copasreceivefrom(client, size)
local s, err, port
if not size or size == 0 then
size = UDP_DATAGRAM_MAX
end
repeat
-- upon success err holds ip address
s, err, port = client:receivefrom(size)
if s or err ~= "timeout" then
_reading_log[client] = nil
return s, err, port
end
_reading_log[client] = gettime()
yieldcoroutine(client, _reading)
until false
end
-- Same as above but with special treatment when reading chunks, unblocks on any
-- data received.
local function copasreceivepartial(client, pattern, part)
if not pattern or pattern == "" then
pattern = "*l"
end
local logger = _reading_log
local queue = _reading
local s, err
repeat
s, err, part = client:receive(pattern, part)
if s or (type(pattern) == "number" and part ~= "" and part) or not _is_timeout[err] then
logger[client] = nil
return s, err, part
end
if err == "wantwrite" then
logger = _writing_log
queue = _writing
else
logger = _reading_log
queue = _reading
end
logger[client] = gettime()
yieldcoroutine(client, queue)
until false
end
-- Sends data to a client. The operation is buffered and yields to the writing set
-- on timeouts Note: from and to parameters will be ignored by/for UDP sockets
local function copassend(client, data, from, to)
if not from then
from = 1
end
local lastIndex = from - 1
local logger = _writing_log
local queue = _writing
local s, err
repeat
s, err, lastIndex = client:send(data, lastIndex + 1, to)
-- Adds extra coroutine swap and garantees that high throughput doesn't take
-- other threads to starvation.
if random(100) > 90 then
logger[client] = gettime()
yieldcoroutine(client, queue)
end
if s or not _is_timeout[err] then
logger[client] = nil
return s, err,lastIndex
end
if err == "wantread" then
logger = _reading_log
queue = _reading
else
logger = _writing_log
queue = _writing
end
logger[client] = gettime()
yieldcoroutine(client, queue)
until false
end
-- Sends data to a client over UDP. Not available for TCP. (this is a copy of send()
-- method, adapted for sendto() use).
local function copassendto(client, data, ip, port)
repeat
local s, err = client:sendto(data, ip, port)
-- Adds extra coroutine swap and garantees that high throughput doesn't
-- take other threads to starvation.
if random(100) > 90 then
_writing_log[client] = gettime()
yieldcoroutine(client, _writing)
end
if s or err ~= "timeout" then
_writing_log[client] = nil
return s, err
end
_writing_log[client] = gettime()
yieldcoroutine(client, _writing)
until false
end
-- Waits until connection is completed.
local function copasconnect(skt, host, port)
skt:settimeout(0)
local ret, err, tried_more_than_once
repeat
ret, err = skt:connect (host, port)
-- A non-blocking connect on Windows results in error "Operation already in
-- progress" to indicate that it is completing the request async. So
-- essentially it is the same as "timeout".
if ret or (err ~= "timeout" and err ~= "Operation already in progress") then
-- Once the async connect completes, Windows returns the error "already
-- connected" to indicate it is done, so that error should be ignored.
-- Except when it is the first call to connect, then it was already
-- connected to something else and the error should be returned.
if not ret and err == "already connected" and tried_more_than_once then
ret = 1
err = nil
end
_writing_log[skt] = nil
return ret, err
end
tried_more_than_once = tried_more_than_once or true
_writing_log[skt] = gettime()
yieldcoroutine(skt, _writing)
until false
end
-- Peforms an (async) ssl handshake on a connected TCP client socket. Replacec all
-- previous socket references, with the returned new ssl wrapped socket Throws error
-- and does not return nil+error, as that might silently fail in code like this.
local function copasdohandshake(skt, sslt) -- extra ssl parameters
if not ssl then
ssl = require("ssl")
end
if not ssl then
report("error: no ssl library")
return
end
local nskt, err = ssl.wrap(skt, sslt)
if not nskt then
report("error: %s",tostring(err))
return
end
nskt:settimeout(0)
local queue
repeat
local success, err = nskt:dohandshake()
if success then
return nskt
elseif err == "wantwrite" then
queue = _writing
elseif err == "wantread" then
queue = _reading
else
report("error: %s",tostring(err))
return
end
yieldcoroutine(nskt, queue)
until false
end
-- Flushes a client write buffer.
local function copasflush(client)
end
-- Public.
copas.connect = copassconnect
copas.send = copassend
copas.sendto = copassendto
copas.receive = copasreceive
copas.receivefrom = copasreceivefrom
copas.copasreceivepartial = copasreceivepartial
copas.copasreceivePartial = copasreceivepartial
copas.dohandshake = copasdohandshake
copas.flush = copasflush
-- Wraps a TCP socket to use Copas methods (send, receive, flush and settimeout).
local function _skt_mt_tostring(self)
return tostring(self.socket) .. " (copas wrapped)"
end
local _skt_mt_tcp_index = {
send =
function(self, data, from, to)
return copassend (self.socket, data, from, to)
end,
receive =
function (self, pattern, prefix)
if self.timeout == 0 then
return copasreceivePartial(self.socket, pattern, prefix)
else
return copasreceive(self.socket, pattern, prefix)
end
end,
flush =
function (self)
return copasflush(self.socket)
end,
settimeout =
function (self, time)
self.timeout = time
return true
end,
-- TODO: socket.connect is a shortcut, and must be provided with an alternative
-- if ssl parameters are available, it will also include a handshake
connect =
function(self, ...)
local res, err = copasconnect(self.socket, ...)
if res and self.ssl_params then
res, err = self:dohandshake()
end
return res, err
end,
close =
function(self, ...)
return self.socket:close(...)
end,
-- TODO: socket.bind is a shortcut, and must be provided with an alternative
bind =
function(self, ...)
return self.socket:bind(...)
end,
-- TODO: is this DNS related? hence blocking?
getsockname =
function(self, ...)
return self.socket:getsockname(...)
end,
getstats =
function(self, ...)
return self.socket:getstats(...)
end,
setstats =
function(self, ...)
return self.socket:setstats(...)
end,
listen =
function(self, ...)
return self.socket:listen(...)
end,
accept =
function(self, ...)
return self.socket:accept(...)
end,
setoption =
function(self, ...)
return self.socket:setoption(...)
end,
-- TODO: is this DNS related? hence blocking?
getpeername =
function(self, ...)
return self.socket:getpeername(...)
end,
shutdown =
function(self, ...)
return self.socket:shutdown(...)
end,
dohandshake =
function(self, sslt)
self.ssl_params = sslt or self.ssl_params
local nskt, err = copasdohandshake(self.socket, self.ssl_params)
if not nskt then
return nskt, err
end
self.socket = nskt
return self
end,
}
local _skt_mt_tcp = {
__tostring = _skt_mt_tostring,
__index = _skt_mt_tcp_index,
}
-- wraps a UDP socket, copy of TCP one adapted for UDP.
local _skt_mt_udp_index = {
-- UDP sending is non-blocking, but we provide starvation prevention, so replace
-- anyway.
sendto =
function (self, ...)
return copassendto(self.socket,...)
end,
receive =
function (self, size)
return copasreceive(self.socket, size or UDP_DATAGRAM_MAX)
end,
receivefrom =
function (self, size)
return copasreceivefrom(self.socket, size or UDP_DATAGRAM_MAX)
end,
-- TODO: is this DNS related? hence blocking?
setpeername =
function(self, ...)
return self.socket:getpeername(...)
end,
setsockname =
function(self, ...)
return self.socket:setsockname(...)
end,
-- do not close client, as it is also the server for udp.
close =
function(self, ...)
return true
end
}
local _skt_mt_udp = {
__tostring = _skt_mt_tostring,
__index = _skt_mt_udp_index,
}
for k, v in next, _skt_mt_tcp_index do
if not _skt_mt_udp_index[k] then
_skt_mt_udp_index[k] = v
end
end
-- Wraps a LuaSocket socket object in an async Copas based socket object.
-- @param skt the socket to wrap
-- @sslt (optional) Table with ssl parameters, use an empty table to use ssl with defaults
-- @return wrapped socket object
local function wrap(skt, sslt)
if getmetatable(skt) == _skt_mt_tcp or getmetatable(skt) == _skt_mt_udp then
return skt -- already wrapped
end
skt:settimeout(0)
if isTCP(skt) then
return setmetatable ({ socket = skt, ssl_params = sslt }, _skt_mt_tcp)
else
return setmetatable ({ socket = skt }, _skt_mt_udp)
end
end
copas.wrap = wrap
-- Wraps a handler in a function that deals with wrapping the socket and doing
-- the optional ssl handshake.
function copas.handler(handler, sslparams)
return function (skt,...)
skt = wrap(skt)
if sslparams then
skt:dohandshake(sslparams)
end
return handler(skt,...)
end
end
-- Error handling (a handler per coroutine).
local _errhandlers = { }
function copas.setErrorHandler(err)
local co = runningcoroutine()
if co then
_errhandlers[co] = err
end
end
local function _deferror (msg, co, skt)
report("%s (%s) (%s)", msg, tostring(co), tostring(skt))
end
-- Thread handling
local function _doTick (co, skt, ...)
if not co then
return
end
local ok, res, new_q = resumecoroutine(co, skt, ...)
if ok and res and new_q then
new_q:insert(res)
new_q:push(res, co)
else
if not ok then
pcall(_errhandlers[co] or _deferror, res, co, skt)
end
-- Do not auto-close UDP sockets, as the handler socket is also the server socket.
if skt and copas.autoclose and isTCP(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 = createcoroutine(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.
local function addTCPserver(server, handler, timeout)
server:settimeout(timeout or 0)
_servers[server] = handler
_reading:insert(server)
end
local function addUDPserver(server, handler, timeout)
server:settimeout(timeout or 0)
local co = createcoroutine(handler)
_reading:insert(server)
_doTick(co, server)
end
function copas.addserver(server, handler, timeout)
if isTCP(server) then
addTCPserver(server, handler, timeout)
else
addUDPserver(server, handler, timeout)
end
end
function copas.removeserver(server, keep_open)
local s = server
local mt = getmetatable(server)
if mt == _skt_mt_tcp or mt == _skt_mt_udp then
s = server.socket
end
_servers[s] = nil
_reading:remove(s)
if keep_open then
return true
end
return server:close()
end
-- Adds an new coroutine thread to Copas dispatcher. Create a coroutine that skips
-- the first argument, which is always the socket passed by the scheduler, but `nil`
-- in case of a task/thread
function copas.addthread(handler, ...)
local thread = createcoroutine(function(_, ...) return handler(...) end)
_doTick(thread, nil, ...)
return thread
end
-- tasks registering
local _tasks = { }
-- Lets tasks call the default _tick().
local function addtaskRead(task)
task.def_tick = _tickRead
_tasks[task] = true
end
-- Lets tasks call the default _tick().
local function addtaskWrite(task)
task.def_tick = _tickWrite
_tasks[task] = true
end
local function tasks()
return next, _tasks
end
-- 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)
--sleeping threads task
local _sleeping_t = {
tick = function(self, time, ...)
_doTick(_sleeping:pop(time), ...)
end
}
-- yields the current coroutine and wakes it after 'sleeptime' seconds.
-- If sleeptime<0 then it sleeps until explicitly woken up using 'wakeup'
function copas.sleep(sleeptime)
yieldcoroutine((sleeptime or 0), _sleeping)
end
-- Wakes up a sleeping coroutine 'co'.
function copas.wakeup(co)
_sleeping:wakeup(co)
end
-- Checks for reads and writes on sockets
local last_cleansing = 0
local function _select(timeout)
local now = gettime()
local r_evs, w_evs, err = selectsocket(_reading, _writing, timeout)
_readable_t._evs = r_evs
_writable_t._evs = w_evs
if (last_cleansing - now) > WATCH_DOG_TIMEOUT then
last_cleansing = now
-- Check all sockets selected for reading, and check how long they have been
-- waiting for data already, without select returning them as readable.
for skt, time in next, _reading_log do
if not r_evs[skt] and (time - now) > WATCH_DOG_TIMEOUT then
-- This one timedout while waiting to become readable, so move it in
-- the readable list and try and read anyway, despite not having
-- been returned by select.
local n = #r_evs + 1
_reading_log[skt] = nil
r_evs[n] = skt
r_evs[skt] = n
end
end
-- Do the same for writing.
for skt, time in next, _writing_log do
if not w_evs[skt] and (time - now) > WATCH_DOG_TIMEOUT then
local n = #w_evs + 1
_writing_log[skt] = nil
w_evs[n] = skt
w_evs[skt] = n
end
end
end
if err == "timeout" and #r_evs + #w_evs > 0 then
return nil
else
return err
end
end
-- Check whether there is something to do. It returns false if there are no sockets
-- for read/write nor tasks scheduled (which means Copas is in an empty spin).
local function copasfinished()
return not (next(_reading) or next(_writing) or _sleeping:getnext())
end
-- Dispatcher loop step. It listens to client requests and handles them and returns
-- false if no data was handled (timeout), or true if there was data handled (or nil
-- + error message).
local function copasstep(timeout)
_sleeping_t:tick(gettime())
local nextwait = _sleeping:getnext()
if nextwait then
timeout = timeout and min(nextwait,timeout) or nextwait
elseif copasfinished() then
return false
end
local err = _select(timeout)
if err then
if err == "timeout" then
return false
end
return nil, err
end
for task in tasks() do
for event in task:events() do
task:tick(event)
end
end
return true
end
copas.finished = copasfinished
copas.step = copasstep
-- Dispatcher endless loop. It listens to client requests and handles them forever.
function copas.loop(timeout)
copas.running = true
while not copasfinished() do
copasstep(timeout)
end
copas.running = false
end
-- _G.copas = copas
package.loaded["copas"] = copas
return copas
