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 if not self._buffer and self._write_fut is None: 

109 # Nothing left to flush: no more data will be sent. 

110 self._conn_lost += 1 

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

112 if self._read_fut is not None: 

113 self._read_fut.cancel() 

114 self._read_fut = None 

115 

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

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

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

119 self._sock.close() 

120 

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

122 try: 

123 if isinstance(exc, OSError): 

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

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

126 else: 

127 self._loop.call_exception_handler({ 

128 'message': message, 

129 'exception': exc, 

130 'transport': self, 

131 'protocol': self._protocol, 

132 }) 

133 finally: 

134 self._force_close(exc) 

135 

136 def _force_close(self, exc): 

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

138 if exc is None: 

139 self._empty_waiter.set_result(None) 

140 else: 

141 self._empty_waiter.set_exception(exc) 

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

143 return 

144 self._closing = True 

145 self._conn_lost += 1 

146 if self._write_fut: 

147 self._write_fut.cancel() 

148 self._write_fut = None 

149 if self._read_fut: 

150 self._read_fut.cancel() 

151 self._read_fut = None 

152 self._pending_write = 0 

153 self._buffer = None 

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

155 

156 def _call_connection_lost(self, exc): 

157 if self._called_connection_lost: 

158 return 

159 try: 

160 self._protocol.connection_lost(exc) 

161 finally: 

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

163 # end then it may fail with ERROR_NETNAME_DELETED if we 

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

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

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

167 self._sock.shutdown(socket.SHUT_RDWR) 

168 self._sock.close() 

169 self._sock = None 

170 server = self._server 

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

172 server._detach(self) 

173 self._server = None 

174 self._called_connection_lost = True 

175 

176 def get_write_buffer_size(self): 

177 size = self._pending_write 

178 if self._buffer is not None: 

179 size += len(self._buffer) 

180 return size 

181 

182 

183class _ProactorReadPipeTransport(_ProactorBasePipeTransport, 

184 transports.ReadTransport): 

185 """Transport for read pipes.""" 

186 

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

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

189 self._pending_data_length = -1 

190 self._paused = True 

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

192 

193 self._data = bytearray(buffer_size) 

194 self._loop.call_soon(self._loop_reading) 

195 self._paused = False 

196 

197 def is_reading(self): 

198 return not self._paused and not self._closing 

199 

200 def pause_reading(self): 

201 if self._closing or self._paused: 

202 return 

203 self._paused = True 

204 

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

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

207 # implementation. 

208 # 

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

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

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

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

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

214 # WSARecv(). 

215 

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

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

218 

219 def resume_reading(self): 

220 if self._closing or not self._paused: 

221 return 

222 

223 self._paused = False 

224 if self._read_fut is None: 

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

226 

227 length = self._pending_data_length 

228 self._pending_data_length = -1 

229 if length > -1: 

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

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

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

233 

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

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

236 

237 def _eof_received(self): 

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

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

240 

241 try: 

242 keep_open = self._protocol.eof_received() 

243 except (SystemExit, KeyboardInterrupt): 

244 raise 

245 except BaseException as exc: 

246 self._fatal_error( 

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

248 return 

249 

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

251 self.close() 

252 

253 def _data_received(self, data, length): 

254 if self._paused: 

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

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

257 assert self._pending_data_length == -1 

258 self._pending_data_length = length 

259 return 

260 

261 if length == 0: 

262 self._eof_received() 

263 return 

264 

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

266 try: 

267 protocols._feed_data_to_buffered_proto(self._protocol, data) 

268 except (SystemExit, KeyboardInterrupt): 

269 raise 

270 except BaseException as exc: 

271 self._fatal_error(exc, 

272 'Fatal error: protocol.buffer_updated() ' 

273 'call failed.') 

274 return 

275 else: 

276 self._protocol.data_received(data) 

277 

278 def _loop_reading(self, fut=None): 

279 length = -1 

280 data = None 

281 try: 

282 if fut is not None: 

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

284 self._closing) 

285 self._read_fut = None 

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

287 # deliver data later in "finally" clause 

288 length = fut.result() 

289 if length == 0: 

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

291 return 

292 

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

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

295 else: 

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

297 fut.cancel() 

298 

299 if self._closing: 

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

301 return 

302 

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

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

305 

306 if not self._paused: 

307 # reschedule a new read 

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

309 except ConnectionAbortedError as exc: 

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

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

312 elif self._loop.get_debug(): 

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

314 exc_info=True) 

315 except ConnectionResetError as exc: 

316 self._force_close(exc) 

317 except OSError as exc: 

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

319 except exceptions.CancelledError: 

320 if not self._closing: 

321 raise 

322 else: 

323 if not self._paused: 

324 self._read_fut.add_done_callback(self._loop_reading) 

325 finally: 

326 if length > -1: 

327 self._data_received(data, length) 

328 

329 

330class _ProactorBaseWritePipeTransport(_ProactorBasePipeTransport, 

331 transports.WriteTransport): 

332 """Transport for write pipes.""" 

333 

334 _start_tls_compatible = True 

335 

336 def write(self, data): 

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

338 raise TypeError( 

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

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

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

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

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

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

345 

346 if not data: 

347 return 

348 

349 if self._conn_lost: 

350 if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES: 

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

352 self._conn_lost += 1 

353 return 

354 

355 # Observable states: 

356 # 1. IDLE: _write_fut and _buffer both None 

357 # 2. WRITING: _write_fut set; _buffer None 

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

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

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

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

362 assert self._buffer is None 

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

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

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

366 # Make a mutable copy which we can extend. 

367 self._buffer = bytearray(data) 

368 self._maybe_pause_protocol() 

369 else: # BACKED UP 

370 # Append to buffer (also copies). 

371 self._buffer.extend(data) 

372 self._maybe_pause_protocol() 

373 

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

375 try: 

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

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

378 # it has set self._write_fut to None. 

379 return 

380 assert f is self._write_fut 

381 self._write_fut = None 

382 self._pending_write = 0 

383 if f: 

384 f.result() 

385 if data is None: 

386 data = self._buffer 

387 self._buffer = None 

388 if not data: 

389 if self._closing: 

390 self._conn_lost += 1 

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

392 if self._eof_written: 

393 self._sock.shutdown(socket.SHUT_WR) 

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

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

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

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

398 # protocol to be paused again). 

399 self._maybe_resume_protocol() 

400 else: 

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

402 if not self._write_fut.done(): 

403 assert self._pending_write == 0 

404 self._pending_write = len(data) 

405 self._write_fut.add_done_callback(self._loop_writing) 

406 self._maybe_pause_protocol() 

407 else: 

408 self._write_fut.add_done_callback(self._loop_writing) 

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

410 self._empty_waiter.set_result(None) 

411 except ConnectionResetError as exc: 

412 self._force_close(exc) 

413 except OSError as exc: 

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

415 

416 def can_write_eof(self): 

417 return True 

418 

419 def write_eof(self): 

420 self.close() 

421 

422 def abort(self): 

423 self._force_close(None) 

424 

425 def _make_empty_waiter(self): 

426 if self._empty_waiter is not None: 

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

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

429 if self._write_fut is None: 

430 self._empty_waiter.set_result(None) 

431 return self._empty_waiter 

432 

433 def _reset_empty_waiter(self): 

434 self._empty_waiter = None 

435 

436 

437class _ProactorWritePipeTransport(_ProactorBaseWritePipeTransport): 

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

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

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

441 self._read_fut.add_done_callback(self._pipe_closed) 

442 

443 def _pipe_closed(self, fut): 

444 if fut.cancelled(): 

445 # the transport has been closed 

446 return 

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

448 if self._closing: 

449 assert self._read_fut is None 

450 return 

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

452 self._read_fut = None 

453 if self._write_fut is not None: 

454 self._force_close(BrokenPipeError()) 

455 else: 

456 self.close() 

457 

458 

459class _ProactorDatagramTransport(_ProactorBasePipeTransport, 

460 transports.DatagramTransport): 

461 max_size = 256 * 1024 

462 _header_size = 8 

463 

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

465 waiter=None, extra=None): 

466 self._address = address 

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 _force_close(self, exc): 

486 # The base class drops the buffer; the size is tracked separately. 

487 self._buffer_size = 0 

488 super()._force_close(exc) 

489 

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

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

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

493 type(data)) 

494 

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

496 raise ValueError( 

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

498 

499 if self._conn_lost and self._address: 

500 if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES: 

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

502 self._conn_lost += 1 

503 return 

504 

505 # Ensure that what we buffer is immutable. 

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

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

508 

509 if self._write_fut is None: 

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

511 self._loop_writing() 

512 # else: A write operation is already kicked off 

513 

514 self._maybe_pause_protocol() 

515 

516 def _loop_writing(self, fut=None): 

517 try: 

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

519 # No more data will be sent: either everything buffered has 

520 # already been flushed, or _force_close() dropped it. 

521 return 

522 

523 assert fut is self._write_fut 

524 self._write_fut = None 

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

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

527 fut.result() 

528 

529 if not self._buffer: 

530 # Everything buffered has been sent 

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

532 self._conn_lost += 1 

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

534 return 

535 

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

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

538 if self._address is not None: 

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

540 data) 

541 else: 

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

543 data, 

544 addr=addr) 

545 except OSError as exc: 

546 self._protocol.error_received(exc) 

547 # error_received() is arbitrary protocol code: it may have sent 

548 # (scheduling a write of its own, directly or via call_soon()), 

549 # closed, or aborted the transport. 

550 if self._buffer or self._closing: 550 ↛ 555line 550 didn't jump to line 555 because the condition on line 550 was never true

551 # Either data is still queued, or a close() is waiting on 

552 # the write loop to drain it and call connection_lost(). 

553 # This write failed, so there is no completion callback 

554 # pending to re-enter the loop -- schedule one (gh-156698). 

555 def write_next(): 

556 # error_received() may have scheduled a write of its own, 

557 # directly or with call_soon(); its completion callback 

558 # will drain the rest of the buffer. 

559 if self._write_fut is None: 

560 self._loop_writing() 

561 

562 self._loop.call_soon(write_next) 

563 else: 

564 # Nothing left to write, so a paused protocol has to be 

565 # resumed here: the next entry into _loop_writing() returns 

566 # early on an empty buffer without doing it. 

567 self._maybe_resume_protocol() 

568 except Exception as exc: 

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

570 else: 

571 self._write_fut.add_done_callback(self._loop_writing) 

572 self._maybe_resume_protocol() 

573 

574 def _loop_reading(self, fut=None): 

575 data = None 

576 try: 

577 if self._closing: 

578 return 

579 

580 assert self._read_fut is fut 

581 

582 self._read_fut = None 

583 if fut is not None: 

584 res = fut.result() 

585 

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

587 data, addr = res, self._address 

588 else: 

589 data, addr = res 

590 

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

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

593 self.max_size) 

594 else: 

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

596 self.max_size) 

597 except ConnectionResetError as exc: 

598 # WSARecvFrom() reports a stale ICMP port unreachable 

599 # notification as a synchronous ConnectionResetError when the 

600 # same socket was used to send to an address that is not 

601 # listening. This is transient, so reschedule the read loop 

602 # instead of leaving it dead. 

603 self._protocol.error_received(exc) 

604 if not self._closing: 

605 self._loop.call_soon(self._loop_reading) 

606 except OSError as exc: 

607 self._protocol.error_received(exc) 

608 except exceptions.CancelledError: 

609 if not self._closing: 

610 raise 

611 else: 

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

613 self._read_fut.add_done_callback(self._loop_reading) 

614 finally: 

615 if data: 

616 self._protocol.datagram_received(data, addr) 

617 

618 

619class _ProactorDuplexPipeTransport(_ProactorReadPipeTransport, 

620 _ProactorBaseWritePipeTransport, 

621 transports.Transport): 

622 """Transport for duplex pipes.""" 

623 

624 def can_write_eof(self): 

625 return False 

626 

627 def write_eof(self): 

628 raise NotImplementedError 

629 

630 

631class _ProactorSocketTransport(_ProactorReadPipeTransport, 

632 _ProactorBaseWritePipeTransport, 

633 transports.Transport): 

634 """Transport for connected sockets.""" 

635 

636 _sendfile_compatible = constants._SendfileMode.TRY_NATIVE 

637 

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

639 extra=None, server=None): 

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

641 base_events._set_nodelay(sock) 

642 

643 def _set_extra(self, sock): 

644 _set_socket_extra(self, sock) 

645 

646 def can_write_eof(self): 

647 return True 

648 

649 def write_eof(self): 

650 if self._closing or self._eof_written: 

651 return 

652 self._eof_written = True 

653 if self._write_fut is None: 

654 self._sock.shutdown(socket.SHUT_WR) 

655 

656 

657class BaseProactorEventLoop(base_events.BaseEventLoop): 

658 

659 def __init__(self, proactor): 

660 super().__init__() 

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

662 self._proactor = proactor 

663 self._selector = proactor # convenient alias 

664 self._self_reading_future = None 

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

666 proactor.set_loop(self) 

667 self._make_self_pipe() 

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

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

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

671 

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

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

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

675 extra, server) 

676 

677 def _make_ssl_transport( 

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

679 *, server_side=False, server_hostname=None, 

680 extra=None, server=None, 

681 ssl_handshake_timeout=None, 

682 ssl_shutdown_timeout=None, context=None): 

683 ssl_protocol = sslproto.SSLProtocol( 

684 self, protocol, sslcontext, waiter, 

685 server_side, server_hostname, 

686 ssl_handshake_timeout=ssl_handshake_timeout, 

687 ssl_shutdown_timeout=ssl_shutdown_timeout) 

688 _ProactorSocketTransport(self, rawsock, ssl_protocol, 

689 extra=extra, server=server) 

690 return ssl_protocol._app_transport 

691 

692 def _make_datagram_transport(self, sock, protocol, 

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

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

695 waiter, extra) 

696 

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

698 extra=None): 

699 return _ProactorDuplexPipeTransport(self, 

700 sock, protocol, waiter, extra) 

701 

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

703 extra=None): 

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

705 

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

707 extra=None): 

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

709 return _ProactorWritePipeTransport(self, 

710 sock, protocol, waiter, extra) 

711 

712 def close(self): 

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

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

715 if self.is_closed(): 

716 return 

717 

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

719 signal.set_wakeup_fd(-1) 

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

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

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

723 self._stop_accept_futures() 

724 self._close_self_pipe() 

725 self._proactor.close() 

726 self._proactor = None 

727 self._selector = None 

728 

729 # Close the event loop 

730 super().close() 

731 

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

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

734 

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

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

737 

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

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

740 

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

742 if not nbytes: 

743 nbytes = len(buf) 

744 

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

746 

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

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

749 

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

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

752 

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

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

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

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

757 

758 async def sock_accept(self, sock): 

759 return await self._proactor.accept(sock) 

760 

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

762 try: 

763 fileno = file.fileno() 

764 except (AttributeError, io.UnsupportedOperation): 

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

766 try: 

767 fsize = os.fstat(fileno).st_size 

768 except OSError: 

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

770 blocksize = count if count else fsize 

771 if not blocksize: 

772 return 0 # empty file 

773 

774 blocksize = min(blocksize, 0xffff_ffff) 

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

776 offset = min(offset, fsize) 

777 total_sent = 0 

778 try: 

779 while True: 

780 blocksize = min(end_pos - offset, blocksize) 

781 if blocksize <= 0: 

782 return total_sent 

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

784 offset += blocksize 

785 total_sent += blocksize 

786 finally: 

787 file.seek(offset)