net.server_event: Check the buffer *length*, not the buffer itself (Fixes 100% cpu...
[prosody.git] / net / server_event.lua
index f2626f5f4d7c5896a6e83642e4805e360ab5953d..882d10ed95d0e45d3436ed663c73c84258bd4ddb 100644 (file)
@@ -6,7 +6,6 @@
                        notes:
                        -- when using luaevent, never register 2 or more EV_READ at one socket, same for EV_WRITE
                        -- you cant even register a new EV_READ/EV_WRITE callback inside another one
-                       -- never call eventcallback:close( ) from inside eventcallback
                        -- to do some of the above, use timeout events or something what will called from outside
                        -- dont let garbagecollect eventcallbacks, as long they are running
                        -- when using luasec, there are 4 cases of timeout errors: wantread or wantwrite during reading or writing
@@ -24,6 +23,7 @@ local cfg = {
        HANDSHAKE_TIMEOUT     = 60,  -- timeout in seconds per handshake attempt
        MAX_READ_LENGTH       = 1024 * 1024 * 1024 * 1024,  -- max bytes allowed to read from sockets
        MAX_SEND_LENGTH       = 1024 * 1024 * 1024 * 1024,  -- max bytes size of write buffer (for writing on sockets)
+       ACCEPT_QUEUE          = 128,  -- might influence the length of the pending sockets queue
        ACCEPT_DELAY          = 10,  -- seconds to wait until the next attempt of a full server to accept
        READ_TIMEOUT          = 60 * 60 * 6,  -- timeout in seconds for read data from socket
        WRITE_TIMEOUT         = 180,  -- timeout in seconds for write data on socket
@@ -41,6 +41,9 @@ local tostring = use "tostring"
 local coroutine = use "coroutine"
 local setmetatable = use "setmetatable"
 
+local t_insert = table.insert
+local t_concat = table.concat
+
 local ssl = use "ssl"
 local socket = use "socket" or require "socket"
 
@@ -121,14 +124,8 @@ do
                        self.position = new_position or self.position
                        return self.position;
        end
-       function interface_mt:_close() -- regs event to start self:_destroy()
-                       local callback = function( )
-                               self:_destroy();
-                               self.eventclose = nil
-                               return -1
-                       end
-                       self.eventclose = addevent( base, nil, EV_TIMEOUT, callback, 0 )
-                       return true
+       function interface_mt:_close()
+               return self:_destroy();
        end
        
        function interface_mt:_start_connection(plainssl) -- should be called from addclient
@@ -248,7 +245,7 @@ do
                        debug( "closing client with id:", self.id, self.fatalerror )
                        self:_lock( true, true, true )  -- first of all, lock the interface to avoid further actions
                        local _
-                       _ = self.eventread and self.eventread:close( )  -- close events; this must be called outside of the event callbacks!
+                       _ = self.eventread and self.eventread:close( )
                        if self.type == "client" then
                                _ = self.eventwrite and self.eventwrite:close( )
                                _ = self.eventhandshake and self.eventhandshake:close( )
@@ -258,7 +255,7 @@ do
                                _ = self.eventwritetimeout and self.eventwritetimeout:close( )
                                _ = self.eventreadtimeout and self.eventreadtimeout:close( )
                                _ = self.ondisconnect and self:ondisconnect( self.fatalerror ~= "client to close" and self.fatalerror)  -- call ondisconnect listener (wont be the case if handshake failed on connect)
-                               _ = self.conn and self.conn:close( ) -- close connection, must also be called outside of any socket registered events!
+                               _ = self.conn and self.conn:close( ) -- close connection
                                _ = self._server and self._server:counter(-1);
                                self.eventread, self.eventwrite = nil, nil
                                self.eventstarthandshake, self.eventhandshake, self.eventclose = nil, nil, nil
@@ -292,8 +289,9 @@ do
 
        function interface_mt:resume()
                self:_lock(self.nointerface, false, self.nowriting);
-               if not self.eventread then
+               if self.readcallback and not self.eventread then
                        self.eventread = addevent( base, self.conn, EV_READ, self.readcallback, cfg.READ_TIMEOUT );  -- register callback
+                       return true;
                end
        end
 
@@ -316,7 +314,7 @@ do
                        debug( "error:", err )  -- to much, check your app
                        return nil, err
                end
-               self.writebuffer = self.writebuffer .. data -- new buffer
+               t_insert(self.writebuffer, data) -- new buffer
                self.writebufferlen = total
                if not self.eventwrite then  -- register new write event
                        --vdebug( "register new write event" )
@@ -342,7 +340,7 @@ do
                        debug( "try to close server with id:", tostring(self.id))
                        self.fatalerror = "server to close"
                        self:_lock( true )
-                       self:_close( 0 )  -- add new event to remove the server interface
+                       self:_close( 0 )
                        return true
                end
        end
@@ -370,6 +368,7 @@ do
        function interface_mt:ssl()
                return self._usingssl
        end
+       interface_mt.clientport = interface_mt.port -- COMPAT server_select
 
        function interface_mt:type()
                return self._type or "client"
@@ -440,8 +439,11 @@ do
        end
        
        function interface_mt:setlistener(listener)
-               self.onconnect, self.ondisconnect, self.onincoming, self.ontimeout, self.onstatus
-                       = listener.onconnect, listener.ondisconnect, listener.onincoming, listener.ontimeout, listener.onstatus;
+               self:ondetach(); -- Notify listener that it is no longer responsible for this connection
+               self.onconnect, self.ondisconnect, self.onincoming,
+               self.ontimeout, self.onstatus, self.ondetach
+                       = listener.onconnect, listener.ondisconnect, listener.onincoming,
+                       listener.ontimeout, listener.onstatus, listener.ondetach;
        end
        
        -- Stub handlers
@@ -455,6 +457,8 @@ do
        end
        function interface_mt:ondrain()
        end
+       function interface_mt:ondetach()
+       end
        function interface_mt:onstatus()
        end
 end
@@ -464,7 +468,6 @@ end
 local handleclient;
 do
        local string_sub = string.sub  -- caching table lookups
-       local string_len = string.len
        local addevent = base.addevent
        local socket_gettime = socket.gettime
        function handleclient( client, ip, port, server, pattern, listener, sslctx )  -- creates an client interface
@@ -473,7 +476,7 @@ do
                        type = "client";
                        conn = client;
                        currenttime = socket_gettime( );  -- safe the origin
-                       writebuffer = "";  -- writebuffer
+                       writebuffer = {};  -- writebuffer
                        writebufferlen = 0;  -- length of writebuffer
                        send = client.send;  -- caching table lookups
                        receive = client.receive;
@@ -481,6 +484,8 @@ do
                        ondisconnect = listener.ondisconnect;  -- will be called when client disconnects
                        onincoming = listener.onincoming;  -- will be called when client sends data
                        ontimeout = listener.ontimeout; -- called when fatal socket timeout occurs
+                       ondrain = listener.ondrain; -- called when writebuffer is empty
+                       ondetach = listener.ondetach; -- called when disassociating this listener from this connection
                        onstatus = listener.onstatus; -- called for status changes (e.g. of SSL/TLS)
                        eventread = false, eventwrite = false, eventclose = false,
                        eventhandshake = false, eventstarthandshake = false;  -- event handler
@@ -527,10 +532,11 @@ do
                                                interface.eventwritetimeout = false
                                        end
                                end
-                               local succ, err, byte = interface.conn:send( interface.writebuffer, 1, interface.writebufferlen )
+                               interface.writebuffer = { t_concat(interface.writebuffer) }
+                               local succ, err, byte = interface.conn:send( interface.writebuffer[1], 1, interface.writebufferlen )
                                --vdebug( "write data:", interface.writebuffer, "error:", err, "part:", byte )
                                if succ then  -- writing succesful
-                                       interface.writebuffer = ""
+                                       interface.writebuffer[1] = nil
                                        interface.writebufferlen = 0
                                        interface:ondrain();
                                        if interface.fatalerror then
@@ -539,14 +545,17 @@ do
                                        elseif interface.startsslcallback then  -- start ssl connection if needed
                                                debug "starting ssl handshake after writing"
                                                interface.eventstarthandshake = addevent( base, nil, EV_TIMEOUT, interface.startsslcallback, 0 )
+                                       elseif interface.writebufferlen ~= 0 then
+                                               -- data possibly written from ondrain
+                                               return EV_WRITE, cfg.WRITE_TIMEOUT
                                        elseif interface.eventreadtimeout then
-                                               return EV_WRITE, EV_TIMEOUT
+                                               return EV_WRITE, cfg.WRITE_TIMEOUT
                                        end
                                        interface.eventwrite = nil
                                        return -1
                                elseif byte and (err == "timeout" or err == "wantwrite") then  -- want write again
                                        --vdebug( "writebuffer is not empty:", err )
-                                       interface.writebuffer = string_sub( interface.writebuffer, byte + 1, interface.writebufferlen )  -- new buffer
+                                       interface.writebuffer[1] = string_sub( interface.writebuffer[1], byte + 1, interface.writebufferlen )  -- new buffer
                                        interface.writebufferlen = interface.writebufferlen - byte
                                        if "wantread" == err then  -- happens only with luasec
                                                local callback = function( )
@@ -848,6 +857,7 @@ local function link(sender, receiver, buffersize)
                        sender:pause();
                end
        end
+       sender:set_mode("*a");
 end
 
 return {