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

1"""Various Windows specific bits and pieces.""" 

2 

3import sys 

4 

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

6 raise ImportError('win32 only') 

7 

8import _winapi 

9import itertools 

10import msvcrt 

11import os 

12import subprocess 

13import warnings 

14 

15 

16__all__ = 'pipe', 'Popen', 'PIPE', 'PipeHandle' 

17 

18 

19# Constants/globals 

20 

21 

22BUFSIZE = 8192 

23PIPE = subprocess.PIPE 

24STDOUT = subprocess.STDOUT 

25_mmap_counter = itertools.count() 

26_MAX_PIPE_ATTEMPTS = 20 

27 

28 

29# Replacement for os.pipe() using handles instead of fds 

30 

31 

32def pipe(*, duplex=False, overlapped=(True, True), bufsize=BUFSIZE): 

33 """Like os.pipe() but with overlapped support and using handles not fds.""" 

34 if duplex: 

35 openmode = _winapi.PIPE_ACCESS_DUPLEX 

36 access = _winapi.GENERIC_READ | _winapi.GENERIC_WRITE 

37 obsize, ibsize = bufsize, bufsize 

38 else: 

39 openmode = _winapi.PIPE_ACCESS_INBOUND 

40 access = _winapi.GENERIC_WRITE 

41 obsize, ibsize = 0, bufsize 

42 

43 openmode |= _winapi.FILE_FLAG_FIRST_PIPE_INSTANCE 

44 

45 if overlapped[0]: 

46 openmode |= _winapi.FILE_FLAG_OVERLAPPED 

47 

48 if overlapped[1]: 

49 flags_and_attribs = _winapi.FILE_FLAG_OVERLAPPED 

50 else: 

51 flags_and_attribs = 0 

52 

53 h1 = h2 = None 

54 try: 

55 for attempts in itertools.count(): 

56 address = r'\\.\pipe\python-pipe-{:d}-{:d}-{}'.format( 

57 os.getpid(), next(_mmap_counter), os.urandom(8).hex()) 

58 try: 

59 h1 = _winapi.CreateNamedPipe( 

60 address, openmode, _winapi.PIPE_WAIT, 

61 1, obsize, ibsize, _winapi.NMPWAIT_WAIT_FOREVER, _winapi.NULL) 

62 break 

63 except OSError as e: 

64 if attempts >= _MAX_PIPE_ATTEMPTS: 

65 raise 

66 if e.winerror not in (_winapi.ERROR_PIPE_BUSY, 

67 _winapi.ERROR_ACCESS_DENIED): 

68 raise 

69 

70 h2 = _winapi.CreateFile( 

71 address, access, 0, _winapi.NULL, _winapi.OPEN_EXISTING, 

72 flags_and_attribs, _winapi.NULL) 

73 

74 ov = _winapi.ConnectNamedPipe(h1, overlapped=True) 

75 ov.GetOverlappedResult(True) 

76 return h1, h2 

77 except: 

78 if h1 is not None: 

79 _winapi.CloseHandle(h1) 

80 if h2 is not None: 

81 _winapi.CloseHandle(h2) 

82 raise 

83 

84 

85# Wrapper for a pipe handle 

86 

87 

88class PipeHandle: 

89 """Wrapper for an overlapped pipe handle which is vaguely file-object like. 

90 

91 The IOCP event loop can use these instead of socket objects. 

92 """ 

93 def __init__(self, handle): 

94 self._handle = handle 

95 

96 def __repr__(self): 

97 if self._handle is not None: 

98 handle = f'handle={self._handle!r}' 

99 else: 

100 handle = 'closed' 

101 return f'<{self.__class__.__name__} {handle}>' 

102 

103 @property 

104 def handle(self): 

105 return self._handle 

106 

107 def fileno(self): 

108 if self._handle is None: 

109 raise ValueError("I/O operation on closed pipe") 

110 return self._handle 

111 

112 def close(self, *, CloseHandle=_winapi.CloseHandle): 

113 if self._handle is not None: 

114 handle = self._handle 

115 self._handle = None 

116 CloseHandle(handle) 

117 

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

119 if self._handle is not None: 

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

121 self.close() 

122 

123 def __enter__(self): 

124 return self 

125 

126 def __exit__(self, t, v, tb): 

127 self.close() 

128 

129 

130# Replacement for subprocess.Popen using overlapped pipe handles 

131 

132 

133class Popen(subprocess.Popen): 

134 """Replacement for subprocess.Popen using overlapped pipe handles. 

135 

136 The stdin, stdout, stderr are None or instances of PipeHandle. 

137 """ 

138 def __init__(self, args, stdin=None, stdout=None, stderr=None, **kwds): 

139 assert not kwds.get('universal_newlines') 

140 assert kwds.get('bufsize', 0) == 0 

141 stdin_rfd = stdout_wfd = stderr_wfd = None 

142 stdin_wh = stdout_rh = stderr_rh = None 

143 if stdin == PIPE: 

144 stdin_rh, stdin_wh = pipe(overlapped=(False, True), duplex=True) 

145 stdin_rfd = msvcrt.open_osfhandle(stdin_rh, os.O_RDONLY) 

146 else: 

147 stdin_rfd = stdin 

148 if stdout == PIPE: 

149 stdout_rh, stdout_wh = pipe(overlapped=(True, False)) 

150 stdout_wfd = msvcrt.open_osfhandle(stdout_wh, 0) 

151 else: 

152 stdout_wfd = stdout 

153 if stderr == PIPE: 

154 stderr_rh, stderr_wh = pipe(overlapped=(True, False)) 

155 stderr_wfd = msvcrt.open_osfhandle(stderr_wh, 0) 

156 elif stderr == STDOUT: 

157 stderr_wfd = stdout_wfd 

158 else: 

159 stderr_wfd = stderr 

160 try: 

161 super().__init__(args, stdin=stdin_rfd, stdout=stdout_wfd, 

162 stderr=stderr_wfd, **kwds) 

163 except: 

164 for h in (stdin_wh, stdout_rh, stderr_rh): 

165 if h is not None: 

166 _winapi.CloseHandle(h) 

167 raise 

168 else: 

169 if stdin_wh is not None: 

170 self.stdin = PipeHandle(stdin_wh) 

171 if stdout_rh is not None: 

172 self.stdout = PipeHandle(stdout_rh) 

173 if stderr_rh is not None: 

174 self.stderr = PipeHandle(stderr_rh) 

175 finally: 

176 if stdin == PIPE: 

177 os.close(stdin_rfd) 

178 if stdout == PIPE: 

179 os.close(stdout_wfd) 

180 if stderr == PIPE: 

181 os.close(stderr_wfd)