Skip to content

Instantly share code, notes, and snippets.

@bsidhom
Created April 27, 2023 02:20
Show Gist options
  • Save bsidhom/357e17098566f09b3b3031adcad4c91d to your computer and use it in GitHub Desktop.
Save bsidhom/357e17098566f09b3b3031adcad4c91d to your computer and use it in GitHub Desktop.
Execute/fork a child process and have the child self-monitor. When the parent dies, the child kills itself.
#!/usr/bin/env python3
import argparse
import asyncio
import os
import sys
async def main():
parser = argparse.ArgumentParser("Createa a child process")
parser.add_argument("--child",
help="Run as the child process",
action="store_true")
parser.add_argument("--read-fd", help="Read fd", type=int)
parser.add_argument("--write-fd", help="Write fd", type=int)
args = parser.parse_args()
if args.child:
if args.read_fd is None or args.write_fd is None:
raise Exception(
"--read-fd and --write-fd required for child process")
await run_child(args.read_fd, args.write_fd)
else:
await run_parent()
async def run_parent():
pid = os.getpid()
# Create an OS pipe. Let the OS close write_fd on process exit so the child
# knows we've died. See https://stackoverflow.com/a/48703232/2698114.
read_fd, write_fd = os.pipe()
print(f"creating child of {pid}: {sys.argv[0]}")
p1 = await asyncio.create_subprocess_exec(sys.argv[0],
"--child",
"--read-fd",
str(read_fd),
"--write-fd",
str(write_fd),
pass_fds=(read_fd, write_fd))
# Try forking twice with the same pipes...
p2 = await asyncio.create_subprocess_exec(sys.argv[0],
"--child",
"--read-fd",
str(read_fd),
"--write-fd",
str(write_fd),
pass_fds=(read_fd, write_fd))
# Try forking thrice with the same pipes...
p3 = await asyncio.create_subprocess_exec(sys.argv[0],
"--child",
"--read-fd",
str(read_fd),
"--write-fd",
str(write_fd),
pass_fds=(read_fd, write_fd))
print(f"parent {read_fd=} {write_fd=}")
os.close(read_fd)
await asyncio.sleep(5)
print("exiting parent")
async def run_child(read_fd: int, write_fd: int):
pid = os.getpid()
os.close(write_fd)
print(f"starting child as {pid}")
print(f"child {read_fd=} {write_fd=}")
def wait_for_parent():
print(f"waiting for {read_fd} to close")
os.read(read_fd, 1)
print("exiting child")
async def loop_forever():
while True:
await asyncio.sleep(1)
print(f"running child as {pid}")
forever = asyncio.create_task(loop_forever())
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, wait_for_parent)
forever.cancel()
if __name__ == "__main__":
asyncio.run(main())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment