Coverage for /pythoncovmergedfiles/medio/medio/usr/lib/python3.9/asyncio/base_events.py: 13%

1075 statements  

« prev     ^ index     » next       coverage.py v7.3.1, created at 2023-09-25 06:05 +0000

1"""Base implementation of event loop. 

2 

3The event loop can be broken up into a multiplexer (the part 

4responsible for notifying us of I/O events) and the event loop proper, 

5which wraps a multiplexer with functionality for scheduling callbacks, 

6immediately or at a given time in the future. 

7 

8Whenever a public API takes a callback, subsequent positional 

9arguments will be passed to the callback if/when it is called. This 

10avoids the proliferation of trivial lambdas implementing closures. 

11Keyword arguments for the callback are not supported; this is a 

12conscious design decision, leaving the door open for keyword arguments 

13to modify the meaning of the API call itself. 

14""" 

15 

16import collections 

17import collections.abc 

18import concurrent.futures 

19import functools 

20import heapq 

21import itertools 

22import os 

23import socket 

24import stat 

25import subprocess 

26import threading 

27import time 

28import traceback 

29import sys 

30import warnings 

31import weakref 

32 

33try: 

34 import ssl 

35except ImportError: # pragma: no cover 

36 ssl = None 

37 

38from . import constants 

39from . import coroutines 

40from . import events 

41from . import exceptions 

42from . import futures 

43from . import protocols 

44from . import sslproto 

45from . import staggered 

46from . import tasks 

47from . import transports 

48from . import trsock 

49from .log import logger 

50 

51 

52__all__ = 'BaseEventLoop', 

53 

54 

55# Minimum number of _scheduled timer handles before cleanup of 

56# cancelled handles is performed. 

57_MIN_SCHEDULED_TIMER_HANDLES = 100 

58 

59# Minimum fraction of _scheduled timer handles that are cancelled 

60# before cleanup of cancelled handles is performed. 

61_MIN_CANCELLED_TIMER_HANDLES_FRACTION = 0.5 

62 

63 

64_HAS_IPv6 = hasattr(socket, 'AF_INET6') 

65 

66# Maximum timeout passed to select to avoid OS limitations 

67MAXIMUM_SELECT_TIMEOUT = 24 * 3600 

68 

69# Used for deprecation and removal of `loop.create_datagram_endpoint()`'s 

70# *reuse_address* parameter 

71_unset = object() 

72 

73 

74def _format_handle(handle): 

75 cb = handle._callback 

76 if isinstance(getattr(cb, '__self__', None), tasks.Task): 

77 # format the task 

78 return repr(cb.__self__) 

79 else: 

80 return str(handle) 

81 

82 

83def _format_pipe(fd): 

84 if fd == subprocess.PIPE: 

85 return '<pipe>' 

86 elif fd == subprocess.STDOUT: 

87 return '<stdout>' 

88 else: 

89 return repr(fd) 

90 

91 

92def _set_reuseport(sock): 

93 if not hasattr(socket, 'SO_REUSEPORT'): 

94 raise ValueError('reuse_port not supported by socket module') 

95 else: 

96 try: 

97 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) 

98 except OSError: 

99 raise ValueError('reuse_port not supported by socket module, ' 

100 'SO_REUSEPORT defined but not implemented.') 

101 

102 

103def _ipaddr_info(host, port, family, type, proto, flowinfo=0, scopeid=0): 

104 # Try to skip getaddrinfo if "host" is already an IP. Users might have 

105 # handled name resolution in their own code and pass in resolved IPs. 

106 if not hasattr(socket, 'inet_pton'): 

107 return 

108 

109 if proto not in {0, socket.IPPROTO_TCP, socket.IPPROTO_UDP} or \ 

110 host is None: 

111 return None 

112 

113 if type == socket.SOCK_STREAM: 

114 proto = socket.IPPROTO_TCP 

115 elif type == socket.SOCK_DGRAM: 

116 proto = socket.IPPROTO_UDP 

117 else: 

118 return None 

119 

120 if port is None: 

121 port = 0 

122 elif isinstance(port, bytes) and port == b'': 

123 port = 0 

124 elif isinstance(port, str) and port == '': 

125 port = 0 

126 else: 

127 # If port's a service name like "http", don't skip getaddrinfo. 

128 try: 

129 port = int(port) 

130 except (TypeError, ValueError): 

131 return None 

132 

133 if family == socket.AF_UNSPEC: 

134 afs = [socket.AF_INET] 

135 if _HAS_IPv6: 

136 afs.append(socket.AF_INET6) 

137 else: 

138 afs = [family] 

139 

140 if isinstance(host, bytes): 

141 host = host.decode('idna') 

142 if '%' in host: 

143 # Linux's inet_pton doesn't accept an IPv6 zone index after host, 

144 # like '::1%lo0'. 

145 return None 

146 

147 for af in afs: 

148 try: 

149 socket.inet_pton(af, host) 

150 # The host has already been resolved. 

151 if _HAS_IPv6 and af == socket.AF_INET6: 

152 return af, type, proto, '', (host, port, flowinfo, scopeid) 

153 else: 

154 return af, type, proto, '', (host, port) 

155 except OSError: 

156 pass 

157 

158 # "host" is not an IP address. 

159 return None 

160 

161 

162def _interleave_addrinfos(addrinfos, first_address_family_count=1): 

163 """Interleave list of addrinfo tuples by family.""" 

164 # Group addresses by family 

165 addrinfos_by_family = collections.OrderedDict() 

166 for addr in addrinfos: 

167 family = addr[0] 

168 if family not in addrinfos_by_family: 

169 addrinfos_by_family[family] = [] 

170 addrinfos_by_family[family].append(addr) 

171 addrinfos_lists = list(addrinfos_by_family.values()) 

172 

173 reordered = [] 

174 if first_address_family_count > 1: 

175 reordered.extend(addrinfos_lists[0][:first_address_family_count - 1]) 

176 del addrinfos_lists[0][:first_address_family_count - 1] 

177 reordered.extend( 

178 a for a in itertools.chain.from_iterable( 

179 itertools.zip_longest(*addrinfos_lists) 

180 ) if a is not None) 

181 return reordered 

182 

183 

184def _run_until_complete_cb(fut): 

185 if not fut.cancelled(): 

186 exc = fut.exception() 

187 if isinstance(exc, (SystemExit, KeyboardInterrupt)): 

188 # Issue #22429: run_forever() already finished, no need to 

189 # stop it. 

190 return 

191 futures._get_loop(fut).stop() 

192 

193 

194if hasattr(socket, 'TCP_NODELAY'): 

195 def _set_nodelay(sock): 

196 if (sock.family in {socket.AF_INET, socket.AF_INET6} and 

197 sock.type == socket.SOCK_STREAM and 

198 sock.proto == socket.IPPROTO_TCP): 

199 sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) 

200else: 

201 def _set_nodelay(sock): 

202 pass 

203 

204 

205class _SendfileFallbackProtocol(protocols.Protocol): 

206 def __init__(self, transp): 

207 if not isinstance(transp, transports._FlowControlMixin): 

208 raise TypeError("transport should be _FlowControlMixin instance") 

209 self._transport = transp 

210 self._proto = transp.get_protocol() 

211 self._should_resume_reading = transp.is_reading() 

212 self._should_resume_writing = transp._protocol_paused 

213 transp.pause_reading() 

214 transp.set_protocol(self) 

215 if self._should_resume_writing: 

216 self._write_ready_fut = self._transport._loop.create_future() 

217 else: 

218 self._write_ready_fut = None 

219 

220 async def drain(self): 

221 if self._transport.is_closing(): 

222 raise ConnectionError("Connection closed by peer") 

223 fut = self._write_ready_fut 

224 if fut is None: 

225 return 

226 await fut 

227 

228 def connection_made(self, transport): 

229 raise RuntimeError("Invalid state: " 

230 "connection should have been established already.") 

231 

232 def connection_lost(self, exc): 

233 if self._write_ready_fut is not None: 

234 # Never happens if peer disconnects after sending the whole content 

235 # Thus disconnection is always an exception from user perspective 

236 if exc is None: 

237 self._write_ready_fut.set_exception( 

238 ConnectionError("Connection is closed by peer")) 

239 else: 

240 self._write_ready_fut.set_exception(exc) 

241 self._proto.connection_lost(exc) 

242 

243 def pause_writing(self): 

244 if self._write_ready_fut is not None: 

245 return 

246 self._write_ready_fut = self._transport._loop.create_future() 

247 

248 def resume_writing(self): 

249 if self._write_ready_fut is None: 

250 return 

251 self._write_ready_fut.set_result(False) 

252 self._write_ready_fut = None 

253 

254 def data_received(self, data): 

255 raise RuntimeError("Invalid state: reading should be paused") 

256 

257 def eof_received(self): 

258 raise RuntimeError("Invalid state: reading should be paused") 

259 

260 async def restore(self): 

261 self._transport.set_protocol(self._proto) 

262 if self._should_resume_reading: 

263 self._transport.resume_reading() 

264 if self._write_ready_fut is not None: 

265 # Cancel the future. 

266 # Basically it has no effect because protocol is switched back, 

267 # no code should wait for it anymore. 

268 self._write_ready_fut.cancel() 

269 if self._should_resume_writing: 

270 self._proto.resume_writing() 

271 

272 

273class Server(events.AbstractServer): 

274 

275 def __init__(self, loop, sockets, protocol_factory, ssl_context, backlog, 

276 ssl_handshake_timeout): 

277 self._loop = loop 

278 self._sockets = sockets 

279 self._active_count = 0 

280 self._waiters = [] 

281 self._protocol_factory = protocol_factory 

282 self._backlog = backlog 

283 self._ssl_context = ssl_context 

284 self._ssl_handshake_timeout = ssl_handshake_timeout 

285 self._serving = False 

286 self._serving_forever_fut = None 

287 

288 def __repr__(self): 

289 return f'<{self.__class__.__name__} sockets={self.sockets!r}>' 

290 

291 def _attach(self): 

292 assert self._sockets is not None 

293 self._active_count += 1 

294 

295 def _detach(self): 

296 assert self._active_count > 0 

297 self._active_count -= 1 

298 if self._active_count == 0 and self._sockets is None: 

299 self._wakeup() 

300 

301 def _wakeup(self): 

302 waiters = self._waiters 

303 self._waiters = None 

304 for waiter in waiters: 

305 if not waiter.done(): 

306 waiter.set_result(waiter) 

307 

308 def _start_serving(self): 

309 if self._serving: 

310 return 

311 self._serving = True 

312 for sock in self._sockets: 

313 sock.listen(self._backlog) 

314 self._loop._start_serving( 

315 self._protocol_factory, sock, self._ssl_context, 

316 self, self._backlog, self._ssl_handshake_timeout) 

317 

318 def get_loop(self): 

319 return self._loop 

320 

321 def is_serving(self): 

322 return self._serving 

323 

324 @property 

325 def sockets(self): 

326 if self._sockets is None: 

327 return () 

328 return tuple(trsock.TransportSocket(s) for s in self._sockets) 

329 

330 def close(self): 

331 sockets = self._sockets 

332 if sockets is None: 

333 return 

334 self._sockets = None 

335 

336 for sock in sockets: 

337 self._loop._stop_serving(sock) 

338 

339 self._serving = False 

340 

341 if (self._serving_forever_fut is not None and 

342 not self._serving_forever_fut.done()): 

343 self._serving_forever_fut.cancel() 

344 self._serving_forever_fut = None 

345 

346 if self._active_count == 0: 

347 self._wakeup() 

348 

349 async def start_serving(self): 

350 self._start_serving() 

351 # Skip one loop iteration so that all 'loop.add_reader' 

352 # go through. 

353 await tasks.sleep(0, loop=self._loop) 

354 

355 async def serve_forever(self): 

356 if self._serving_forever_fut is not None: 

357 raise RuntimeError( 

358 f'server {self!r} is already being awaited on serve_forever()') 

359 if self._sockets is None: 

360 raise RuntimeError(f'server {self!r} is closed') 

361 

362 self._start_serving() 

363 self._serving_forever_fut = self._loop.create_future() 

364 

365 try: 

366 await self._serving_forever_fut 

367 except exceptions.CancelledError: 

368 try: 

369 self.close() 

370 await self.wait_closed() 

371 finally: 

372 raise 

373 finally: 

374 self._serving_forever_fut = None 

375 

376 async def wait_closed(self): 

377 if self._sockets is None or self._waiters is None: 

378 return 

379 waiter = self._loop.create_future() 

380 self._waiters.append(waiter) 

381 await waiter 

382 

383 

384class BaseEventLoop(events.AbstractEventLoop): 

385 

386 def __init__(self): 

387 self._timer_cancelled_count = 0 

388 self._closed = False 

389 self._stopping = False 

390 self._ready = collections.deque() 

391 self._scheduled = [] 

392 self._default_executor = None 

393 self._internal_fds = 0 

394 # Identifier of the thread running the event loop, or None if the 

395 # event loop is not running 

396 self._thread_id = None 

397 self._clock_resolution = time.get_clock_info('monotonic').resolution 

398 self._exception_handler = None 

399 self.set_debug(coroutines._is_debug_mode()) 

400 # In debug mode, if the execution of a callback or a step of a task 

401 # exceed this duration in seconds, the slow callback/task is logged. 

402 self.slow_callback_duration = 0.1 

403 self._current_handle = None 

404 self._task_factory = None 

405 self._coroutine_origin_tracking_enabled = False 

406 self._coroutine_origin_tracking_saved_depth = None 

407 

408 # A weak set of all asynchronous generators that are 

409 # being iterated by the loop. 

410 self._asyncgens = weakref.WeakSet() 

411 # Set to True when `loop.shutdown_asyncgens` is called. 

412 self._asyncgens_shutdown_called = False 

413 # Set to True when `loop.shutdown_default_executor` is called. 

414 self._executor_shutdown_called = False 

415 

416 def __repr__(self): 

417 return ( 

418 f'<{self.__class__.__name__} running={self.is_running()} ' 

419 f'closed={self.is_closed()} debug={self.get_debug()}>' 

420 ) 

421 

422 def create_future(self): 

423 """Create a Future object attached to the loop.""" 

424 return futures.Future(loop=self) 

425 

426 def create_task(self, coro, *, name=None): 

427 """Schedule a coroutine object. 

428 

429 Return a task object. 

430 """ 

431 self._check_closed() 

432 if self._task_factory is None: 

433 task = tasks.Task(coro, loop=self, name=name) 

434 if task._source_traceback: 

435 del task._source_traceback[-1] 

436 else: 

437 task = self._task_factory(self, coro) 

438 tasks._set_task_name(task, name) 

439 

440 return task 

441 

442 def set_task_factory(self, factory): 

443 """Set a task factory that will be used by loop.create_task(). 

444 

445 If factory is None the default task factory will be set. 

446 

447 If factory is a callable, it should have a signature matching 

448 '(loop, coro)', where 'loop' will be a reference to the active 

449 event loop, 'coro' will be a coroutine object. The callable 

450 must return a Future. 

451 """ 

452 if factory is not None and not callable(factory): 

453 raise TypeError('task factory must be a callable or None') 

454 self._task_factory = factory 

455 

456 def get_task_factory(self): 

457 """Return a task factory, or None if the default one is in use.""" 

458 return self._task_factory 

459 

460 def _make_socket_transport(self, sock, protocol, waiter=None, *, 

461 extra=None, server=None): 

462 """Create socket transport.""" 

463 raise NotImplementedError 

464 

465 def _make_ssl_transport( 

466 self, rawsock, protocol, sslcontext, waiter=None, 

467 *, server_side=False, server_hostname=None, 

468 extra=None, server=None, 

469 ssl_handshake_timeout=None, 

470 call_connection_made=True): 

471 """Create SSL transport.""" 

472 raise NotImplementedError 

473 

474 def _make_datagram_transport(self, sock, protocol, 

475 address=None, waiter=None, extra=None): 

476 """Create datagram transport.""" 

477 raise NotImplementedError 

478 

479 def _make_read_pipe_transport(self, pipe, protocol, waiter=None, 

480 extra=None): 

481 """Create read pipe transport.""" 

482 raise NotImplementedError 

483 

484 def _make_write_pipe_transport(self, pipe, protocol, waiter=None, 

485 extra=None): 

486 """Create write pipe transport.""" 

487 raise NotImplementedError 

488 

489 async def _make_subprocess_transport(self, protocol, args, shell, 

490 stdin, stdout, stderr, bufsize, 

491 extra=None, **kwargs): 

492 """Create subprocess transport.""" 

493 raise NotImplementedError 

494 

495 def _write_to_self(self): 

496 """Write a byte to self-pipe, to wake up the event loop. 

497 

498 This may be called from a different thread. 

499 

500 The subclass is responsible for implementing the self-pipe. 

501 """ 

502 raise NotImplementedError 

503 

504 def _process_events(self, event_list): 

505 """Process selector events.""" 

506 raise NotImplementedError 

507 

508 def _check_closed(self): 

509 if self._closed: 

510 raise RuntimeError('Event loop is closed') 

511 

512 def _check_default_executor(self): 

513 if self._executor_shutdown_called: 

514 raise RuntimeError('Executor shutdown has been called') 

515 

516 def _asyncgen_finalizer_hook(self, agen): 

517 self._asyncgens.discard(agen) 

518 if not self.is_closed(): 

519 self.call_soon_threadsafe(self.create_task, agen.aclose()) 

520 

521 def _asyncgen_firstiter_hook(self, agen): 

522 if self._asyncgens_shutdown_called: 

523 warnings.warn( 

524 f"asynchronous generator {agen!r} was scheduled after " 

525 f"loop.shutdown_asyncgens() call", 

526 ResourceWarning, source=self) 

527 

528 self._asyncgens.add(agen) 

529 

530 async def shutdown_asyncgens(self): 

531 """Shutdown all active asynchronous generators.""" 

532 self._asyncgens_shutdown_called = True 

533 

534 if not len(self._asyncgens): 

535 # If Python version is <3.6 or we don't have any asynchronous 

536 # generators alive. 

537 return 

538 

539 closing_agens = list(self._asyncgens) 

540 self._asyncgens.clear() 

541 

542 results = await tasks.gather( 

543 *[ag.aclose() for ag in closing_agens], 

544 return_exceptions=True, 

545 loop=self) 

546 

547 for result, agen in zip(results, closing_agens): 

548 if isinstance(result, Exception): 

549 self.call_exception_handler({ 

550 'message': f'an error occurred during closing of ' 

551 f'asynchronous generator {agen!r}', 

552 'exception': result, 

553 'asyncgen': agen 

554 }) 

555 

556 async def shutdown_default_executor(self): 

557 """Schedule the shutdown of the default executor.""" 

558 self._executor_shutdown_called = True 

559 if self._default_executor is None: 

560 return 

561 future = self.create_future() 

562 thread = threading.Thread(target=self._do_shutdown, args=(future,)) 

563 thread.start() 

564 try: 

565 await future 

566 finally: 

567 thread.join() 

568 

569 def _do_shutdown(self, future): 

570 try: 

571 self._default_executor.shutdown(wait=True) 

572 self.call_soon_threadsafe(future.set_result, None) 

573 except Exception as ex: 

574 self.call_soon_threadsafe(future.set_exception, ex) 

575 

576 def _check_running(self): 

577 if self.is_running(): 

578 raise RuntimeError('This event loop is already running') 

579 if events._get_running_loop() is not None: 

580 raise RuntimeError( 

581 'Cannot run the event loop while another loop is running') 

582 

583 def run_forever(self): 

584 """Run until stop() is called.""" 

585 self._check_closed() 

586 self._check_running() 

587 self._set_coroutine_origin_tracking(self._debug) 

588 self._thread_id = threading.get_ident() 

589 

590 old_agen_hooks = sys.get_asyncgen_hooks() 

591 sys.set_asyncgen_hooks(firstiter=self._asyncgen_firstiter_hook, 

592 finalizer=self._asyncgen_finalizer_hook) 

593 try: 

594 events._set_running_loop(self) 

595 while True: 

596 self._run_once() 

597 if self._stopping: 

598 break 

599 finally: 

600 self._stopping = False 

601 self._thread_id = None 

602 events._set_running_loop(None) 

603 self._set_coroutine_origin_tracking(False) 

604 sys.set_asyncgen_hooks(*old_agen_hooks) 

605 

606 def run_until_complete(self, future): 

607 """Run until the Future is done. 

608 

609 If the argument is a coroutine, it is wrapped in a Task. 

610 

611 WARNING: It would be disastrous to call run_until_complete() 

612 with the same coroutine twice -- it would wrap it in two 

613 different Tasks and that can't be good. 

614 

615 Return the Future's result, or raise its exception. 

616 """ 

617 self._check_closed() 

618 self._check_running() 

619 

620 new_task = not futures.isfuture(future) 

621 future = tasks.ensure_future(future, loop=self) 

622 if new_task: 

623 # An exception is raised if the future didn't complete, so there 

624 # is no need to log the "destroy pending task" message 

625 future._log_destroy_pending = False 

626 

627 future.add_done_callback(_run_until_complete_cb) 

628 try: 

629 self.run_forever() 

630 except: 

631 if new_task and future.done() and not future.cancelled(): 

632 # The coroutine raised a BaseException. Consume the exception 

633 # to not log a warning, the caller doesn't have access to the 

634 # local task. 

635 future.exception() 

636 raise 

637 finally: 

638 future.remove_done_callback(_run_until_complete_cb) 

639 if not future.done(): 

640 raise RuntimeError('Event loop stopped before Future completed.') 

641 

642 return future.result() 

643 

644 def stop(self): 

645 """Stop running the event loop. 

646 

647 Every callback already scheduled will still run. This simply informs 

648 run_forever to stop looping after a complete iteration. 

649 """ 

650 self._stopping = True 

651 

652 def close(self): 

653 """Close the event loop. 

654 

655 This clears the queues and shuts down the executor, 

656 but does not wait for the executor to finish. 

657 

658 The event loop must not be running. 

659 """ 

660 if self.is_running(): 

661 raise RuntimeError("Cannot close a running event loop") 

662 if self._closed: 

663 return 

664 if self._debug: 

665 logger.debug("Close %r", self) 

666 self._closed = True 

667 self._ready.clear() 

668 self._scheduled.clear() 

669 self._executor_shutdown_called = True 

670 executor = self._default_executor 

671 if executor is not None: 

672 self._default_executor = None 

673 executor.shutdown(wait=False) 

674 

675 def is_closed(self): 

676 """Returns True if the event loop was closed.""" 

677 return self._closed 

678 

679 def __del__(self, _warn=warnings.warn): 

680 if not self.is_closed(): 

681 _warn(f"unclosed event loop {self!r}", ResourceWarning, source=self) 

682 if not self.is_running(): 

683 self.close() 

684 

685 def is_running(self): 

686 """Returns True if the event loop is running.""" 

687 return (self._thread_id is not None) 

688 

689 def time(self): 

690 """Return the time according to the event loop's clock. 

691 

692 This is a float expressed in seconds since an epoch, but the 

693 epoch, precision, accuracy and drift are unspecified and may 

694 differ per event loop. 

695 """ 

696 return time.monotonic() 

697 

698 def call_later(self, delay, callback, *args, context=None): 

699 """Arrange for a callback to be called at a given time. 

700 

701 Return a Handle: an opaque object with a cancel() method that 

702 can be used to cancel the call. 

703 

704 The delay can be an int or float, expressed in seconds. It is 

705 always relative to the current time. 

706 

707 Each callback will be called exactly once. If two callbacks 

708 are scheduled for exactly the same time, it undefined which 

709 will be called first. 

710 

711 Any positional arguments after the callback will be passed to 

712 the callback when it is called. 

713 """ 

714 timer = self.call_at(self.time() + delay, callback, *args, 

715 context=context) 

716 if timer._source_traceback: 

717 del timer._source_traceback[-1] 

718 return timer 

719 

720 def call_at(self, when, callback, *args, context=None): 

721 """Like call_later(), but uses an absolute time. 

722 

723 Absolute time corresponds to the event loop's time() method. 

724 """ 

725 self._check_closed() 

726 if self._debug: 

727 self._check_thread() 

728 self._check_callback(callback, 'call_at') 

729 timer = events.TimerHandle(when, callback, args, self, context) 

730 if timer._source_traceback: 

731 del timer._source_traceback[-1] 

732 heapq.heappush(self._scheduled, timer) 

733 timer._scheduled = True 

734 return timer 

735 

736 def call_soon(self, callback, *args, context=None): 

737 """Arrange for a callback to be called as soon as possible. 

738 

739 This operates as a FIFO queue: callbacks are called in the 

740 order in which they are registered. Each callback will be 

741 called exactly once. 

742 

743 Any positional arguments after the callback will be passed to 

744 the callback when it is called. 

745 """ 

746 self._check_closed() 

747 if self._debug: 

748 self._check_thread() 

749 self._check_callback(callback, 'call_soon') 

750 handle = self._call_soon(callback, args, context) 

751 if handle._source_traceback: 

752 del handle._source_traceback[-1] 

753 return handle 

754 

755 def _check_callback(self, callback, method): 

756 if (coroutines.iscoroutine(callback) or 

757 coroutines.iscoroutinefunction(callback)): 

758 raise TypeError( 

759 f"coroutines cannot be used with {method}()") 

760 if not callable(callback): 

761 raise TypeError( 

762 f'a callable object was expected by {method}(), ' 

763 f'got {callback!r}') 

764 

765 def _call_soon(self, callback, args, context): 

766 handle = events.Handle(callback, args, self, context) 

767 if handle._source_traceback: 

768 del handle._source_traceback[-1] 

769 self._ready.append(handle) 

770 return handle 

771 

772 def _check_thread(self): 

773 """Check that the current thread is the thread running the event loop. 

774 

775 Non-thread-safe methods of this class make this assumption and will 

776 likely behave incorrectly when the assumption is violated. 

777 

778 Should only be called when (self._debug == True). The caller is 

779 responsible for checking this condition for performance reasons. 

780 """ 

781 if self._thread_id is None: 

782 return 

783 thread_id = threading.get_ident() 

784 if thread_id != self._thread_id: 

785 raise RuntimeError( 

786 "Non-thread-safe operation invoked on an event loop other " 

787 "than the current one") 

788 

789 def call_soon_threadsafe(self, callback, *args, context=None): 

790 """Like call_soon(), but thread-safe.""" 

791 self._check_closed() 

792 if self._debug: 

793 self._check_callback(callback, 'call_soon_threadsafe') 

794 handle = self._call_soon(callback, args, context) 

795 if handle._source_traceback: 

796 del handle._source_traceback[-1] 

797 self._write_to_self() 

798 return handle 

799 

800 def run_in_executor(self, executor, func, *args): 

801 self._check_closed() 

802 if self._debug: 

803 self._check_callback(func, 'run_in_executor') 

804 if executor is None: 

805 executor = self._default_executor 

806 # Only check when the default executor is being used 

807 self._check_default_executor() 

808 if executor is None: 

809 executor = concurrent.futures.ThreadPoolExecutor( 

810 thread_name_prefix='asyncio' 

811 ) 

812 self._default_executor = executor 

813 return futures.wrap_future( 

814 executor.submit(func, *args), loop=self) 

815 

816 def set_default_executor(self, executor): 

817 if not isinstance(executor, concurrent.futures.ThreadPoolExecutor): 

818 warnings.warn( 

819 'Using the default executor that is not an instance of ' 

820 'ThreadPoolExecutor is deprecated and will be prohibited ' 

821 'in Python 3.9', 

822 DeprecationWarning, 2) 

823 self._default_executor = executor 

824 

825 def _getaddrinfo_debug(self, host, port, family, type, proto, flags): 

826 msg = [f"{host}:{port!r}"] 

827 if family: 

828 msg.append(f'family={family!r}') 

829 if type: 

830 msg.append(f'type={type!r}') 

831 if proto: 

832 msg.append(f'proto={proto!r}') 

833 if flags: 

834 msg.append(f'flags={flags!r}') 

835 msg = ', '.join(msg) 

836 logger.debug('Get address info %s', msg) 

837 

838 t0 = self.time() 

839 addrinfo = socket.getaddrinfo(host, port, family, type, proto, flags) 

840 dt = self.time() - t0 

841 

842 msg = f'Getting address info {msg} took {dt * 1e3:.3f}ms: {addrinfo!r}' 

843 if dt >= self.slow_callback_duration: 

844 logger.info(msg) 

845 else: 

846 logger.debug(msg) 

847 return addrinfo 

848 

849 async def getaddrinfo(self, host, port, *, 

850 family=0, type=0, proto=0, flags=0): 

851 if self._debug: 

852 getaddr_func = self._getaddrinfo_debug 

853 else: 

854 getaddr_func = socket.getaddrinfo 

855 

856 return await self.run_in_executor( 

857 None, getaddr_func, host, port, family, type, proto, flags) 

858 

859 async def getnameinfo(self, sockaddr, flags=0): 

860 return await self.run_in_executor( 

861 None, socket.getnameinfo, sockaddr, flags) 

862 

863 async def sock_sendfile(self, sock, file, offset=0, count=None, 

864 *, fallback=True): 

865 if self._debug and sock.gettimeout() != 0: 

866 raise ValueError("the socket must be non-blocking") 

867 self._check_sendfile_params(sock, file, offset, count) 

868 try: 

869 return await self._sock_sendfile_native(sock, file, 

870 offset, count) 

871 except exceptions.SendfileNotAvailableError as exc: 

872 if not fallback: 

873 raise 

874 return await self._sock_sendfile_fallback(sock, file, 

875 offset, count) 

876 

877 async def _sock_sendfile_native(self, sock, file, offset, count): 

878 # NB: sendfile syscall is not supported for SSL sockets and 

879 # non-mmap files even if sendfile is supported by OS 

880 raise exceptions.SendfileNotAvailableError( 

881 f"syscall sendfile is not available for socket {sock!r} " 

882 "and file {file!r} combination") 

883 

884 async def _sock_sendfile_fallback(self, sock, file, offset, count): 

885 if offset: 

886 file.seek(offset) 

887 blocksize = ( 

888 min(count, constants.SENDFILE_FALLBACK_READBUFFER_SIZE) 

889 if count else constants.SENDFILE_FALLBACK_READBUFFER_SIZE 

890 ) 

891 buf = bytearray(blocksize) 

892 total_sent = 0 

893 try: 

894 while True: 

895 if count: 

896 blocksize = min(count - total_sent, blocksize) 

897 if blocksize <= 0: 

898 break 

899 view = memoryview(buf)[:blocksize] 

900 read = await self.run_in_executor(None, file.readinto, view) 

901 if not read: 

902 break # EOF 

903 await self.sock_sendall(sock, view[:read]) 

904 total_sent += read 

905 return total_sent 

906 finally: 

907 if total_sent > 0 and hasattr(file, 'seek'): 

908 file.seek(offset + total_sent) 

909 

910 def _check_sendfile_params(self, sock, file, offset, count): 

911 if 'b' not in getattr(file, 'mode', 'b'): 

912 raise ValueError("file should be opened in binary mode") 

913 if not sock.type == socket.SOCK_STREAM: 

914 raise ValueError("only SOCK_STREAM type sockets are supported") 

915 if count is not None: 

916 if not isinstance(count, int): 

917 raise TypeError( 

918 "count must be a positive integer (got {!r})".format(count)) 

919 if count <= 0: 

920 raise ValueError( 

921 "count must be a positive integer (got {!r})".format(count)) 

922 if not isinstance(offset, int): 

923 raise TypeError( 

924 "offset must be a non-negative integer (got {!r})".format( 

925 offset)) 

926 if offset < 0: 

927 raise ValueError( 

928 "offset must be a non-negative integer (got {!r})".format( 

929 offset)) 

930 

931 async def _connect_sock(self, exceptions, addr_info, local_addr_infos=None): 

932 """Create, bind and connect one socket.""" 

933 my_exceptions = [] 

934 exceptions.append(my_exceptions) 

935 family, type_, proto, _, address = addr_info 

936 sock = None 

937 try: 

938 sock = socket.socket(family=family, type=type_, proto=proto) 

939 sock.setblocking(False) 

940 if local_addr_infos is not None: 

941 for _, _, _, _, laddr in local_addr_infos: 

942 try: 

943 sock.bind(laddr) 

944 break 

945 except OSError as exc: 

946 msg = ( 

947 f'error while attempting to bind on ' 

948 f'address {laddr!r}: ' 

949 f'{exc.strerror.lower()}' 

950 ) 

951 exc = OSError(exc.errno, msg) 

952 my_exceptions.append(exc) 

953 else: # all bind attempts failed 

954 raise my_exceptions.pop() 

955 await self.sock_connect(sock, address) 

956 return sock 

957 except OSError as exc: 

958 my_exceptions.append(exc) 

959 if sock is not None: 

960 sock.close() 

961 raise 

962 except: 

963 if sock is not None: 

964 sock.close() 

965 raise 

966 

967 async def create_connection( 

968 self, protocol_factory, host=None, port=None, 

969 *, ssl=None, family=0, 

970 proto=0, flags=0, sock=None, 

971 local_addr=None, server_hostname=None, 

972 ssl_handshake_timeout=None, 

973 happy_eyeballs_delay=None, interleave=None): 

974 """Connect to a TCP server. 

975 

976 Create a streaming transport connection to a given Internet host and 

977 port: socket family AF_INET or socket.AF_INET6 depending on host (or 

978 family if specified), socket type SOCK_STREAM. protocol_factory must be 

979 a callable returning a protocol instance. 

980 

981 This method is a coroutine which will try to establish the connection 

982 in the background. When successful, the coroutine returns a 

983 (transport, protocol) pair. 

984 """ 

985 if server_hostname is not None and not ssl: 

986 raise ValueError('server_hostname is only meaningful with ssl') 

987 

988 if server_hostname is None and ssl: 

989 # Use host as default for server_hostname. It is an error 

990 # if host is empty or not set, e.g. when an 

991 # already-connected socket was passed or when only a port 

992 # is given. To avoid this error, you can pass 

993 # server_hostname='' -- this will bypass the hostname 

994 # check. (This also means that if host is a numeric 

995 # IP/IPv6 address, we will attempt to verify that exact 

996 # address; this will probably fail, but it is possible to 

997 # create a certificate for a specific IP address, so we 

998 # don't judge it here.) 

999 if not host: 

1000 raise ValueError('You must set server_hostname ' 

1001 'when using ssl without a host') 

1002 server_hostname = host 

1003 

1004 if ssl_handshake_timeout is not None and not ssl: 

1005 raise ValueError( 

1006 'ssl_handshake_timeout is only meaningful with ssl') 

1007 

1008 if happy_eyeballs_delay is not None and interleave is None: 

1009 # If using happy eyeballs, default to interleave addresses by family 

1010 interleave = 1 

1011 

1012 if host is not None or port is not None: 

1013 if sock is not None: 

1014 raise ValueError( 

1015 'host/port and sock can not be specified at the same time') 

1016 

1017 infos = await self._ensure_resolved( 

1018 (host, port), family=family, 

1019 type=socket.SOCK_STREAM, proto=proto, flags=flags, loop=self) 

1020 if not infos: 

1021 raise OSError('getaddrinfo() returned empty list') 

1022 

1023 if local_addr is not None: 

1024 laddr_infos = await self._ensure_resolved( 

1025 local_addr, family=family, 

1026 type=socket.SOCK_STREAM, proto=proto, 

1027 flags=flags, loop=self) 

1028 if not laddr_infos: 

1029 raise OSError('getaddrinfo() returned empty list') 

1030 else: 

1031 laddr_infos = None 

1032 

1033 if interleave: 

1034 infos = _interleave_addrinfos(infos, interleave) 

1035 

1036 exceptions = [] 

1037 if happy_eyeballs_delay is None: 

1038 # not using happy eyeballs 

1039 for addrinfo in infos: 

1040 try: 

1041 sock = await self._connect_sock( 

1042 exceptions, addrinfo, laddr_infos) 

1043 break 

1044 except OSError: 

1045 continue 

1046 else: # using happy eyeballs 

1047 sock, _, _ = await staggered.staggered_race( 

1048 (functools.partial(self._connect_sock, 

1049 exceptions, addrinfo, laddr_infos) 

1050 for addrinfo in infos), 

1051 happy_eyeballs_delay, loop=self) 

1052 

1053 if sock is None: 

1054 exceptions = [exc for sub in exceptions for exc in sub] 

1055 if len(exceptions) == 1: 

1056 raise exceptions[0] 

1057 else: 

1058 # If they all have the same str(), raise one. 

1059 model = str(exceptions[0]) 

1060 if all(str(exc) == model for exc in exceptions): 

1061 raise exceptions[0] 

1062 # Raise a combined exception so the user can see all 

1063 # the various error messages. 

1064 raise OSError('Multiple exceptions: {}'.format( 

1065 ', '.join(str(exc) for exc in exceptions))) 

1066 

1067 else: 

1068 if sock is None: 

1069 raise ValueError( 

1070 'host and port was not specified and no sock specified') 

1071 if sock.type != socket.SOCK_STREAM: 

1072 # We allow AF_INET, AF_INET6, AF_UNIX as long as they 

1073 # are SOCK_STREAM. 

1074 # We support passing AF_UNIX sockets even though we have 

1075 # a dedicated API for that: create_unix_connection. 

1076 # Disallowing AF_UNIX in this method, breaks backwards 

1077 # compatibility. 

1078 raise ValueError( 

1079 f'A Stream Socket was expected, got {sock!r}') 

1080 

1081 transport, protocol = await self._create_connection_transport( 

1082 sock, protocol_factory, ssl, server_hostname, 

1083 ssl_handshake_timeout=ssl_handshake_timeout) 

1084 if self._debug: 

1085 # Get the socket from the transport because SSL transport closes 

1086 # the old socket and creates a new SSL socket 

1087 sock = transport.get_extra_info('socket') 

1088 logger.debug("%r connected to %s:%r: (%r, %r)", 

1089 sock, host, port, transport, protocol) 

1090 return transport, protocol 

1091 

1092 async def _create_connection_transport( 

1093 self, sock, protocol_factory, ssl, 

1094 server_hostname, server_side=False, 

1095 ssl_handshake_timeout=None): 

1096 

1097 sock.setblocking(False) 

1098 

1099 protocol = protocol_factory() 

1100 waiter = self.create_future() 

1101 if ssl: 

1102 sslcontext = None if isinstance(ssl, bool) else ssl 

1103 transport = self._make_ssl_transport( 

1104 sock, protocol, sslcontext, waiter, 

1105 server_side=server_side, server_hostname=server_hostname, 

1106 ssl_handshake_timeout=ssl_handshake_timeout) 

1107 else: 

1108 transport = self._make_socket_transport(sock, protocol, waiter) 

1109 

1110 try: 

1111 await waiter 

1112 except: 

1113 transport.close() 

1114 raise 

1115 

1116 return transport, protocol 

1117 

1118 async def sendfile(self, transport, file, offset=0, count=None, 

1119 *, fallback=True): 

1120 """Send a file to transport. 

1121 

1122 Return the total number of bytes which were sent. 

1123 

1124 The method uses high-performance os.sendfile if available. 

1125 

1126 file must be a regular file object opened in binary mode. 

1127 

1128 offset tells from where to start reading the file. If specified, 

1129 count is the total number of bytes to transmit as opposed to 

1130 sending the file until EOF is reached. File position is updated on 

1131 return or also in case of error in which case file.tell() 

1132 can be used to figure out the number of bytes 

1133 which were sent. 

1134 

1135 fallback set to True makes asyncio to manually read and send 

1136 the file when the platform does not support the sendfile syscall 

1137 (e.g. Windows or SSL socket on Unix). 

1138 

1139 Raise SendfileNotAvailableError if the system does not support 

1140 sendfile syscall and fallback is False. 

1141 """ 

1142 if transport.is_closing(): 

1143 raise RuntimeError("Transport is closing") 

1144 mode = getattr(transport, '_sendfile_compatible', 

1145 constants._SendfileMode.UNSUPPORTED) 

1146 if mode is constants._SendfileMode.UNSUPPORTED: 

1147 raise RuntimeError( 

1148 f"sendfile is not supported for transport {transport!r}") 

1149 if mode is constants._SendfileMode.TRY_NATIVE: 

1150 try: 

1151 return await self._sendfile_native(transport, file, 

1152 offset, count) 

1153 except exceptions.SendfileNotAvailableError as exc: 

1154 if not fallback: 

1155 raise 

1156 

1157 if not fallback: 

1158 raise RuntimeError( 

1159 f"fallback is disabled and native sendfile is not " 

1160 f"supported for transport {transport!r}") 

1161 

1162 return await self._sendfile_fallback(transport, file, 

1163 offset, count) 

1164 

1165 async def _sendfile_native(self, transp, file, offset, count): 

1166 raise exceptions.SendfileNotAvailableError( 

1167 "sendfile syscall is not supported") 

1168 

1169 async def _sendfile_fallback(self, transp, file, offset, count): 

1170 if offset: 

1171 file.seek(offset) 

1172 blocksize = min(count, 16384) if count else 16384 

1173 buf = bytearray(blocksize) 

1174 total_sent = 0 

1175 proto = _SendfileFallbackProtocol(transp) 

1176 try: 

1177 while True: 

1178 if count: 

1179 blocksize = min(count - total_sent, blocksize) 

1180 if blocksize <= 0: 

1181 return total_sent 

1182 view = memoryview(buf)[:blocksize] 

1183 read = await self.run_in_executor(None, file.readinto, view) 

1184 if not read: 

1185 return total_sent # EOF 

1186 await proto.drain() 

1187 transp.write(view[:read]) 

1188 total_sent += read 

1189 finally: 

1190 if total_sent > 0 and hasattr(file, 'seek'): 

1191 file.seek(offset + total_sent) 

1192 await proto.restore() 

1193 

1194 async def start_tls(self, transport, protocol, sslcontext, *, 

1195 server_side=False, 

1196 server_hostname=None, 

1197 ssl_handshake_timeout=None): 

1198 """Upgrade transport to TLS. 

1199 

1200 Return a new transport that *protocol* should start using 

1201 immediately. 

1202 """ 

1203 if ssl is None: 

1204 raise RuntimeError('Python ssl module is not available') 

1205 

1206 if not isinstance(sslcontext, ssl.SSLContext): 

1207 raise TypeError( 

1208 f'sslcontext is expected to be an instance of ssl.SSLContext, ' 

1209 f'got {sslcontext!r}') 

1210 

1211 if not getattr(transport, '_start_tls_compatible', False): 

1212 raise TypeError( 

1213 f'transport {transport!r} is not supported by start_tls()') 

1214 

1215 waiter = self.create_future() 

1216 ssl_protocol = sslproto.SSLProtocol( 

1217 self, protocol, sslcontext, waiter, 

1218 server_side, server_hostname, 

1219 ssl_handshake_timeout=ssl_handshake_timeout, 

1220 call_connection_made=False) 

1221 

1222 # Pause early so that "ssl_protocol.data_received()" doesn't 

1223 # have a chance to get called before "ssl_protocol.connection_made()". 

1224 transport.pause_reading() 

1225 

1226 transport.set_protocol(ssl_protocol) 

1227 conmade_cb = self.call_soon(ssl_protocol.connection_made, transport) 

1228 resume_cb = self.call_soon(transport.resume_reading) 

1229 

1230 try: 

1231 await waiter 

1232 except BaseException: 

1233 transport.close() 

1234 conmade_cb.cancel() 

1235 resume_cb.cancel() 

1236 raise 

1237 

1238 return ssl_protocol._app_transport 

1239 

1240 async def create_datagram_endpoint(self, protocol_factory, 

1241 local_addr=None, remote_addr=None, *, 

1242 family=0, proto=0, flags=0, 

1243 reuse_address=_unset, reuse_port=None, 

1244 allow_broadcast=None, sock=None): 

1245 """Create datagram connection.""" 

1246 if sock is not None: 

1247 if sock.type != socket.SOCK_DGRAM: 

1248 raise ValueError( 

1249 f'A UDP Socket was expected, got {sock!r}') 

1250 if (local_addr or remote_addr or 

1251 family or proto or flags or 

1252 reuse_port or allow_broadcast): 

1253 # show the problematic kwargs in exception msg 

1254 opts = dict(local_addr=local_addr, remote_addr=remote_addr, 

1255 family=family, proto=proto, flags=flags, 

1256 reuse_address=reuse_address, reuse_port=reuse_port, 

1257 allow_broadcast=allow_broadcast) 

1258 problems = ', '.join(f'{k}={v}' for k, v in opts.items() if v) 

1259 raise ValueError( 

1260 f'socket modifier keyword arguments can not be used ' 

1261 f'when sock is specified. ({problems})') 

1262 sock.setblocking(False) 

1263 r_addr = None 

1264 else: 

1265 if not (local_addr or remote_addr): 

1266 if family == 0: 

1267 raise ValueError('unexpected address family') 

1268 addr_pairs_info = (((family, proto), (None, None)),) 

1269 elif hasattr(socket, 'AF_UNIX') and family == socket.AF_UNIX: 

1270 for addr in (local_addr, remote_addr): 

1271 if addr is not None and not isinstance(addr, str): 

1272 raise TypeError('string is expected') 

1273 

1274 if local_addr and local_addr[0] not in (0, '\x00'): 

1275 try: 

1276 if stat.S_ISSOCK(os.stat(local_addr).st_mode): 

1277 os.remove(local_addr) 

1278 except FileNotFoundError: 

1279 pass 

1280 except OSError as err: 

1281 # Directory may have permissions only to create socket. 

1282 logger.error('Unable to check or remove stale UNIX ' 

1283 'socket %r: %r', 

1284 local_addr, err) 

1285 

1286 addr_pairs_info = (((family, proto), 

1287 (local_addr, remote_addr)), ) 

1288 else: 

1289 # join address by (family, protocol) 

1290 addr_infos = {} # Using order preserving dict 

1291 for idx, addr in ((0, local_addr), (1, remote_addr)): 

1292 if addr is not None: 

1293 assert isinstance(addr, tuple) and len(addr) == 2, ( 

1294 '2-tuple is expected') 

1295 

1296 infos = await self._ensure_resolved( 

1297 addr, family=family, type=socket.SOCK_DGRAM, 

1298 proto=proto, flags=flags, loop=self) 

1299 if not infos: 

1300 raise OSError('getaddrinfo() returned empty list') 

1301 

1302 for fam, _, pro, _, address in infos: 

1303 key = (fam, pro) 

1304 if key not in addr_infos: 

1305 addr_infos[key] = [None, None] 

1306 addr_infos[key][idx] = address 

1307 

1308 # each addr has to have info for each (family, proto) pair 

1309 addr_pairs_info = [ 

1310 (key, addr_pair) for key, addr_pair in addr_infos.items() 

1311 if not ((local_addr and addr_pair[0] is None) or 

1312 (remote_addr and addr_pair[1] is None))] 

1313 

1314 if not addr_pairs_info: 

1315 raise ValueError('can not get address information') 

1316 

1317 exceptions = [] 

1318 

1319 # bpo-37228 

1320 if reuse_address is not _unset: 

1321 if reuse_address: 

1322 raise ValueError("Passing `reuse_address=True` is no " 

1323 "longer supported, as the usage of " 

1324 "SO_REUSEPORT in UDP poses a significant " 

1325 "security concern.") 

1326 else: 

1327 warnings.warn("The *reuse_address* parameter has been " 

1328 "deprecated as of 3.5.10 and is scheduled " 

1329 "for removal in 3.11.", DeprecationWarning, 

1330 stacklevel=2) 

1331 

1332 for ((family, proto), 

1333 (local_address, remote_address)) in addr_pairs_info: 

1334 sock = None 

1335 r_addr = None 

1336 try: 

1337 sock = socket.socket( 

1338 family=family, type=socket.SOCK_DGRAM, proto=proto) 

1339 if reuse_port: 

1340 _set_reuseport(sock) 

1341 if allow_broadcast: 

1342 sock.setsockopt( 

1343 socket.SOL_SOCKET, socket.SO_BROADCAST, 1) 

1344 sock.setblocking(False) 

1345 

1346 if local_addr: 

1347 sock.bind(local_address) 

1348 if remote_addr: 

1349 if not allow_broadcast: 

1350 await self.sock_connect(sock, remote_address) 

1351 r_addr = remote_address 

1352 except OSError as exc: 

1353 if sock is not None: 

1354 sock.close() 

1355 exceptions.append(exc) 

1356 except: 

1357 if sock is not None: 

1358 sock.close() 

1359 raise 

1360 else: 

1361 break 

1362 else: 

1363 raise exceptions[0] 

1364 

1365 protocol = protocol_factory() 

1366 waiter = self.create_future() 

1367 transport = self._make_datagram_transport( 

1368 sock, protocol, r_addr, waiter) 

1369 if self._debug: 

1370 if local_addr: 

1371 logger.info("Datagram endpoint local_addr=%r remote_addr=%r " 

1372 "created: (%r, %r)", 

1373 local_addr, remote_addr, transport, protocol) 

1374 else: 

1375 logger.debug("Datagram endpoint remote_addr=%r created: " 

1376 "(%r, %r)", 

1377 remote_addr, transport, protocol) 

1378 

1379 try: 

1380 await waiter 

1381 except: 

1382 transport.close() 

1383 raise 

1384 

1385 return transport, protocol 

1386 

1387 async def _ensure_resolved(self, address, *, 

1388 family=0, type=socket.SOCK_STREAM, 

1389 proto=0, flags=0, loop): 

1390 host, port = address[:2] 

1391 info = _ipaddr_info(host, port, family, type, proto, *address[2:]) 

1392 if info is not None: 

1393 # "host" is already a resolved IP. 

1394 return [info] 

1395 else: 

1396 return await loop.getaddrinfo(host, port, family=family, type=type, 

1397 proto=proto, flags=flags) 

1398 

1399 async def _create_server_getaddrinfo(self, host, port, family, flags): 

1400 infos = await self._ensure_resolved((host, port), family=family, 

1401 type=socket.SOCK_STREAM, 

1402 flags=flags, loop=self) 

1403 if not infos: 

1404 raise OSError(f'getaddrinfo({host!r}) returned empty list') 

1405 return infos 

1406 

1407 async def create_server( 

1408 self, protocol_factory, host=None, port=None, 

1409 *, 

1410 family=socket.AF_UNSPEC, 

1411 flags=socket.AI_PASSIVE, 

1412 sock=None, 

1413 backlog=100, 

1414 ssl=None, 

1415 reuse_address=None, 

1416 reuse_port=None, 

1417 ssl_handshake_timeout=None, 

1418 start_serving=True): 

1419 """Create a TCP server. 

1420 

1421 The host parameter can be a string, in that case the TCP server is 

1422 bound to host and port. 

1423 

1424 The host parameter can also be a sequence of strings and in that case 

1425 the TCP server is bound to all hosts of the sequence. If a host 

1426 appears multiple times (possibly indirectly e.g. when hostnames 

1427 resolve to the same IP address), the server is only bound once to that 

1428 host. 

1429 

1430 Return a Server object which can be used to stop the service. 

1431 

1432 This method is a coroutine. 

1433 """ 

1434 if isinstance(ssl, bool): 

1435 raise TypeError('ssl argument must be an SSLContext or None') 

1436 

1437 if ssl_handshake_timeout is not None and ssl is None: 

1438 raise ValueError( 

1439 'ssl_handshake_timeout is only meaningful with ssl') 

1440 

1441 if host is not None or port is not None: 

1442 if sock is not None: 

1443 raise ValueError( 

1444 'host/port and sock can not be specified at the same time') 

1445 

1446 if reuse_address is None: 

1447 reuse_address = os.name == 'posix' and sys.platform != 'cygwin' 

1448 sockets = [] 

1449 if host == '': 

1450 hosts = [None] 

1451 elif (isinstance(host, str) or 

1452 not isinstance(host, collections.abc.Iterable)): 

1453 hosts = [host] 

1454 else: 

1455 hosts = host 

1456 

1457 fs = [self._create_server_getaddrinfo(host, port, family=family, 

1458 flags=flags) 

1459 for host in hosts] 

1460 infos = await tasks.gather(*fs, loop=self) 

1461 infos = set(itertools.chain.from_iterable(infos)) 

1462 

1463 completed = False 

1464 try: 

1465 for res in infos: 

1466 af, socktype, proto, canonname, sa = res 

1467 try: 

1468 sock = socket.socket(af, socktype, proto) 

1469 except socket.error: 

1470 # Assume it's a bad family/type/protocol combination. 

1471 if self._debug: 

1472 logger.warning('create_server() failed to create ' 

1473 'socket.socket(%r, %r, %r)', 

1474 af, socktype, proto, exc_info=True) 

1475 continue 

1476 sockets.append(sock) 

1477 if reuse_address: 

1478 sock.setsockopt( 

1479 socket.SOL_SOCKET, socket.SO_REUSEADDR, True) 

1480 if reuse_port: 

1481 _set_reuseport(sock) 

1482 # Disable IPv4/IPv6 dual stack support (enabled by 

1483 # default on Linux) which makes a single socket 

1484 # listen on both address families. 

1485 if (_HAS_IPv6 and 

1486 af == socket.AF_INET6 and 

1487 hasattr(socket, 'IPPROTO_IPV6')): 

1488 sock.setsockopt(socket.IPPROTO_IPV6, 

1489 socket.IPV6_V6ONLY, 

1490 True) 

1491 try: 

1492 sock.bind(sa) 

1493 except OSError as err: 

1494 raise OSError(err.errno, 'error while attempting ' 

1495 'to bind on address %r: %s' 

1496 % (sa, err.strerror.lower())) from None 

1497 completed = True 

1498 finally: 

1499 if not completed: 

1500 for sock in sockets: 

1501 sock.close() 

1502 else: 

1503 if sock is None: 

1504 raise ValueError('Neither host/port nor sock were specified') 

1505 if sock.type != socket.SOCK_STREAM: 

1506 raise ValueError(f'A Stream Socket was expected, got {sock!r}') 

1507 sockets = [sock] 

1508 

1509 for sock in sockets: 

1510 sock.setblocking(False) 

1511 

1512 server = Server(self, sockets, protocol_factory, 

1513 ssl, backlog, ssl_handshake_timeout) 

1514 if start_serving: 

1515 server._start_serving() 

1516 # Skip one loop iteration so that all 'loop.add_reader' 

1517 # go through. 

1518 await tasks.sleep(0, loop=self) 

1519 

1520 if self._debug: 

1521 logger.info("%r is serving", server) 

1522 return server 

1523 

1524 async def connect_accepted_socket( 

1525 self, protocol_factory, sock, 

1526 *, ssl=None, 

1527 ssl_handshake_timeout=None): 

1528 """Handle an accepted connection. 

1529 

1530 This is used by servers that accept connections outside of 

1531 asyncio but that use asyncio to handle connections. 

1532 

1533 This method is a coroutine. When completed, the coroutine 

1534 returns a (transport, protocol) pair. 

1535 """ 

1536 if sock.type != socket.SOCK_STREAM: 

1537 raise ValueError(f'A Stream Socket was expected, got {sock!r}') 

1538 

1539 if ssl_handshake_timeout is not None and not ssl: 

1540 raise ValueError( 

1541 'ssl_handshake_timeout is only meaningful with ssl') 

1542 

1543 transport, protocol = await self._create_connection_transport( 

1544 sock, protocol_factory, ssl, '', server_side=True, 

1545 ssl_handshake_timeout=ssl_handshake_timeout) 

1546 if self._debug: 

1547 # Get the socket from the transport because SSL transport closes 

1548 # the old socket and creates a new SSL socket 

1549 sock = transport.get_extra_info('socket') 

1550 logger.debug("%r handled: (%r, %r)", sock, transport, protocol) 

1551 return transport, protocol 

1552 

1553 async def connect_read_pipe(self, protocol_factory, pipe): 

1554 protocol = protocol_factory() 

1555 waiter = self.create_future() 

1556 transport = self._make_read_pipe_transport(pipe, protocol, waiter) 

1557 

1558 try: 

1559 await waiter 

1560 except: 

1561 transport.close() 

1562 raise 

1563 

1564 if self._debug: 

1565 logger.debug('Read pipe %r connected: (%r, %r)', 

1566 pipe.fileno(), transport, protocol) 

1567 return transport, protocol 

1568 

1569 async def connect_write_pipe(self, protocol_factory, pipe): 

1570 protocol = protocol_factory() 

1571 waiter = self.create_future() 

1572 transport = self._make_write_pipe_transport(pipe, protocol, waiter) 

1573 

1574 try: 

1575 await waiter 

1576 except: 

1577 transport.close() 

1578 raise 

1579 

1580 if self._debug: 

1581 logger.debug('Write pipe %r connected: (%r, %r)', 

1582 pipe.fileno(), transport, protocol) 

1583 return transport, protocol 

1584 

1585 def _log_subprocess(self, msg, stdin, stdout, stderr): 

1586 info = [msg] 

1587 if stdin is not None: 

1588 info.append(f'stdin={_format_pipe(stdin)}') 

1589 if stdout is not None and stderr == subprocess.STDOUT: 

1590 info.append(f'stdout=stderr={_format_pipe(stdout)}') 

1591 else: 

1592 if stdout is not None: 

1593 info.append(f'stdout={_format_pipe(stdout)}') 

1594 if stderr is not None: 

1595 info.append(f'stderr={_format_pipe(stderr)}') 

1596 logger.debug(' '.join(info)) 

1597 

1598 async def subprocess_shell(self, protocol_factory, cmd, *, 

1599 stdin=subprocess.PIPE, 

1600 stdout=subprocess.PIPE, 

1601 stderr=subprocess.PIPE, 

1602 universal_newlines=False, 

1603 shell=True, bufsize=0, 

1604 encoding=None, errors=None, text=None, 

1605 **kwargs): 

1606 if not isinstance(cmd, (bytes, str)): 

1607 raise ValueError("cmd must be a string") 

1608 if universal_newlines: 

1609 raise ValueError("universal_newlines must be False") 

1610 if not shell: 

1611 raise ValueError("shell must be True") 

1612 if bufsize != 0: 

1613 raise ValueError("bufsize must be 0") 

1614 if text: 

1615 raise ValueError("text must be False") 

1616 if encoding is not None: 

1617 raise ValueError("encoding must be None") 

1618 if errors is not None: 

1619 raise ValueError("errors must be None") 

1620 

1621 protocol = protocol_factory() 

1622 debug_log = None 

1623 if self._debug: 

1624 # don't log parameters: they may contain sensitive information 

1625 # (password) and may be too long 

1626 debug_log = 'run shell command %r' % cmd 

1627 self._log_subprocess(debug_log, stdin, stdout, stderr) 

1628 transport = await self._make_subprocess_transport( 

1629 protocol, cmd, True, stdin, stdout, stderr, bufsize, **kwargs) 

1630 if self._debug and debug_log is not None: 

1631 logger.info('%s: %r', debug_log, transport) 

1632 return transport, protocol 

1633 

1634 async def subprocess_exec(self, protocol_factory, program, *args, 

1635 stdin=subprocess.PIPE, stdout=subprocess.PIPE, 

1636 stderr=subprocess.PIPE, universal_newlines=False, 

1637 shell=False, bufsize=0, 

1638 encoding=None, errors=None, text=None, 

1639 **kwargs): 

1640 if universal_newlines: 

1641 raise ValueError("universal_newlines must be False") 

1642 if shell: 

1643 raise ValueError("shell must be False") 

1644 if bufsize != 0: 

1645 raise ValueError("bufsize must be 0") 

1646 if text: 

1647 raise ValueError("text must be False") 

1648 if encoding is not None: 

1649 raise ValueError("encoding must be None") 

1650 if errors is not None: 

1651 raise ValueError("errors must be None") 

1652 

1653 popen_args = (program,) + args 

1654 protocol = protocol_factory() 

1655 debug_log = None 

1656 if self._debug: 

1657 # don't log parameters: they may contain sensitive information 

1658 # (password) and may be too long 

1659 debug_log = f'execute program {program!r}' 

1660 self._log_subprocess(debug_log, stdin, stdout, stderr) 

1661 transport = await self._make_subprocess_transport( 

1662 protocol, popen_args, False, stdin, stdout, stderr, 

1663 bufsize, **kwargs) 

1664 if self._debug and debug_log is not None: 

1665 logger.info('%s: %r', debug_log, transport) 

1666 return transport, protocol 

1667 

1668 def get_exception_handler(self): 

1669 """Return an exception handler, or None if the default one is in use. 

1670 """ 

1671 return self._exception_handler 

1672 

1673 def set_exception_handler(self, handler): 

1674 """Set handler as the new event loop exception handler. 

1675 

1676 If handler is None, the default exception handler will 

1677 be set. 

1678 

1679 If handler is a callable object, it should have a 

1680 signature matching '(loop, context)', where 'loop' 

1681 will be a reference to the active event loop, 'context' 

1682 will be a dict object (see `call_exception_handler()` 

1683 documentation for details about context). 

1684 """ 

1685 if handler is not None and not callable(handler): 

1686 raise TypeError(f'A callable object or None is expected, ' 

1687 f'got {handler!r}') 

1688 self._exception_handler = handler 

1689 

1690 def default_exception_handler(self, context): 

1691 """Default exception handler. 

1692 

1693 This is called when an exception occurs and no exception 

1694 handler is set, and can be called by a custom exception 

1695 handler that wants to defer to the default behavior. 

1696 

1697 This default handler logs the error message and other 

1698 context-dependent information. In debug mode, a truncated 

1699 stack trace is also appended showing where the given object 

1700 (e.g. a handle or future or task) was created, if any. 

1701 

1702 The context parameter has the same meaning as in 

1703 `call_exception_handler()`. 

1704 """ 

1705 message = context.get('message') 

1706 if not message: 

1707 message = 'Unhandled exception in event loop' 

1708 

1709 exception = context.get('exception') 

1710 if exception is not None: 

1711 exc_info = (type(exception), exception, exception.__traceback__) 

1712 else: 

1713 exc_info = False 

1714 

1715 if ('source_traceback' not in context and 

1716 self._current_handle is not None and 

1717 self._current_handle._source_traceback): 

1718 context['handle_traceback'] = \ 

1719 self._current_handle._source_traceback 

1720 

1721 log_lines = [message] 

1722 for key in sorted(context): 

1723 if key in {'message', 'exception'}: 

1724 continue 

1725 value = context[key] 

1726 if key == 'source_traceback': 

1727 tb = ''.join(traceback.format_list(value)) 

1728 value = 'Object created at (most recent call last):\n' 

1729 value += tb.rstrip() 

1730 elif key == 'handle_traceback': 

1731 tb = ''.join(traceback.format_list(value)) 

1732 value = 'Handle created at (most recent call last):\n' 

1733 value += tb.rstrip() 

1734 else: 

1735 value = repr(value) 

1736 log_lines.append(f'{key}: {value}') 

1737 

1738 logger.error('\n'.join(log_lines), exc_info=exc_info) 

1739 

1740 def call_exception_handler(self, context): 

1741 """Call the current event loop's exception handler. 

1742 

1743 The context argument is a dict containing the following keys: 

1744 

1745 - 'message': Error message; 

1746 - 'exception' (optional): Exception object; 

1747 - 'future' (optional): Future instance; 

1748 - 'task' (optional): Task instance; 

1749 - 'handle' (optional): Handle instance; 

1750 - 'protocol' (optional): Protocol instance; 

1751 - 'transport' (optional): Transport instance; 

1752 - 'socket' (optional): Socket instance; 

1753 - 'asyncgen' (optional): Asynchronous generator that caused 

1754 the exception. 

1755 

1756 New keys maybe introduced in the future. 

1757 

1758 Note: do not overload this method in an event loop subclass. 

1759 For custom exception handling, use the 

1760 `set_exception_handler()` method. 

1761 """ 

1762 if self._exception_handler is None: 

1763 try: 

1764 self.default_exception_handler(context) 

1765 except (SystemExit, KeyboardInterrupt): 

1766 raise 

1767 except BaseException: 

1768 # Second protection layer for unexpected errors 

1769 # in the default implementation, as well as for subclassed 

1770 # event loops with overloaded "default_exception_handler". 

1771 logger.error('Exception in default exception handler', 

1772 exc_info=True) 

1773 else: 

1774 try: 

1775 self._exception_handler(self, context) 

1776 except (SystemExit, KeyboardInterrupt): 

1777 raise 

1778 except BaseException as exc: 

1779 # Exception in the user set custom exception handler. 

1780 try: 

1781 # Let's try default handler. 

1782 self.default_exception_handler({ 

1783 'message': 'Unhandled error in exception handler', 

1784 'exception': exc, 

1785 'context': context, 

1786 }) 

1787 except (SystemExit, KeyboardInterrupt): 

1788 raise 

1789 except BaseException: 

1790 # Guard 'default_exception_handler' in case it is 

1791 # overloaded. 

1792 logger.error('Exception in default exception handler ' 

1793 'while handling an unexpected error ' 

1794 'in custom exception handler', 

1795 exc_info=True) 

1796 

1797 def _add_callback(self, handle): 

1798 """Add a Handle to _scheduled (TimerHandle) or _ready.""" 

1799 assert isinstance(handle, events.Handle), 'A Handle is required here' 

1800 if handle._cancelled: 

1801 return 

1802 assert not isinstance(handle, events.TimerHandle) 

1803 self._ready.append(handle) 

1804 

1805 def _add_callback_signalsafe(self, handle): 

1806 """Like _add_callback() but called from a signal handler.""" 

1807 self._add_callback(handle) 

1808 self._write_to_self() 

1809 

1810 def _timer_handle_cancelled(self, handle): 

1811 """Notification that a TimerHandle has been cancelled.""" 

1812 if handle._scheduled: 

1813 self._timer_cancelled_count += 1 

1814 

1815 def _run_once(self): 

1816 """Run one full iteration of the event loop. 

1817 

1818 This calls all currently ready callbacks, polls for I/O, 

1819 schedules the resulting callbacks, and finally schedules 

1820 'call_later' callbacks. 

1821 """ 

1822 

1823 sched_count = len(self._scheduled) 

1824 if (sched_count > _MIN_SCHEDULED_TIMER_HANDLES and 

1825 self._timer_cancelled_count / sched_count > 

1826 _MIN_CANCELLED_TIMER_HANDLES_FRACTION): 

1827 # Remove delayed calls that were cancelled if their number 

1828 # is too high 

1829 new_scheduled = [] 

1830 for handle in self._scheduled: 

1831 if handle._cancelled: 

1832 handle._scheduled = False 

1833 else: 

1834 new_scheduled.append(handle) 

1835 

1836 heapq.heapify(new_scheduled) 

1837 self._scheduled = new_scheduled 

1838 self._timer_cancelled_count = 0 

1839 else: 

1840 # Remove delayed calls that were cancelled from head of queue. 

1841 while self._scheduled and self._scheduled[0]._cancelled: 

1842 self._timer_cancelled_count -= 1 

1843 handle = heapq.heappop(self._scheduled) 

1844 handle._scheduled = False 

1845 

1846 timeout = None 

1847 if self._ready or self._stopping: 

1848 timeout = 0 

1849 elif self._scheduled: 

1850 # Compute the desired timeout. 

1851 when = self._scheduled[0]._when 

1852 timeout = min(max(0, when - self.time()), MAXIMUM_SELECT_TIMEOUT) 

1853 

1854 event_list = self._selector.select(timeout) 

1855 self._process_events(event_list) 

1856 

1857 # Handle 'later' callbacks that are ready. 

1858 end_time = self.time() + self._clock_resolution 

1859 while self._scheduled: 

1860 handle = self._scheduled[0] 

1861 if handle._when >= end_time: 

1862 break 

1863 handle = heapq.heappop(self._scheduled) 

1864 handle._scheduled = False 

1865 self._ready.append(handle) 

1866 

1867 # This is the only place where callbacks are actually *called*. 

1868 # All other places just add them to ready. 

1869 # Note: We run all currently scheduled callbacks, but not any 

1870 # callbacks scheduled by callbacks run this time around -- 

1871 # they will be run the next time (after another I/O poll). 

1872 # Use an idiom that is thread-safe without using locks. 

1873 ntodo = len(self._ready) 

1874 for i in range(ntodo): 

1875 handle = self._ready.popleft() 

1876 if handle._cancelled: 

1877 continue 

1878 if self._debug: 

1879 try: 

1880 self._current_handle = handle 

1881 t0 = self.time() 

1882 handle._run() 

1883 dt = self.time() - t0 

1884 if dt >= self.slow_callback_duration: 

1885 logger.warning('Executing %s took %.3f seconds', 

1886 _format_handle(handle), dt) 

1887 finally: 

1888 self._current_handle = None 

1889 else: 

1890 handle._run() 

1891 handle = None # Needed to break cycles when an exception occurs. 

1892 

1893 def _set_coroutine_origin_tracking(self, enabled): 

1894 if bool(enabled) == bool(self._coroutine_origin_tracking_enabled): 

1895 return 

1896 

1897 if enabled: 

1898 self._coroutine_origin_tracking_saved_depth = ( 

1899 sys.get_coroutine_origin_tracking_depth()) 

1900 sys.set_coroutine_origin_tracking_depth( 

1901 constants.DEBUG_STACK_DEPTH) 

1902 else: 

1903 sys.set_coroutine_origin_tracking_depth( 

1904 self._coroutine_origin_tracking_saved_depth) 

1905 

1906 self._coroutine_origin_tracking_enabled = enabled 

1907 

1908 def get_debug(self): 

1909 return self._debug 

1910 

1911 def set_debug(self, enabled): 

1912 self._debug = enabled 

1913 

1914 if self.is_running(): 

1915 self.call_soon_threadsafe(self._set_coroutine_origin_tracking, enabled)