我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用msvcrt.get_osfhandle()。
def _recv(self, which, maxsize): conn, maxsize = self.get_conn_maxsize(which, maxsize) if conn is None: return None try: x = msvcrt.get_osfhandle(conn.fileno()) (read, nAvail, nMessage) = PeekNamedPipe(x, 0) if maxsize < nAvail: nAvail = maxsize if nAvail > 0: (errCode, read) = ReadFile(x, nAvail, None) except ValueError: return self._close(which) except (subprocess.pywintypes.error, Exception): if geterror()[0] in (109, errno.ESHUTDOWN): return self._close(which) raise if self.universal_newlines: # Translate newlines. For Python 3.x assume read is text. # If bytes then another solution is needed. read = read.replace("\r\n", "\n").replace("\r", "\n") return read
def isatty(): if isatty.no_tty: return False f = sys.stdout if f.isatty(): return True if not iswindows: return False # Check for a cygwin ssh pipe buf = ctypes.create_string_buffer(1024) h = msvcrt.get_osfhandle(f.fileno()) if get_file_type(h) != 3: return False ret = get_file_info_by_handle(h, 2, buf, ctypes.sizeof(buf)) if not ret: raise ctypes.WinError() data = buf.raw name = data[4:].decode('utf-16').rstrip(u'\0') parts = name.split('-') return parts[0] == r'\cygwin' and parts[2].startswith('pty') and parts[4] == 'master'
def test_fd_transfer(self): if self.TYPE != 'processes': self.skipTest("only makes sense with processes") conn, child_conn = self.Pipe(duplex=True) p = self.Process(target=self._writefd, args=(child_conn, b"foo")) p.daemon = True p.start() self.addCleanup(test.support.unlink, test.support.TESTFN) with open(test.support.TESTFN, "wb") as f: fd = f.fileno() if msvcrt: fd = msvcrt.get_osfhandle(fd) reduction.send_handle(conn, fd, p.pid) p.join() with open(test.support.TESTFN, "rb") as f: self.assertEqual(f.read(), b"foo")
def test_fd_transfer(self): if self.TYPE != 'processes': self.skipTest("only makes sense with processes") conn, child_conn = self.Pipe(duplex=True) p = self.Process(target=self._writefd, args=(child_conn, b"foo")) p.daemon = True p.start() with open(test_support.TESTFN, "wb") as f: fd = f.fileno() if msvcrt: fd = msvcrt.get_osfhandle(fd) reduction.send_handle(conn, fd, p.pid) p.join() with open(test_support.TESTFN, "rb") as f: self.assertEqual(f.read(), b"foo")
def _recv(self, which, maxsize): conn, maxsize = self.get_conn_maxsize(which, maxsize) if conn is None: return None try: x = msvcrt.get_osfhandle(conn.fileno()) (read, nAvail, nMessage) = PeekNamedPipe(x, 0) if maxsize < nAvail: nAvail = maxsize if nAvail > 0: (errCode, read) = ReadFile(x, nAvail, None) except (ValueError, NameError): return self._close(which) except (subprocess.pywintypes.error, Exception), why: if why[0] in (109, errno.ESHUTDOWN): return self._close(which) raise if self.universal_newlines: read = self._translate_newlines(read) return read
def _make_non_blocking(file_obj): """make file object non-blocking Windows doesn't have the fcntl module, but someone on stack overflow supplied this code as an answer, and it works http://stackoverflow.com/a/34504971/2893090""" if USING_WINDOWS: LPDWORD = POINTER(DWORD) PIPE_NOWAIT = wintypes.DWORD(0x00000001) SetNamedPipeHandleState = windll.kernel32.SetNamedPipeHandleState SetNamedPipeHandleState.argtypes = [HANDLE, LPDWORD, LPDWORD, LPDWORD] SetNamedPipeHandleState.restype = BOOL h = msvcrt.get_osfhandle(file_obj.fileno()) res = windll.kernel32.SetNamedPipeHandleState(h, byref(PIPE_NOWAIT), None, None) if res == 0: raise ValueError(WinError()) else: # Set the file status flag (F_SETFL) on the pipes to be non-blocking # so we can attempt to read from a pipe with no new data without locking # the program up fcntl.fcntl(file_obj, fcntl.F_SETFL, os.O_NONBLOCK)
def run(cmd, mSec=None, stdin=None, stdout=None, stderr=None, **kw): """ Run cmd as a child process and return exit code. mSec: terminate cmd after specified number of milliseconds stdin, stdout, stderr: file objects for child I/O (use hStdin etc. to attach handles instead of files); default is caller's stdin, stdout & stderr; kw: see Process.__init__ for more keyword options """ if stdin is not None: kw['hStdin'] = msvcrt.get_osfhandle(stdin.fileno()) if stdout is not None: kw['hStdout'] = msvcrt.get_osfhandle(stdout.fileno()) if stderr is not None: kw['hStderr'] = msvcrt.get_osfhandle(stderr.fileno()) child = Process(cmd, **kw) if child.wait(mSec) != win32event.WAIT_OBJECT_0: child.kill() raise WindowsError('process timeout exceeded') return child.exitCode()
def _recv(self, which, maxsize): conn, maxsize = self.get_conn_maxsize(which, maxsize) if conn is None: return None try: x = msvcrt.get_osfhandle(conn.fileno()) (read, nAvail, nMessage) = PeekNamedPipe(x, 0) if maxsize < nAvail: nAvail = maxsize if nAvail > 0: (errCode, read) = ReadFile(x, nAvail, None) except ValueError: return self._close(which) except (subprocess.pywintypes.error, Exception), why: if why[0] in (109, errno.ESHUTDOWN): return self._close(which) raise #if self.universal_newlines: # read = self._translate_newlines(read) return read
def _recv(self, which, maxsize): conn, maxsize = self.get_conn_maxsize(which, maxsize) if conn is None: return None try: x = msvcrt.get_osfhandle(conn.fileno()) (read, nAvail, nMessage) = PeekNamedPipe(x, 0) if maxsize < nAvail: nAvail = maxsize if nAvail > 0: (errCode, read) = ReadFile(x, nAvail, None) except ValueError: return self._close(which) except (subprocess.pywintypes.error, Exception) as why: if why[0] in (109, errno.ESHUTDOWN): return self._close(which) raise if self.universal_newlines: read = self._translate_newlines(read) return read
def __init__(self, process_obj): # create pipe for communication with child rfd, wfd = os.pipe() # get handle for read end of the pipe and make it inheritable rhandle = duplicate(msvcrt.get_osfhandle(rfd), inheritable=True) os.close(rfd) # start process cmd = get_command_line() + [rhandle] cmd = ' '.join('"%s"' % x for x in cmd) hp, ht, pid, tid = _subprocess.CreateProcess( _python_exe, cmd, None, None, 1, 0, None, None, None ) ht.Close() close(rhandle) # set attributes of self self.pid = pid self.returncode = None self._handle = hp # send information to child prep_data = get_preparation_data(process_obj._name) to_child = os.fdopen(wfd, 'wb') Popen._tls.process_handle = int(hp) try: dump(prep_data, to_child, HIGHEST_PROTOCOL) dump(process_obj, to_child, HIGHEST_PROTOCOL) finally: del Popen._tls.process_handle to_child.close()
def send(self, input): if not self.stdin: return None try: x = msvcrt.get_osfhandle(self.stdin.fileno()) (errCode, written) = WriteFile(x, input) except ValueError: return self._close('stdin') except (subprocess.pywintypes.error, Exception): if geterror()[0] in (109, errno.ESHUTDOWN): return self._close('stdin') raise return written
def get_handle_from_file(f): """Get the ``Windows`` HANDLE of a python :class:`file`""" return msvcrt.get_osfhandle(f.fileno())
def lock(f, flags): hfile = msvcrt.get_osfhandle(_fd(f)) overlapped = OVERLAPPED() ret = LockFileEx(hfile, flags, 0, 0, 0xFFFF0000, byref(overlapped)) return bool(ret)
def unlock(f): hfile = msvcrt.get_osfhandle(_fd(f)) overlapped = OVERLAPPED() ret = UnlockFileEx(hfile, 0, 0, 0xFFFF0000, byref(overlapped)) return bool(ret)
def _lock_file(f, exclusive): overlapped = OVERLAPPED() overlapped.Offset = 0 overlapped.OffsetHigh = 0 overlapped.hEvent = 0 f._lock_file_overlapped_p = ctypes.pointer(overlapped) handle = msvcrt.get_osfhandle(f.fileno()) if not LockFileEx(handle, 0x2 if exclusive else 0x0, 0, whole_low, whole_high, f._lock_file_overlapped_p): raise OSError('Locking file failed: %r' % ctypes.FormatError())
def _unlock_file(f): assert f._lock_file_overlapped_p handle = msvcrt.get_osfhandle(f.fileno()) if not UnlockFileEx(handle, 0, whole_low, whole_high, f._lock_file_overlapped_p): raise OSError('Unlocking file failed: %r' % ctypes.FormatError())
def lock(file, flags): hfile = msvcrt.get_osfhandle(file.fileno()) overlapped = OVERLAPPED() LockFileEx(hfile, flags, 0, 0, 0xFFFF0000, ctypes.byref(overlapped))
def unlock(file): hfile = msvcrt.get_osfhandle(file.fileno()) overlapped = OVERLAPPED() UnlockFileEx(hfile, 0, 0, 0xFFFF0000, ctypes.byref(overlapped))
def pipe_non_blocking_set(fd): # Constant could define globally but avoid polluting the name-space # thanks to: http://stackoverflow.com/questions/34504970 import msvcrt from ctypes import windll, byref, wintypes, WinError, POINTER from ctypes.wintypes import HANDLE, DWORD, BOOL LPDWORD = POINTER(DWORD) PIPE_NOWAIT = wintypes.DWORD(0x00000001) def pipe_no_wait(pipefd): SetNamedPipeHandleState = windll.kernel32.SetNamedPipeHandleState SetNamedPipeHandleState.argtypes = [HANDLE, LPDWORD, LPDWORD, LPDWORD] SetNamedPipeHandleState.restype = BOOL h = msvcrt.get_osfhandle(pipefd) res = windll.kernel32.SetNamedPipeHandleState(h, byref(PIPE_NOWAIT), None, None) if res == 0: print(WinError()) return False return True return pipe_no_wait(fd)
def add_reader(self, fd, callback): " Start watching the file descriptor for read availability. " h = msvcrt.get_osfhandle(fd) self._read_fds[h] = callback
def remove_reader(self, fd): " Stop watching the file descriptor for read availability. " h = msvcrt.get_osfhandle(fd) if h in self._read_fds: del self._read_fds[h]
def __init__(self, recognize_paste=True): self._fdcon = None self.recognize_paste = recognize_paste # When stdin is a tty, use that handle, otherwise, create a handle from # CONIN$. if sys.stdin.isatty(): self.handle = windll.kernel32.GetStdHandle(STD_INPUT_HANDLE) else: self._fdcon = os.open('CONIN$', os.O_RDWR | os.O_BINARY) self.handle = msvcrt.get_osfhandle(self._fdcon)
def _stdin_raw_nonblock(self): """Use the raw Win32 handle of sys.stdin to do non-blocking reads""" # WARNING: This is experimental, and produces inconsistent results. # It's possible for the handle not to be appropriate for use # with WaitForSingleObject, among other things. handle = msvcrt.get_osfhandle(sys.stdin.fileno()) result = WaitForSingleObject(handle, 100) if result == WAIT_FAILED: raise ctypes.WinError() elif result == WAIT_TIMEOUT: print(".", end='') return None else: data = ctypes.create_string_buffer(256) bytesRead = DWORD(0) print('?', end='') if not ReadFile(handle, data, 256, ctypes.byref(bytesRead), None): raise ctypes.WinError() # This ensures the non-blocking works with an actual console # Not checking the error, so the processing will still work with # other handle types FlushConsoleInputBuffer(handle) data = data.value data = data.replace('\r\n', '\n') data = data.replace('\r', '\n') print(repr(data) + " ", end='') return data
def _run_stdio(self): """Runs the process using the system standard I/O. IMPORTANT: stdin needs to be asynchronous, so the Python sys.stdin object is not used. Instead, msvcrt.kbhit/getwch are used asynchronously. """ # Disable Line and Echo mode #lpMode = DWORD() #handle = msvcrt.get_osfhandle(sys.stdin.fileno()) #if GetConsoleMode(handle, ctypes.byref(lpMode)): # set_console_mode = True # if not SetConsoleMode(handle, lpMode.value & # ~(ENABLE_ECHO_INPUT | ENABLE_LINE_INPUT | ENABLE_PROCESSED_INPUT)): # raise ctypes.WinError() if self.mergeout: return self.run(stdout_func = self._stdout_raw, stdin_func = self._stdin_raw_block) else: return self.run(stdout_func = self._stdout_raw, stdin_func = self._stdin_raw_block, stderr_func = self._stderr_raw) # Restore the previous console mode #if set_console_mode: # if not SetConsoleMode(handle, lpMode.value): # raise ctypes.WinError()
