iframe-proxy | Sunbelt Computer Software

History

100 lines (86 loc) · 3.04 KB

Raw

100

import atexit

import os

import signal

from . import util

__all__ = ['Popen']

# Start child process using fork

class Popen(object):

method = 'fork'

def __init__(self, process_obj):

util._flush_std_streams()

self.returncode = None

self.finalizer = None

self._launch(process_obj)

def duplicate_for_child(self, fd):

return fd

def poll(self, flag=os.WNOHANG):

if self.returncode is None:

try:

pid, sts = os.waitpid(self.pid, flag)

except OSError:

# Child process not yet created. See #1731717

# e.errno == errno.ECHILD == 10

return None

if pid == self.pid:

self.returncode = os.waitstatus_to_exitcode(sts)

return self.returncode

def wait(self, timeout=None):

if self.returncode is None:

if timeout is not None:

from multiprocessing.connection import wait

if not wait([self.sentinel], timeout):

return None

# This shouldn't block if wait() returned successfully.

return self.poll(os.WNOHANG if timeout == 0.0 else 0)

return self.returncode

def _send_signal(self, sig):

if self.returncode is None:

try:

os.kill(self.pid, sig)

except ProcessLookupError:

pass

except OSError:

if self.wait(timeout=0.1) is None:

raise

def interrupt(self):

self._send_signal(signal.SIGINT)

def terminate(self):

self._send_signal(signal.SIGTERM)

def kill(self):

self._send_signal(signal.SIGKILL)

def _launch(self, process_obj):

code = 1

parent_r, child_w = os.pipe()

child_r, parent_w = os.pipe()

# gh-146313: Tell the resource tracker's at-fork handler to keep

# the inherited pipe fd so this child reuses the parent's tracker

# (gh-80849) rather than closing it and launching its own.

from .resource_tracker import _fork_intent

_fork_intent.preserve_fd = True

try:

self.pid = os.fork()

finally:

# Reset in both parent and child so the flag does not leak

# into a subsequent raw os.fork() or nested Process launch.

_fork_intent.preserve_fd = False

if self.pid == 0:

try:

atexit._clear()

atexit.register(util._exit_function)

os.close(parent_r)

os.close(parent_w)

code = process_obj._bootstrap(parent_sentinel=child_r)

finally:

atexit._run_exitfuncs()

os._exit(code)

else:

os.close(child_w)

os.close(child_r)

self.finalizer = util.Finalize(self, util.close_fds,

(parent_r, parent_w,))

self.sentinel = parent_r

def close(self):

if self.finalizer is not None:

self.finalizer()

Sunbelt Computer Software

PL/B Language Development and Support

Navigation Menu

Uh oh!

Uh oh!

Sunbelt Computer Software

PL/B Language Development and Support

Uh oh!

Uh oh!

FilesExpand file tree

popen_fork.py

Latest commit

History

popen_fork.py

File metadata and controls