Coverage for Lib/asyncio/proactor_events.py: 73%

1"""Event loop using a proactor and related classes. 

2 

3A proactor is a "notify-on-completion" multiplexer. Currently a 

4proactor is only implemented on Windows with IOCP. 

5""" 

6 

7__all__ = 'BaseProactorEventLoop', 

8 

9import io 

10import os 

11import socket 

12import warnings 

13import signal 

14import threading 

15import collections 

16 

17from . import base_events 

18from . import constants 

19from . import futures 

20from . import exceptions 

21from . import protocols 

22from . import sslproto 

23from . import transports 

24from . import trsock 

25from .log import logger 

26 

27 

28def _set_socket_extra(transport, sock): 

29 transport._extra['socket'] = trsock.TransportSocket(sock) 

30 

31 try: 

32 transport._extra['sockname'] = sock.getsockname() 

33 except socket.error: 

34 if transport._loop.get_debug(): 

35 logger.warning( 

36 "getsockname() failed on %r", sock, exc_info=True) 

37 

38 if 'peername' not in transport._extra: 38 ↛ exitline 38 didn't return from function '_set_socket_extra' because the condition on line 38 was always true

39 try: 

40 transport._extra['peername'] = sock.getpeername() 

41 except socket.error: 

42 # UDP sockets may not have a peer name 

43 transport._extra['peername'] = None 

44 

45 

46class _ProactorBasePipeTransport(transports._FlowControlMixin, 

47 transports.BaseTransport): 

48 """Base class for pipe and socket transports.""" 

49 

50 def __init__(self, loop, sock, protocol, waiter=None, 

51 extra=None, server=None): 

52 super().__init__(extra, loop) 

53 self._set_extra(sock) 

54 self._sock = sock 

55 self.set_protocol(protocol) 

56 self._server = server 

57 self._buffer = None # None or bytearray. 

58 self._read_fut = None 

59 self._write_fut = None 

60 self._pending_write = 0 

61 self._conn_lost = 0 

62 self._closing = False # Set when close() called. 

63 self._called_connection_lost = False 

64 self._eof_written = False 

65 self._empty_waiter = None 

66 if self._server is not None: 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true

67 self._server._attach(self) 

68 self._loop.call_soon(self._protocol.connection_made, self) 

69 if waiter is not None: 

70 # only wake up the waiter when connection_made() has been called 

71 self._loop.call_soon(futures._set_result_unless_cancelled, 

72 waiter, None) 

73 

74 def __repr__(self): 

75 info = [self.__class__.__name__] 

76 if self._sock is None: 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true

77 info.append('closed') 

78 elif self._closing: 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true

79 info.append('closing') 

80 if self._sock is not None: 80 ↛ 82line 80 didn't jump to line 82 because the condition on line 80 was always true

81 info.append(f'fd={self._sock.fileno()}') 

82 if self._read_fut is not None: 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true

83 info.append(f'read={self._read_fut!r}') 

84 if self._write_fut is not None: 84 ↛ 85line 84 didn't jump to line 85 because the condition on line 84 was never true

85 info.append(f'write={self._write_fut!r}') 

86 if self._buffer: 86 ↛ 87line 86 didn't jump to line 87 because the condition on line 86 was never true

87 info.append(f'write_bufsize={len(self._buffer)}') 

88 if self._eof_written: 88 ↛ 89line 88 didn't jump to line 89 because the condition on line 88 was never true

89 info.append('EOF written') 

90 return '<{}>'.format(' '.join(info)) 

91 

92 def _set_extra(self, sock): 

93 self._extra['pipe'] = sock 

94 

95 def set_protocol(self, protocol): 

96 self._protocol = protocol 

97 

98 def get_protocol(self): 

99 return self._protocol 

100 

101 def is_closing(self): 

102 return self._closing 

103 

104 def close(self): 

105 if self._closing: 

106 return 

107 self._closing = True 

108 self._conn_lost += 1 

109 if not self._buffer and self._write_fut is None: 

110 self._loop.call_soon(self._call_connection_lost, None) 

111 if self._read_fut is not None: 

112 self._read_fut.cancel() 

113 self._read_fut = None 

114 

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

116 if self._sock is not None: 116 ↛ 117line 116 didn't jump to line 117 because the condition on line 116 was never true

117 _warn(f"unclosed transport {self!r}", ResourceWarning, source=self) 

118 self._sock.close() 

119 

120 def _fatal_error(self, exc, message='Fatal error on pipe transport'): 

121 try: 

122 if isinstance(exc, OSError): 

123 if self._loop.get_debug(): 123 ↛ 124line 123 didn't jump to line 124 because the condition on line 123 was never true

124 logger.debug("%r: %s", self, message, exc_info=True) 

125 else: 

126 self._loop.call_exception_handler({ 

127 'message': message, 

128 'exception': exc, 

129 'transport': self, 

130 'protocol': self._protocol, 

131 }) 

132 finally: 

133 self._force_close(exc) 

134 

135 def _force_close(self, exc): 

136 if self._empty_waiter is not None and not self._empty_waiter.done(): 136 ↛ 137line 136 didn't jump to line 137 because the condition on line 136 was never true

137 if exc is None: 

138 self._empty_waiter.set_result(None) 

139 else: 

140 self._empty_waiter.set_exception(exc) 

141 if self._closing and self._called_connection_lost: 141 ↛ 142line 141 didn't jump to line 142 because the condition on line 141 was never true

142 return 

143 self._closing = True 

144 self._conn_lost += 1 

145 if self._write_fut: 

146 self._write_fut.cancel() 

147 self._write_fut = None 

148 if self._read_fut: 

149 self._read_fut.cancel() 

150 self._read_fut = None 

151 self._pending_write = 0 

152 self._buffer = None 

153 self._loop.call_soon(self._call_connection_lost, exc) 

154 

155 def _call_connection_lost(self, exc): 

156 if self._called_connection_lost: 

157 return 

158 try: 

159 self._protocol.connection_lost(exc) 

160 finally: 

161 # XXX If there is a pending overlapped read on the other 

162 # end then it may fail with ERROR_NETNAME_DELETED if we 

163 # just close our end. First calling shutdown() seems to 

164 # cure it, but maybe using DisconnectEx() would be better. 

165 if hasattr(self._sock, 'shutdown') and self._sock.fileno() != -1: 

166 self._sock.shutdown(socket.SHUT_RDWR) 

167 self._sock.close() 

168 self._sock = None 

169 server = self._server 

170 if server is not None: 170 ↛ 171line 170 didn't jump to line 171 because the condition on line 170 was never true

171 server._detach(self) 

172 self._server = None 

173 self._called_connection_lost = True 

174 

175 def get_write_buffer_size(self): 

176 size = self._pending_write 

177 if self._buffer is not None: 

178 size += len(self._buffer) 

179 return size 

180 

181 

182class _ProactorReadPipeTransport(_ProactorBasePipeTransport, 

183 transports.ReadTransport): 

184 """Transport for read pipes.""" 

185 

186 def __init__(self, loop, sock, protocol, waiter=None, 

187 extra=None, server=None, buffer_size=65536): 

188 self._pending_data_length = -1 

189 self._paused = True 

190 super().__init__(loop, sock, protocol, waiter, extra, server) 

191 

192 self._data = bytearray(buffer_size) 

193 self._loop.call_soon(self._loop_reading) 

194 self._paused = False 

195 

196 def is_reading(self): 

197 return not self._paused and not self._closing 

198 

199 def pause_reading(self): 

200 if self._closing or self._paused: 

201 return 

202 self._paused = True 

203 

204 # bpo-33694: Don't cancel self._read_fut because cancelling an 

205 # overlapped WSASend() loss silently data with the current proactor 

206 # implementation. 

207 # 

208 # If CancelIoEx() fails with ERROR_NOT_FOUND, it means that WSASend() 

209 # completed (even if HasOverlappedIoCompleted() returns 0), but 

210 # Overlapped.cancel() currently silently ignores the ERROR_NOT_FOUND 

211 # error. Once the overlapped is ignored, the IOCP loop will ignores the 

212 # completion I/O event and so not read the result of the overlapped 

213 # WSARecv(). 

214 

215 if self._loop.get_debug(): 215 ↛ 216line 215 didn't jump to line 216 because the condition on line 215 was never true

216 logger.debug("%r pauses reading", self) 

217 

218 def resume_reading(self): 

219 if self._closing or not self._paused: 

220 return 

221 

222 self._paused = False 

223 if self._read_fut is None: 

224 self._loop.call_soon(self._loop_reading, None) 

225 

226 length = self._pending_data_length 

227 self._pending_data_length = -1 

228 if length > -1: 

229 # Call the protocol method after calling _loop_reading(), 

230 # since the protocol can decide to pause reading again. 

231 self._loop.call_soon(self._data_received, self._data[:length], length) 

232 

233 if self._loop.get_debug(): 233 ↛ 234line 233 didn't jump to line 234 because the condition on line 233 was never true

234 logger.debug("%r resumes reading", self) 

235 

236 def _eof_received(self): 

237 if self._loop.get_debug(): 237 ↛ 238line 237 didn't jump to line 238 because the condition on line 237 was never true

238 logger.debug("%r received EOF", self) 

239 

240 try: 

241 keep_open = self._protocol.eof_received() 

242 except (SystemExit, KeyboardInterrupt): 

243 raise 

244 except BaseException as exc: 

245 self._fatal_error( 

246 exc, 'Fatal error: protocol.eof_received() call failed.') 

247 return 

248 

249 if not keep_open: 249 ↛ exitline 249 didn't return from function '_eof_received' because the condition on line 249 was always true

250 self.close() 

251 

252 def _data_received(self, data, length): 

253 if self._paused: 

254 # Don't call any protocol method while reading is paused. 

255 # The protocol will be called on resume_reading(). 

256 assert self._pending_data_length == -1 

257 self._pending_data_length = length 

258 return 

259 

260 if length == 0: 

261 self._eof_received() 

262 return 

263 

264 if isinstance(self._protocol, protocols.BufferedProtocol): 264 ↛ 265line 264 didn't jump to line 265 because the condition on line 264 was never true

265 try: 

266 protocols._feed_data_to_buffered_proto(self._protocol, data) 

267 except (SystemExit, KeyboardInterrupt): 

268 raise 

269 except BaseException as exc: 

270 self._fatal_error(exc, 

271 'Fatal error: protocol.buffer_updated() ' 

272 'call failed.') 

273 return 

274 else: 

275 self._protocol.data_received(data) 

276 

277 def _loop_reading(self, fut=None): 

278 length = -1 

279 data = None 

280 try: 

281 if fut is not None: 

282 assert self._read_fut is fut or (self._read_fut is None and 

283 self._closing) 

284 self._read_fut = None 

285 if fut.done(): 285 ↛ 296line 285 didn't jump to line 296 because the condition on line 285 was always true

286 # deliver data later in "finally" clause 

287 length = fut.result() 

288 if length == 0: 

289 # we got end-of-file so no need to reschedule a new read 

290 return 

291 

292 # It's a new slice so make it immutable so protocols upstream don't have problems 

293 data = bytes(memoryview(self._data)[:length]) 

294 else: 

295 # the future will be replaced by next proactor.recv call 

296 fut.cancel() 

297 

298 if self._closing: 

299 # since close() has been called we ignore any read data 

300 return 

301 

302 # bpo-33694: buffer_updated() has currently no fast path because of 

303 # a data loss issue caused by overlapped WSASend() cancellation. 

304 

305 if not self._paused: 

306 # reschedule a new read 

307 self._read_fut = self._loop._proactor.recv_into(self._sock, self._data) 

308 except ConnectionAbortedError as exc: 

309 if not self._closing: 309 ↛ 311line 309 didn't jump to line 311 because the condition on line 309 was always true

310 self._fatal_error(exc, 'Fatal read error on pipe transport') 

311 elif self._loop.get_debug(): 

312 logger.debug("Read error on pipe transport while closing", 

313 exc_info=True) 

314 except ConnectionResetError as exc: 

315 self._force_close(exc) 

316 except OSError as exc: 

317 self._fatal_error(exc, 'Fatal read error on pipe transport') 

318 except exceptions.CancelledError: 

319 if not self._closing: 

320 raise 

321 else: 

322 if not self._paused: 

323 self._read_fut.add_done_callback(self._loop_reading) 

324 finally: 

325 if length > -1: 

326 self._data_received(data, length) 

327 

328 

329class _ProactorBaseWritePipeTransport(_ProactorBasePipeTransport, 

330 transports.WriteTransport): 

331 """Transport for write pipes.""" 

332 

333 _start_tls_compatible = True 

334 

335 def write(self, data): 

336 if not isinstance(data, (bytes, bytearray, memoryview)): 336 ↛ 337line 336 didn't jump to line 337 because the condition on line 336 was never true

337 raise TypeError( 

338 f"data argument must be a bytes-like object, " 

339 f"not {type(data).__name__}") 

340 if self._eof_written: 340 ↛ 341line 340 didn't jump to line 341 because the condition on line 340 was never true

341 raise RuntimeError('write_eof() already called') 

342 if self._empty_waiter is not None: 342 ↛ 343line 342 didn't jump to line 343 because the condition on line 342 was never true

343 raise RuntimeError('unable to write; sendfile is in progress') 

344 

345 if not data: 

346 return 

347 

348 if self._conn_lost: 

349 if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES: 

350 logger.warning('socket.send() raised exception.') 

351 self._conn_lost += 1 

352 return 

353 

354 # Observable states: 

355 # 1. IDLE: _write_fut and _buffer both None 

356 # 2. WRITING: _write_fut set; _buffer None 

357 # 3. BACKED UP: _write_fut set; _buffer a bytearray 

358 # We always copy the data, so the caller can't modify it 

359 # while we're still waiting for the I/O to happen. 

360 if self._write_fut is None: # IDLE -> WRITING 

361 assert self._buffer is None 

362 # Pass a copy, except if it's already immutable. 

363 self._loop_writing(data=bytes(data)) 

364 elif not self._buffer: # WRITING -> BACKED UP 

365 # Make a mutable copy which we can extend. 

366 self._buffer = bytearray(data) 

367 self._maybe_pause_protocol() 

368 else: # BACKED UP 

369 # Append to buffer (also copies). 

370 self._buffer.extend(data) 

371 self._maybe_pause_protocol() 

372 

373 def _loop_writing(self, f=None, data=None): 

374 try: 

375 if f is not None and self._write_fut is None and self._closing: 

376 # XXX most likely self._force_close() has been called, and 

377 # it has set self._write_fut to None. 

378 return 

379 assert f is self._write_fut 

380 self._write_fut = None 

381 self._pending_write = 0 

382 if f: 

383 f.result() 

384 if data is None: 

385 data = self._buffer 

386 self._buffer = None 

387 if not data: 

388 if self._closing: 

389 self._loop.call_soon(self._call_connection_lost, None) 

390 if self._eof_written: 

391 self._sock.shutdown(socket.SHUT_WR) 

392 # Now that we've reduced the buffer size, tell the 

393 # protocol to resume writing if it was paused. Note that 

394 # we do this last since the callback is called immediately 

395 # and it may add more data to the buffer (even causing the 

396 # protocol to be paused again). 

397 self._maybe_resume_protocol() 

398 else: 

399 self._write_fut = self._loop._proactor.send(self._sock, data) 

400 if not self._write_fut.done(): 

401 assert self._pending_write == 0 

402 self._pending_write = len(data) 

403 self._write_fut.add_done_callback(self._loop_writing) 

404 self._maybe_pause_protocol() 

405 else: 

406 self._write_fut.add_done_callback(self._loop_writing) 

407 if self._empty_waiter is not None and self._write_fut is None: 407 ↛ 408line 407 didn't jump to line 408 because the condition on line 407 was never true

408 self._empty_waiter.set_result(None) 

409 except ConnectionResetError as exc: 

410 self._force_close(exc) 

411 except OSError as exc: 

412 self._fatal_error(exc, 'Fatal write error on pipe transport') 

413 

414 def can_write_eof(self): 

415 return True 

416 

417 def write_eof(self): 

418 self.close() 

419 

420 def abort(self): 

421 self._force_close(None) 

422 

423 def _make_empty_waiter(self): 

424 if self._empty_waiter is not None: 

425 raise RuntimeError("Empty waiter is already set") 

426 self._empty_waiter = self._loop.create_future() 

427 if self._write_fut is None: 

428 self._empty_waiter.set_result(None) 

429 return self._empty_waiter 

430 

431 def _reset_empty_waiter(self): 

432 self._empty_waiter = None 

433 

434 

435class _ProactorWritePipeTransport(_ProactorBaseWritePipeTransport): 

436 def __init__(self, *args, **kw): 

437 super().__init__(*args, **kw) 

438 self._read_fut = self._loop._proactor.recv(self._sock, 16) 

439 self._read_fut.add_done_callback(self._pipe_closed) 

440 

441 def _pipe_closed(self, fut): 

442 if fut.cancelled(): 

443 # the transport has been closed 

444 return 

445 assert fut.result() == b'' 

446 if self._closing: 

447 assert self._read_fut is None 

448 return 

449 assert fut is self._read_fut, (fut, self._read_fut) 

450 self._read_fut = None 

451 if self._write_fut is not None: 

452 self._force_close(BrokenPipeError()) 

453 else: 

454 self.close() 

455 

456 

457class _ProactorDatagramTransport(_ProactorBasePipeTransport, 

458 transports.DatagramTransport): 

459 max_size = 256 * 1024 

460 _header_size = 8 

461 

462 def __init__(self, loop, sock, protocol, address=None, 

463 waiter=None, extra=None): 

464 self._address = address 

465 self._buffer_size = 0 

466 # We don't need to call _protocol.connection_made() since our base 

467 # constructor does it for us. 

468 super().__init__(loop, sock, protocol, waiter=waiter, extra=extra) 

469 

470 # The base constructor sets _buffer = None, so we set it here 

471 self._buffer = collections.deque() 

472 self._loop.call_soon(self._loop_reading) 

473 

474 def _set_extra(self, sock): 

475 _set_socket_extra(self, sock) 

476 

477 def get_write_buffer_size(self): 

478 return self._buffer_size 

479 

480 def abort(self): 

481 self._force_close(None) 

482 

483 def sendto(self, data, addr=None): 

484 if not isinstance(data, (bytes, bytearray, memoryview)): 

485 raise TypeError('data argument must be bytes-like object (%r)', 

486 type(data)) 

487 

488 if self._address is not None and addr not in (None, self._address): 

489 raise ValueError( 

490 f'Invalid address: must be None or {self._address}') 

491 

492 if self._conn_lost and self._address: 

493 if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES: 

494 logger.warning('socket.sendto() raised exception.') 

495 self._conn_lost += 1 

496 return 

497 

498 # Ensure that what we buffer is immutable. 

499 self._buffer.append((bytes(data), addr)) 

500 self._buffer_size += len(data) + self._header_size 

501 

502 if self._write_fut is None: 

503 # No current write operations are active, kick one off 

504 self._loop_writing() 

505 # else: A write operation is already kicked off 

506 

507 self._maybe_pause_protocol() 

508 

509 def _loop_writing(self, fut=None): 

510 try: 

511 if self._conn_lost: 511 ↛ 512line 511 didn't jump to line 512 because the condition on line 511 was never true

512 return 

513 

514 assert fut is self._write_fut 

515 self._write_fut = None 

516 if fut: 516 ↛ 518line 516 didn't jump to line 518 because the condition on line 516 was never true

517 # We are in a _loop_writing() done callback, get the result 

518 fut.result() 

519 

520 if not self._buffer or (self._conn_lost and self._address): 

521 # The connection has been closed 

522 if self._closing: 522 ↛ 524line 522 didn't jump to line 524 because the condition on line 522 was always true

523 self._loop.call_soon(self._call_connection_lost, None) 

524 return 

525 

526 data, addr = self._buffer.popleft() 

527 self._buffer_size -= len(data) + self._header_size 

528 if self._address is not None: 

529 self._write_fut = self._loop._proactor.send(self._sock, 

530 data) 

531 else: 

532 self._write_fut = self._loop._proactor.sendto(self._sock, 

533 data, 

534 addr=addr) 

535 except OSError as exc: 

536 self._protocol.error_received(exc) 

537 except Exception as exc: 

538 self._fatal_error(exc, 'Fatal write error on datagram transport') 

539 else: 

540 self._write_fut.add_done_callback(self._loop_writing) 

541 self._maybe_resume_protocol() 

542 

543 def _loop_reading(self, fut=None): 

544 data = None 

545 try: 

546 if self._conn_lost: 546 ↛ 547line 546 didn't jump to line 547 because the condition on line 546 was never true

547 return 

548 

549 assert self._read_fut is fut or (self._read_fut is None and 

550 self._closing) 

551 

552 self._read_fut = None 

553 if fut is not None: 

554 res = fut.result() 

555 

556 if self._closing: 556 ↛ 558line 556 didn't jump to line 558 because the condition on line 556 was never true

557 # since close() has been called we ignore any read data 

558 data = None 

559 return 

560 

561 if self._address is not None: 561 ↛ 562line 561 didn't jump to line 562 because the condition on line 561 was never true

562 data, addr = res, self._address 

563 else: 

564 data, addr = res 

565 

566 if self._conn_lost: 566 ↛ 567line 566 didn't jump to line 567 because the condition on line 566 was never true

567 return 

568 if self._address is not None: 568 ↛ 569line 568 didn't jump to line 569 because the condition on line 568 was never true

569 self._read_fut = self._loop._proactor.recv(self._sock, 

570 self.max_size) 

571 else: 

572 self._read_fut = self._loop._proactor.recvfrom(self._sock, 

573 self.max_size) 

574 except OSError as exc: 

575 self._protocol.error_received(exc) 

576 except exceptions.CancelledError: 

577 if not self._closing: 

578 raise 

579 else: 

580 if self._read_fut is not None: 580 ↛ 583line 580 didn't jump to line 583 because the condition on line 580 was always true

581 self._read_fut.add_done_callback(self._loop_reading) 

582 finally: 

583 if data: 

584 self._protocol.datagram_received(data, addr) 

585 

586 

587class _ProactorDuplexPipeTransport(_ProactorReadPipeTransport, 

588 _ProactorBaseWritePipeTransport, 

589 transports.Transport): 

590 """Transport for duplex pipes.""" 

591 

592 def can_write_eof(self): 

593 return False 

594 

595 def write_eof(self): 

596 raise NotImplementedError 

597 

598 

599class _ProactorSocketTransport(_ProactorReadPipeTransport, 

600 _ProactorBaseWritePipeTransport, 

601 transports.Transport): 

602 """Transport for connected sockets.""" 

603 

604 _sendfile_compatible = constants._SendfileMode.TRY_NATIVE 

605 

606 def __init__(self, loop, sock, protocol, waiter=None, 

607 extra=None, server=None): 

608 super().__init__(loop, sock, protocol, waiter, extra, server) 

609 base_events._set_nodelay(sock) 

610 

611 def _set_extra(self, sock): 

612 _set_socket_extra(self, sock) 

613 

614 def can_write_eof(self): 

615 return True 

616 

617 def write_eof(self): 

618 if self._closing or self._eof_written: 

619 return 

620 self._eof_written = True 

621 if self._write_fut is None: 

622 self._sock.shutdown(socket.SHUT_WR) 

623 

624 

625class BaseProactorEventLoop(base_events.BaseEventLoop): 

626 

627 def __init__(self, proactor): 

628 super().__init__() 

629 logger.debug('Using proactor: %s', proactor.__class__.__name__) 

630 self._proactor = proactor 

631 self._selector = proactor # convenient alias 

632 self._self_reading_future = None 

633 self._accept_futures = {} # socket file descriptor => Future 

634 proactor.set_loop(self) 

635 self._make_self_pipe() 

636 if threading.current_thread() is threading.main_thread(): 636 ↛ exitline 636 didn't return from function '__init__' because the condition on line 636 was always true

637 # wakeup fd can only be installed to a file descriptor from the main thread 

638 signal.set_wakeup_fd(self._csock.fileno()) 

639 

640 def _make_socket_transport(self, sock, protocol, waiter=None, 

641 extra=None, server=None, context=None): 

642 return _ProactorSocketTransport(self, sock, protocol, waiter, 

643 extra, server) 

644 

645 def _make_ssl_transport( 

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

647 *, server_side=False, server_hostname=None, 

648 extra=None, server=None, 

649 ssl_handshake_timeout=None, 

650 ssl_shutdown_timeout=None, context=None): 

651 ssl_protocol = sslproto.SSLProtocol( 

652 self, protocol, sslcontext, waiter, 

653 server_side, server_hostname, 

654 ssl_handshake_timeout=ssl_handshake_timeout, 

655 ssl_shutdown_timeout=ssl_shutdown_timeout) 

656 _ProactorSocketTransport(self, rawsock, ssl_protocol, 

657 extra=extra, server=server) 

658 return ssl_protocol._app_transport 

659 

660 def _make_datagram_transport(self, sock, protocol, 

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

662 return _ProactorDatagramTransport(self, sock, protocol, address, 

663 waiter, extra) 

664 

665 def _make_duplex_pipe_transport(self, sock, protocol, waiter=None, 

666 extra=None): 

667 return _ProactorDuplexPipeTransport(self, 

668 sock, protocol, waiter, extra) 

669 

670 def _make_read_pipe_transport(self, sock, protocol, waiter=None, 

671 extra=None): 

672 return _ProactorReadPipeTransport(self, sock, protocol, waiter, extra) 

673 

674 def _make_write_pipe_transport(self, sock, protocol, waiter=None, 

675 extra=None): 

676 # We want connection_lost() to be called when other end closes 

677 return _ProactorWritePipeTransport(self, 

678 sock, protocol, waiter, extra) 

679 

680 def close(self): 

681 if self.is_running(): 681 ↛ 682line 681 didn't jump to line 682 because the condition on line 681 was never true

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

683 if self.is_closed(): 

684 return 

685 

686 if threading.current_thread() is threading.main_thread(): 686 ↛ 691line 686 didn't jump to line 691 because the condition on line 686 was always true

687 signal.set_wakeup_fd(-1) 

688 # Call these methods before closing the event loop (before calling 

689 # BaseEventLoop.close), because they can schedule callbacks with 

690 # call_soon(), which is forbidden when the event loop is closed. 

691 self._stop_accept_futures() 

692 self._close_self_pipe() 

693 self._proactor.close() 

694 self._proactor = None 

695 self._selector = None 

696 

697 # Close the event loop 

698 super().close() 

699 

700 async def sock_recv(self, sock, n): 

701 return await self._proactor.recv(sock, n) 

702 

703 async def sock_recv_into(self, sock, buf): 

704 return await self._proactor.recv_into(sock, buf) 

705 

706 async def sock_recvfrom(self, sock, bufsize): 

707 return await self._proactor.recvfrom(sock, bufsize) 

708 

709 async def sock_recvfrom_into(self, sock, buf, nbytes=0): 

710 if not nbytes: 

711 nbytes = len(buf) 

712 

713 return await self._proactor.recvfrom_into(sock, buf, nbytes) 

714 

715 async def sock_sendall(self, sock, data): 

716 return await self._proactor.send(sock, data) 

717 

718 async def sock_sendto(self, sock, data, address): 

719 return await self._proactor.sendto(sock, data, 0, address) 

720 

721 async def sock_connect(self, sock, address): 

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

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

724 return await self._proactor.connect(sock, address) 

725 

726 async def sock_accept(self, sock): 

727 return await self._proactor.accept(sock) 

728 

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

730 try: 

731 fileno = file.fileno() 

732 except (AttributeError, io.UnsupportedOperation): 

733 raise exceptions.SendfileNotAvailableError("not a regular file") 

734 try: 

735 fsize = os.fstat(fileno).st_size 

736 except OSError: 

737 raise exceptions.SendfileNotAvailableError("not a regular file") 

738 blocksize = count if count else fsize 

739 if not blocksize: 

740 return 0 # empty file 

741 

742 blocksize = min(blocksize, 0xffff_ffff) 

743 end_pos = min(offset + count, fsize) if count else fsize 

744 offset = min(offset, fsize) 

745 total_sent = 0 

746 try: 

747 while True: 

748 blocksize = min(end_pos - offset, blocksize) 

749 if blocksize <= 0: 

750 return total_sent 

751 await self._proactor.sendfile(sock, file, offset, blocksize) 

752 offset += blocksize 

753 total_sent += blocksize 

754 finally: 

755 file.seek(offset) 

756 

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

758 resume_reading = transp.is_reading() 

759 transp.pause_reading() 

760 await transp._make_empty_waiter() 

761 try: 

762 return await self.sock_sendfile(transp._sock, file, offset, count, 

763 fallback=False) 

764 finally: 

765 transp._reset_empty_waiter() 

766 if resume_reading: 

767 transp.resume_reading() 

768 

769 def _close_self_pipe(self): 

770 if self._self_reading_future is not None: 

771 self._self_reading_future.cancel() 

772 self._self_reading_future = None 

773 self._ssock.close() 

774 self._ssock = None 

775 self._csock.close() 

776 self._csock = None 

777 self._internal_fds -= 1