Coverage for Lib/asyncio/unix_events.py: 87%

1"""Selector event loop for Unix with signal handling.""" 

2 

3import errno 

4import io 

5import itertools 

6import os 

7import selectors 

8import signal 

9import socket 

10import stat 

11import subprocess 

12import sys 

13import threading 

14import warnings 

15import inspect 

16 

17from . import base_events 

18from . import base_subprocess 

19from . import constants 

20from . import coroutines 

21from . import events 

22from . import exceptions 

23from . import futures 

24from . import selector_events 

25from . import tasks 

26from . import transports 

27from .log import logger 

28 

29 

30__all__ = ( 

31 'SelectorEventLoop', 

32 'EventLoop', 

33) 

34 

35 

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

37 raise ImportError('Signals are not really supported on Windows') 

38 

39 

40def _sighandler_noop(signum, frame): 

41 """Dummy signal handler.""" 

42 pass 

43 

44 

45def waitstatus_to_exitcode(status): 

46 try: 

47 return os.waitstatus_to_exitcode(status) 

48 except ValueError: 

49 # The child exited, but we don't understand its status. 

50 # This shouldn't happen, but if it does, let's just 

51 # return that status; perhaps that helps debug it. 

52 return status 

53 

54 

55class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop): 

56 """Unix event loop. 

57 

58 Adds signal handling and UNIX Domain Socket support to 

59 SelectorEventLoop. 

60 """ 

61 

62 def __init__(self, selector=None): 

63 super().__init__(selector) 

64 self._signal_handlers = {} 

65 self._unix_server_sockets = {} 

66 if can_use_pidfd(): 

67 self._watcher = _PidfdChildWatcher() 

68 else: 

69 self._watcher = _ThreadedChildWatcher() 

70 

71 def close(self): 

72 super().close() 

73 if not sys.is_finalizing(): 

74 for sig in list(self._signal_handlers): 

75 self.remove_signal_handler(sig) 

76 else: 

77 if self._signal_handlers: 77 ↛ exitline 77 didn't return from function 'close' because the condition on line 77 was always true

78 warnings.warn(f"Closing the loop {self!r} " 

79 f"on interpreter shutdown " 

80 f"stage, skipping signal handlers removal", 

81 ResourceWarning, 

82 source=self) 

83 self._signal_handlers.clear() 

84 

85 def _process_self_data(self, data): 

86 for signum in data: 

87 if not signum: 

88 # ignore null bytes written by _write_to_self() 

89 continue 

90 self._handle_signal(signum) 

91 

92 def add_signal_handler(self, sig, callback, *args): 

93 """Add a handler for a signal. UNIX only. 

94 

95 Raise ValueError if the signal number is invalid or uncatchable. 

96 Raise RuntimeError if there is a problem setting up the handler. 

97 """ 

98 if (coroutines.iscoroutine(callback) or 

99 inspect.iscoroutinefunction(callback)): 

100 raise TypeError("coroutines cannot be used " 

101 "with add_signal_handler()") 

102 self._check_signal(sig) 

103 self._check_closed() 

104 try: 

105 # set_wakeup_fd() raises ValueError if this is not the 

106 # main thread. By calling it early we ensure that an 

107 # event loop running in another thread cannot add a signal 

108 # handler. 

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

110 except (ValueError, OSError) as exc: 

111 raise RuntimeError(str(exc)) 

112 

113 handle = events.Handle(callback, args, self, None) 

114 self._signal_handlers[sig] = handle 

115 

116 try: 

117 # Register a dummy signal handler to ask Python to write the signal 

118 # number in the wakeup file descriptor. _process_self_data() will 

119 # read signal numbers from this file descriptor to handle signals. 

120 signal.signal(sig, _sighandler_noop) 

121 

122 # Set SA_RESTART to limit EINTR occurrences. 

123 signal.siginterrupt(sig, False) 

124 except OSError as exc: 

125 del self._signal_handlers[sig] 

126 if not self._signal_handlers: 

127 try: 

128 signal.set_wakeup_fd(-1) 

129 except (ValueError, OSError) as nexc: 

130 logger.info('set_wakeup_fd(-1) failed: %s', nexc) 

131 

132 if exc.errno == errno.EINVAL: 

133 raise RuntimeError(f'sig {sig} cannot be caught') 

134 else: 

135 raise 

136 

137 def _handle_signal(self, sig): 

138 """Internal helper that is the actual signal handler.""" 

139 handle = self._signal_handlers.get(sig) 

140 if handle is None: 

141 return # Assume it's some race condition. 

142 if handle._cancelled: 

143 self.remove_signal_handler(sig) # Remove it properly. 

144 else: 

145 self._add_callback_signalsafe(handle) 

146 

147 def remove_signal_handler(self, sig): 

148 """Remove a handler for a signal. UNIX only. 

149 

150 Return True if a signal handler was removed, False if not. 

151 """ 

152 self._check_signal(sig) 

153 try: 

154 del self._signal_handlers[sig] 

155 except KeyError: 

156 return False 

157 

158 if sig == signal.SIGINT: 

159 handler = signal.default_int_handler 

160 else: 

161 handler = signal.SIG_DFL 

162 

163 try: 

164 signal.signal(sig, handler) 

165 except OSError as exc: 

166 if exc.errno == errno.EINVAL: 

167 raise RuntimeError(f'sig {sig} cannot be caught') 

168 else: 

169 raise 

170 

171 if not self._signal_handlers: 

172 try: 

173 signal.set_wakeup_fd(-1) 

174 except (ValueError, OSError) as exc: 

175 logger.info('set_wakeup_fd(-1) failed: %s', exc) 

176 

177 return True 

178 

179 def _check_signal(self, sig): 

180 """Internal helper to validate a signal. 

181 

182 Raise ValueError if the signal number is invalid or uncatchable. 

183 Raise RuntimeError if there is a problem setting up the handler. 

184 """ 

185 if not isinstance(sig, int): 

186 raise TypeError(f'sig must be an int, not {sig!r}') 

187 

188 if sig not in signal.valid_signals(): 

189 raise ValueError(f'invalid signal number {sig}') 

190 

191 def _make_read_pipe_transport(self, pipe, protocol, waiter=None, 

192 extra=None): 

193 return _UnixReadPipeTransport(self, pipe, protocol, waiter, extra) 

194 

195 def _make_write_pipe_transport(self, pipe, protocol, waiter=None, 

196 extra=None): 

197 return _UnixWritePipeTransport(self, pipe, protocol, waiter, extra) 

198 

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

200 stdin, stdout, stderr, bufsize, 

201 extra=None, **kwargs): 

202 watcher = self._watcher 

203 waiter = self.create_future() 

204 transp = _UnixSubprocessTransport(self, protocol, args, shell, 

205 stdin, stdout, stderr, bufsize, 

206 waiter=waiter, extra=extra, 

207 **kwargs) 

208 watcher.add_child_handler(transp.get_pid(), 

209 self._child_watcher_callback, transp) 

210 try: 

211 await waiter 

212 except (SystemExit, KeyboardInterrupt): 

213 raise 

214 except BaseException: 

215 transp.close() 

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

217 raise 

218 

219 return transp 

220 

221 def _child_watcher_callback(self, pid, returncode, transp): 

222 transp._process_exited(returncode) 

223 

224 async def create_unix_connection( 

225 self, protocol_factory, path=None, *, 

226 ssl=None, sock=None, 

227 server_hostname=None, 

228 ssl_handshake_timeout=None, 

229 ssl_shutdown_timeout=None): 

230 assert server_hostname is None or isinstance(server_hostname, str) 

231 if ssl: 

232 if server_hostname is None: 

233 raise ValueError( 

234 'you have to pass server_hostname when using ssl') 

235 else: 

236 if server_hostname is not None: 

237 raise ValueError('server_hostname is only meaningful with ssl') 

238 if ssl_handshake_timeout is not None: 

239 raise ValueError( 

240 'ssl_handshake_timeout is only meaningful with ssl') 

241 if ssl_shutdown_timeout is not None: 241 ↛ 242line 241 didn't jump to line 242 because the condition on line 241 was never true

242 raise ValueError( 

243 'ssl_shutdown_timeout is only meaningful with ssl') 

244 

245 if path is not None: 

246 if sock is not None: 

247 raise ValueError( 

248 'path and sock can not be specified at the same time') 

249 

250 path = os.fspath(path) 

251 sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM, 0) 

252 try: 

253 sock.setblocking(False) 

254 await self.sock_connect(sock, path) 

255 except: 

256 sock.close() 

257 raise 

258 

259 else: 

260 if sock is None: 

261 raise ValueError('no path and sock were specified') 

262 if (sock.family != socket.AF_UNIX or 262 ↛ 266line 262 didn't jump to line 266 because the condition on line 262 was always true

263 sock.type != socket.SOCK_STREAM): 

264 raise ValueError( 

265 f'A UNIX Domain Stream Socket was expected, got {sock!r}') 

266 sock.setblocking(False) 

267 

268 transport, protocol = await self._create_connection_transport( 

269 sock, protocol_factory, ssl, server_hostname, 

270 ssl_handshake_timeout=ssl_handshake_timeout, 

271 ssl_shutdown_timeout=ssl_shutdown_timeout) 

272 return transport, protocol 

273 

274 async def create_unix_server( 

275 self, protocol_factory, path=None, *, 

276 sock=None, backlog=100, ssl=None, 

277 ssl_handshake_timeout=None, 

278 ssl_shutdown_timeout=None, 

279 start_serving=True, cleanup_socket=True, mode=None): 

280 if isinstance(ssl, bool): 

281 raise TypeError('ssl argument must be an SSLContext or None') 

282 

283 if ssl_handshake_timeout is not None and not ssl: 

284 raise ValueError( 

285 'ssl_handshake_timeout is only meaningful with ssl') 

286 

287 if ssl_shutdown_timeout is not None and not ssl: 287 ↛ 288line 287 didn't jump to line 288 because the condition on line 287 was never true

288 raise ValueError( 

289 'ssl_shutdown_timeout is only meaningful with ssl') 

290 

291 if path is not None: 

292 if sock is not None: 

293 raise ValueError( 

294 'path and sock can not be specified at the same time') 

295 

296 path = os.fspath(path) 

297 if mode is not None and path and path[0] in (0, '\x00'): 

298 raise ValueError( 

299 'mode is not supported for abstract sockets') 

300 sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) 

301 

302 # Check for abstract socket. `str` and `bytes` paths are supported. 

303 if path[0] not in (0, '\x00'): 303 ↛ 314line 303 didn't jump to line 314 because the condition on line 303 was always true

304 try: 

305 if stat.S_ISSOCK(os.stat(path).st_mode): 

306 os.remove(path) 

307 except FileNotFoundError: 

308 pass 

309 except OSError as err: 

310 # Directory may have permissions only to create socket. 

311 logger.error('Unable to check or remove stale UNIX socket ' 

312 '%r: %r', path, err) 

313 

314 try: 

315 sock.bind(path) 

316 except OSError as exc: 

317 sock.close() 

318 if exc.errno == errno.EADDRINUSE: 

319 # Let's improve the error message by adding 

320 # with what exact address it occurs. 

321 msg = f'Address {path!r} is already in use' 

322 raise OSError(errno.EADDRINUSE, msg) from None 

323 else: 

324 raise 

325 except: 

326 sock.close() 

327 raise 

328 

329 if mode is not None: 

330 # The socket cannot accept connections until listen() is 

331 # called, which happens later in Server._start_serving(), 

332 # so no connection can be accepted while the socket still 

333 # has the default permissions. 

334 try: 

335 os.chmod(path, mode) 

336 except: 

337 sock.close() 

338 raise 

339 else: 

340 if sock is None: 

341 raise ValueError( 

342 'path was not specified, and no sock specified') 

343 

344 if mode is not None: 

345 raise ValueError( 

346 'mode is only meaningful with path') 

347 

348 if (sock.family != socket.AF_UNIX or 

349 sock.type != socket.SOCK_STREAM): 

350 raise ValueError( 

351 f'A UNIX Domain Stream Socket was expected, got {sock!r}') 

352 

353 if cleanup_socket: 

354 path = sock.getsockname() 

355 # Check for abstract socket. `str` and `bytes` paths are supported. 

356 if path[0] not in (0, '\x00'): 356 ↛ 362line 356 didn't jump to line 362 because the condition on line 356 was always true

357 try: 

358 self._unix_server_sockets[sock] = os.stat(path).st_ino 

359 except FileNotFoundError: 

360 pass 

361 

362 sock.setblocking(False) 

363 server = base_events.Server(self, [sock], protocol_factory, 

364 ssl, backlog, ssl_handshake_timeout, 

365 ssl_shutdown_timeout) 

366 if start_serving: 

367 server._start_serving() 

368 # Skip one loop iteration so that all 'loop.add_reader' 

369 # go through. 

370 await tasks.sleep(0) 

371 

372 return server 

373 

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

375 try: 

376 os.sendfile 

377 except AttributeError: 

378 raise exceptions.SendfileNotAvailableError( 

379 "os.sendfile() is not available") 

380 try: 

381 fileno = file.fileno() 

382 except (AttributeError, io.UnsupportedOperation): 

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

384 try: 

385 fsize = os.fstat(fileno).st_size 

386 except OSError: 

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

388 blocksize = count if count else fsize 

389 if not blocksize: 

390 return 0 # empty file 

391 

392 fut = self.create_future() 

393 self._sock_sendfile_native_impl(fut, None, sock, fileno, 

394 offset, count, blocksize, 0) 

395 return await fut 

396 

397 def _sock_sendfile_native_impl(self, fut, registered_fd, sock, fileno, 

398 offset, count, blocksize, total_sent): 

399 fd = sock.fileno() 

400 if registered_fd is not None: 

401 # Remove the callback early. It should be rare that the 

402 # selector says the fd is ready but the call still returns 

403 # EAGAIN, and I am willing to take a hit in that case in 

404 # order to simplify the common case. 

405 self.remove_writer(registered_fd) 

406 if fut.cancelled(): 

407 self._sock_sendfile_update_filepos(fileno, offset) 

408 return 

409 if count: 

410 blocksize = count - total_sent 

411 if blocksize <= 0: 

412 self._sock_sendfile_update_filepos(fileno, offset) 

413 fut.set_result(total_sent) 

414 return 

415 

416 # On 32-bit architectures truncate to 1GiB to avoid OverflowError 

417 blocksize = min(blocksize, sys.maxsize//2 + 1) 

418 

419 try: 

420 sent = os.sendfile(fd, fileno, offset, blocksize) 

421 except (BlockingIOError, InterruptedError): 

422 if registered_fd is None: 422 ↛ 424line 422 didn't jump to line 424 because the condition on line 422 was always true

423 self._sock_add_cancellation_callback(fut, sock) 

424 self.add_writer(fd, self._sock_sendfile_native_impl, fut, 

425 fd, sock, fileno, 

426 offset, count, blocksize, total_sent) 

427 except OSError as exc: 

428 if (registered_fd is not None and 428 ↛ 435line 428 didn't jump to line 435 because the condition on line 428 was never true

429 exc.errno == errno.ENOTCONN and 

430 type(exc) is not ConnectionError): 

431 # If we have an ENOTCONN and this isn't a first call to 

432 # sendfile(), i.e. the connection was closed in the middle 

433 # of the operation, normalize the error to ConnectionError 

434 # to make it consistent across all Posix systems. 

435 new_exc = ConnectionError( 

436 "socket is not connected", errno.ENOTCONN) 

437 new_exc.__cause__ = exc 

438 exc = new_exc 

439 if total_sent == 0: 

440 # We can get here for different reasons, the main 

441 # one being 'file' is not a regular mmap(2)-like 

442 # file, in which case we'll fall back on using 

443 # plain send(). 

444 err = exceptions.SendfileNotAvailableError( 

445 "os.sendfile call failed") 

446 self._sock_sendfile_update_filepos(fileno, offset) 

447 fut.set_exception(err) 

448 else: 

449 self._sock_sendfile_update_filepos(fileno, offset) 

450 fut.set_exception(exc) 

451 except (SystemExit, KeyboardInterrupt): 

452 raise 

453 except BaseException as exc: 

454 self._sock_sendfile_update_filepos(fileno, offset) 

455 fut.set_exception(exc) 

456 else: 

457 if sent == 0: 

458 # EOF 

459 self._sock_sendfile_update_filepos(fileno, offset) 

460 fut.set_result(total_sent) 

461 else: 

462 offset += sent 

463 total_sent += sent 

464 if registered_fd is None: 

465 self._sock_add_cancellation_callback(fut, sock) 

466 self.add_writer(fd, self._sock_sendfile_native_impl, fut, 

467 fd, sock, fileno, 

468 offset, count, blocksize, total_sent) 

469 

470 def _sock_sendfile_update_filepos(self, fileno, offset): 

471 # After this helper runs, the source fd's lseek pointer is at offset." 

472 os.lseek(fileno, offset, os.SEEK_SET) 

473 

474 def _sock_add_cancellation_callback(self, fut, sock): 

475 def cb(fut): 

476 if fut.cancelled(): 

477 fd = sock.fileno() 

478 if fd != -1: 478 ↛ exitline 478 didn't return from function 'cb' because the condition on line 478 was always true

479 self.remove_writer(fd) 

480 fut.add_done_callback(cb) 

481 

482 def _stop_serving(self, sock): 

483 # Is this a unix socket that needs cleanup? 

484 if sock in self._unix_server_sockets: 

485 path = sock.getsockname() 

486 else: 

487 path = None 

488 

489 super()._stop_serving(sock) 

490 

491 if path is not None: 

492 prev_ino = self._unix_server_sockets[sock] 

493 del self._unix_server_sockets[sock] 

494 try: 

495 if os.stat(path).st_ino == prev_ino: 

496 os.unlink(path) 

497 except FileNotFoundError: 

498 pass 

499 except OSError as err: 

500 logger.error('Unable to clean up listening UNIX socket ' 

501 '%r: %r', path, err) 

502 

503 

504class _UnixReadPipeTransport(transports.ReadTransport): 

505 

506 max_size = 256 * 1024 # max bytes we read in one event loop iteration 

507 

508 def __init__(self, loop, pipe, protocol, waiter=None, extra=None): 

509 super().__init__(extra) 

510 self._extra['pipe'] = pipe 

511 self._loop = loop 

512 self._pipe = pipe 

513 self._fileno = pipe.fileno() 

514 self._protocol = protocol 

515 self._closing = False 

516 self._paused = False 

517 

518 mode = os.fstat(self._fileno).st_mode 

519 if not (stat.S_ISFIFO(mode) or 

520 stat.S_ISSOCK(mode) or 

521 stat.S_ISCHR(mode)): 

522 self._pipe = None 

523 self._fileno = None 

524 self._protocol = None 

525 raise ValueError("Pipe transport is for pipes/sockets only.") 

526 

527 os.set_blocking(self._fileno, False) 

528 

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

530 # only start reading when connection_made() has been called 

531 self._loop.call_soon(self._add_reader, 

532 self._fileno, self._read_ready) 

533 if waiter is not None: 

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

535 self._loop.call_soon(futures._set_result_unless_cancelled, 

536 waiter, None) 

537 

538 def _add_reader(self, fd, callback): 

539 if not self.is_reading(): 

540 return 

541 self._loop._add_reader(fd, callback) 

542 

543 def is_reading(self): 

544 return not self._paused and not self._closing 

545 

546 def __repr__(self): 

547 info = [self.__class__.__name__] 

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

549 info.append('closed') 

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

551 info.append('closing') 

552 info.append(f'fd={self._fileno}') 

553 selector = getattr(self._loop, '_selector', None) 

554 if self._pipe is not None and selector is not None: 

555 polling = selector_events._test_selector_event( 

556 selector, self._fileno, selectors.EVENT_READ) 

557 if polling: 557 ↛ 558line 557 didn't jump to line 558 because the condition on line 557 was never true

558 info.append('polling') 

559 else: 

560 info.append('idle') 

561 elif self._pipe is not None: 561 ↛ 564line 561 didn't jump to line 564 because the condition on line 561 was always true

562 info.append('open') 

563 else: 

564 info.append('closed') 

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

566 

567 def _read_ready(self): 

568 try: 

569 data = os.read(self._fileno, self.max_size) 

570 except (BlockingIOError, InterruptedError): 

571 pass 

572 except OSError as exc: 

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

574 else: 

575 if data: 

576 self._protocol.data_received(data) 

577 else: 

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

579 logger.info("%r was closed by peer", self) 

580 self._closing = True 

581 self._loop._remove_reader(self._fileno) 

582 self._loop.call_soon(self._protocol.eof_received) 

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

584 

585 def pause_reading(self): 

586 if not self.is_reading(): 

587 return 

588 self._paused = True 

589 self._loop._remove_reader(self._fileno) 

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

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

592 

593 def resume_reading(self): 

594 if self._closing or not self._paused: 

595 return 

596 self._paused = False 

597 self._loop._add_reader(self._fileno, self._read_ready) 

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

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

600 

601 def set_protocol(self, protocol): 

602 self._protocol = protocol 

603 

604 def get_protocol(self): 

605 return self._protocol 

606 

607 def is_closing(self): 

608 return self._closing 

609 

610 def close(self): 

611 if not self._closing: 

612 self._close(None) 

613 

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

615 if self._pipe is not None: 615 ↛ 616line 615 didn't jump to line 616 because the condition on line 615 was never true

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

617 self._pipe.close() 

618 

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

620 # should be called by exception handler only 

621 if (isinstance(exc, OSError) and exc.errno == errno.EIO): 621 ↛ 622line 621 didn't jump to line 622 because the condition on line 621 was never true

622 if self._loop.get_debug(): 

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

624 else: 

625 self._loop.call_exception_handler({ 

626 'message': message, 

627 'exception': exc, 

628 'transport': self, 

629 'protocol': self._protocol, 

630 }) 

631 self._close(exc) 

632 

633 def _close(self, exc): 

634 self._closing = True 

635 self._loop._remove_reader(self._fileno) 

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

637 

638 def _call_connection_lost(self, exc): 

639 try: 

640 self._protocol.connection_lost(exc) 

641 finally: 

642 self._pipe.close() 

643 self._pipe = None 

644 self._protocol = None 

645 self._loop = None 

646 

647 

648class _UnixWritePipeTransport(transports._FlowControlMixin, 

649 transports.WriteTransport): 

650 

651 def __init__(self, loop, pipe, protocol, waiter=None, extra=None): 

652 super().__init__(extra, loop) 

653 self._extra['pipe'] = pipe 

654 self._pipe = pipe 

655 self._fileno = pipe.fileno() 

656 self._protocol = protocol 

657 self._buffer = bytearray() 

658 self._conn_lost = 0 

659 self._closing = False # Set when close() or write_eof() called. 

660 

661 pipe_stat = os.fstat(self._fileno) 

662 mode = pipe_stat.st_mode 

663 is_char = stat.S_ISCHR(mode) 

664 is_fifo = stat.S_ISFIFO(mode) 

665 is_socket = stat.S_ISSOCK(mode) 

666 if not (is_char or is_fifo or is_socket): 

667 self._pipe = None 

668 self._fileno = None 

669 self._protocol = None 

670 raise ValueError("Pipe transport is only for " 

671 "pipes, sockets and character devices") 

672 

673 os.set_blocking(self._fileno, False) 

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

675 

676 # On AIX, the reader trick (to be notified when the read end of the 

677 # socket is closed) only works for sockets. On other platforms it 

678 # works for pipes and sockets. (Exception: OS X 10.4? Issue #19294.) 

679 # On macOS and Solaris, the trick misfires for named FIFOs (but not for 

680 # pipes created with os.pipe(), which have st_nlink == 0): the write 

681 # end polls as readable whenever unread data sits in the FIFO, and no 

682 # event is delivered when the read end is closed, so it can only 

683 # ever report a false disconnection (gh-145030). The same XNU 

684 # behaviour applies on iOS/tvOS/watchOS (sys.platform is not 

685 # "darwin" there). 

686 is_named_fifo_without_close_event = ( 

687 sys.platform in {"darwin", "ios", "tvos", "watchos", "sunos5"} 

688 and is_fifo and pipe_stat.st_nlink > 0) 

689 if is_socket or (is_fifo 

690 and not sys.platform.startswith("aix") 

691 and not is_named_fifo_without_close_event): 

692 # only start reading when connection_made() has been called 

693 self._loop.call_soon(self._loop._add_reader, 

694 self._fileno, self._read_ready) 

695 

696 if waiter is not None: 

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

698 self._loop.call_soon(futures._set_result_unless_cancelled, 

699 waiter, None) 

700 

701 def __repr__(self): 

702 info = [self.__class__.__name__] 

703 if self._pipe is None: 703 ↛ 704line 703 didn't jump to line 704 because the condition on line 703 was never true

704 info.append('closed') 

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

706 info.append('closing') 

707 info.append(f'fd={self._fileno}') 

708 selector = getattr(self._loop, '_selector', None) 

709 if self._pipe is not None and selector is not None: 709 ↛ 710line 709 didn't jump to line 710 because the condition on line 709 was never true

710 polling = selector_events._test_selector_event( 

711 selector, self._fileno, selectors.EVENT_WRITE) 

712 if polling: 

713 info.append('polling') 

714 else: 

715 info.append('idle') 

716 

717 bufsize = self.get_write_buffer_size() 

718 info.append(f'bufsize={bufsize}') 

719 elif self._pipe is not None: 719 ↛ 722line 719 didn't jump to line 722 because the condition on line 719 was always true

720 info.append('open') 

721 else: 

722 info.append('closed') 

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

724 

725 def get_write_buffer_size(self): 

726 return len(self._buffer) 

727 

728 def _read_ready(self): 

729 # Pipe was closed by peer. 

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

731 logger.info("%r was closed by peer", self) 

732 if self._buffer: 

733 self._close(BrokenPipeError()) 

734 else: 

735 self._close() 

736 

737 def write(self, data): 

738 assert isinstance(data, (bytes, bytearray, memoryview)), repr(data) 

739 if isinstance(data, bytearray): 739 ↛ 740line 739 didn't jump to line 740 because the condition on line 739 was never true

740 data = memoryview(data) 

741 if not data: 

742 return 

743 

744 if self._conn_lost or self._closing: 

745 if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES: 

746 logger.warning('pipe closed by peer or ' 

747 'os.write(pipe, data) raised exception.') 

748 self._conn_lost += 1 

749 return 

750 

751 if not self._buffer: 

752 # Attempt to send it right away first. 

753 try: 

754 n = os.write(self._fileno, data) 

755 except (BlockingIOError, InterruptedError): 

756 n = 0 

757 except (SystemExit, KeyboardInterrupt): 

758 raise 

759 except BaseException as exc: 

760 self._conn_lost += 1 

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

762 return 

763 if n == len(data): 

764 return 

765 elif n > 0: 

766 data = memoryview(data)[n:] 

767 self._loop._add_writer(self._fileno, self._write_ready) 

768 

769 self._buffer += data 

770 self._maybe_pause_protocol() 

771 

772 def _write_ready(self): 

773 assert self._buffer, 'Data should not be empty' 

774 

775 try: 

776 n = os.write(self._fileno, self._buffer) 

777 except (BlockingIOError, InterruptedError): 

778 pass 

779 except (SystemExit, KeyboardInterrupt): 

780 raise 

781 except BaseException as exc: 

782 self._buffer.clear() 

783 self._conn_lost += 1 

784 # Remove writer here, _fatal_error() doesn't it 

785 # because _buffer is empty. 

786 self._loop._remove_writer(self._fileno) 

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

788 else: 

789 if n == len(self._buffer): 

790 self._buffer.clear() 

791 self._loop._remove_writer(self._fileno) 

792 self._maybe_resume_protocol() # May append to buffer. 

793 if self._closing: 

794 self._loop._remove_reader(self._fileno) 

795 self._call_connection_lost(None) 

796 return 

797 elif n > 0: 

798 del self._buffer[:n] 

799 

800 def can_write_eof(self): 

801 return True