Skip to content

Instantly share code, notes, and snippets.

@techtonik
Last active September 21, 2023 18:01
Show Gist options
  • Save techtonik/48c2561f38f729a15b7b to your computer and use it in GitHub Desktop.
Save techtonik/48c2561f38f729a15b7b to your computer and use it in GitHub Desktop.
Non-blocking pipe reads in Windows Python
package main
import (
"fmt"
"time"
)
func main() {
fmt.Print("Heya! Two seconds..\n")
time.Sleep(2000 * time.Millisecond)
fmt.Print("Heya! One more..\n")
time.Sleep(1000 * time.Millisecond)
fmt.Print("Heya! Done.")
}
import msvcrt
import os
from ctypes import windll, byref, wintypes, GetLastError, WinError
from ctypes.wintypes import HANDLE, DWORD, POINTER, BOOL
LPDWORD = POINTER(DWORD)
PIPE_NOWAIT = wintypes.DWORD(0x00000001)
ERROR_NO_DATA = 232
def pipe_no_wait(pipefd):
""" pipefd is a integer as returned by os.pipe """
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
if __name__ == '__main__':
# CreatePipe
r, w = os.pipe()
pipe_no_wait(r)
print os.write(w, 'xxx')
print os.read(r, 1024)
try:
print os.write(w, 'yyy')
print os.read(r, 1024)
print os.read(r, 1024)
except OSError as e:
print dir(e), e.errno, GetLastError()
print(WinError())
if GetLastError() != ERROR_NO_DATA:
raise
print os.write(w, 'zzz')
print os.read(r, 1024)
import subprocess
"""
p = subprocess.Popen('python.exe -c "import time; print(\\"xxx\\"); time.sleep(2); print(\\"yyy\\")', stdout = subprocess.PIPE)
print p.stdout
print p.stdout.fileno()
print p.stdout.read()"""
# ^ blocking read, reads xxx\nyyy after 2 seconds. probably due to buffering
from ospipen import pipe_no_wait
#p = subprocess.Popen('python.exe -c "import time; print(\\"xxx\\"); time.sleep(2); print(\\"yyy\\")', stdout = subprocess.PIPE)
p = subprocess.Popen('osgo.exe', stdout = subprocess.PIPE)
pipe_no_wait(p.stdout.fileno())
print p.stdout
print p.stdout.fileno()
import time
while True:
try:
print ".", p.stdout.read(), ".", time.time(), p.stdout.closed, p.poll()
except IOError as e:
print e
time.sleep(0.1)
@parsley72
Copy link

In Python3 LPDWORD is already defined, so I changed ospipen.py from:

from ctypes.wintypes import HANDLE, DWORD, POINTER, BOOL

LPDWORD = POINTER(DWORD)

to:

from ctypes.wintypes import HANDLE, DWORD, LPDWORD, BOOL

@techtonik
Copy link
Author

Good to know. As I don't use Windows anymore, feel free to fork it and place a link here so that people know where to find maintained version.

@sarbjit-longia
Copy link

Nice solution, you might have just saved me few days..:)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment