Mercurial > prosody-hg

comparison net/server_epoll.lua @ 11200:bf8f2da84007

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Merge 0.11->trunk
author Kim Alvefur <zash@zash.se>
date Thu, 05 Nov 2020 22:31:25 +0100
parents 988ddd57e851
children 06a5919e2496
comparison
equal deleted inserted replaced
11199:6c7c50a4de32 11200:bf8f2da84007
7 7
8 8
9 local t_insert = table.insert; 9 local t_insert = table.insert;
10 local t_concat = table.concat; 10 local t_concat = table.concat;
11 local setmetatable = setmetatable; 11 local setmetatable = setmetatable;
12 local tostring = tostring;
13 local pcall = pcall; 12 local pcall = pcall;
14 local type = type; 13 local type = type;
15 local next = next; 14 local next = next;
16 local pairs = pairs; 15 local pairs = pairs;
17 local log = require "util.logger".init("server_epoll"); 16 local traceback = debug.traceback;
17 local logger = require "util.logger";
18 local log = logger.init("server_epoll");
18 local socket = require "socket"; 19 local socket = require "socket";
19 local luasec = require "ssl"; 20 local luasec = require "ssl";
20 local gettime = require "util.time".now; 21 local realtime = require "util.time".now;
22 local monotonic = require "util.time".monotonic;
21 local indexedbheap = require "util.indexedbheap"; 23 local indexedbheap = require "util.indexedbheap";
22 local createtable = require "util.table".create; 24 local createtable = require "util.table".create;
23 local inet = require "util.net"; 25 local inet = require "util.net";
24 local inet_pton = inet.pton; 26 local inet_pton = inet.pton;
25 local _SOCKETINVALID = socket._SOCKETINVALID or -1; 27 local _SOCKETINVALID = socket._SOCKETINVALID or -1;
28 local new_id = require "util.id".medium;
29 local xpcall = require "util.xpcall".xpcall;
26 30
27 local poller = require "util.poll" 31 local poller = require "util.poll"
28 local EEXIST = poller.EEXIST; 32 local EEXIST = poller.EEXIST;
29 local ENOENT = poller.ENOENT; 33 local ENOENT = poller.ENOENT;
30 34
36 local default_config = { __index = { 40 local default_config = { __index = {
37 -- If a connection is silent for this long, close it unless onreadtimeout says not to 41 -- If a connection is silent for this long, close it unless onreadtimeout says not to
38 read_timeout = 14 * 60; 42 read_timeout = 14 * 60;
39 43
40 -- How long to wait for a socket to become writable after queuing data to send 44 -- How long to wait for a socket to become writable after queuing data to send
41 send_timeout = 60; 45 send_timeout = 180;
46
47 -- How long to wait for a socket to become writable after creation
48 connect_timeout = 20;
42 49
43 -- Some number possibly influencing how many pending connections can be accepted 50 -- Some number possibly influencing how many pending connections can be accepted
44 tcp_backlog = 128; 51 tcp_backlog = 128;
45 52
46 -- If accepting a new incoming connection fails, wait this long before trying again 53 -- If accepting a new incoming connection fails, wait this long before trying again
56 ssl_handshake_timeout = 60; 63 ssl_handshake_timeout = 60;
57 64
58 -- Maximum and minimum amount of time to sleep waiting for events (adjusted for pending timers) 65 -- Maximum and minimum amount of time to sleep waiting for events (adjusted for pending timers)
59 max_wait = 86400; 66 max_wait = 86400;
60 min_wait = 1e-06; 67 min_wait = 1e-06;
68
69 -- Enable extra noisy debug logging
70 -- TODO disable once considered stable
71 verbose = true;
72
73 -- EXPERIMENTAL
74 -- Whether to kill connections in case of callback errors.
75 fatal_errors = false;
76
77 -- Or disable protection (like server_select) for potential performance gains
78 protect_listeners = true;
79
80 -- Attempt writes instantly
81 opportunistic_writes = false;
61 }}; 82 }};
62 local cfg = default_config.__index; 83 local cfg = default_config.__index;
63 84
64 local fds = createtable(10, 0); -- FD -> conn 85 local fds = createtable(10, 0); -- FD -> conn
65 86
66 -- Timer and scheduling -- 87 -- Timer and scheduling --
67 88
68 local timers = indexedbheap.create(); 89 local timers = indexedbheap.create();
69 90
70 local function noop() end 91 local function noop() end
71 local function closetimer(t) 92 local function closetimer(id)
72 t[1] = 0; 93 timers:remove(id);
73 t[2] = noop; 94 end
74 timers:remove(t.id); 95
75 end 96 local function reschedule(id, time)
76 97 time = monotonic() + time;
77 local function reschedule(t, time) 98 timers:reprioritize(id, time);
78 t[1] = time;
79 timers:reprioritize(t.id, time);
80 end
81
82 -- Add absolute timer
83 local function at(time, f)
84 local timer = { time, f, close = closetimer, reschedule = reschedule, id = nil };
85 timer.id = timers:insert(timer, time);
86 return timer;
87 end 99 end
88 100
89 -- Add relative timer 101 -- Add relative timer
90 local function addtimer(timeout, f) 102 local function addtimer(timeout, f, param)
91 return at(gettime() + timeout, f); 103 local time = monotonic() + timeout;
104 if param ~= nil then
105 local timer_callback = f
106 function f(current_time, timer_id)
107 local t = timer_callback(current_time, timer_id, param)
108 return t;
109 end
110 end
111 local id = timers:insert(f, time);
112 return id;
92 end 113 end
93 114
94 -- Run callbacks of expired timers 115 -- Run callbacks of expired timers
95 -- Return time until next timeout 116 -- Return time until next timeout
96 local function runtimers(next_delay, min_wait) 117 local function runtimers(next_delay, min_wait)
97 -- Any timers at all? 118 -- Any timers at all?
98 local now = gettime(); 119 local elapsed = monotonic();
120 local now = realtime();
99 local peek = timers:peek(); 121 local peek = timers:peek();
100 while peek do 122 while peek do
101 123
102 if peek > now then 124 if peek > elapsed then
103 next_delay = peek - now; 125 next_delay = peek - elapsed;
104 break; 126 break;
105 end 127 end
106 128
107 local _, timer, id = timers:pop(); 129 local _, timer, id = timers:pop();
108 local ok, ret = pcall(timer[2], now); 130 local ok, ret = xpcall(timer, traceback, now, id);
109 if ok and type(ret) == "number" then 131 if ok and type(ret) == "number" then
110 local next_time = now+ret; 132 local next_time = elapsed+ret;
111 timer[1] = next_time;
112 timers:insert(timer, next_time); 133 timers:insert(timer, next_time);
134 elseif not ok then
135 log("error", "Error in timer: %s", ret);
113 end 136 end
114 137
115 peek = timers:peek(); 138 peek = timers:peek();
116 end 139 end
117 if peek == nil then 140 if peek == nil then
136 return ("FD %d (%s, %d)"):format(self:getfd(), self.sockname or self.peername, self.sockport or self.peerport); 159 return ("FD %d (%s, %d)"):format(self:getfd(), self.sockname or self.peername, self.sockport or self.peerport);
137 end 160 end
138 return ("FD %d"):format(self:getfd()); 161 return ("FD %d"):format(self:getfd());
139 end 162 end
140 163
164 interface.log = log;
165 function interface:debug(msg, ...) --luacheck: ignore 212/self
166 self.log("debug", msg, ...);
167 end
168
169 interface.noise = interface.debug;
170 function interface:noise(msg, ...) --luacheck: ignore 212/self
171 if cfg.verbose then
172 return self:debug(msg, ...);
173 end
174 end
175
176 function interface:error(msg, ...) --luacheck: ignore 212/self
177 self.log("error", msg, ...);
178 end
179
141 -- Replace the listener and tell the old one 180 -- Replace the listener and tell the old one
142 function interface:setlistener(listeners, data) 181 function interface:setlistener(listeners, data)
143 self:on("detach"); 182 self:on("detach");
144 self.listeners = listeners; 183 self.listeners = listeners;
145 self:on("attach", data); 184 self:on("attach", data);
146 end 185 end
147 186
148 -- Call a listener callback 187 -- Call a listener callback
149 function interface:on(what, ...) 188 function interface:on(what, ...)
150 if not self.listeners then 189 if not self.listeners then
151 log("error", "%s has no listeners", self); 190 self:error("Interface is missing listener callbacks");
152 return; 191 return;
153 end 192 end
154 local listener = self.listeners["on"..what]; 193 local listener = self.listeners["on"..what];
155 if not listener then 194 if not listener then
156 -- log("debug", "Missing listener 'on%s'", what); -- uncomment for development and debugging 195 self:noise("Missing listener 'on%s'", what); -- uncomment for development and debugging
157 return; 196 return;
158 end 197 end
159 local ok, err = pcall(listener, self, ...); 198 if not cfg.protect_listeners then
199 return listener(self, ...);
200 end
201 local onerror = self.listeners.onerror or traceback;
202 local ok, err = xpcall(listener, onerror, self, ...);
160 if not ok then 203 if not ok then
161 log("error", "Error calling on%s: %s", what, err); 204 if cfg.fatal_errors then
205 self:error("Closing due to error calling on%s: %s", what, err);
206 self:destroy();
207 else
208 self:debug("Error calling on%s: %s", what, err);
209 end
210 return nil, err;
162 end 211 end
163 return err; 212 return err;
213 end
214
215 -- Allow this one to be overridden
216 function interface:onincoming(...)
217 return self:on("incoming", ...);
164 end 218 end
165 219
166 -- Return the file descriptor number 220 -- Return the file descriptor number
167 function interface:getfd() 221 function interface:getfd()
168 if self.conn then 222 if self.conn then
217 271
218 -- Timeout for detecting dead or idle sockets 272 -- Timeout for detecting dead or idle sockets
219 function interface:setreadtimeout(t) 273 function interface:setreadtimeout(t)
220 if t == false then 274 if t == false then
221 if self._readtimeout then 275 if self._readtimeout then
222 self._readtimeout:close(); 276 closetimer(self._readtimeout);
223 self._readtimeout = nil; 277 self._readtimeout = nil;
224 end 278 end
225 return 279 return
226 end 280 end
227 t = t or cfg.read_timeout; 281 t = t or cfg.read_timeout;
228 if self._readtimeout then 282 if self._readtimeout then
229 self._readtimeout:reschedule(gettime() + t); 283 reschedule(self._readtimeout, t);
230 else 284 else
231 self._readtimeout = addtimer(t, function () 285 self._readtimeout = addtimer(t, function ()
232 if self:on("readtimeout") then 286 if self:on("readtimeout") then
287 self:noise("Read timeout handled");
233 return cfg.read_timeout; 288 return cfg.read_timeout;
234 else 289 else
290 self:debug("Read timeout not handled, disconnecting");
235 self:on("disconnect", "read timeout"); 291 self:on("disconnect", "read timeout");
236 self:destroy(); 292 self:destroy();
237 end 293 end
238 end); 294 end);
239 end 295 end
241 297
242 -- Timeout for detecting dead sockets 298 -- Timeout for detecting dead sockets
243 function interface:setwritetimeout(t) 299 function interface:setwritetimeout(t)
244 if t == false then 300 if t == false then
245 if self._writetimeout then 301 if self._writetimeout then
246 self._writetimeout:close(); 302 closetimer(self._writetimeout);
247 self._writetimeout = nil; 303 self._writetimeout = nil;
248 end 304 end
249 return 305 return
250 end 306 end
251 t = t or cfg.send_timeout; 307 t = t or cfg.send_timeout;
252 if self._writetimeout then 308 if self._writetimeout then
253 self._writetimeout:reschedule(gettime() + t); 309 reschedule(self._writetimeout, t);
254 else 310 else
255 self._writetimeout = addtimer(t, function () 311 self._writetimeout = addtimer(t, function ()
256 self:on("disconnect", "write timeout"); 312 self:noise("Write timeout");
313 self:on("disconnect", self._connected and "write timeout" or "connection timeout");
257 self:destroy(); 314 self:destroy();
258 end); 315 end);
259 end 316 end
260 end 317 end
261 318
267 if r == nil then r = self._wantread; end 324 if r == nil then r = self._wantread; end
268 if w == nil then w = self._wantwrite; end 325 if w == nil then w = self._wantwrite; end
269 local ok, err, errno = poll:add(fd, r, w); 326 local ok, err, errno = poll:add(fd, r, w);
270 if not ok then 327 if not ok then
271 if errno == EEXIST then 328 if errno == EEXIST then
272 log("debug", "%s already registered!", self); 329 self:debug("FD already registered in poller! (EEXIST)");
273 return self:set(r, w); -- So try to change its flags 330 return self:set(r, w); -- So try to change its flags
274 end 331 end
275 log("error", "Could not register %s: %s(%d)", self, err, errno); 332 self:debug("Could not register in poller: %s(%d)", err, errno);
276 return ok, err; 333 return ok, err;
277 end 334 end
278 self._wantread, self._wantwrite = r, w; 335 self._wantread, self._wantwrite = r, w;
279 fds[fd] = self; 336 fds[fd] = self;
280 log("debug", "Watching %s", self); 337 self:noise("Registered in poller");
281 return true; 338 return true;
282 end 339 end
283 340
284 function interface:set(r, w) 341 function interface:set(r, w)
285 local fd = self:getfd(); 342 local fd = self:getfd();
288 end 345 end
289 if r == nil then r = self._wantread; end 346 if r == nil then r = self._wantread; end
290 if w == nil then w = self._wantwrite; end 347 if w == nil then w = self._wantwrite; end
291 local ok, err, errno = poll:set(fd, r, w); 348 local ok, err, errno = poll:set(fd, r, w);
292 if not ok then 349 if not ok then
293 log("error", "Could not update poller state %s: %s(%d)", self, err, errno); 350 self:debug("Could not update poller state: %s(%d)", err, errno);
294 return ok, err; 351 return ok, err;
295 end 352 end
296 self._wantread, self._wantwrite = r, w; 353 self._wantread, self._wantwrite = r, w;
297 return true; 354 return true;
298 end 355 end
305 if fds[fd] ~= self then 362 if fds[fd] ~= self then
306 return nil, "unregistered fd"; 363 return nil, "unregistered fd";
307 end 364 end
308 local ok, err, errno = poll:del(fd); 365 local ok, err, errno = poll:del(fd);
309 if not ok and errno ~= ENOENT then 366 if not ok and errno ~= ENOENT then
310 log("error", "Could not unregister %s: %s(%d)", self, err, errno); 367 self:debug("Could not unregister: %s(%d)", err, errno);
311 return ok, err; 368 return ok, err;
312 end 369 end
313 self._wantread, self._wantwrite = nil, nil; 370 self._wantread, self._wantwrite = nil, nil;
314 fds[fd] = nil; 371 fds[fd] = nil;
315 log("debug", "Unwatched %s", self); 372 self:noise("Unregistered from poller");
316 return true; 373 return true;
317 end 374 end
318 375
319 function interface:setflags(r, w) 376 function interface:setflags(r, w)
320 if not(self._wantread or self._wantwrite) then 377 if not(self._wantread or self._wantwrite) then
332 -- Called when socket is readable 389 -- Called when socket is readable
333 function interface:onreadable() 390 function interface:onreadable()
334 local data, err, partial = self.conn:receive(self.read_size or cfg.read_size); 391 local data, err, partial = self.conn:receive(self.read_size or cfg.read_size);
335 if data then 392 if data then
336 self:onconnect(); 393 self:onconnect();
337 self:on("incoming", data); 394 self:onincoming(data);
338 else 395 else
339 if err == "wantread" then 396 if err == "wantread" then
340 self:set(true, nil); 397 self:set(true, nil);
341 err = "timeout"; 398 err = "timeout";
342 elseif err == "wantwrite" then 399 elseif err == "wantwrite" then
343 self:set(nil, true); 400 self:set(nil, true);
344 err = "timeout"; 401 err = "timeout";
345 end 402 end
346 if partial and partial ~= "" then 403 if partial and partial ~= "" then
347 self:onconnect(); 404 self:onconnect();
348 self:on("incoming", partial, err); 405 self:onincoming(partial, err);
349 end 406 end
350 if err ~= "timeout" then 407 if err ~= "timeout" then
408 if err == "closed" then
409 self:debug("Connection closed by remote");
410 else
411 self:debug("Read error, closing (%s)", err);
412 end
351 self:on("disconnect", err); 413 self:on("disconnect", err);
352 self:destroy() 414 self:destroy()
353 return; 415 return;
354 end 416 end
355 end 417 end
356 if not self.conn then return; end 418 if not self.conn then return; end
419 if self._limit and (data or partial) then
420 local cost = self._limit * #(data or partial);
421 if cost > cfg.min_wait then
422 self:setreadtimeout(false);
423 self:pausefor(cost);
424 return;
425 end
426 end
357 if self._wantread and self.conn:dirty() then 427 if self._wantread and self.conn:dirty() then
358 self:setreadtimeout(false); 428 self:setreadtimeout(false);
359 self:pausefor(cfg.read_retry_delay); 429 self:pausefor(cfg.read_retry_delay);
360 else 430 else
361 self:setreadtimeout(); 431 self:setreadtimeout();
365 -- Called when socket is writable 435 -- Called when socket is writable
366 function interface:onwritable() 436 function interface:onwritable()
367 self:onconnect(); 437 self:onconnect();
368 if not self.conn then return; end -- could have been closed in onconnect 438 if not self.conn then return; end -- could have been closed in onconnect
369 local buffer = self.writebuffer; 439 local buffer = self.writebuffer;
370 local data = t_concat(buffer); 440 local data = #buffer == 1 and buffer[1] or t_concat(buffer);
371 local ok, err, partial = self.conn:send(data); 441 local ok, err, partial = self.conn:send(data);
372 if ok then 442 if ok then
373 self:set(nil, false); 443 self:set(nil, false);
374 for i = #buffer, 1, -1 do 444 for i = #buffer, 1, -1 do
375 buffer[i] = nil; 445 buffer[i] = nil;
376 end 446 end
377 self:setwritetimeout(false); 447 self:setwritetimeout(false);
378 self:ondrain(); -- Be aware of writes in ondrain 448 self:ondrain(); -- Be aware of writes in ondrain
379 return; 449 return;
380 elseif partial then 450 elseif partial then
451 self:debug("Sent %d out of %d buffered bytes", partial, #data);
381 buffer[1] = data:sub(partial+1); 452 buffer[1] = data:sub(partial+1);
382 for i = #buffer, 2, -1 do 453 for i = #buffer, 2, -1 do
383 buffer[i] = nil; 454 buffer[i] = nil;
384 end 455 end
456 self:set(nil, true);
385 self:setwritetimeout(); 457 self:setwritetimeout();
386 end 458 end
387 if err == "wantwrite" or err == "timeout" then 459 if err == "wantwrite" or err == "timeout" then
388 self:set(nil, true); 460 self:set(nil, true);
389 elseif err == "wantread" then 461 elseif err == "wantread" then
405 if buffer then 477 if buffer then
406 t_insert(buffer, data); 478 t_insert(buffer, data);
407 else 479 else
408 self.writebuffer = { data }; 480 self.writebuffer = { data };
409 end 481 end
410 self:setwritetimeout(); 482 if not self._write_lock then
411 self:set(nil, true); 483 if cfg.opportunistic_writes then
484 self:onwritable();
485 return #data;
486 end
487 self:setwritetimeout();
488 self:set(nil, true);
489 end
412 return #data; 490 return #data;
413 end 491 end
414 interface.send = interface.write; 492 interface.send = interface.write;
415 493
416 -- Close, possibly after writing is done 494 -- Close, possibly after writing is done
417 function interface:close() 495 function interface:close()
418 if self.writebuffer and self.writebuffer[1] then 496 if self.writebuffer and self.writebuffer[1] then
419 self:set(false, true); -- Flush final buffer contents 497 self:set(false, true); -- Flush final buffer contents
420 self.write, self.send = noop, noop; -- No more writing 498 self.write, self.send = noop, noop; -- No more writing
421 log("debug", "Close %s after writing", self); 499 self:debug("Close after writing remaining buffered data");
422 self.ondrain = interface.close; 500 self.ondrain = interface.close;
423 else 501 else
424 log("debug", "Close %s now", self); 502 self:debug("Closing now");
425 self.write, self.send = noop, noop; 503 self.write, self.send = noop, noop;
426 self.close = noop; 504 self.close = noop;
427 self:on("disconnect"); 505 self:on("disconnect");
428 self:destroy(); 506 self:destroy();
429 end 507 end
448 526
449 function interface:starttls(tls_ctx) 527 function interface:starttls(tls_ctx)
450 if tls_ctx then self.tls_ctx = tls_ctx; end 528 if tls_ctx then self.tls_ctx = tls_ctx; end
451 self.starttls = false; 529 self.starttls = false;
452 if self.writebuffer and self.writebuffer[1] then 530 if self.writebuffer and self.writebuffer[1] then
453 log("debug", "Start TLS on %s after write", self); 531 self:debug("Start TLS after write");
454 self.ondrain = interface.starttls; 532 self.ondrain = interface.starttls;
455 self:set(nil, true); -- make sure wantwrite is set 533 self:set(nil, true); -- make sure wantwrite is set
456 else 534 else
457 if self.ondrain == interface.starttls then 535 if self.ondrain == interface.starttls then
458 self.ondrain = nil; 536 self.ondrain = nil;
459 end 537 end
460 self.onwritable = interface.tlshandskake; 538 self.onwritable = interface.tlshandshake;
461 self.onreadable = interface.tlshandskake; 539 self.onreadable = interface.tlshandshake;
462 self:set(true, true); 540 self:set(true, true);
463 log("debug", "Prepare to start TLS on %s", self); 541 self:debug("Prepared to start TLS");
464 end 542 end
465 end 543 end
466 544
467 function interface:tlshandskake() 545 function interface:tlshandshake()
468 self:setwritetimeout(false); 546 self:setwritetimeout(false);
469 self:setreadtimeout(false); 547 self:setreadtimeout(false);
470 if not self._tls then 548 if not self._tls then
471 self._tls = true; 549 self._tls = true;
472 log("debug", "Start TLS on %s now", self); 550 self:debug("Starting TLS now");
473 self:del(); 551 self:del();
552 self:updatenames(); -- Can't getpeer/sockname after wrap()
474 local ok, conn, err = pcall(luasec.wrap, self.conn, self.tls_ctx); 553 local ok, conn, err = pcall(luasec.wrap, self.conn, self.tls_ctx);
475 if not ok then 554 if not ok then
476 conn, err = ok, conn; 555 conn, err = ok, conn;
477 log("error", "Failed to initialize TLS: %s", err); 556 self:debug("Failed to initialize TLS: %s", err);
478 end 557 end
479 if not conn then 558 if not conn then
480 self:on("disconnect", err); 559 self:on("disconnect", err);
481 self:destroy(); 560 self:destroy();
482 return conn, err; 561 return conn, err;
483 end 562 end
484 conn:settimeout(0); 563 conn:settimeout(0);
485 self.conn = conn; 564 self.conn = conn;
486 if conn.sni and self.servername then 565 if conn.sni then
487 conn:sni(self.servername); 566 if self.servername then
567 conn:sni(self.servername);
568 elseif self._server and type(self._server.hosts) == "table" and next(self._server.hosts) ~= nil then
569 conn:sni(self._server.hosts, true);
570 end
488 end 571 end
489 self:on("starttls"); 572 self:on("starttls");
490 self.ondrain = nil; 573 self.ondrain = nil;
491 self.onwritable = interface.tlshandskake; 574 self.onwritable = interface.tlshandshake;
492 self.onreadable = interface.tlshandskake; 575 self.onreadable = interface.tlshandshake;
493 return self:init(); 576 return self:init();
494 end 577 end
578 self:noise("Continuing TLS handshake");
495 local ok, err = self.conn:dohandshake(); 579 local ok, err = self.conn:dohandshake();
496 if ok then 580 if ok then
497 log("debug", "TLS handshake on %s complete", self); 581 local info = self.conn.info and self.conn:info();
582 if type(info) == "table" then
583 self:debug("TLS handshake complete (%s with %s)", info.protocol, info.cipher);
584 else
585 self:debug("TLS handshake complete");
586 end
498 self.onwritable = nil; 587 self.onwritable = nil;
499 self.onreadable = nil; 588 self.onreadable = nil;
500 self:on("status", "ssl-handshake-complete"); 589 self:on("status", "ssl-handshake-complete");
501 self:setwritetimeout(); 590 self:setwritetimeout();
502 self:set(true, true); 591 self:set(true, true);
503 elseif err == "wantread" then 592 elseif err == "wantread" then
504 log("debug", "TLS handshake on %s to wait until readable", self); 593 self:noise("TLS handshake to wait until readable");
505 self:set(true, false); 594 self:set(true, false);
506 self:setreadtimeout(cfg.ssl_handshake_timeout); 595 self:setreadtimeout(cfg.ssl_handshake_timeout);
507 elseif err == "wantwrite" then 596 elseif err == "wantwrite" then
508 log("debug", "TLS handshake on %s to wait until writable", self); 597 self:noise("TLS handshake to wait until writable");
509 self:set(false, true); 598 self:set(false, true);
510 self:setwritetimeout(cfg.ssl_handshake_timeout); 599 self:setwritetimeout(cfg.ssl_handshake_timeout);
511 else 600 else
512 log("debug", "TLS handshake error on %s: %s", self, err); 601 self:debug("TLS handshake error: %s", err);
513 self:on("disconnect", err); 602 self:on("disconnect", err);
514 self:destroy(); 603 self:destroy();
515 end 604 end
516 end 605 end
517 606
518 local function wrapsocket(client, server, read_size, listeners, tls_ctx, extra) -- luasocket object -> interface object 607 local function wrapsocket(client, server, read_size, listeners, tls_ctx, extra) -- luasocket object -> interface object
519 client:settimeout(0); 608 client:settimeout(0);
609 local conn_id = ("conn%s"):format(new_id());
520 local conn = setmetatable({ 610 local conn = setmetatable({
521 conn = client; 611 conn = client;
522 _server = server; 612 _server = server;
523 created = gettime(); 613 created = realtime();
524 listeners = listeners; 614 listeners = listeners;
525 read_size = read_size or (server and server.read_size); 615 read_size = read_size or (server and server.read_size);
526 writebuffer = {}; 616 writebuffer = {};
527 tls_ctx = tls_ctx or (server and server.tls_ctx); 617 tls_ctx = tls_ctx or (server and server.tls_ctx);
528 tls_direct = server and server.tls_direct; 618 tls_direct = server and server.tls_direct;
619 id = conn_id;
620 log = logger.init(conn_id);
529 extra = extra; 621 extra = extra;
530 }, interface_mt); 622 }, interface_mt);
531 623
532 if extra then 624 if extra then
533 if extra.servername then 625 if extra.servername then
540 end 632 end
541 633
542 function interface:updatenames() 634 function interface:updatenames()
543 local conn = self.conn; 635 local conn = self.conn;
544 local ok, peername, peerport = pcall(conn.getpeername, conn); 636 local ok, peername, peerport = pcall(conn.getpeername, conn);
545 if ok then 637 if ok and peername then
546 self.peername, self.peerport = peername, peerport; 638 self.peername, self.peerport = peername, peerport or 0;
547 end 639 end
548 local ok, sockname, sockport = pcall(conn.getsockname, conn); 640 local ok, sockname, sockport = pcall(conn.getsockname, conn);
549 if ok then 641 if ok and sockname then
550 self.sockname, self.sockport = sockname, sockport; 642 self.sockname, self.sockport = sockname, sockport or 0;
551 end 643 end
552 end 644 end
553 645
554 -- A server interface has new incoming connections waiting 646 -- A server interface has new incoming connections waiting
555 -- This replaces the onreadable callback 647 -- This replaces the onreadable callback
556 function interface:onacceptable() 648 function interface:onacceptable()
557 local conn, err = self.conn:accept(); 649 local conn, err = self.conn:accept();
558 if not conn then 650 if not conn then
559 log("debug", "Error accepting new client: %s, server will be paused for %ds", err, cfg.accept_retry_interval); 651 self:debug("Error accepting new client: %s, server will be paused for %ds", err, cfg.accept_retry_interval);
560 self:pausefor(cfg.accept_retry_interval); 652 self:pausefor(cfg.accept_retry_interval);
561 return; 653 return;
562 end 654 end
563 local client = wrapsocket(conn, self, nil, self.listeners); 655 local client = wrapsocket(conn, self, nil, self.listeners);
564 log("debug", "New connection %s", tostring(client)); 656 client:debug("New connection %s on server %s", client, self);
565 client:init(); 657 client:init();
566 if self.tls_direct then 658 if self.tls_direct then
567 client:starttls(self.tls_ctx); 659 client:starttls(self.tls_ctx);
660 else
661 client:onconnect();
568 end 662 end
569 end 663 end
570 664
571 -- Initialization 665 -- Initialization
572 function interface:init() 666 function interface:init()
573 self:setwritetimeout(); 667 self:setwritetimeout(cfg.connect_timeout);
574 return self:add(true, true); 668 return self:add(true, true);
575 end 669 end
576 670
577 function interface:pause() 671 function interface:pause()
672 self:noise("Pause reading");
578 return self:set(false); 673 return self:set(false);
579 end 674 end
580 675
581 function interface:resume() 676 function interface:resume()
677 self:noise("Resume reading");
582 return self:set(true); 678 return self:set(true);
583 end 679 end
584 680
585 -- Pause connection for some time 681 -- Pause connection for some time
586 function interface:pausefor(t) 682 function interface:pausefor(t)
683 self:noise("Pause for %fs", t);
587 if self._pausefor then 684 if self._pausefor then
588 self._pausefor:close(); 685 closetimer(self._pausefor);
686 self._pausefor = nil;
589 end 687 end
590 if t == false then return; end 688 if t == false then return; end
591 self:set(false); 689 self:set(false);
592 self._pausefor = addtimer(t, function () 690 self._pausefor = addtimer(t, function ()
593 self._pausefor = nil; 691 self._pausefor = nil;
594 self:set(true); 692 self:set(true);
693 self:noise("Resuming after pause, connection is %s", not self.conn and "missing" or self.conn:dirty() and "dirty" or "clean");
595 if self.conn and self.conn:dirty() then 694 if self.conn and self.conn:dirty() then
596 self:onreadable(); 695 self:onreadable();
597 end 696 end
598 end); 697 end);
599 end 698 end
600 699
700 function interface:setlimit(Bps)
701 if Bps > 0 then
702 self._limit = 1/Bps;
703 else
704 self._limit = nil;
705 end
706 end
707
708 function interface:pause_writes()
709 if self._write_lock then
710 return
711 end
712 self:noise("Pause writes");
713 self._write_lock = true;
714 self:setwritetimeout(false);
715 self:set(nil, false);
716 end
717
718 function interface:resume_writes()
719 if not self._write_lock then
720 return
721 end
722 self:noise("Resume writes");
723 self._write_lock = nil;
724 if self.writebuffer[1] then
725 self:setwritetimeout();
726 self:set(nil, true);
727 end
728 end
729
601 -- Connected! 730 -- Connected!
602 function interface:onconnect() 731 function interface:onconnect()
603 if self.conn and not self.peername and self.conn.getpeername then 732 self._connected = true;
604 self.peername, self.peerport = self.conn:getpeername(); 733 self:updatenames();
605 end 734 self:debug("Connected (%s)", self);
606 self.onconnect = noop; 735 self.onconnect = noop;
607 self:on("connect"); 736 self:on("connect");
608 end 737 end
609 738
610 local function addserver(addr, port, listeners, read_size, tls_ctx) 739 local function wrapserver(conn, addr, port, listeners, config)
740 local server = setmetatable({
741 conn = conn;
742 created = realtime();
743 listeners = listeners;
744 read_size = config and config.read_size;
745 onreadable = interface.onacceptable;
746 tls_ctx = config and config.tls_ctx;
747 tls_direct = config and config.tls_direct;
748 hosts = config and config.sni_hosts;
749 sockname = addr;
750 sockport = port;
751 log = logger.init(("serv%s"):format(new_id()));
752 }, interface_mt);
753 server:debug("Server %s created", server);
754 server:add(true, false);
755 return server;
756 end
757
758 local function listen(addr, port, listeners, config)
611 local conn, err = socket.bind(addr, port, cfg.tcp_backlog); 759 local conn, err = socket.bind(addr, port, cfg.tcp_backlog);
612 if not conn then return conn, err; end 760 if not conn then return conn, err; end
613 conn:settimeout(0); 761 conn:settimeout(0);
614 local server = setmetatable({ 762 return wrapserver(conn, addr, port, listeners, config);
615 conn = conn; 763 end
616 created = gettime(); 764
617 listeners = listeners; 765 -- COMPAT
766 local function addserver(addr, port, listeners, read_size, tls_ctx)
767 return listen(addr, port, listeners, {
618 read_size = read_size; 768 read_size = read_size;
619 onreadable = interface.onacceptable;
620 tls_ctx = tls_ctx; 769 tls_ctx = tls_ctx;
621 tls_direct = tls_ctx and true or false; 770 tls_direct = tls_ctx and true or false;
622 sockname = addr; 771 });
623 sockport = port;
624 }, interface_mt);
625 server:add(true, false);
626 return server;
627 end 772 end
628 773
629 -- COMPAT 774 -- COMPAT
630 local function wrapclient(conn, addr, port, listeners, read_size, tls_ctx, extra) 775 local function wrapclient(conn, addr, port, listeners, read_size, tls_ctx, extra)
631 local client = wrapsocket(conn, nil, read_size, listeners, tls_ctx, extra); 776 local client = wrapsocket(conn, nil, read_size, listeners, tls_ctx, extra);
657 end 802 end
658 if type(create) ~= "function" then 803 if type(create) ~= "function" then
659 return nil, "invalid socket type"; 804 return nil, "invalid socket type";
660 end 805 end
661 local conn, err = create(); 806 local conn, err = create();
807 if not conn then return conn, err; end
662 local ok, err = conn:settimeout(0); 808 local ok, err = conn:settimeout(0);
663 if not ok then return ok, err; end 809 if not ok then return ok, err; end
664 local ok, err = conn:setpeername(addr, port); 810 local ok, err = conn:setpeername(addr, port);
665 if not ok and err ~= "timeout" then return ok, err; end 811 if not ok and err ~= "timeout" then return ok, err; end
666 local client = wrapsocket(conn, nil, read_size, listeners, tls_ctx, extra) 812 local client = wrapsocket(conn, nil, read_size, listeners, tls_ctx, extra)
667 local ok, err = client:init(); 813 local ok, err = client:init();
814 if not client.peername then
815 -- otherwise not set until connected
816 client.peername, client.peerport = addr, port;
817 end
668 if not ok then return ok, err; end 818 if not ok then return ok, err; end
819 client:debug("Client %s created", client);
669 if tls_ctx then 820 if tls_ctx then
670 client:starttls(tls_ctx); 821 client:starttls(tls_ctx);
671 end 822 end
672 return client, conn; 823 return client, conn;
673 end 824 end
685 conn.getfd = function () 836 conn.getfd = function ()
686 return fd; 837 return fd;
687 end; 838 end;
688 -- Otherwise it'll need to be something LuaSocket-compatible 839 -- Otherwise it'll need to be something LuaSocket-compatible
689 end 840 end
841 conn.id = new_id();
842 conn.log = logger.init(("fdwatch%s"):format(conn.id));
690 conn:add(onreadable, onwritable); 843 conn:add(onreadable, onwritable);
691 return conn; 844 return conn;
692 end; 845 end;
693 846
694 -- Dump all data from one connection into another 847 -- Dump all data from one connection into another
695 local function link(from, to) 848 local function link(from, to, read_size)
696 from.listeners = setmetatable({ 849 from:debug("Linking to %s", to.id);
697 onincoming = function (_, data) 850 function from:onincoming(data)
698 from:pause(); 851 self:pause();
699 to:write(data); 852 to:write(data);
700 end, 853 end
701 }, {__index=from.listeners}); 854 function to:ondrain() -- luacheck: ignore 212/self
702 to.listeners = setmetatable({ 855 from:resume();
703 ondrain = function () 856 end
704 from:resume(); 857 from:set_mode(read_size);
705 end,
706 }, {__index=to.listeners});
707 from:set(true, nil); 858 from:set(true, nil);
708 to:set(nil, true); 859 to:set(nil, true);
709 end 860 end
710 861
711 -- COMPAT 862 -- COMPAT
760 return { 911 return {
761 get_backend = function () return "epoll"; end; 912 get_backend = function () return "epoll"; end;
762 addserver = addserver; 913 addserver = addserver;
763 addclient = addclient; 914 addclient = addclient;
764 add_task = addtimer; 915 add_task = addtimer;
765 at = at; 916 timer = {
917 -- API-compatible with util.timer
918 add_task = addtimer;
919 stop = closetimer;
920 reschedule = reschedule;
921 to_absolute_time = function (t)
922 return t-monotonic()+realtime();
923 end;
924 };
925 listen = listen;
766 loop = loop; 926 loop = loop;
767 closeall = closeall; 927 closeall = closeall;
768 setquitting = setquitting; 928 setquitting = setquitting;
769 wrapclient = wrapclient; 929 wrapclient = wrapclient;
930 wrapserver = wrapserver;
770 watchfd = watchfd; 931 watchfd = watchfd;
771 link = link; 932 link = link;
772 set_config = function (newconfig) 933 set_config = function (newconfig)
773 cfg = setmetatable(newconfig, default_config); 934 cfg = setmetatable(newconfig, default_config);
774 end; 935 end;
775 936
776 -- libevent emulation 937 -- libevent emulation
777 event = { EV_READ = "r", EV_WRITE = "w", EV_READWRITE = "rw", EV_LEAVE = -1 }; 938 event = { EV_READ = "r", EV_WRITE = "w", EV_READWRITE = "rw", EV_LEAVE = -1 };
778 addevent = function (fd, mode, callback) 939 addevent = function (fd, mode, callback)
940 log("warn", "Using deprecated libevent emulation, please update code to use watchfd API instead");
779 local function onevent(self) 941 local function onevent(self)
780 local ret = self:callback(); 942 local ret = self:callback();
781 if ret == -1 then 943 if ret == -1 then
782 self:set(false, false); 944 self:set(false, false);
783 elseif ret then 945 elseif ret then
793 close = function (self) 955 close = function (self)
794 self:del(); 956 self:del();
795 fds[fd] = nil; 957 fds[fd] = nil;
796 end; 958 end;
797 }, interface_mt); 959 }, interface_mt);
960 conn.id = conn:getfd();
961 conn.log = logger.init(("fdwatch%d"):format(conn.id));
798 local ok, err = conn:add(mode == "r" or mode == "rw", mode == "w" or mode == "rw"); 962 local ok, err = conn:add(mode == "r" or mode == "rw", mode == "w" or mode == "rw");
799 if not ok then return ok, err; end 963 if not ok then return ok, err; end
800 return conn; 964 return conn;
801 end; 965 end;
802 }; 966 };