Skip to content

Instantly share code, notes, and snippets.

@thediveo
Created December 21, 2023 19:50
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save thediveo/45259091c42ffb3e7a56c400e054c496 to your computer and use it in GitHub Desktop.
Save thediveo/45259091c42ffb3e7a56c400e054c496 to your computer and use it in GitHub Desktop.
Detecting the reader of a named pipe disconnecting (or not, I'm looking at you, macOS)
import os
import tempfile
import atexit
import shutil
import time
import select
def writer(fifoname):
print("writer: start...")
w = open(fifoname, 'w')
poller = select.poll()
poller.register(w, select.POLLERR | select.POLLHUP)
poll = poller.poll(4.0*1000)
if len(poll) == 0:
print("ERROR: timed out, disconnect not detected")
return
print("reader has disconnected", poll)
def reader(fifoname):
print("reader: start...")
r = open(fifoname, 'r')
print("sleeping...")
time.sleep(2.0)
r.close()
print("reader closed")
tmpdir = tempfile.mkdtemp(prefix="pype-")
fifoname = os.path.join(tmpdir, "fifo")
print("named pipe", fifoname)
os.mkfifo(fifoname, 0o600)
atexit.register(shutil.rmtree, tmpdir)
childpid = os.fork()
if childpid == 0:
writer(fifoname)
else:
atexit.unregister(shutil.rmtree)
reader(fifoname)
print("waiting for writer to terminate...")
os.waitpid(childpid, 0)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment