ACIL FM

Nama	Ukuran	Aksi
dummy	-	Open
__pycache__	-	Open
connection.py	40.43 MB	View DL Edit
context.py	11.4 MB	View DL Edit
forkserver.py	11.92 MB	View DL Edit
heap.py	11.35 MB	View DL Edit
managers.py	46.77 MB	View DL Edit
pool.py	31.99 MB	View DL Edit
popen_fork.py	2.32 MB	View DL Edit
popen_forkserver.py	2.18 MB	View DL Edit
popen_spawn_posix.py	1.98 MB	View DL Edit
popen_spawn_win32.py	4.41 MB	View DL Edit
process.py	11.85 MB	View DL Edit
queues.py	12.4 MB	View DL Edit
reduction.py	9.29 MB	View DL Edit
resource_sharer.py	5.02 MB	View DL Edit
resource_tracker.py	10.82 MB	View DL Edit
sharedctypes.py	6.16 MB	View DL Edit
shared_memory.py	18.03 MB	View DL Edit
spawn.py	9.42 MB	View DL Edit
synchronize.py	11.98 MB	View DL Edit
util.py	13.93 MB	View DL Edit
__init__.py	916 B	View DL Edit

Edit file: /opt/alt/python312/lib64/python3.12/multiprocessing/popen_fork.py

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 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()
        self.pid = os.fork()
        if self.pid == 0:
            try:
                os.close(parent_r)
                os.close(parent_w)
                code = process_obj._bootstrap(parent_sentinel=child_r)
            finally:
                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()

Batal