Unnamed Fossil Project

-------------------------------------------------------------------------------
-- Copas - Coroutine Oriented Portable Asynchronous Services
--
-- A dispatcher based on coroutines that can be used by TCP/IP servers.
-- Uses LuaSocket as the interface with the TCP/IP stack.
--
-- Authors: Andre Carregal, Javier Guerra, and Fabio Mascarenhas
-- 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.37 2009/04/07 22:09:52 carregal Exp $

-------------------------------------------------------------------------------

if package.loaded["socket.http"] then
  error("you must require copas before require'ing socket.http")
end

local socket = require "socket"


local coxpcall = require "coxpcall"

local WATCH_DOG_TIMEOUT = 120
local UDP_DATAGRAM_MAX = 8192

-- 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
  local err = (...)
  if type(err) == "table" then
    return nil, err[1]
  else
    error(err)
  end
end

function socket.protect(func)
  return function (...)
           return statusHandler(coxpcall.pcall(func, ...))
         end
end

function socket.newtry(finalizer)
  return function (...)
           local status = (...)
           if not status then
             coxpcall.pcall(finalizer, select(2, ...))
             error({ (select(2, ...)) }, 0)
           end
           return ...
         end
end

-- end of LuaSocket redefinitions

local copas = {}


-- Meta information is public even if beginning with an "_"
copas._COPYRIGHT = "Copyright (C) 2005-2010 Kepler Project"
copas._DESCRIPTION = "Coroutine Oriented Portable Asynchronous Services"
copas._VERSION = "Copas 1.1.7"

-- Close the socket associated with the current connection after the handler finishes
copas.autoclose = true

-------------------------------------------------------------------------------
-- 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 = {}

                                end
                              end
                    }})
  return set
end

local fnil = function()end
local _sleeping = {
    times = {}, cos = {},
    lethargy = {} -- потоки в летаргическом сне

    , insert = fnil, remove = fnil

    , push = function(self, sleeptime, co)
        if not co then return end
        if sleeptime<0 then
            --sleep until explicit wakeup through coroutine.wakeup
            self.lethargy[co] = true
            return
        else
            sleeptime = os.time() + sleeptime
        end
        local t, c = self.times, self.cos
        local i, cou = 1, #t
        --TODO: сделать бинарный поиск
        while i<=cou and t[i]<sleeptime do i=i+1 end
        table.insert(t, i, sleeptime)
        table.insert(c, i, co)
      end
        --найти нить, которая должна проснуться ко времени time
    , pop = function(self, time)
        local t, c = self.times, self.cos
        if #t==0 or time<t[1] then return end
        local co = c[1]
        table.remove(t, 1)
        table.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
            let = self.cos
            for i=1,#let do
                if let[i]==co then
                    table.remove(let, i)
                    local tm = self.times[i]
                    table.remove(self.times, i)
                    self:push(0, co)
                    return
                end

            end


        end
      end
} --_sleeping

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
-- 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.
function copas.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

-- receives data from a client over UDP. Not available for TCP.
-- (this is a copy of receive() method, adapted for receivefrom() use)
function copas.receivefrom(client, size)
  local s, err, port
  size = size or UDP_DATAGRAM_MAX
  repeat
    s, err, port = client:receivefrom(size) -- upon success err holds ip address
    if s or err ~= "timeout" then
      _reading_log[client] = nil
      return s, err, port
    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 copas.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
-- Note: from and to parameters will be ignored by/for UDP sockets
function copas.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

-- sends data to a client over UDP. Not available for TCP.
-- (this is a copy of send() method, adapted for sendto() use)
function copas.sendto(client, data, ip, port)
  local s, err,sent

  repeat
    s, err = client:sendto(data, ip, port)
    -- 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
    end
    _writing_log[client] = os.time()
    coroutine.yield(client, _writing)
  until false
end

-- waits until connection is completed
function copas.connect(skt, host, port)
  skt:settimeout(0)
  local ret, err
  repeat
    ret, err = skt:connect (host, port)
    if ret or err ~= "timeout" then
      _writing_log[skt] = nil
      return ret, err
    end
    _writing_log[skt] = os.time()
    coroutine.yield(skt, _writing)
  until false
  return ret, err
end

-- flushes a client write buffer (deprecated)
function copas.flush(client)
end

-- wraps a TCP socket to use Copas methods (send, receive, flush and settimeout)
local _skt_mt = {__index = {
                   send = function (self, data, from, to)
                            return copas.send (self.socket, data, from, to)
                          end,

                   receive = function (self, pattern)
                               if (self.timeout==0) then
                                 return copas.receivePartial(self.socket, pattern)
                               end
                               return copas.receive(self.socket, pattern)
                             end,

                   flush = function (self)
                             return copas.flush(self.socket)
                           end,

                   settimeout = function (self,time)
                                  self.timeout=time
                                  return
                                end,
               }}

-- wraps a UDP socket, copy of TCP one adapted for UDP.
-- Mainly adds sendto() and receivefrom()
local _skt_mt_udp = {__index = {
                   send = function (self, data)
                            return copas.send (self.socket, data)
                          end,

                   sendto = function (self, data, ip, port)
                            return copas.sendto (self.socket, data, ip, port)
                          end,

                   receive = function (self, size)
                               return copas.receive (self.socket, (size or UDP_DATAGRAM_MAX))
                             end,

                   receivefrom = function (self, size)
                               return copas.receivefrom (self.socket, (size or UDP_DATAGRAM_MAX))
                             end,

                   flush = function (self)
                             return copas.flush (self.socket)
                           end,

                   settimeout = function (self,time)
                                  self.timeout=time
                                  return
                                end,
               }}

function copas.wrap (skt)
  if string.sub(tostring(skt),1,3) == "udp" then
    return setmetatable ({socket = skt}, _skt_mt_udp)
  else
    return setmetatable ({socket = skt}, _skt_mt)
  end
end

--------------------------------------------------
-- Error handling
--------------------------------------------------

local _errhandlers = {} -- error handler per coroutine

function copas.setErrorHandler (err)
  local co = coroutine.running()
  if co then
    _errhandlers [co] = err
  end
end

local function _deferror (msg, co, skt)

  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 coxpcall.pcall (_errhandlers [co] or _deferror, res, co, skt) end
    if skt and copas.autoclose then skt:close() end
    _errhandlers [co] = nil
  end
end

-- accepts a connection on socket input
local function _accept(input, handler)
  local client = input:accept()

local function addUDPserver(server, handler, timeout)
    server:settimeout(timeout or 0)
    local co = coroutine.create(handler)
    _reading:insert(server)
    _doTick (co, server)
end

function copas.addserver(server, handler, timeout)
    if string.sub(tostring(server),1,3) == "udp" then
        addUDPserver(server, handler, timeout)
    else
        addTCPserver(server, handler, timeout)
    end
end
-------------------------------------------------------------------------------
-- Adds an new courotine thread to Copas dispatcher
-------------------------------------------------------------------------------
function copas.addthread(thread, ...)
  if type(thread) ~= "thread" then
    thread = coroutine.create(thread)
  end
  _doTick (thread, nil, ...)
  return thread
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
}

function copas.sleep(sleeptime)
    coroutine.yield((sleeptime or 0), _sleeping)
end

function copas.wakeup(co)
    _sleeping:wakeup(co)
end

local last_cleansing = 0

-------------------------------------------------------------------------------
-- Checks for reads and writes on sockets
-------------------------------------------------------------------------------
local function _select (timeout)
  local err




  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

-------------------------------------------------------------------------------
-- Dispatcher loop step.
-- Listen to client requests and handles them
-- Returns false if no data was handled (timeout), or true if there was data
-- handled (or nil + error message)
-------------------------------------------------------------------------------
function copas.step(timeout)
  _sleeping_t:tick(os.time())

  local err = _select (timeout)
  if err == "timeout" then return false end

  if err then
    error(err)
  end

  for tsk in tasks() do
    for ev in tsk:events() do
      tsk:tick (ev)
    end
  end
  return true
end

-------------------------------------------------------------------------------
-- Dispatcher endless loop.
-- Listen to client requests and handles them forever
-------------------------------------------------------------------------------
function copas.loop(timeout)
  while true do
    copas.step(timeout)
  end
end

return copas