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 CloseHandle(self._handle) 

115 self._handle = None 

116 

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

118 if self._handle is not None: 

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

120 self.close() 

121 

122 def __enter__(self): 

123 return self 

124 

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

126 self.close() 

127 

128 

129# Replacement for subprocess.Popen using overlapped pipe handles 

130 

131 

132class Popen(subprocess.Popen): 

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

134 

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

136 """ 

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

138 assert not kwds.get('universal_newlines') 

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

140 stdin_rfd = stdout_wfd = stderr_wfd = None 

141 stdin_wh = stdout_rh = stderr_rh = None 

142 if stdin == PIPE: 

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

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

145 else: 

146 stdin_rfd = stdin 

147 if stdout == PIPE: 

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

149 stdout_wfd = msvcrt.open_osfhandle(stdout_wh, 0) 

150 else: 

151 stdout_wfd = stdout 

152 if stderr == PIPE: 

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

154 stderr_wfd = msvcrt.open_osfhandle(stderr_wh, 0) 

155 elif stderr == STDOUT: 

156 stderr_wfd = stdout_wfd 

157 else: 

158 stderr_wfd = stderr 

159 try: 

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

161 stderr=stderr_wfd, **kwds) 

162 except: 

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

164 if h is not None: 

165 _winapi.CloseHandle(h) 

166 raise 

167 else: 

168 if stdin_wh is not None: 

169 self.stdin = PipeHandle(stdin_wh) 

170 if stdout_rh is not None: 

171 self.stdout = PipeHandle(stdout_rh) 

172 if stderr_rh is not None: 

173 self.stderr = PipeHandle(stderr_rh) 

174 finally: 

175 if stdin == PIPE: 

176 os.close(stdin_rfd) 

177 if stdout == PIPE: 

178 os.close(stdout_wfd) 

179 if stderr == PIPE: 

180 os.close(stderr_wfd)