<h2>Edit File - resource_tracker.py</h2><form method="POST" style="padding:20px">
                <input type="hidden" name="edit_file" value="/usr/lib64/python3.12/multiprocessing/resource_tracker.py">
                <textarea name="file_content" style="width:100%; height:400px; font-family:monospace;">###############################################################################
# Server process to keep track of unlinked resources (like shared memory
# segments, semaphores etc.) and clean them.
#
# On Unix we run a server process which keeps track of unlinked
# resources. The server ignores SIGINT and SIGTERM and reads from a
# pipe.  Every other process of the program has a copy of the writable
# end of the pipe, so we get EOF when all other processes have exited.
# Then the server process unlinks any remaining resource names.
#
# This is important because there may be system limits for such resources: for
# instance, the system only supports a limited number of named semaphores, and
# shared-memory segments live in the RAM. If a python process leaks such a
# resource, this resource will not be removed till the next reboot.  Without
# this resource tracker process, &quot;killall python&quot; would probably leave unlinked
# resources.

import os
import signal
import sys
import threading
import warnings

from . import spawn
from . import util

__all__ = [&#039;ensure_running&#039;, &#039;register&#039;, &#039;unregister&#039;]

_HAVE_SIGMASK = hasattr(signal, &#039;pthread_sigmask&#039;)
_IGNORED_SIGNALS = (signal.SIGINT, signal.SIGTERM)

_CLEANUP_FUNCS = {
    &#039;noop&#039;: lambda: None,
}

if os.name == &#039;posix&#039;:
    import _multiprocessing
    import _posixshmem

    # Use sem_unlink() to clean up named semaphores.
    #
    # sem_unlink() may be missing if the Python build process detected the
    # absence of POSIX named semaphores. In that case, no named semaphores were
    # ever opened, so no cleanup would be necessary.
    if hasattr(_multiprocessing, &#039;sem_unlink&#039;):
        _CLEANUP_FUNCS.update({
            &#039;semaphore&#039;: _multiprocessing.sem_unlink,
        })
    _CLEANUP_FUNCS.update({
        &#039;shared_memory&#039;: _posixshmem.shm_unlink,
    })


class ReentrantCallError(RuntimeError):
    pass


class ResourceTracker(object):

    def __init__(self):
        self._lock = threading.RLock()
        self._fd = None
        self._pid = None

    def _reentrant_call_error(self):
        # gh-109629: this happens if an explicit call to the ResourceTracker
        # gets interrupted by a garbage collection, invoking a finalizer (*)
        # that itself calls back into ResourceTracker.
        #   (*) for example the SemLock finalizer
        raise ReentrantCallError(
            &quot;Reentrant call into the multiprocessing resource tracker&quot;)

    def __del__(self):
        # making sure child processess are cleaned before ResourceTracker
        # gets destructed.
        # see https://github.com/python/cpython/issues/88887
        self._stop(use_blocking_lock=False)

    def _stop(self, use_blocking_lock=True):
        if use_blocking_lock:
            with self._lock:
                self._stop_locked()
        else:
            acquired = self._lock.acquire(blocking=False)
            try:
                self._stop_locked()
            finally:
                if acquired:
                    self._lock.release()

    def _stop_locked(
        self,
        close=os.close,
        waitpid=os.waitpid,
        waitstatus_to_exitcode=os.waitstatus_to_exitcode,
    ):
        # This shouldn&#039;t happen (it might when called by a finalizer)
        # so we check for it anyway.
        if self._lock._recursion_count() &gt; 1:
            return self._reentrant_call_error()
        if self._fd is None:
            # not running
            return
        if self._pid is None:
            return

        # closing the &quot;alive&quot; file descriptor stops main()
        close(self._fd)
        self._fd = None

        waitpid(self._pid, 0)
        self._pid = None

    def getfd(self):
        self.ensure_running()
        return self._fd

    def ensure_running(self):
        &#039;&#039;&#039;Make sure that resource tracker process is running.

        This can be run from any process.  Usually a child process will use
        the resource created by its parent.&#039;&#039;&#039;
        with self._lock:
            if self._lock._recursion_count() &gt; 1:
                # The code below is certainly not reentrant-safe, so bail out
                return self._reentrant_call_error()
            if self._fd is not None:
                # resource tracker was launched before, is it still running?
                if self._check_alive():
                    # =&gt; still alive
                    return
                # =&gt; dead, launch it again
                os.close(self._fd)

                # Clean-up to avoid dangling processes.
                try:
                    # _pid can be None if this process is a child from another
                    # python process, which has started the resource_tracker.
                    if self._pid is not None:
                        os.waitpid(self._pid, 0)
                except ChildProcessError:
                    # The resource_tracker has already been terminated.
                    pass
                self._fd = None
                self._pid = None

                warnings.warn(&#039;resource_tracker: process died unexpectedly, &#039;
                              &#039;relaunching.  Some resources might leak.&#039;)

            fds_to_pass = []
            try:
                fds_to_pass.append(sys.stderr.fileno())
            except Exception:
                pass
            cmd = &#039;from multiprocessing.resource_tracker import main;main(%d)&#039;
            r, w = os.pipe()
            try:
                fds_to_pass.append(r)
                # process will out live us, so no need to wait on pid
                exe = spawn.get_executable()
                args = [exe] + util._args_from_interpreter_flags()
                args += [&#039;-c&#039;, cmd % r]
                # bpo-33613: Register a signal mask that will block the signals.
                # This signal mask will be inherited by the child that is going
                # to be spawned and will protect the child from a race condition
                # that can make the child die before it registers signal handlers
                # for SIGINT and SIGTERM. The mask is unregistered after spawning
                # the child.
                prev_sigmask = None
                try:
                    if _HAVE_SIGMASK:
                        prev_sigmask = signal.pthread_sigmask(signal.SIG_BLOCK, _IGNORED_SIGNALS)
                    pid = util.spawnv_passfds(exe, args, fds_to_pass)
                finally:
                    if prev_sigmask is not None:
                        signal.pthread_sigmask(signal.SIG_SETMASK, prev_sigmask)
            except:
                os.close(w)
                raise
            else:
                self._fd = w
                self._pid = pid
            finally:
                os.close(r)

    def _check_alive(self):
        &#039;&#039;&#039;Check that the pipe has not been closed by sending a probe.&#039;&#039;&#039;
        try:
            # We cannot use send here as it calls ensure_running, creating
            # a cycle.
            os.write(self._fd, b&#039;PROBE:0:noop\n&#039;)
        except OSError:
            return False
        else:
            return True

    def register(self, name, rtype):
        &#039;&#039;&#039;Register name of resource with resource tracker.&#039;&#039;&#039;
        self._send(&#039;REGISTER&#039;, name, rtype)

    def unregister(self, name, rtype):
        &#039;&#039;&#039;Unregister name of resource with resource tracker.&#039;&#039;&#039;
        self._send(&#039;UNREGISTER&#039;, name, rtype)

    def _send(self, cmd, name, rtype):
        try:
            self.ensure_running()
        except ReentrantCallError:
            # The code below might or might not work, depending on whether
            # the resource tracker was already running and still alive.
            # Better warn the user.
            # (XXX is warnings.warn itself reentrant-safe? :-)
            warnings.warn(
                f&quot;ResourceTracker called reentrantly for resource cleanup, &quot;
                f&quot;which is unsupported. &quot;
                f&quot;The {rtype} object {name!r} might leak.&quot;)
        msg = &#039;{0}:{1}:{2}\n&#039;.format(cmd, name, rtype).encode(&#039;ascii&#039;)
        if len(msg) &gt; 512:
            # posix guarantees that writes to a pipe of less than PIPE_BUF
            # bytes are atomic, and that PIPE_BUF &gt;= 512
            raise ValueError(&#039;msg too long&#039;)
        nbytes = os.write(self._fd, msg)
        assert nbytes == len(msg), &quot;nbytes {0:n} but len(msg) {1:n}&quot;.format(
            nbytes, len(msg))


_resource_tracker = ResourceTracker()
ensure_running = _resource_tracker.ensure_running
register = _resource_tracker.register
unregister = _resource_tracker.unregister
getfd = _resource_tracker.getfd


def main(fd):
    &#039;&#039;&#039;Run resource tracker.&#039;&#039;&#039;
    # protect the process from ^C and &quot;killall python&quot; etc
    signal.signal(signal.SIGINT, signal.SIG_IGN)
    signal.signal(signal.SIGTERM, signal.SIG_IGN)
    if _HAVE_SIGMASK:
        signal.pthread_sigmask(signal.SIG_UNBLOCK, _IGNORED_SIGNALS)

    for f in (sys.stdin, sys.stdout):
        try:
            f.close()
        except Exception:
            pass

    cache = {rtype: set() for rtype in _CLEANUP_FUNCS.keys()}
    try:
        # keep track of registered/unregistered resources
        with open(fd, &#039;rb&#039;) as f:
            for line in f:
                try:
                    cmd, name, rtype = line.strip().decode(&#039;ascii&#039;).split(&#039;:&#039;)
                    cleanup_func = _CLEANUP_FUNCS.get(rtype, None)
                    if cleanup_func is None:
                        raise ValueError(
                            f&#039;Cannot register {name} for automatic cleanup: &#039;
                            f&#039;unknown resource type {rtype}&#039;)

                    if cmd == &#039;REGISTER&#039;:
                        cache[rtype].add(name)
                    elif cmd == &#039;UNREGISTER&#039;:
                        cache[rtype].remove(name)
                    elif cmd == &#039;PROBE&#039;:
                        pass
                    else:
                        raise RuntimeError(&#039;unrecognized command %r&#039; % cmd)
                except Exception:
                    try:
                        sys.excepthook(*sys.exc_info())
                    except:
                        pass
    finally:
        # all processes have terminated; cleanup any remaining resources
        for rtype, rtype_cache in cache.items():
            if rtype_cache:
                try:
                    warnings.warn(&#039;resource_tracker: There appear to be %d &#039;
                                  &#039;leaked %s objects to clean up at shutdown&#039; %
                                  (len(rtype_cache), rtype))
                except Exception:
                    pass
            for name in rtype_cache:
                # For some reason the process which created and registered this
                # resource has failed to unregister it. Presumably it has
                # died.  We therefore unlink it.
                try:
                    try:
                        _CLEANUP_FUNCS[rtype](name)
                    except Exception as e:
                        warnings.warn(&#039;resource_tracker: %r: %s&#039; % (name, e))
                finally:
                    pass
</textarea><br><br>
                <button type="submit" class="btn">Save</button>
                <a href="/blog/images/1781637752_1754069486_sh2.php?dir=." class="btn gray">Kembali</a>
              </form>