local use = function( what )
return _G[ what ]
end
-local clean = function( tbl )
- for i, k in pairs( tbl ) do
- tbl[ i ] = nil
- end
-end
local log, table_concat = require ("util.logger").init("socket"), table.concat;
local out_put = function (...) return log("debug", table_concat{...}); end
local out_error = function (...) return log("warn", table_concat{...}); end
-local mem_free = collectgarbage
----------------------------------// DECLARATION //--
local type = use "type"
local pairs = use "pairs"
local ipairs = use "ipairs"
+local tonumber = use "tonumber"
local tostring = use "tostring"
-local collectgarbage = use "collectgarbage"
--// lua libs //--
local math_min = math.min
local math_huge = math.huge
local table_concat = table.concat
-local table_remove = table.remove
-local string_len = string.len
local string_sub = string.sub
local coroutine_wrap = coroutine.wrap
local coroutine_yield = coroutine.yield
local socket_bind = luasocket.bind
local socket_sleep = luasocket.sleep
local socket_select = luasocket.select
-local ssl_newcontext = ( luasec and luasec.newcontext )
--// functions //--
local loop
local stats
local idfalse
-local addtimer
local closeall
+local addsocket
local addserver
+local addtimer
local getserver
local wrapserver
local getsettings
local closesocket
local removesocket
local removeserver
-local changetimeout
local wrapconnection
local changesettings
local _selecttimeout
local _sleeptime
+local _tcpbacklog
local _starttime
local _currenttime
local _sendtimeout
local _readtimeout
-local _cleanqueue
-
local _timer
-local _maxclientsperserver
+local _maxselectlen
+local _maxfd
+
+local _maxsslhandshake
----------------------------------// DEFINITION //--
_selecttimeout = 1 -- timeout of socket.select
_sleeptime = 0 -- time to wait at the end of every loop
+_tcpbacklog = 128 -- some kind of hint to the OS
_maxsendlen = 51000 * 1024 -- max len of send buffer
_maxreadlen = 25000 * 1024 -- max len of read buffer
_sendtimeout = 60000 -- allowed send idle time in secs
_readtimeout = 6 * 60 * 60 -- allowed read idle time in secs
-_cleanqueue = false -- clean bufferqueue after using
-
-_maxclientsperserver = 1000
+local is_windows = package.config:sub(1,1) == "\\" -- check the directory separator, to detemine whether this is Windows
+_maxfd = (is_windows and math.huge) or luasocket._SETSIZE or 1024 -- max fd number, limit to 1024 by default to prevent glibc buffer overflow, but not on Windows
+_maxselectlen = luasocket._SETSIZE or 1024 -- But this still applies on Windows
_maxsslhandshake = 30 -- max handshake round-trips
----------------------------------// PRIVATE //--
-wrapserver = function( listeners, socket, ip, serverport, pattern, sslctx, maxconnections ) -- this function wraps a server
+wrapserver = function( listeners, socket, ip, serverport, pattern, sslctx ) -- this function wraps a server -- FIXME Make sure FD < _maxfd
- maxconnections = maxconnections or _maxclientsperserver
+ if socket:getfd() >= _maxfd then
+ out_error("server.lua: Disallowed FD number: "..socket:getfd())
+ socket:close()
+ return nil, "fd-too-large"
+ end
local connections = 0
- local dispatch, disconnect = listeners.onconnect or listeners.onincoming, listeners.ondisconnect
+ local dispatch, disconnect = listeners.onconnect, listeners.ondisconnect
local accept = socket.accept
end
handler.remove = function( )
connections = connections - 1
- end
- handler.close = function( )
- for _, handler in pairs( _socketlist ) do
- if handler.serverport == serverport then
- handler.disconnect( handler, "server closed" )
- handler:close( true )
- end
+ if handler then
+ handler.resume( )
end
+ end
+ handler.close = function()
socket:close( )
_sendlistlen = removesocket( _sendlist, socket, _sendlistlen )
_readlistlen = removesocket( _readlist, socket, _readlistlen )
+ _server[ip..":"..serverport] = nil;
_socketlist[ socket ] = nil
handler = nil
socket = nil
--mem_free( )
out_put "server.lua: closed server handler and removed sockets from list"
end
+ handler.pause = function( hard )
+ if not handler.paused then
+ _readlistlen = removesocket( _readlist, socket, _readlistlen )
+ if hard then
+ _socketlist[ socket ] = nil
+ socket:close( )
+ socket = nil;
+ end
+ handler.paused = true;
+ end
+ end
+ handler.resume = function( )
+ if handler.paused then
+ if not socket then
+ socket = socket_bind( ip, serverport, _tcpbacklog );
+ socket:settimeout( 0 )
+ end
+ _readlistlen = addsocket(_readlist, socket, _readlistlen)
+ _socketlist[ socket ] = handler
+ handler.paused = false;
+ end
+ end
handler.ip = function( )
return ip
end
return socket
end
handler.readbuffer = function( )
- if connections > maxconnections then
+ if _readlistlen >= _maxselectlen or _sendlistlen >= _maxselectlen then
+ handler.pause( )
out_put( "server.lua: refused new client connection: server full" )
return false
end
local client, err = accept( socket ) -- try to accept
if client then
local ip, clientport = client:getpeername( )
- client:settimeout( 0 )
local handler, client, err = wrapconnection( handler, listeners, client, ip, serverport, clientport, pattern, sslctx ) -- wrap new client socket
if err then -- error while wrapping ssl socket
return false
end
connections = connections + 1
out_put( "server.lua: accepted new client connection from ", tostring(ip), ":", tostring(clientport), " to ", tostring(serverport))
- return dispatch( handler )
+ if dispatch and not sslctx then -- SSL connections will notify onconnect when handshake completes
+ return dispatch( handler );
+ end
+ return;
elseif err then -- maybe timeout or something else
out_put( "server.lua: error with new client connection: ", tostring(err) )
return false
wrapconnection = function( server, listeners, socket, ip, serverport, clientport, pattern, sslctx ) -- this function wraps a client to a handler object
+ if socket:getfd() >= _maxfd then
+ out_error("server.lua: Disallowed FD number: "..socket:getfd()) -- PROTIP: Switch to libevent
+ socket:close( ) -- Should we send some kind of error here?
+ if server then
+ server.pause( )
+ end
+ return nil, nil, "fd-too-large"
+ end
socket:settimeout( 0 )
--// local import of socket methods //--
end
return false, "setoption not implemented";
end
- handler.close = function( self, forced )
+ handler.force_close = function ( self, err )
+ if bufferqueuelen ~= 0 then
+ out_put("server.lua: discarding unwritten data for ", tostring(ip), ":", tostring(clientport))
+ bufferqueuelen = 0;
+ end
+ return self:close(err);
+ end
+ handler.close = function( self, err )
if not handler then return true; end
_readlistlen = removesocket( _readlist, socket, _readlistlen )
_readtimes[ handler ] = nil
if bufferqueuelen ~= 0 then
- if not ( forced or fatalerror ) then
- handler.sendbuffer( )
- if bufferqueuelen ~= 0 then -- try again...
- if handler then
- handler.write = nil -- ... but no further writing allowed
- end
- toclose = true
- return false
+ handler.sendbuffer() -- Try now to send any outstanding data
+ if bufferqueuelen ~= 0 then -- Still not empty, so we'll try again later
+ if handler then
+ handler.write = nil -- ... but no further writing allowed
end
- else
- send( socket, table_concat( bufferqueue, "", 1, bufferqueuelen ), 1, bufferlen ) -- forced send
+ toclose = true
+ return false
end
end
if socket then
if handler then
_writetimes[ handler ] = nil
_closelist[ handler ] = nil
+ local _handler = handler;
handler = nil
+ if disconnect then
+ disconnect(_handler, err or false);
+ disconnect = nil
+ end
end
if server then
server.remove( )
return clientport
end
local write = function( self, data )
- bufferlen = bufferlen + string_len( data )
+ bufferlen = bufferlen + #data
if bufferlen > maxsendlen then
_closelist[ handler ] = "send buffer exceeded" -- cannot close the client at the moment, have to wait to the end of the cycle
handler.write = idfalse -- dont write anymore
local buffer, err, part = receive( socket, pattern ) -- receive buffer with "pattern"
if not err or (err == "wantread" or err == "timeout") then -- received something
local buffer = buffer or part or ""
- local len = string_len( buffer )
+ local len = #buffer
if len > maxreadlen then
- disconnect( handler, "receive buffer exceeded" )
- handler:close( true )
+ handler:close( "receive buffer exceeded" )
return false
end
local count = len * STAT_UNIT
else -- connections was closed or fatal error
out_put( "server.lua: client ", tostring(ip), ":", tostring(clientport), " read error: ", tostring(err) )
fatalerror = true
- disconnect( handler, err )
- _ = handler and handler:close( )
+ _ = handler and handler:force_close( err )
return false
end
end
local _sendbuffer = function( ) -- this function sends data
local succ, err, byte, buffer, count;
- local count;
if socket then
buffer = table_concat( bufferqueue, "", 1, bufferqueuelen )
succ, err, byte = send( socket, buffer, 1, bufferlen )
count = ( succ or byte or 0 ) * STAT_UNIT
sendtraffic = sendtraffic + count
_sendtraffic = _sendtraffic + count
- _ = _cleanqueue and clean( bufferqueue )
+ for i = bufferqueuelen,1,-1 do
+ bufferqueue[ i ] = nil
+ end
--out_put( "server.lua: sended '", buffer, "', bytes: ", tostring(succ), ", error: ", tostring(err), ", part: ", tostring(byte), ", to: ", tostring(ip), ":", tostring(clientport) )
else
- succ, err, count = false, "closed", 0;
+ succ, err, count = false, "unexpected close", 0;
end
if succ then -- sending succesful
bufferqueuelen = 0
drain(handler)
end
_ = needtls and handler:starttls(nil)
- _ = toclose and handler:close( )
+ _ = toclose and handler:force_close( )
return true
elseif byte and ( err == "timeout" or err == "wantwrite" ) then -- want write
buffer = string_sub( buffer, byte + 1, bufferlen ) -- new buffer
else -- connection was closed during sending or fatal error
out_put( "server.lua: client ", tostring(ip), ":", tostring(clientport), " write error: ", tostring(err) )
fatalerror = true
- disconnect( handler, err )
- _ = handler and handler:close( )
+ _ = handler and handler:force_close( err )
return false
end
end
-- Set the sslctx
local handshake;
function handler.set_sslctx(self, new_sslctx)
- ssl = true
sslctx = new_sslctx;
- local wrote
- local read
+ local read, wrote
handshake = coroutine_wrap( function( client ) -- create handshake coroutine
local err
for i = 1, _maxsslhandshake do
handler.readbuffer = _readbuffer -- when handshake is done, replace the handshake function with regular functions
handler.sendbuffer = _sendbuffer
_ = status and status( handler, "ssl-handshake-complete" )
+ if self.autostart_ssl and listeners.onconnect then
+ listeners.onconnect(self);
+ end
_readlistlen = addsocket(_readlist, client, _readlistlen)
return true
else
- out_put( "server.lua: error during ssl handshake: ", tostring(err) )
- if err == "wantwrite" and not wrote then
+ if err == "wantwrite" then
_sendlistlen = addsocket(_sendlist, client, _sendlistlen)
wrote = true
- elseif err == "wantread" and not read then
+ elseif err == "wantread" then
_readlistlen = addsocket(_readlist, client, _readlistlen)
read = true
else
break;
end
- --coroutine_yield( handler, nil, err ) -- handshake not finished
- coroutine_yield( )
+ err = nil;
+ coroutine_yield( ) -- handshake not finished
end
end
- disconnect( handler, "ssl handshake failed" )
- _ = handler and handler:close( true ) -- forced disconnect
- return false -- handshake failed
+ out_put( "server.lua: ssl handshake error: ", tostring(err or "handshake too long") )
+ _ = handler and handler:force_close("ssl handshake failed")
+ return false, err -- handshake failed
end
)
end
if luasec then
- if sslctx then -- ssl?
- handler:set_sslctx(sslctx);
- out_put("server.lua: ", "starting ssl handshake")
- local err
- socket, err = ssl_wrap( socket, sslctx ) -- wrap socket
- if err then
- out_put( "server.lua: ssl error: ", tostring(err) )
- --mem_free( )
- return nil, nil, err -- fatal error
+ handler.starttls = function( self, _sslctx)
+ if _sslctx then
+ handler:set_sslctx(_sslctx);
end
- socket:settimeout( 0 )
- handler.readbuffer = handshake
- handler.sendbuffer = handshake
- handshake( socket ) -- do handshake
+ if bufferqueuelen > 0 then
+ out_put "server.lua: we need to do tls, but delaying until send buffer empty"
+ needtls = true
+ return
+ end
+ out_put( "server.lua: attempting to start tls on " .. tostring( socket ) )
+ local oldsocket, err = socket
+ socket, err = ssl_wrap( socket, sslctx ) -- wrap socket
if not socket then
- return nil, nil, "ssl handshake failed";
+ out_put( "server.lua: error while starting tls on client: ", tostring(err or "unknown error") )
+ return nil, err -- fatal error
end
- else
- local sslctx;
- handler.starttls = function( self, _sslctx)
- if _sslctx then
- sslctx = _sslctx;
- handler:set_sslctx(sslctx);
- end
- if bufferqueuelen > 0 then
- out_put "server.lua: we need to do tls, but delaying until send buffer empty"
- needtls = true
- return
- end
- out_put( "server.lua: attempting to start tls on " .. tostring( socket ) )
- local oldsocket, err = socket
- socket, err = ssl_wrap( socket, sslctx ) -- wrap socket
- --out_put( "server.lua: sslwrapped socket is " .. tostring( socket ) )
- if err then
- out_put( "server.lua: error while starting tls on client: ", tostring(err) )
- return nil, err -- fatal error
- end
-
- socket:settimeout( 0 )
-
- -- add the new socket to our system
-
- send = socket.send
- receive = socket.receive
- shutdown = id
-
- _socketlist[ socket ] = handler
- _readlistlen = addsocket(_readlist, socket, _readlistlen)
- -- remove traces of the old socket
+ socket:settimeout( 0 )
- _readlistlen = removesocket( _readlist, oldsocket, _readlistlen )
- _sendlistlen = removesocket( _sendlist, oldsocket, _sendlistlen )
- _socketlist[ oldsocket ] = nil
+ -- add the new socket to our system
+ send = socket.send
+ receive = socket.receive
+ shutdown = id
+ _socketlist[ socket ] = handler
+ _readlistlen = addsocket(_readlist, socket, _readlistlen)
+
+ -- remove traces of the old socket
+ _readlistlen = removesocket( _readlist, oldsocket, _readlistlen )
+ _sendlistlen = removesocket( _sendlist, oldsocket, _sendlistlen )
+ _socketlist[ oldsocket ] = nil
- handler.starttls = nil
- needtls = nil
+ handler.starttls = nil
+ needtls = nil
- -- Secure now
- ssl = true
+ -- Secure now (if handshake fails connection will close)
+ ssl = true
- handler.readbuffer = handshake
- handler.sendbuffer = handshake
- handshake( socket ) -- do handshake
- end
- handler.readbuffer = _readbuffer
- handler.sendbuffer = _sendbuffer
+ handler.readbuffer = handshake
+ handler.sendbuffer = handshake
+ return handshake( socket ) -- do handshake
end
- else
- handler.readbuffer = _readbuffer
- handler.sendbuffer = _sendbuffer
end
+
+ handler.readbuffer = _readbuffer
+ handler.sendbuffer = _sendbuffer
send = socket.send
receive = socket.receive
shutdown = ( ssl and id ) or socket.shutdown
_socketlist[ socket ] = handler
_readlistlen = addsocket(_readlist, socket, _readlistlen)
+
+ if sslctx and luasec then
+ out_put "server.lua: auto-starting ssl negotiation..."
+ handler.autostart_ssl = true;
+ local ok, err = handler:starttls(sslctx);
+ if ok == false then
+ return nil, nil, err
+ end
+ end
+
return handler, socket
end
end
if type( port ) ~= "number" or not ( port >= 0 and port <= 65535 ) then
err = "invalid port"
- elseif _server[ port ] then
- err = "listeners on port '" .. port .. "' already exist"
+ elseif _server[ addr..":"..port ] then
+ err = "listeners on '[" .. addr .. "]:" .. port .. "' already exist"
elseif sslctx and not luasec then
err = "luasec not found"
end
if err then
- out_error( "server.lua, port ", port, ": ", err )
+ out_error( "server.lua, [", addr, "]:", port, ": ", err )
return nil, err
end
addr = addr or "*"
- local server, err = socket_bind( addr, port )
+ local server, err = socket_bind( addr, port, _tcpbacklog )
if err then
- out_error( "server.lua, port ", port, ": ", err )
+ out_error( "server.lua, [", addr, "]:", port, ": ", err )
return nil, err
end
- local handler, err = wrapserver( listeners, server, addr, port, pattern, sslctx, _maxclientsperserver ) -- wrap new server socket
+ local handler, err = wrapserver( listeners, server, addr, port, pattern, sslctx ) -- wrap new server socket
if not handler then
server:close( )
return nil, err
end
server:settimeout( 0 )
_readlistlen = addsocket(_readlist, server, _readlistlen)
- _server[ port ] = handler
+ _server[ addr..":"..port ] = handler
_socketlist[ server ] = handler
- out_put( "server.lua: new "..(sslctx and "ssl " or "").."server listener on '", addr, ":", port, "'" )
+ out_put( "server.lua: new "..(sslctx and "ssl " or "").."server listener on '[", addr, "]:", port, "'" )
return handler
end
-getserver = function ( port )
- return _server[ port ];
+getserver = function ( addr, port )
+ return _server[ addr..":"..port ];
end
-removeserver = function( port )
- local handler = _server[ port ]
+removeserver = function( addr, port )
+ local handler = _server[ addr..":"..port ]
if not handler then
- return nil, "no server found on port '" .. tostring( port ) .. "'"
+ return nil, "no server found on '[" .. addr .. "]:" .. tostring( port ) .. "'"
end
handler:close( )
- _server[ port ] = nil
+ _server[ addr..":"..port ] = nil
return true
end
end
getsettings = function( )
- return _selecttimeout, _sleeptime, _maxsendlen, _maxreadlen, _checkinterval, _sendtimeout, _readtimeout, _cleanqueue, _maxclientsperserver, _maxsslhandshake
+ return {
+ select_timeout = _selecttimeout;
+ select_sleep_time = _sleeptime;
+ tcp_backlog = _tcpbacklog;
+ max_send_buffer_size = _maxsendlen;
+ max_receive_buffer_size = _maxreadlen;
+ select_idle_check_interval = _checkinterval;
+ send_timeout = _sendtimeout;
+ read_timeout = _readtimeout;
+ max_connections = _maxselectlen;
+ max_ssl_handshake_roundtrips = _maxsslhandshake;
+ highest_allowed_fd = _maxfd;
+ }
end
changesettings = function( new )
if type( new ) ~= "table" then
return nil, "invalid settings table"
end
- _selecttimeout = tonumber( new.timeout ) or _selecttimeout
- _sleeptime = tonumber( new.sleeptime ) or _sleeptime
- _maxsendlen = tonumber( new.maxsendlen ) or _maxsendlen
- _maxreadlen = tonumber( new.maxreadlen ) or _maxreadlen
- _checkinterval = tonumber( new.checkinterval ) or _checkinterval
- _sendtimeout = tonumber( new.sendtimeout ) or _sendtimeout
- _readtimeout = tonumber( new.readtimeout ) or _readtimeout
- _cleanqueue = new.cleanqueue
- _maxclientsperserver = new._maxclientsperserver or _maxclientsperserver
- _maxsslhandshake = new._maxsslhandshake or _maxsslhandshake
+ _selecttimeout = tonumber( new.select_timeout ) or _selecttimeout
+ _sleeptime = tonumber( new.select_sleep_time ) or _sleeptime
+ _maxsendlen = tonumber( new.max_send_buffer_size ) or _maxsendlen
+ _maxreadlen = tonumber( new.max_receive_buffer_size ) or _maxreadlen
+ _checkinterval = tonumber( new.select_idle_check_interval ) or _checkinterval
+ _tcpbacklog = tonumber( new.tcp_backlog ) or _tcpbacklog
+ _sendtimeout = tonumber( new.send_timeout ) or _sendtimeout
+ _readtimeout = tonumber( new.read_timeout ) or _readtimeout
+ _maxselectlen = new.max_connections or _maxselectlen
+ _maxsslhandshake = new.max_ssl_handshake_roundtrips or _maxsslhandshake
+ _maxfd = new.highest_allowed_fd or _maxfd
return true
end
local quitting;
-setquitting = function (quit)
+local function setquitting(quit)
quitting = not not quit;
end
end
for handler, err in pairs( _closelist ) do
handler.disconnect( )( handler, err )
- handler:close( true ) -- forced disconnect
+ handler:force_close() -- forced disconnect
+ _closelist[ handler ] = nil;
end
- clean( _closelist )
_currenttime = luasocket_gettime( )
+
+ -- Check for socket timeouts
+ local difftime = os_difftime( _currenttime - _starttime )
+ if difftime > _checkinterval then
+ _starttime = _currenttime
+ for handler, timestamp in pairs( _writetimes ) do
+ if os_difftime( _currenttime - timestamp ) > _sendtimeout then
+ --_writetimes[ handler ] = nil
+ handler.disconnect( )( handler, "send timeout" )
+ handler:force_close() -- forced disconnect
+ end
+ end
+ for handler, timestamp in pairs( _readtimes ) do
+ if os_difftime( _currenttime - timestamp ) > _readtimeout then
+ --_readtimes[ handler ] = nil
+ handler.disconnect( )( handler, "read timeout" )
+ handler:close( ) -- forced disconnect?
+ end
+ end
+ end
+
+ -- Fire timers
if _currenttime - _timer >= math_min(next_timer_time, 1) then
next_timer_time = math_huge;
for i = 1, _timerlistlen do
else
next_timer_time = next_timer_time - (_currenttime - _timer);
end
- socket_sleep( _sleeptime ) -- wait some time
- --collectgarbage( )
+
+ -- wait some time (0 by default)
+ socket_sleep( _sleeptime )
until quitting;
if once and quitting == "once" then quitting = nil; return; end
return "quitting"
end
-step = function ()
+local function step()
return loop(true);
end
--// EXPERIMENTAL //--
local wrapclient = function( socket, ip, serverport, listeners, pattern, sslctx )
- local handler = wrapconnection( nil, listeners, socket, ip, serverport, "clientport", pattern, sslctx )
+ local handler, socket, err = wrapconnection( nil, listeners, socket, ip, serverport, "clientport", pattern, sslctx )
+ if not handler then return nil, err end
_socketlist[ socket ] = handler
- _sendlistlen = addsocket(_sendlist, socket, _sendlistlen)
- if listeners.onconnect then
- -- When socket is writeable, call onconnect
- local _sendbuffer = handler.sendbuffer;
- handler.sendbuffer = function ()
- handler.sendbuffer = _sendbuffer;
- listeners.onconnect(handler);
- -- If there was data with the incoming packet, handle it now.
- if #handler:bufferqueue() > 0 then
- return _sendbuffer();
+ if not sslctx then
+ _sendlistlen = addsocket(_sendlist, socket, _sendlistlen)
+ if listeners.onconnect then
+ -- When socket is writeable, call onconnect
+ local _sendbuffer = handler.sendbuffer;
+ handler.sendbuffer = function ()
+ handler.sendbuffer = _sendbuffer;
+ listeners.onconnect(handler);
+ return _sendbuffer(); -- Send any queued outgoing data
end
end
end
_timer = luasocket_gettime( )
_starttime = luasocket_gettime( )
-addtimer( function( )
- local difftime = os_difftime( _currenttime - _starttime )
- if difftime > _checkinterval then
- _starttime = _currenttime
- for handler, timestamp in pairs( _writetimes ) do
- if os_difftime( _currenttime - timestamp ) > _sendtimeout then
- --_writetimes[ handler ] = nil
- handler.disconnect( )( handler, "send timeout" )
- handler:close( true ) -- forced disconnect
- end
- end
- for handler, timestamp in pairs( _readtimes ) do
- if os_difftime( _currenttime - timestamp ) > _readtimeout then
- --_readtimes[ handler ] = nil
- handler.disconnect( )( handler, "read timeout" )
- handler:close( ) -- forced disconnect?
- end
- end
- end
- end
-)
-
local function setlogger(new_logger)
local old_logger = log;
if new_logger then
----------------------------------// PUBLIC INTERFACE //--
return {
+ _addtimer = addtimer,
addclient = addclient,
wrapclient = wrapclient,
loop = loop,
link = link,
+ step = step,
stats = stats,
closeall = closeall,
- addtimer = addtimer,
addserver = addserver,
getserver = getserver,
setlogger = setlogger,
projects / prosody.git / blobdiff