prosody-hg: net/server_epoll.lua comparison

comparison net/server_epoll.lua @ 9314:c9eef7e8ee65

net.server_epoll: Use util.poll

author	Kim Alvefur <zash@zash.se>
date	Wed, 16 May 2018 23:57:09 +0200
parents	9b0604fe01f1
children	7c954c75b6ac

comparison

equal deleted inserted replaced

-:b95ef295c66d
+:c9eef7e8ee65
 --
 -- This project is MIT/X11 licensed. Please see the
 -- COPYING file in the source package for more information.
 --
--- server_epoll
---  Server backend based on https://luarocks.org/modules/zash/lua-epoll
 local t_sort = table.sort;
 local t_insert = table.insert;
 local t_remove = table.remove;
 local t_concat = table.concat;
 local pcall = pcall;
 local type = type;
 local next = next;
 local pairs = pairs;
 local log = require "util.logger".init("server_epoll");
-local epoll = require "epoll";
 local socket = require "socket";
 local luasec = require "ssl";
 local gettime = require "util.time".now;
 local createtable = require "util.table".create;
 local _SOCKETINVALID = socket._SOCKETINVALID or -1;
-assert(socket.tcp6 and socket.tcp4, "Incompatible LuaSocket version");
+local poll = require "util.poll".new();
 local _ENV = nil;
 -- luacheck: std none
 local default_config = { __index = {
 			self:destroy();
 		end);
 	end
 end
--- lua-epoll flag for currently requested poll state
+function interface:add(r, w)
-function interface:flags()
-	if self._wantread then
-		if self._wantwrite then
-			return "rw";
-		end
-		return "r";
-	elseif self._wantwrite then
-		return "w";
-	end
-end
--- Add or remove sockets or modify epoll flags
-function interface:setflags(r, w)
-	if r ~= nil then self._wantread = r; end
-	if w ~= nil then self._wantwrite = w; end
-	local flags = self:flags();
-	local currentflags = self._flags;
-	if flags == currentflags then
-		return true;
-	end
 	local fd = self:getfd();
 	if fd < 0 then
-		self._wantread, self._wantwrite = nil, nil;
 		return nil, "invalid fd";
 	end
-	local op = "mod";
+	if r == nil then r = self._wantread; end
-	if not flags then
+	if w == nil then w = self._wantwrite; end
-		op = "del";
+	local ok, err = poll:add(fd, r, w);
-	elseif not currentflags then
+	if not ok then
-		op = "add";
+		log("error", "Could not register %s: %s", self, err);
-	end
+		return ok, err;
-	local ok, err = epoll.ctl(op, fd, flags);
+	end
---	log("debug", "epoll_ctl(%q, %d, %q) -> %s" .. (err and ", %q" or ""),
+	self._wantread, self._wantwrite = r, w;
---		op, fd, flags or "", tostring(ok), err);
+	fds[fd] = self;
-	if not ok then return ok, err end
+	log("debug", "Registered %s", self);
-	if op == "add" then
+	return true;
-		fds[fd] = self;
+end
-	elseif op == "del" then
-		fds[fd] = nil;
+function interface:set(r, w)
-	end
+	local fd = self:getfd();
-	self._flags = flags;
+	if fd < 0 then
+		return nil, "invalid fd";
+	end
+	if r == nil then r = self._wantread; end
+	if w == nil then w = self._wantwrite; end
+	local ok, err = poll:set(fd, r, w);
+	if not ok then
+		log("error", "Could not update poller state %s: %s", self, err);
+		return ok, err;
+	end
+	self._wantread, self._wantwrite = r, w;
+	return true;
+end
+function interface:del()
+	local fd = self:getfd();
+	if fd < 0 then
+		return nil, "invalid fd";
+	end
+	if fds[fd] ~= self then
+		return nil, "unregistered fd";
+	end
+	local ok, err = poll:del(fd);
+	if not ok then
+		log("error", "Could not unregister %s: %s", self, err);
+		return ok, err;
+	end
+	self._wantread, self._wantwrite = nil, nil;
+	fds[fd] = nil;
+	log("debug", "Unregistered %s", self);
 	return true;
 end
 -- Called when socket is readable
 function interface:onreadable()
 		if partial and partial ~= "" then
 			self:onconnect();
 			self:on("incoming", partial, err);
 		end
 		if err == "wantread" then
-			self:setflags(true, nil);
+			self:set(true, nil);
 		elseif err == "wantwrite" then
-			self:setflags(nil, true);
+			self:set(nil, true);
 		elseif err ~= "timeout" then
 			self:on("disconnect", err);
 			self:destroy()
 			return;
 		end
 	if not self.conn then return; end -- could have been closed in onconnect
 	local buffer = self.writebuffer;
 	local data = t_concat(buffer);
 	local ok, err, partial = self.conn:send(data);
 	if ok then
-		self:setflags(nil, false);
+		self:set(nil, false);
 		for i = #buffer, 1, -1 do
 			buffer[i] = nil;
 		end
 		self:setwritetimeout(false);
 		self:ondrain(); -- Be aware of writes in ondrain
 			buffer[i] = nil;
 		end
 		self:setwritetimeout();
 	end
 	if err == "wantwrite" or err == "timeout" then
-		self:setflags(nil, true);
+		self:set(nil, true);
 	elseif err == "wantread" then
-		self:setflags(true, nil);
+		self:set(true, nil);
 	elseif err ~= "timeout" then
 		self:on("disconnect", err);
 		self:destroy();
 	end
 end
 		t_insert(buffer, data);
 	else
 		self.writebuffer = { data };
 	end
 	self:setwritetimeout();
-	self:setflags(nil, true);
+	self:set(nil, true);
 	return #data;
 end
 interface.send = interface.write;
 -- Close, possibly after writing is done
 function interface:close()
 	if self.writebuffer and self.writebuffer[1] then
-		self:setflags(false, true); -- Flush final buffer contents
+		self:set(false, true); -- Flush final buffer contents
 		self.write, self.send = noop, noop; -- No more writing
 		log("debug", "Close %s after writing", self);
 		self.ondrain = interface.close;
 	else
 		log("debug", "Close %s now", self);
 		self:destroy();
 	end
 end
 function interface:destroy()
-	self:setflags(false, false);
+	self:del();
 	self:setwritetimeout(false);
 	self:setreadtimeout(false);
 	self.onreadable = noop;
 	self.onwritable = noop;
 	self.destroy = noop;
 	if tls_ctx then self.tls_ctx = tls_ctx; end
 	if self.writebuffer and self.writebuffer[1] then
 		log("debug", "Start TLS on %s after write", self);
 		self.ondrain = interface.starttls;
 		self.starttls = false;
-		self:setflags(nil, true); -- make sure wantwrite is set
+		self:set(nil, true); -- make sure wantwrite is set
 	else
 		log("debug", "Start TLS on %s now", self);
-		self:setflags(false, false);
+		self:del();
 		local conn, err = luasec.wrap(self.conn, tls_ctx or self.tls_ctx);
 		if not conn then
 			self:on("disconnect", err);
 			self:destroy();
 			return conn, err;
 		conn:settimeout(0);
 		self.conn = conn;
 		self.ondrain = nil;
 		self.onwritable = interface.tlshandskake;
 		self.onreadable = interface.tlshandskake;
-		self:setflags(true, true);
+		return self:init();
-		self:setwritetimeout(cfg.handshake_timeout);
 	end
 end
 function interface:tlshandskake()
 	self:setwritetimeout(false);
 		log("debug", "TLS handshake on %s complete", self);
 		self.onwritable = nil;
 		self.onreadable = nil;
 		self._tls = true;
 		self:on("status", "ssl-handshake-complete");
-		self:init();
+		self:setwritetimeout();
+		self:set(true, true);
 	elseif err == "wantread" then
 		log("debug", "TLS handshake on %s to wait until readable", self);
-		self:setflags(true, false);
+		self:set(true, false);
 		self:setreadtimeout(cfg.handshake_timeout);
 	elseif err == "wantwrite" then
 		log("debug", "TLS handshake on %s to wait until writable", self);
-		self:setflags(false, true);
+		self:set(false, true);
 		self:setwritetimeout(cfg.handshake_timeout);
 	else
 		log("debug", "TLS handshake error on %s: %s", self, err);
 		self:on("disconnect", err);
 		self:destroy();
 end
 -- Initialization
 function interface:init()
 	self:setwritetimeout();
-	return self:setflags(true, true);
+	return self:add(true, true);
 end
 function interface:pause()
-	return self:setflags(false);
+	return self:set(false);
 end
 function interface:resume()
-	return self:setflags(true);
+	return self:set(true);
 end
 -- Pause connection for some time
 function interface:pausefor(t)
 	if self._pausefor then
 		self._pausefor:close();
 	end
 	if t == false then return; end
-	self:setflags(false);
+	self:set(false);
 	self._pausefor = addtimer(t, function ()
 		self._pausefor = nil;
 		if self.conn and self.conn:dirty() then
 			self:onreadable();
 		end
-		self:setflags(true);
+		self:set(true);
 	end);
 end
 -- Connected!
 function interface:onconnect()
 		tls_ctx = tls_ctx;
 		tls_direct = tls_ctx and true or false;
 		sockname = addr;
 		sockport = port;
 	}, interface_mt);
-	server:setflags(true, false);
+	server:add(true, false);
 	return server;
 end
 -- COMPAT
 local function wrapclient(conn, addr, port, listeners, read_size, tls_ctx)
 	local conn = setmetatable({
 		conn = fd;
 		onreadable = onreadable;
 		onwriteable = onwriteable;
 		close = function (self)
-			self:setflags(false, false);
+			self:del();
 		end
 	}, interface_mt);
 	if type(fd) == "number" then
 		conn.getfd = function ()
 			return fd;
 		end;
 		-- Otherwise it'll need to be something LuaSocket-compatible
 	end
-	conn:setflags(onreadable, onwriteable);
+	conn:add(onreadable, onwriteable);
 	return conn;
 end;
 -- Dump all data from one connection into another
 local function link(from, to)
 	to.listeners = setmetatable({
 		ondrain = function ()
 			from:resume();
 		end,
 	}, {__index=to.listeners});
-	from:setflags(true, nil);
+	from:set(true, nil);
-	to:setflags(nil, true);
+	to:set(nil, true);
 end
 -- XXX What uses this?
 -- net.adns
 function interface:set_send(new_send)
 -- Main loop
 local function loop(once)
 	repeat
 		local t = runtimers(cfg.max_wait, cfg.min_wait);
-		local fd, r, w = epoll.wait(t);
+		local fd, r, w = poll:wait(t);
 		if fd then
 			local conn = fds[fd];
 			if conn then
 				if r then
 					conn:onreadable();
 				if w then
 					conn:onwritable();
 				end
 			else
 				log("debug", "Removing unknown fd %d", fd);
-				epoll.ctl("del", fd);
+				poll:del(fd);
 			end
 		elseif r ~= "timeout" then
 			log("debug", "epoll_wait error: %s", tostring(r));
 		end
 	until once or (quitting and next(fds) == nil);
 	event = { EV_READ = "r", EV_WRITE = "w", EV_READWRITE = "rw", EV_LEAVE = -1 };
 	addevent = function (fd, mode, callback)
 		local function onevent(self)
 			local ret = self:callback();
 			if ret == -1 then
-				self:setflags(false, false);
+				self:set(false, false);
 			elseif ret then
-				self:setflags(mode == "r" or mode == "rw", mode == "w" or mode == "rw");
+				self:set(mode == "r" or mode == "rw", mode == "w" or mode == "rw");
 			end
 		end
 		local conn = setmetatable({
 			getfd = function () return fd; end;
 			callback = callback;
 			onreadable = onevent;
 			onwritable = onevent;
 			close = function (self)
-				self:setflags(false, false);
+				self:del();
 				fds[fd] = nil;
 			end;
 		}, interface_mt);
-		local ok, err = conn:setflags(mode == "r" or mode == "rw", mode == "w" or mode == "rw");
+		local ok, err = conn:add(mode == "r" or mode == "rw", mode == "w" or mode == "rw");
 		if not ok then return ok, err; end
 		return conn;
 	end;
 };

Mercurial > prosody-hg

comparison net/server_epoll.lua @ 9314:c9eef7e8ee65