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 def __init__(self, loop, sock, protocol, address=None, 

464 waiter=None, extra=None): 

465 self._address = address 

466 self._empty_waiter = None 

467 self._buffer_size = 0 

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

469 # constructor does it for us. 

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

471 

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

473 self._buffer = collections.deque() 

474 self._loop.call_soon(self._loop_reading) 

475 

476 def _set_extra(self, sock): 

477 _set_socket_extra(self, sock) 

478 

479 def get_write_buffer_size(self): 

480 return self._buffer_size 

481 

482 def abort(self): 

483 self._force_close(None) 

484 

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

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

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

488 type(data)) 

489 

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

491 raise ValueError( 

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

493 

494 if self._conn_lost and self._address: 

495 if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES: 

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

497 self._conn_lost += 1 

498 return 

499 

500 # Ensure that what we buffer is immutable. 

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

502 self._buffer_size += len(data) + 8 # include header bytes 

503 

504 if self._write_fut is None: 

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

506 self._loop_writing() 

507 # else: A write operation is already kicked off 

508 

509 self._maybe_pause_protocol() 

510 

511 def _loop_writing(self, fut=None): 

512 try: 

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

514 return 

515 

516 assert fut is self._write_fut 

517 self._write_fut = None 

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

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

520 fut.result() 

521 

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

523 # The connection has been closed 

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

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

526 return 

527 

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

529 self._buffer_size -= len(data) 

530 if self._address is not None: 

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

532 data) 

533 else: 

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

535 data, 

536 addr=addr) 

537 except OSError as exc: 

538 self._protocol.error_received(exc) 

539 except Exception as exc: 

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

541 else: 

542 self._write_fut.add_done_callback(self._loop_writing) 

543 self._maybe_resume_protocol() 

544 

545 def _loop_reading(self, fut=None): 

546 data = None 

547 try: 

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

549 return 

550 

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

552 self._closing) 

553 

554 self._read_fut = None 

555 if fut is not None: 

556 res = fut.result() 

557 

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

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

560 data = None 

561 return 

562 

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

564 data, addr = res, self._address 

565 else: 

566 data, addr = res 

567 

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

569 return 

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

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

572 self.max_size) 

573 else: 

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

575 self.max_size) 

576 except OSError as exc: 

577 self._protocol.error_received(exc) 

578 except exceptions.CancelledError: 

579 if not self._closing: 

580 raise 

581 else: 

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

583 self._read_fut.add_done_callback(self._loop_reading) 

584 finally: 

585 if data: 

586 self._protocol.datagram_received(data, addr) 

587 

588 

589class _ProactorDuplexPipeTransport(_ProactorReadPipeTransport, 

590 _ProactorBaseWritePipeTransport, 

591 transports.Transport): 

592 """Transport for duplex pipes.""" 

593 

594 def can_write_eof(self): 

595 return False 

596 

597 def write_eof(self): 

598 raise NotImplementedError 

599 

600 

601class _ProactorSocketTransport(_ProactorReadPipeTransport, 

602 _ProactorBaseWritePipeTransport, 

603 transports.Transport): 

604 """Transport for connected sockets.""" 

605 

606 _sendfile_compatible = constants._SendfileMode.TRY_NATIVE 

607 

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

609 extra=None, server=None): 

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

611 base_events._set_nodelay(sock) 

612 

613 def _set_extra(self, sock): 

614 _set_socket_extra(self, sock) 

615 

616 def can_write_eof(self): 

617 return True 

618 

619 def write_eof(self): 

620 if self._closing or self._eof_written: 

621 return 

622 self._eof_written = True 

623 if self._write_fut is None: 

624 self._sock.shutdown(socket.SHUT_WR) 

625 

626 

627class BaseProactorEventLoop(base_events.BaseEventLoop): 

628 

629 def __init__(self, proactor): 

630 super().__init__() 

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

632 self._proactor = proactor 

633 self._selector = proactor # convenient alias 

634 self._self_reading_future = None 

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

636 proactor.set_loop(self) 

637 self._make_self_pipe() 

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

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

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

641 

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

643 extra=None, server=None): 

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

645 extra, server) 

646 

647 def _make_ssl_transport( 

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

649 *, server_side=False, server_hostname=None, 

650 extra=None, server=None, 

651 ssl_handshake_timeout=None, 

652 ssl_shutdown_timeout=None): 

653 ssl_protocol = sslproto.SSLProtocol( 

654 self, protocol, sslcontext, waiter, 

655 server_side, server_hostname, 

656 ssl_handshake_timeout=ssl_handshake_timeout, 

657 ssl_shutdown_timeout=ssl_shutdown_timeout) 

658 _ProactorSocketTransport(self, rawsock, ssl_protocol, 

659 extra=extra, server=server) 

660 return ssl_protocol._app_transport 

661 

662 def _make_datagram_transport(self, sock, protocol, 

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

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

665 waiter, extra) 

666 

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

668 extra=None): 

669 return _ProactorDuplexPipeTransport(self, 

670 sock, protocol, waiter, extra) 

671 

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

673 extra=None): 

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

675 

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

677 extra=None): 

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

679 return _ProactorWritePipeTransport(self, 

680 sock, protocol, waiter, extra) 

681 

682 def close(self): 

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

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

685 if self.is_closed(): 

686 return 

687 

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

689 signal.set_wakeup_fd(-1) 

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

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

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

693 self._stop_accept_futures() 

694 self._close_self_pipe() 

695 self._proactor.close() 

696 self._proactor = None 

697 self._selector = None 

698 

699 # Close the event loop 

700 super().close() 

701 

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

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

704 

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

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

707 

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

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

710 

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

712 if not nbytes: 

713 nbytes = len(buf) 

714 

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

716 

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

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

719 

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

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

722 

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

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

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

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

727 

728 async def sock_accept(self, sock): 

729 return await self._proactor.accept(sock) 

730 

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

732 try: 

733 fileno = file.fileno() 

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

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

736 try: 

737 fsize = os.fstat(fileno).st_size 

738 except OSError: 

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

740 blocksize = count if count else fsize 

741 if not blocksize: 

742 return 0 # empty file 

743 

744 blocksize = min(blocksize, 0xffff_ffff) 

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

746 offset = min(offset, fsize) 

747 total_sent = 0 

748 try: 

749 while True: 

750 blocksize = min(end_pos - offset, blocksize) 

751 if blocksize <= 0: 

752 return total_sent 

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

754 offset += blocksize 

755 total_sent += blocksize 

756 finally: 

757 if total_sent > 0: 

758 file.seek(offset) 

759 

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

761 resume_reading = transp.is_reading() 

762 transp.pause_reading() 

763 await transp._make_empty_waiter() 

764 try: 

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

766 fallback=False) 

767 finally: 

768 transp._reset_empty_waiter() 

769 if resume_reading: 

770 transp.resume_reading() 

771 

772 def _close_self_pipe(self): 

773 if self._self_reading_future is not None: 

774 self._self_reading_future.cancel() 

775 self._self_reading_future = None 

776 self._ssock.close() 

777 self._ssock = None