Coverage for Lib/asyncio/windows_events.py: 0%

1"""Selector and proactor event loops for Windows.""" 

2 

3import sys 

4 

5if sys.platform != 'win32': # pragma: no cover 

6 raise ImportError('win32 only') 

7 

8import _overlapped 

9import _winapi 

10import errno 

11from functools import partial 

12import math 

13import msvcrt 

14import socket 

15import struct 

16import time 

17import weakref 

18 

19from . import base_subprocess 

20from . import futures 

21from . import exceptions 

22from . import proactor_events 

23from . import selector_events 

24from . import tasks 

25from . import windows_utils 

26from .log import logger 

27 

28 

29__all__ = ( 

30 'SelectorEventLoop', 'ProactorEventLoop', 'IocpProactor', 

31 'EventLoop', 

32) 

33 

34 

35NULL = _winapi.NULL 

36INFINITE = _winapi.INFINITE 

37ERROR_CONNECTION_REFUSED = 1225 

38ERROR_CONNECTION_ABORTED = 1236 

39 

40# Initial delay in seconds for connect_pipe() before retrying to connect 

41CONNECT_PIPE_INIT_DELAY = 0.001 

42 

43# Maximum delay in seconds for connect_pipe() before retrying to connect 

44CONNECT_PIPE_MAX_DELAY = 0.100 

45 

46 

47class _OverlappedFuture(futures.Future): 

48 """Subclass of Future which represents an overlapped operation. 

49 

50 Cancelling it will immediately cancel the overlapped operation. 

51 """ 

52 

53 def __init__(self, ov, *, loop=None): 

54 super().__init__(loop=loop) 

55 if self._source_traceback: 

56 del self._source_traceback[-1] 

57 self._ov = ov 

58 

59 def _repr_info(self): 

60 info = super()._repr_info() 

61 if self._ov is not None: 

62 state = 'pending' if self._ov.pending else 'completed' 

63 info.insert(1, f'overlapped=<{state}, {self._ov.address:#x}>') 

64 return info 

65 

66 def _cancel_overlapped(self): 

67 if self._ov is None: 

68 return 

69 try: 

70 self._ov.cancel() 

71 except OSError as exc: 

72 context = { 

73 'message': 'Cancelling an overlapped future failed', 

74 'exception': exc, 

75 'future': self, 

76 } 

77 if self._source_traceback: 

78 context['source_traceback'] = self._source_traceback 

79 self._loop.call_exception_handler(context) 

80 self._ov = None 

81 

82 def cancel(self, msg=None): 

83 self._cancel_overlapped() 

84 return super().cancel(msg=msg) 

85 

86 def set_exception(self, exception): 

87 super().set_exception(exception) 

88 self._cancel_overlapped() 

89 

90 def set_result(self, result): 

91 super().set_result(result) 

92 self._ov = None 

93 

94 

95class _BaseWaitHandleFuture(futures.Future): 

96 """Subclass of Future which represents a wait handle.""" 

97 

98 def __init__(self, ov, handle, wait_handle, *, loop=None): 

99 super().__init__(loop=loop) 

100 if self._source_traceback: 

101 del self._source_traceback[-1] 

102 # Keep a reference to the Overlapped object to keep it alive until the 

103 # wait is unregistered 

104 self._ov = ov 

105 self._handle = handle 

106 self._wait_handle = wait_handle 

107 

108 # Should we call UnregisterWaitEx() if the wait completes 

109 # or is cancelled? 

110 self._registered = True 

111 

112 def _poll(self): 

113 # non-blocking wait: use a timeout of 0 millisecond 

114 return (_winapi.WaitForSingleObject(self._handle, 0) == 

115 _winapi.WAIT_OBJECT_0) 

116 

117 def _repr_info(self): 

118 info = super()._repr_info() 

119 info.append(f'handle={self._handle:#x}') 

120 if self._handle is not None: 

121 state = 'signaled' if self._poll() else 'waiting' 

122 info.append(state) 

123 if self._wait_handle is not None: 

124 info.append(f'wait_handle={self._wait_handle:#x}') 

125 return info 

126 

127 def _unregister_wait_cb(self, fut): 

128 # The wait was unregistered: it's not safe to destroy the Overlapped 

129 # object 

130 self._ov = None 

131 

132 def _unregister_wait(self): 

133 if not self._registered: 

134 return 

135 self._registered = False 

136 

137 wait_handle = self._wait_handle 

138 self._wait_handle = None 

139 try: 

140 _overlapped.UnregisterWait(wait_handle) 

141 except OSError as exc: 

142 if exc.winerror != _overlapped.ERROR_IO_PENDING: 

143 context = { 

144 'message': 'Failed to unregister the wait handle', 

145 'exception': exc, 

146 'future': self, 

147 } 

148 if self._source_traceback: 

149 context['source_traceback'] = self._source_traceback 

150 self._loop.call_exception_handler(context) 

151 return 

152 # ERROR_IO_PENDING means that the unregister is pending 

153 

154 self._unregister_wait_cb(None) 

155 

156 def cancel(self, msg=None): 

157 self._unregister_wait() 

158 return super().cancel(msg=msg) 

159 

160 def set_exception(self, exception): 

161 self._unregister_wait() 

162 super().set_exception(exception) 

163 

164 def set_result(self, result): 

165 self._unregister_wait() 

166 super().set_result(result) 

167 

168 

169class _WaitCancelFuture(_BaseWaitHandleFuture): 

170 """Subclass of Future which represents a wait for the cancellation of a 

171 _WaitHandleFuture using an event. 

172 """ 

173 

174 def __init__(self, ov, event, wait_handle, *, loop=None): 

175 super().__init__(ov, event, wait_handle, loop=loop) 

176 

177 self._done_callback = None 

178 

179 def cancel(self): 

180 raise RuntimeError("_WaitCancelFuture must not be cancelled") 

181 

182 def set_result(self, result): 

183 super().set_result(result) 

184 if self._done_callback is not None: 

185 self._done_callback(self) 

186 

187 def set_exception(self, exception): 

188 super().set_exception(exception) 

189 if self._done_callback is not None: 

190 self._done_callback(self) 

191 

192 

193class _WaitHandleFuture(_BaseWaitHandleFuture): 

194 def __init__(self, ov, handle, wait_handle, proactor, *, loop=None): 

195 super().__init__(ov, handle, wait_handle, loop=loop) 

196 self._proactor = proactor 

197 self._unregister_proactor = True 

198 self._event = _overlapped.CreateEvent(None, True, False, None) 

199 self._event_fut = None 

200 

201 def _unregister_wait_cb(self, fut): 

202 if self._event is not None: 

203 _winapi.CloseHandle(self._event) 

204 self._event = None 

205 self._event_fut = None 

206 

207 # If the wait was cancelled, the wait may never be signalled, so 

208 # it's required to unregister it. Otherwise, IocpProactor.close() will 

209 # wait forever for an event which will never come. 

210 # 

211 # If the IocpProactor already received the event, it's safe to call 

212 # _unregister() because we kept a reference to the Overlapped object 

213 # which is used as a unique key. 

214 self._proactor._unregister(self._ov) 

215 self._proactor = None 

216 

217 super()._unregister_wait_cb(fut) 

218 

219 def _unregister_wait(self): 

220 if not self._registered: 

221 return 

222 self._registered = False 

223 

224 wait_handle = self._wait_handle 

225 self._wait_handle = None 

226 try: 

227 _overlapped.UnregisterWaitEx(wait_handle, self._event) 

228 except OSError as exc: 

229 if exc.winerror != _overlapped.ERROR_IO_PENDING: 

230 context = { 

231 'message': 'Failed to unregister the wait handle', 

232 'exception': exc, 

233 'future': self, 

234 } 

235 if self._source_traceback: 

236 context['source_traceback'] = self._source_traceback 

237 self._loop.call_exception_handler(context) 

238 return 

239 # ERROR_IO_PENDING is not an error, the wait was unregistered 

240 

241 self._event_fut = self._proactor._wait_cancel(self._event, 

242 self._unregister_wait_cb) 

243 

244 

245class PipeServer(object): 

246 """Class representing a pipe server. 

247 

248 This is much like a bound, listening socket. 

249 """ 

250 def __init__(self, address): 

251 self._address = address 

252 self._free_instances = weakref.WeakSet() 

253 # initialize the pipe attribute before calling _server_pipe_handle() 

254 # because this function can raise an exception and the destructor calls 

255 # the close() method 

256 self._pipe = None 

257 self._accept_pipe_future = None 

258 self._pipe = self._server_pipe_handle(True) 

259 

260 def _get_unconnected_pipe(self): 

261 # Create new instance and return previous one. This ensures 

262 # that (until the server is closed) there is always at least 

263 # one pipe handle for address. Therefore if a client attempt 

264 # to connect it will not fail with FileNotFoundError. 

265 tmp, self._pipe = self._pipe, self._server_pipe_handle(False) 

266 return tmp 

267 

268 def _server_pipe_handle(self, first): 

269 # Return a wrapper for a new pipe handle. 

270 if self.closed(): 

271 return None 

272 flags = _winapi.PIPE_ACCESS_DUPLEX | _winapi.FILE_FLAG_OVERLAPPED 

273 if first: 

274 flags |= _winapi.FILE_FLAG_FIRST_PIPE_INSTANCE 

275 h = _winapi.CreateNamedPipe( 

276 self._address, flags, 

277 _winapi.PIPE_TYPE_MESSAGE | _winapi.PIPE_READMODE_MESSAGE | 

278 _winapi.PIPE_WAIT, 

279 _winapi.PIPE_UNLIMITED_INSTANCES, 

280 windows_utils.BUFSIZE, windows_utils.BUFSIZE, 

281 _winapi.NMPWAIT_WAIT_FOREVER, _winapi.NULL) 

282 pipe = windows_utils.PipeHandle(h) 

283 self._free_instances.add(pipe) 

284 return pipe 

285 

286 def closed(self): 

287 return (self._address is None) 

288 

289 def close(self): 

290 if self._accept_pipe_future is not None: 

291 self._accept_pipe_future.cancel() 

292 self._accept_pipe_future = None 

293 # Close all instances which have not been connected to by a client. 

294 if self._address is not None: 

295 for pipe in self._free_instances: 

296 pipe.close() 

297 self._pipe = None 

298 self._address = None 

299 self._free_instances.clear() 

300 

301 __del__ = close 

302 

303 

304class _WindowsSelectorEventLoop(selector_events.BaseSelectorEventLoop): 

305 """Windows version of selector event loop.""" 

306 

307 

308class ProactorEventLoop(proactor_events.BaseProactorEventLoop): 

309 """Windows version of proactor event loop using IOCP.""" 

310 

311 def __init__(self, proactor=None): 

312 if proactor is None: 

313 proactor = IocpProactor() 

314 super().__init__(proactor) 

315 

316 def _run_forever_setup(self): 

317 assert self._self_reading_future is None 

318 self.call_soon(self._loop_self_reading) 

319 super()._run_forever_setup() 

320 

321 def _run_forever_cleanup(self): 

322 super()._run_forever_cleanup() 

323 if self._self_reading_future is not None: 

324 ov = self._self_reading_future._ov 

325 self._self_reading_future.cancel() 

326 # self_reading_future always uses IOCP, so even though it's 

327 # been cancelled, we need to make sure that the IOCP message 

328 # is received so that the kernel is not holding on to the 

329 # memory, possibly causing memory corruption later. Only 

330 # unregister it if IO is complete in all respects. Otherwise 

331 # we need another _poll() later to complete the IO. 

332 if ov is not None and not ov.pending: 

333 self._proactor._unregister(ov) 

334 self._self_reading_future = None 

335 

336 async def create_pipe_connection(self, protocol_factory, address): 

337 f = self._proactor.connect_pipe(address) 

338 pipe = await f 

339 protocol = protocol_factory() 

340 trans = self._make_duplex_pipe_transport(pipe, protocol, 

341 extra={'addr': address}) 

342 return trans, protocol 

343 

344 async def start_serving_pipe(self, protocol_factory, address): 

345 server = PipeServer(address) 

346 

347 def loop_accept_pipe(f=None): 

348 pipe = None 

349 try: 

350 if f: 

351 pipe = f.result() 

352 server._free_instances.discard(pipe) 

353 

354 if server.closed(): 

355 # A client connected before the server was closed: 

356 # drop the client (close the pipe) and exit 

357 pipe.close() 

358 return 

359 

360 protocol = protocol_factory() 

361 self._make_duplex_pipe_transport( 

362 pipe, protocol, extra={'addr': address}) 

363 

364 pipe = server._get_unconnected_pipe() 

365 if pipe is None: 

366 return 

367 

368 f = self._proactor.accept_pipe(pipe) 

369 except BrokenPipeError: 

370 if pipe and pipe.fileno() != -1: 

371 pipe.close() 

372 self.call_soon(loop_accept_pipe) 

373 except OSError as exc: 

374 if pipe and pipe.fileno() != -1: 

375 self.call_exception_handler({ 

376 'message': 'Pipe accept failed', 

377 'exception': exc, 

378 'pipe': pipe, 

379 }) 

380 pipe.close() 

381 elif self._debug: 

382 logger.warning("Accept pipe failed on pipe %r", 

383 pipe, exc_info=True) 

384 self.call_soon(loop_accept_pipe) 

385 except exceptions.CancelledError: 

386 if pipe: 

387 pipe.close() 

388 else: 

389 server._accept_pipe_future = f 

390 f.add_done_callback(loop_accept_pipe) 

391 

392 self.call_soon(loop_accept_pipe) 

393 return [server] 

394 

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

396 stdin, stdout, stderr, bufsize, 

397 extra=None, **kwargs): 

398 waiter = self.create_future() 

399 transp = _WindowsSubprocessTransport(self, protocol, args, shell, 

400 stdin, stdout, stderr, bufsize, 

401 waiter=waiter, extra=extra, 

402 **kwargs) 

403 try: 

404 await waiter 

405 except (SystemExit, KeyboardInterrupt): 

406 raise 

407 except BaseException: 

408 transp.close() 

409 await tasks.shield(transp._wait()) 

410 raise 

411 

412 return transp 

413 

414 

415class IocpProactor: 

416 """Proactor implementation using IOCP.""" 

417 

418 def __init__(self, concurrency=INFINITE): 

419 self._loop = None 

420 self._results = [] 

421 self._iocp = _overlapped.CreateIoCompletionPort( 

422 _overlapped.INVALID_HANDLE_VALUE, NULL, 0, concurrency) 

423 self._cache = {} 

424 self._registered = weakref.WeakSet() 

425 self._unregistered = [] 

426 self._stopped_serving = weakref.WeakSet() 

427 

428 def _check_closed(self): 

429 if self._iocp is None: 

430 raise RuntimeError('IocpProactor is closed') 

431 

432 def __repr__(self): 

433 info = ['overlapped#=%s' % len(self._cache), 

434 'result#=%s' % len(self._results)] 

435 if self._iocp is None: 

436 info.append('closed') 

437 return '<%s %s>' % (self.__class__.__name__, " ".join(info)) 

438 

439 def set_loop(self, loop): 

440 self._loop = loop 

441 

442 def select(self, timeout=None): 

443 if not self._results: 

444 self._poll(timeout) 

445 tmp = self._results 

446 self._results = [] 

447 try: 

448 return tmp 

449 finally: 

450 # Needed to break cycles when an exception occurs. 

451 tmp = None 

452 

453 def _result(self, value): 

454 fut = self._loop.create_future() 

455 fut.set_result(value) 

456 return fut 

457 

458 @staticmethod 

459 def finish_socket_func(trans, key, ov): 

460 try: 

461 return ov.getresult() 

462 except OSError as exc: 

463 if exc.winerror in (_overlapped.ERROR_NETNAME_DELETED, 

464 _overlapped.ERROR_OPERATION_ABORTED): 

465 raise ConnectionResetError(*exc.args) 

466 else: 

467 raise 

468 

469 @classmethod 

470 def _finish_recvfrom(cls, trans, key, ov, *, empty_result): 

471 try: 

472 return cls.finish_socket_func(trans, key, ov) 

473 except OSError as exc: 

474 # WSARecvFrom will report ERROR_PORT_UNREACHABLE when the same 

475 # socket is used to send to an address that is not listening. 

476 if exc.winerror == _overlapped.ERROR_PORT_UNREACHABLE: 

477 return empty_result, None 

478 else: 

479 raise 

480 

481 def recv(self, conn, nbytes, flags=0): 

482 self._register_with_iocp(conn) 

483 ov = _overlapped.Overlapped(NULL) 

484 try: 

485 if isinstance(conn, socket.socket): 

486 ov.WSARecv(conn.fileno(), nbytes, flags) 

487 else: 

488 ov.ReadFile(conn.fileno(), nbytes) 

489 except BrokenPipeError: 

490 return self._result(b'') 

491 

492 return self._register(ov, conn, self.finish_socket_func) 

493 

494 def recv_into(self, conn, buf, flags=0): 

495 self._register_with_iocp(conn) 

496 ov = _overlapped.Overlapped(NULL) 

497 try: 

498 if isinstance(conn, socket.socket): 

499 ov.WSARecvInto(conn.fileno(), buf, flags) 

500 else: 

501 ov.ReadFileInto(conn.fileno(), buf) 

502 except BrokenPipeError: 

503 return self._result(0) 

504 

505 return self._register(ov, conn, self.finish_socket_func) 

506 

507 def recvfrom(self, conn, nbytes, flags=0): 

508 self._register_with_iocp(conn) 

509 ov = _overlapped.Overlapped(NULL) 

510 try: 

511 ov.WSARecvFrom(conn.fileno(), nbytes, flags) 

512 except BrokenPipeError: 

513 return self._result((b'', None)) 

514 

515 return self._register(ov, conn, partial(self._finish_recvfrom, 

516 empty_result=b'')) 

517 

518 def recvfrom_into(self, conn, buf, flags=0): 

519 self._register_with_iocp(conn) 

520 ov = _overlapped.Overlapped(NULL) 

521 try: 

522 ov.WSARecvFromInto(conn.fileno(), buf, flags) 

523 except BrokenPipeError: 

524 return self._result((0, None)) 

525 

526 return self._register(ov, conn, partial(self._finish_recvfrom, 

527 empty_result=0)) 

528 

529 def sendto(self, conn, buf, flags=0, addr=None): 

530 self._register_with_iocp(conn) 

531 ov = _overlapped.Overlapped(NULL) 

532 

533 ov.WSASendTo(conn.fileno(), buf, flags, addr) 

534 

535 return self._register(ov, conn, self.finish_socket_func) 

536 

537 def send(self, conn, buf, flags=0): 

538 self._register_with_iocp(conn) 

539 ov = _overlapped.Overlapped(NULL) 

540 if isinstance(conn, socket.socket): 

541 ov.WSASend(conn.fileno(), buf, flags) 

542 else: 

543 ov.WriteFile(conn.fileno(), buf) 

544 

545 return self._register(ov, conn, self.finish_socket_func) 

546 

547 def accept(self, listener): 

548 self._register_with_iocp(listener) 

549 conn = self._get_accept_socket(listener.family) 

550 ov = _overlapped.Overlapped(NULL) 

551 ov.AcceptEx(listener.fileno(), conn.fileno()) 

552 

553 def finish_accept(trans, key, ov): 

554 ov.getresult() 

555 # Use SO_UPDATE_ACCEPT_CONTEXT so getsockname() etc work. 

556 buf = struct.pack('@P', listener.fileno()) 

557 conn.setsockopt(socket.SOL_SOCKET, 

558 _overlapped.SO_UPDATE_ACCEPT_CONTEXT, buf) 

559 conn.settimeout(listener.gettimeout()) 

560 return conn, conn.getpeername() 

561 

562 async def accept_coro(future, conn): 

563 # Coroutine closing the accept socket if the future is cancelled 

564 try: 

565 await future 

566 except exceptions.CancelledError: 

567 conn.close() 

568 raise 

569 

570 future = self._register(ov, listener, finish_accept) 

571 coro = accept_coro(future, conn) 

572 tasks.ensure_future(coro, loop=self._loop) 

573 return future 

574 

575 def connect(self, conn, address): 

576 if conn.type == socket.SOCK_DGRAM: 

577 # WSAConnect will complete immediately for UDP sockets so we don't 

578 # need to register any IOCP operation 

579 _overlapped.WSAConnect(conn.fileno(), address) 

580 fut = self._loop.create_future() 

581 fut.set_result(None) 

582 return fut 

583 

584 self._register_with_iocp(conn) 

585 # The socket needs to be locally bound before we call ConnectEx(). 

586 try: 

587 _overlapped.BindLocal(conn.fileno(), conn.family) 

588 except OSError as e: 

589 if e.winerror != errno.WSAEINVAL: 

590 raise 

591 # Probably already locally bound; check using getsockname(). 

592 if conn.getsockname()[1] == 0: 

593 raise 

594 ov = _overlapped.Overlapped(NULL) 

595 ov.ConnectEx(conn.fileno(), address) 

596 

597 def finish_connect(trans, key, ov): 

598 ov.getresult() 

599 # Use SO_UPDATE_CONNECT_CONTEXT so getsockname() etc work. 

600 conn.setsockopt(socket.SOL_SOCKET, 

601 _overlapped.SO_UPDATE_CONNECT_CONTEXT, 0) 

602 return conn 

603 

604 return self._register(ov, conn, finish_connect) 

605 

606 def sendfile(self, sock, file, offset, count): 

607 self._register_with_iocp(sock) 

608 ov = _overlapped.Overlapped(NULL) 

609 offset_low = offset & 0xffff_ffff 

610 offset_high = (offset >> 32) & 0xffff_ffff 

611 # TransmitFile ignores OVERLAPPED.Offset for handles not opened with 

612 # FILE_FLAG_OVERLAPPED, so seek the CRT file pointer to match. 

613 file.seek(offset) 

614 ov.TransmitFile(sock.fileno(), 

615 msvcrt.get_osfhandle(file.fileno()), 

616 offset_low, offset_high, 

617 count, 0, 0) 

618 

619 return self._register(ov, sock, self.finish_socket_func) 

620 

621 def accept_pipe(self, pipe): 

622 self._register_with_iocp(pipe) 

623 ov = _overlapped.Overlapped(NULL) 

624 connected = ov.ConnectNamedPipe(pipe.fileno()) 

625 

626 if connected: 

627 # ConnectNamePipe() failed with ERROR_PIPE_CONNECTED which means 

628 # that the pipe is connected. There is no need to wait for the 

629 # completion of the connection. 

630 return self._result(pipe) 

631 

632 def finish_accept_pipe(trans, key, ov): 

633 ov.getresult() 

634 return pipe 

635 

636 return self._register(ov, pipe, finish_accept_pipe) 

637 

638 async def connect_pipe(self, address): 

639 delay = CONNECT_PIPE_INIT_DELAY 

640 while True: 

641 # Unfortunately there is no way to do an overlapped connect to 

642 # a pipe. Call CreateFile() in a loop until it doesn't fail with 

643 # ERROR_PIPE_BUSY. 

644 try: 

645 handle = _overlapped.ConnectPipe(address) 

646 break 

647 except OSError as exc: 

648 if exc.winerror != _overlapped.ERROR_PIPE_BUSY: 

649 raise 

650 

651 # ConnectPipe() failed with ERROR_PIPE_BUSY: retry later 

652 delay = min(delay * 2, CONNECT_PIPE_MAX_DELAY) 

653 await tasks.sleep(delay) 

654 

655 return windows_utils.PipeHandle(handle) 

656 

657 def wait_for_handle(self, handle, timeout=None): 

658 """Wait for a handle. 

659 

660 Return a Future object. The result of the future is True if the wait 

661 completed, or False if the wait did not complete (on timeout). 

662 """ 

663 return self._wait_for_handle(handle, timeout, False) 

664 

665 def _wait_cancel(self, event, done_callback): 

666 fut = self._wait_for_handle(event, None, True) 

667 # add_done_callback() cannot be used because the wait may only complete 

668 # in IocpProactor.close(), while the event loop is not running. 

669 fut._done_callback = done_callback 

670 return fut 

671 

672 def _wait_for_handle(self, handle, timeout, _is_cancel): 

673 self._check_closed() 

674 

675 if timeout is None: 

676 ms = _winapi.INFINITE 

677 else: 

678 # RegisterWaitForSingleObject() has a resolution of 1 millisecond, 

679 # round away from zero to wait *at least* timeout seconds. 

680 ms = math.ceil(timeout * 1e3) 

681 

682 # We only create ov so we can use ov.address as a key for the cache. 

683 ov = _overlapped.Overlapped(NULL) 

684 wait_handle = _overlapped.RegisterWaitWithQueue( 

685 handle, self._iocp, ov.address, ms) 

686 if _is_cancel: 

687 f = _WaitCancelFuture(ov, handle, wait_handle, loop=self._loop) 

688 else: 

689 f = _WaitHandleFuture(ov, handle, wait_handle, self, 

690 loop=self._loop) 

691 if f._source_traceback: 

692 del f._source_traceback[-1] 

693 

694 def finish_wait_for_handle(trans, key, ov): 

695 # Note that this second wait means that we should only use 

696 # this with handles types where a successful wait has no 

697 # effect. So events or processes are all right, but locks 

698 # or semaphores are not. Also note if the handle is 

699 # signalled and then quickly reset, then we may return 

700 # False even though we have not timed out. 

701 return f._poll() 

702 

703 self._cache[ov.address] = (f, ov, 0, finish_wait_for_handle) 

704 return f 

705 

706 def _register_with_iocp(self, obj): 

707 # To get notifications of finished ops on this objects sent to the 

708 # completion port, were must register the handle. 

709 if obj not in self._registered: 

710 self._registered.add(obj) 

711 _overlapped.CreateIoCompletionPort(obj.fileno(), self._iocp, 0, 0) 

712 # XXX We could also use SetFileCompletionNotificationModes() 

713 # to avoid sending notifications to completion port of ops 

714 # that succeed immediately. 

715 

716 def _register(self, ov, obj, callback): 

717 self._check_closed() 

718 

719 # Return a future which will be set with the result of the 

720 # operation when it completes. The future's value is actually 

721 # the value returned by callback(). 

722 f = _OverlappedFuture(ov, loop=self._loop) 

723 if f._source_traceback: 

724 del f._source_traceback[-1] 

725 if not ov.pending: 

726 # The operation has completed, so no need to postpone the 

727 # work. We cannot take this short cut if we need the 

728 # NumberOfBytes, CompletionKey values returned by 

729 # PostQueuedCompletionStatus(). 

730 try: 

731 value = callback(None, None, ov) 

732 except OSError as e: 

733 f.set_exception(e) 

734 else: 

735 f.set_result(value) 

736 # Even if GetOverlappedResult() was called, we have to wait for the 

737 # notification of the completion in GetQueuedCompletionStatus(). 

738 # Register the overlapped operation to keep a reference to the 

739 # OVERLAPPED object, otherwise the memory is freed and Windows may 

740 # read uninitialized memory. 

741 

742 # Register the overlapped operation for later. Note that 

743 # we only store obj to prevent it from being garbage 

744 # collected too early. 

745 self._cache[ov.address] = (f, ov, obj, callback) 

746 return f 

747 

748 def _unregister(self, ov): 

749 """Unregister an overlapped object. 

750 

751 Call this method when its future has been cancelled. The event can 

752 already be signalled (pending in the proactor event queue). It is also 

753 safe if the event is never signalled (because it was cancelled). 

754 """ 

755 self._check_closed() 

756 self._unregistered.append(ov) 

757 

758 def _get_accept_socket(self, family): 

759 s = socket.socket(family) 

760 s.settimeout(0) 

761 return s 

762 

763 def _process_completion_status(self, status): 

764 """Process a single status from the completion port. 

765 

766 A caller that waits on the completion port itself can pass each 

767 status it receives here. 

768 """ 

769 err, transferred, key, address = status 

770 try: 

771 f, ov, obj, callback = self._cache.pop(address) 

772 except KeyError: 

773 if self._loop.get_debug(): 

774 self._loop.call_exception_handler({ 

775 'message': ('GetQueuedCompletionStatus() returned an ' 

776 'unexpected event'), 

777 'status': ('err=%s transferred=%s key=%#x address=%#x' 

778 % (err, transferred, key, address)), 

779 }) 

780 

781 # key is either zero, or it is used to return a pipe 

782 # handle which should be closed to avoid a leak. 

783 if key not in (0, _overlapped.INVALID_HANDLE_VALUE): 

784 _winapi.CloseHandle(key) 

785 return 

786 

787 if obj in self._stopped_serving: 

788 f.cancel() 

789 # Don't call the callback if _register() already read the result or 

790 # if the overlapped has been cancelled 

791 elif not f.done(): 

792 try: 

793 value = callback(transferred, key, ov) 

794 except OSError as e: 

795 f.set_exception(e) 

796 self._results.append(f) 

797 else: 

798 f.set_result(value) 

799 self._results.append(f) 

800 finally: 

801 f = None 

802 

803 def _poll(self, timeout=None): 

804 if timeout is None: 

805 ms = INFINITE 

806 elif timeout < 0: 

807 raise ValueError("negative timeout") 

808 else: 

809 # GetQueuedCompletionStatus() has a resolution of 1 millisecond, 

810 # round away from zero to wait *at least* timeout seconds. 

811 ms = math.ceil(timeout * 1e3) 

812 if ms >= INFINITE: 

813 raise ValueError("timeout too big") 

814 

815 while True: 

816 status = _overlapped.GetQueuedCompletionStatus(self._iocp, ms)