prosody
6c2370f17027
net/server_epoll.lua

Find changesets by keywords by author, files, the commit message, revision number or hash, or revset expression.

Software / code / prosody

Comparison

net/server_epoll.lua @ 9855:6c2370f17027

net.server_epoll: Optimize timer handling
author Kim Alvefur <zash@zash.se>
date Tue, 12 Mar 2019 23:13:51 +0100
parent 9853:9aea8dbb105d
child 9933:aac4c55721f9
comparison
equal deleted inserted replaced
9854:115b5e32d960 9855:6c2370f17027
4 -- This project is MIT/X11 licensed. Please see the 4 -- This project is MIT/X11 licensed. Please see the
5 -- COPYING file in the source package for more information. 5 -- COPYING file in the source package for more information.
6 -- 6 --
7 7
8 8
9 local t_sort = table.sort;
10 local t_insert = table.insert; 9 local t_insert = table.insert;
11 local t_remove = table.remove;
12 local t_concat = table.concat; 10 local t_concat = table.concat;
13 local setmetatable = setmetatable; 11 local setmetatable = setmetatable;
14 local tostring = tostring; 12 local tostring = tostring;
15 local pcall = pcall; 13 local pcall = pcall;
16 local type = type; 14 local type = type;
18 local pairs = pairs; 16 local pairs = pairs;
19 local log = require "util.logger".init("server_epoll"); 17 local log = require "util.logger".init("server_epoll");
20 local socket = require "socket"; 18 local socket = require "socket";
21 local luasec = require "ssl"; 19 local luasec = require "ssl";
22 local gettime = require "util.time".now; 20 local gettime = require "util.time".now;
21 local indexedbheap = require "util.indexedbheap";
23 local createtable = require "util.table".create; 22 local createtable = require "util.table".create;
24 local inet = require "util.net"; 23 local inet = require "util.net";
25 local inet_pton = inet.pton; 24 local inet_pton = inet.pton;
26 local _SOCKETINVALID = socket._SOCKETINVALID or -1; 25 local _SOCKETINVALID = socket._SOCKETINVALID or -1;
27 26
67 66
68 local fds = createtable(10, 0); -- FD -> conn 67 local fds = createtable(10, 0); -- FD -> conn
69 68
70 -- Timer and scheduling -- 69 -- Timer and scheduling --
71 70
72 local timers = {}; 71 local timers = indexedbheap.create();
73 72
74 local function noop() end 73 local function noop() end
75 local function closetimer(t) 74 local function closetimer(t)
76 t[1] = 0; 75 t[1] = 0;
77 t[2] = noop; 76 t[2] = noop;
78 end 77 timers:remove(t.id);
79 78 end
80 -- Set to true when timers have changed 79
81 local resort_timers = false; 80 local function reschedule(t, time)
81 t[1] = time;
82 timers:reprioritize(t.id, time);
83 end
82 84
83 -- Add absolute timer 85 -- Add absolute timer
84 local function at(time, f) 86 local function at(time, f)
85 local timer = { time, f, close = closetimer }; 87 local timer = { time, f, close = closetimer, reschedule = reschedule, id = nil };
86 t_insert(timers, timer); 88 timer.id = timers:insert(timer, time);
87 resort_timers = true;
88 return timer; 89 return timer;
89 end 90 end
90 91
91 -- Add relative timer 92 -- Add relative timer
92 local function addtimer(timeout, f) 93 local function addtimer(timeout, f)
95 96
96 -- Run callbacks of expired timers 97 -- Run callbacks of expired timers
97 -- Return time until next timeout 98 -- Return time until next timeout
98 local function runtimers(next_delay, min_wait) 99 local function runtimers(next_delay, min_wait)
99 -- Any timers at all? 100 -- Any timers at all?
100 if not timers[1] then 101 local now = gettime();
102 local peek = timers:peek();
103 while peek do
104
105 if peek > now then
106 next_delay = peek - now;
107 break;
108 end
109
110 local _, timer, id = timers:pop();
111 local ok, ret = pcall(timer[2], now);
112 if ok and type(ret) == "number" then
113 local next_time = now+ret;
114 timer[1] = next_time;
115 timers:insert(timer, next_time);
116 end
117
118 peek = timers:peek();
119 end
120 if peek == nil then
101 return next_delay; 121 return next_delay;
102 end 122 end
103 123
104 if resort_timers then 124 if next_delay < min_wait then
105 -- Sort earliest timers to the end 125 return min_wait;
106 t_sort(timers, function (a, b) return a[1] > b[1]; end); 126 end
107 resort_timers = false;
108 end
109
110 -- Iterate from the end and remove completed timers
111 for i = #timers, 1, -1 do
112 local timer = timers[i];
113 local t, f = timer[1], timer[2];
114 -- Get time for every iteration to increase accuracy
115 local now = gettime();
116 if t > now then
117 -- This timer should not fire yet
118 local diff = t - now;
119 if diff < next_delay then
120 next_delay = diff;
121 end
122 break;
123 end
124 local new_timeout = f(now);
125 if new_timeout then
126 -- Schedule for 'delay' from the time actually scheduled, not from now,
127 -- in order to prevent timer drift, unless it already drifted way out of sync.
128 if (t + new_timeout) > ( now - new_timeout ) then
129 timer[1] = t + new_timeout;
130 else
131 timer[1] = now + new_timeout;
132 end
133 resort_timers = true;
134 else
135 t_remove(timers, i);
136 end
137 end
138
139 if resort_timers or next_delay < min_wait then
140 -- Timers may be added from within a timer callback.
141 -- Those would not be considered for next_delay,
142 -- and we might sleep for too long, so instead
143 -- we return a shorter timeout so we can
144 -- properly sort all new timers.
145 next_delay = min_wait;
146 end
147
148 return next_delay; 127 return next_delay;
149 end 128 end
150 129
151 -- Socket handler interface 130 -- Socket handler interface
152 131
249 end 228 end
250 return 229 return
251 end 230 end
252 t = t or cfg.read_timeout; 231 t = t or cfg.read_timeout;
253 if self._readtimeout then 232 if self._readtimeout then
254 self._readtimeout[1] = gettime() + t; 233 self._readtimeout:reschedule(gettime() + t);
255 resort_timers = true;
256 else 234 else
257 self._readtimeout = addtimer(t, function () 235 self._readtimeout = addtimer(t, function ()
258 if self:on("readtimeout") then 236 if self:on("readtimeout") then
259 return cfg.read_timeout; 237 return cfg.read_timeout;
260 else 238 else
274 end 252 end
275 return 253 return
276 end 254 end
277 t = t or cfg.send_timeout; 255 t = t or cfg.send_timeout;
278 if self._writetimeout then 256 if self._writetimeout then
279 self._writetimeout[1] = gettime() + t; 257 self._writetimeout:reschedule(gettime() + t);
280 resort_timers = true;
281 else 258 else
282 self._writetimeout = addtimer(t, function () 259 self._writetimeout = addtimer(t, function ()
283 self:on("disconnect", "write timeout"); 260 self:on("disconnect", "write timeout");
284 self:destroy(); 261 self:destroy();
285 end); 262 end);