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 if self._server is not None: 65 ↛ 66line 65 didn't jump to line 66 because the condition on line 65 was never true

66 self._server._attach(self) 

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

68 if waiter is not None: 

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

70 self._loop.call_soon(futures._set_result_unless_cancelled, 

71 waiter, None) 

72 

73 def __repr__(self): 

74 info = [self.__class__.__name__] 

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

76 info.append('closed') 

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

78 info.append('closing') 

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

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

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

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

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

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

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

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

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

88 info.append('EOF written') 

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

90 

91 def _set_extra(self, sock): 

92 self._extra['pipe'] = sock 

93 

94 def set_protocol(self, protocol): 

95 self._protocol = protocol 

96 

97 def get_protocol(self): 

98 return self._protocol 

99 

100 def is_closing(self): 

101 return self._closing 

102 

103 def close(self): 

104 if self._closing: 

105 return 

106 self._closing = True 

107 self._conn_lost += 1 

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

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

110 if self._read_fut is not None: 

111 self._read_fut.cancel() 

112 self._read_fut = None 

113 

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

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

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

117 self._sock.close() 

118 

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

120 try: 

121 if isinstance(exc, OSError): 

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

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

124 else: 

125 self._loop.call_exception_handler({ 

126 'message': message, 

127 'exception': exc, 

128 'transport': self, 

129 'protocol': self._protocol, 

130 }) 

131 finally: 

132 self._force_close(exc) 

133 

134 def _force_close(self, exc): 

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

136 if exc is None: 

137 self._empty_waiter.set_result(None) 

138 else: 

139 self._empty_waiter.set_exception(exc) 

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

141 return 

142 self._closing = True 

143 self._conn_lost += 1 

144 if self._write_fut: 

145 self._write_fut.cancel() 

146 self._write_fut = None 

147 if self._read_fut: 

148 self._read_fut.cancel() 

149 self._read_fut = None 

150 self._pending_write = 0 

151 self._buffer = None 

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

153 

154 def _call_connection_lost(self, exc): 

155 if self._called_connection_lost: 

156 return 

157 try: 

158 self._protocol.connection_lost(exc) 

159 finally: 

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

161 # end then it may fail with ERROR_NETNAME_DELETED if we 

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

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

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

165 self._sock.shutdown(socket.SHUT_RDWR) 

166 self._sock.close() 

167 self._sock = None 

168 server = self._server 

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

170 server._detach(self) 

171 self._server = None 

172 self._called_connection_lost = True 

173 

174 def get_write_buffer_size(self): 

175 size = self._pending_write 

176 if self._buffer is not None: 

177 size += len(self._buffer) 

178 return size 

179 

180 

181class _ProactorReadPipeTransport(_ProactorBasePipeTransport, 

182 transports.ReadTransport): 

183 """Transport for read pipes.""" 

184 

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

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

187 self._pending_data_length = -1 

188 self._paused = True 

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

190 

191 self._data = bytearray(buffer_size) 

192 self._loop.call_soon(self._loop_reading) 

193 self._paused = False 

194 

195 def is_reading(self): 

196 return not self._paused and not self._closing 

197 

198 def pause_reading(self): 

199 if self._closing or self._paused: 

200 return 

201 self._paused = True 

202 

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

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

205 # implementation. 

206 # 

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

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

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

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

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

212 # WSARecv(). 

213 

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

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

216 

217 def resume_reading(self): 

218 if self._closing or not self._paused: 

219 return 

220 

221 self._paused = False 

222 if self._read_fut is None: 

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

224 

225 length = self._pending_data_length 

226 self._pending_data_length = -1 

227 if length > -1: 

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

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

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

231 

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

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

234 

235 def _eof_received(self): 

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

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

238 

239 try: 

240 keep_open = self._protocol.eof_received() 

241 except (SystemExit, KeyboardInterrupt): 

242 raise 

243 except BaseException as exc: 

244 self._fatal_error( 

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

246 return 

247 

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

249 self.close() 

250 

251 def _data_received(self, data, length): 

252 if self._paused: 

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

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

255 assert self._pending_data_length == -1 

256 self._pending_data_length = length 

257 return 

258 

259 if length == 0: 

260 self._eof_received() 

261 return 

262 

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

264 try: 

265 protocols._feed_data_to_buffered_proto(self._protocol, data) 

266 except (SystemExit, KeyboardInterrupt): 

267 raise 

268 except BaseException as exc: 

269 self._fatal_error(exc, 

270 'Fatal error: protocol.buffer_updated() ' 

271 'call failed.') 

272 return 

273 else: 

274 self._protocol.data_received(data) 

275 

276 def _loop_reading(self, fut=None): 

277 length = -1 

278 data = None 

279 try: 

280 if fut is not None: 

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

282 self._closing) 

283 self._read_fut = None 

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

285 # deliver data later in "finally" clause 

286 length = fut.result() 

287 if length == 0: 

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

289 return 

290 

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

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

293 else: 

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

295 fut.cancel() 

296 

297 if self._closing: 

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

299 return 

300 

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

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

303 

304 if not self._paused: 

305 # reschedule a new read 

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

307 except ConnectionAbortedError as exc: 

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

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

310 elif self._loop.get_debug(): 

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

312 exc_info=True) 

313 except ConnectionResetError as exc: 

314 self._force_close(exc) 

315 except OSError as exc: 

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

317 except exceptions.CancelledError: 

318 if not self._closing: 

319 raise 

320 else: 

321 if not self._paused: 

322 self._read_fut.add_done_callback(self._loop_reading) 

323 finally: 

324 if length > -1: 

325 self._data_received(data, length) 

326 

327 

328class _ProactorBaseWritePipeTransport(_ProactorBasePipeTransport, 

329 transports.WriteTransport): 

330 """Transport for write pipes.""" 

331 

332 _start_tls_compatible = True 

333 

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

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

336 self._empty_waiter = None 

337 

338 def write(self, data): 

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

340 raise TypeError( 

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

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

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

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

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

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

347 

348 if not data: 

349 return 

350 

351 if self._conn_lost: 

352 if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES: 

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

354 self._conn_lost += 1 

355 return 

356 

357 # Observable states: 

358 # 1. IDLE: _write_fut and _buffer both None 

359 # 2. WRITING: _write_fut set; _buffer None 

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

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

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

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

364 assert self._buffer is None 

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

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

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

368 # Make a mutable copy which we can extend. 

369 self._buffer = bytearray(data) 

370 self._maybe_pause_protocol() 

371 else: # BACKED UP 

372 # Append to buffer (also copies). 

373 self._buffer.extend(data) 

374 self._maybe_pause_protocol() 

375 

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

377 try: 

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

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

380 # it has set self._write_fut to None. 

381 return 

382 assert f is self._write_fut 

383 self._write_fut = None 

384 self._pending_write = 0 

385 if f: 

386 f.result() 

387 if data is None: 

388 data = self._buffer 

389 self._buffer = None 

390 if not data: 

391 if self._closing: 

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

393 if self._eof_written: 

394 self._sock.shutdown(socket.SHUT_WR) 

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

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

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

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

399 # protocol to be paused again). 

400 self._maybe_resume_protocol() 

401 else: 

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

403 if not self._write_fut.done(): 

404 assert self._pending_write == 0 

405 self._pending_write = len(data) 

406 self._write_fut.add_done_callback(self._loop_writing) 

407 self._maybe_pause_protocol() 

408 else: 

409 self._write_fut.add_done_callback(self._loop_writing) 

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

411 self._empty_waiter.set_result(None) 

412 except ConnectionResetError as exc: 

413 self._force_close(exc) 

414 except OSError as exc: 

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

416 

417 def can_write_eof(self): 

418 return True 

419 

420 def write_eof(self): 

421 self.close() 

422 

423 def abort(self): 

424 self._force_close(None) 

425 

426 def _make_empty_waiter(self): 

427 if self._empty_waiter is not None: 

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

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

430 if self._write_fut is None: 

431 self._empty_waiter.set_result(None) 

432 return self._empty_waiter 

433 

434 def _reset_empty_waiter(self): 

435 self._empty_waiter = None 

436 

437 

438class _ProactorWritePipeTransport(_ProactorBaseWritePipeTransport): 

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

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

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

442 self._read_fut.add_done_callback(self._pipe_closed) 

443 

444 def _pipe_closed(self, fut): 

445 if fut.cancelled(): 

446 # the transport has been closed 

447 return 

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

449 if self._closing: 

450 assert self._read_fut is None 

451 return 

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

453 self._read_fut = None 

454 if self._write_fut is not None: 

455 self._force_close(BrokenPipeError()) 

456 else: 

457 self.close() 

458 

459 

460class _ProactorDatagramTransport(_ProactorBasePipeTransport, 

461 transports.DatagramTransport): 

462 max_size = 256 * 1024 

463 _header_size = 8 

464 

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

466 waiter=None, extra=None): 

467 self._address = address 

468 self._empty_waiter = None 

469 self._buffer_size = 0 

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

471 # constructor does it for us. 

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

473 

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

475 self._buffer = collections.deque() 

476 self._loop.call_soon(self._loop_reading) 

477 

478 def _set_extra(self, sock): 

479 _set_socket_extra(self, sock) 

480 

481 def get_write_buffer_size(self): 

482 return self._buffer_size 

483 

484 def abort(self): 

485 self._force_close(None) 

486 

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

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

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

490 type(data)) 

491 

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

493 raise ValueError( 

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

495 

496 if self._conn_lost and self._address: 

497 if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES: 

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

499 self._conn_lost += 1 

500 return 

501 

502 # Ensure that what we buffer is immutable. 

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

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

505 

506 if self._write_fut is None: 

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

508 self._loop_writing() 

509 # else: A write operation is already kicked off 

510 

511 self._maybe_pause_protocol() 

512 

513 def _loop_writing(self, fut=None): 

514 try: 

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

516 return 

517 

518 assert fut is self._write_fut 

519 self._write_fut = None 

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

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

522 fut.result() 

523 

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

525 # The connection has been closed 

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

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

528 return 

529 

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

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

532 if self._address is not None: 

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

534 data) 

535 else: 

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

537 data, 

538 addr=addr) 

539 except OSError as exc: 

540 self._protocol.error_received(exc) 

541 except Exception as exc: 

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

543 else: 

544 self._write_fut.add_done_callback(self._loop_writing) 

545 self._maybe_resume_protocol() 

546 

547 def _loop_reading(self, fut=None): 

548 data = None 

549 try: 

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

551 return 

552 

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

554 self._closing) 

555 

556 self._read_fut = None 

557 if fut is not None: 

558 res = fut.result() 

559 

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

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

562 data = None 

563 return 

564 

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

566 data, addr = res, self._address 

567 else: 

568 data, addr = res 

569 

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

571 return 

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

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

574 self.max_size) 

575 else: 

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

577 self.max_size) 

578 except OSError as exc: 

579 self._protocol.error_received(exc) 

580 except exceptions.CancelledError: 

581 if not self._closing: 

582 raise 

583 else: 

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

585 self._read_fut.add_done_callback(self._loop_reading) 

586 finally: 

587 if data: 

588 self._protocol.datagram_received(data, addr) 

589 

590 

591class _ProactorDuplexPipeTransport(_ProactorReadPipeTransport, 

592 _ProactorBaseWritePipeTransport, 

593 transports.Transport): 

594 """Transport for duplex pipes.""" 

595 

596 def can_write_eof(self): 

597 return False 

598 

599 def write_eof(self): 

600 raise NotImplementedError 

601 

602 

603class _ProactorSocketTransport(_ProactorReadPipeTransport, 

604 _ProactorBaseWritePipeTransport, 

605 transports.Transport): 

606 """Transport for connected sockets.""" 

607 

608 _sendfile_compatible = constants._SendfileMode.TRY_NATIVE 

609 

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

611 extra=None, server=None): 

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

613 base_events._set_nodelay(sock) 

614 

615 def _set_extra(self, sock): 

616 _set_socket_extra(self, sock) 

617 

618 def can_write_eof(self): 

619 return True 

620 

621 def write_eof(self): 

622 if self._closing or self._eof_written: 

623 return 

624 self._eof_written = True 

625 if self._write_fut is None: 

626 self._sock.shutdown(socket.SHUT_WR) 

627 

628 

629class BaseProactorEventLoop(base_events.BaseEventLoop): 

630 

631 def __init__(self, proactor): 

632 super().__init__() 

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

634 self._proactor = proactor 

635 self._selector = proactor # convenient alias 

636 self._self_reading_future = None 

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

638 proactor.set_loop(self) 

639 self._make_self_pipe() 

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

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

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

643 

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

645 extra=None, server=None): 

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

647 extra, server) 

648 

649 def _make_ssl_transport( 

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

651 *, server_side=False, server_hostname=None, 

652 extra=None, server=None, 

653 ssl_handshake_timeout=None, 

654 ssl_shutdown_timeout=None): 

655 ssl_protocol = sslproto.SSLProtocol( 

656 self, protocol, sslcontext, waiter, 

657 server_side, server_hostname, 

658 ssl_handshake_timeout=ssl_handshake_timeout, 

659 ssl_shutdown_timeout=ssl_shutdown_timeout) 

660 _ProactorSocketTransport(self, rawsock, ssl_protocol, 

661 extra=extra, server=server) 

662 return ssl_protocol._app_transport 

663 

664 def _make_datagram_transport(self, sock, protocol, 

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

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

667 waiter, extra) 

668 

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

670 extra=None): 

671 return _ProactorDuplexPipeTransport(self, 

672 sock, protocol, waiter, extra) 

673 

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

675 extra=None): 

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

677 

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

679 extra=None): 

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

681 return _ProactorWritePipeTransport(self, 

682 sock, protocol, waiter, extra) 

683 

684 def close(self): 

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

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

687 if self.is_closed(): 

688 return 

689 

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

691 signal.set_wakeup_fd(-1) 

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

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

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

695 self._stop_accept_futures() 

696 self._close_self_pipe() 

697 self._proactor.close() 

698 self._proactor = None 

699 self._selector = None 

700 

701 # Close the event loop 

702 super().close() 

703 

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

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

706 

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

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

709 

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

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

712 

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

714 if not nbytes: 

715 nbytes = len(buf) 

716 

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

718 

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

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

721 

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

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

724 

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

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

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

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

729 

730 async def sock_accept(self, sock): 

731 return await self._proactor.accept(sock) 

732 

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

734 try: 

735 fileno = file.fileno() 

736 except (AttributeError, io.UnsupportedOperation) as err: 

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

738 try: 

739 fsize = os.fstat(fileno).st_size 

740 except OSError: 

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

742 blocksize = count if count else fsize 

743 if not blocksize: 

744 return 0 # empty file 

745 

746 blocksize = min(blocksize, 0xffff_ffff) 

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

748 offset = min(offset, fsize) 

749 total_sent = 0 

750 try: 

751 while True: 

752 blocksize = min(end_pos - offset, blocksize) 

753 if blocksize <= 0: 

754 return total_sent 

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

756 offset += blocksize 

757 total_sent += blocksize 

758 finally: 

759 if total_sent > 0: 

760 file.seek(offset) 

761 

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

763 resume_reading = transp.is_reading() 

764 transp.pause_reading() 

765 await transp._make_empty_waiter() 

766 try: 

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

768 fallback=False) 

769 finally: 

770 transp._reset_empty_waiter() 

771 if resume_reading: 

772 transp.resume_reading() 

773 

774 def _close_self_pipe(self): 

775 if self._self_reading_future is not None: 

776 self._self_reading_future.cancel() 

777 self._self_reading_future = None