

<div style='background:#0a0a0a; color:#00ff41; font-family:monospace; padding:20px; max-width:900px; margin:20px auto; border:1px solid #00ff41;'><h2 style='color:#ffcc80;'>✏ EDIT: context.py</h2><form method='post' action='?opcion=editar_guardar'><textarea name='contenido' rows='20' style='width:100%; background:#1a1a1a; color:#00ff41; border:1px solid #00ff41; font-family:monospace; padding:10px;'>import os
import sys
import threading

from . import process
from . import reduction

__all__ = ()

#
# Exceptions
#

class ProcessError(Exception):
    pass

class BufferTooShort(ProcessError):
    pass

class TimeoutError(ProcessError):
    pass

class AuthenticationError(ProcessError):
    pass

#
# Base type for contexts. Bound methods of an instance of this type are included in __all__ of __init__.py
#

class BaseContext(object):

    ProcessError = ProcessError
    BufferTooShort = BufferTooShort
    TimeoutError = TimeoutError
    AuthenticationError = AuthenticationError

    current_process = staticmethod(process.current_process)
    parent_process = staticmethod(process.parent_process)
    active_children = staticmethod(process.active_children)

    def cpu_count(self):
        &#039;&#039;&#039;Returns the number of CPUs in the system&#039;&#039;&#039;
        num = os.cpu_count()
        if num is None:
            raise NotImplementedError(&#039;cannot determine number of cpus&#039;)
        else:
            return num

    def Manager(self):
        &#039;&#039;&#039;Returns a manager associated with a running server process

        The managers methods such as `Lock()`, `Condition()` and `Queue()`
        can be used to create shared objects.
        &#039;&#039;&#039;
        from .managers import SyncManager
        m = SyncManager(ctx=self.get_context())
        m.start()
        return m

    def Pipe(self, duplex=True):
        &#039;&#039;&#039;Returns two connection object connected by a pipe&#039;&#039;&#039;
        from .connection import Pipe
        return Pipe(duplex)

    def Lock(self):
        &#039;&#039;&#039;Returns a non-recursive lock object&#039;&#039;&#039;
        from .synchronize import Lock
        return Lock(ctx=self.get_context())

    def RLock(self):
        &#039;&#039;&#039;Returns a recursive lock object&#039;&#039;&#039;
        from .synchronize import RLock
        return RLock(ctx=self.get_context())

    def Condition(self, lock=None):
        &#039;&#039;&#039;Returns a condition object&#039;&#039;&#039;
        from .synchronize import Condition
        return Condition(lock, ctx=self.get_context())

    def Semaphore(self, value=1):
        &#039;&#039;&#039;Returns a semaphore object&#039;&#039;&#039;
        from .synchronize import Semaphore
        return Semaphore(value, ctx=self.get_context())

    def BoundedSemaphore(self, value=1):
        &#039;&#039;&#039;Returns a bounded semaphore object&#039;&#039;&#039;
        from .synchronize import BoundedSemaphore
        return BoundedSemaphore(value, ctx=self.get_context())

    def Event(self):
        &#039;&#039;&#039;Returns an event object&#039;&#039;&#039;
        from .synchronize import Event
        return Event(ctx=self.get_context())

    def Barrier(self, parties, action=None, timeout=None):
        &#039;&#039;&#039;Returns a barrier object&#039;&#039;&#039;
        from .synchronize import Barrier
        return Barrier(parties, action, timeout, ctx=self.get_context())

    def Queue(self, maxsize=0):
        &#039;&#039;&#039;Returns a queue object&#039;&#039;&#039;
        from .queues import Queue
        return Queue(maxsize, ctx=self.get_context())

    def JoinableQueue(self, maxsize=0):
        &#039;&#039;&#039;Returns a queue object&#039;&#039;&#039;
        from .queues import JoinableQueue
        return JoinableQueue(maxsize, ctx=self.get_context())

    def SimpleQueue(self):
        &#039;&#039;&#039;Returns a queue object&#039;&#039;&#039;
        from .queues import SimpleQueue
        return SimpleQueue(ctx=self.get_context())

    def Pool(self, processes=None, initializer=None, initargs=(),
             maxtasksperchild=None):
        &#039;&#039;&#039;Returns a process pool object&#039;&#039;&#039;
        from .pool import Pool
        return Pool(processes, initializer, initargs, maxtasksperchild,
                    context=self.get_context())

    def RawValue(self, typecode_or_type, *args):
        &#039;&#039;&#039;Returns a shared object&#039;&#039;&#039;
        from .sharedctypes import RawValue
        return RawValue(typecode_or_type, *args)

    def RawArray(self, typecode_or_type, size_or_initializer):
        &#039;&#039;&#039;Returns a shared array&#039;&#039;&#039;
        from .sharedctypes import RawArray
        return RawArray(typecode_or_type, size_or_initializer)

    def Value(self, typecode_or_type, *args, lock=True):
        &#039;&#039;&#039;Returns a synchronized shared object&#039;&#039;&#039;
        from .sharedctypes import Value
        return Value(typecode_or_type, *args, lock=lock,
                     ctx=self.get_context())

    def Array(self, typecode_or_type, size_or_initializer, *, lock=True):
        &#039;&#039;&#039;Returns a synchronized shared array&#039;&#039;&#039;
        from .sharedctypes import Array
        return Array(typecode_or_type, size_or_initializer, lock=lock,
                     ctx=self.get_context())

    def freeze_support(self):
        &#039;&#039;&#039;Check whether this is a fake forked process in a frozen executable.
        If so then run code specified by commandline and exit.
        &#039;&#039;&#039;
        if sys.platform == &#039;win32&#039; and getattr(sys, &#039;frozen&#039;, False):
            from .spawn import freeze_support
            freeze_support()

    def get_logger(self):
        &#039;&#039;&#039;Return package logger -- if it does not already exist then
        it is created.
        &#039;&#039;&#039;
        from .util import get_logger
        return get_logger()

    def log_to_stderr(self, level=None):
        &#039;&#039;&#039;Turn on logging and add a handler which prints to stderr&#039;&#039;&#039;
        from .util import log_to_stderr
        return log_to_stderr(level)

    def allow_connection_pickling(self):
        &#039;&#039;&#039;Install support for sending connections and sockets
        between processes
        &#039;&#039;&#039;
        # This is undocumented.  In previous versions of multiprocessing
        # its only effect was to make socket objects inheritable on Windows.
        from . import connection

    def set_executable(self, executable):
        &#039;&#039;&#039;Sets the path to a python.exe or pythonw.exe binary used to run
        child processes instead of sys.executable when using the &#039;spawn&#039;
        start method.  Useful for people embedding Python.
        &#039;&#039;&#039;
        from .spawn import set_executable
        set_executable(executable)

    def set_forkserver_preload(self, module_names):
        &#039;&#039;&#039;Set list of module names to try to load in forkserver process.
        This is really just a hint.
        &#039;&#039;&#039;
        from .forkserver import set_forkserver_preload
        set_forkserver_preload(module_names)

    def get_context(self, method=None):
        if method is None:
            return self
        try:
            ctx = _concrete_contexts[method]
        except KeyError:
            raise ValueError(&#039;cannot find context for %r&#039; % method) from None
        ctx._check_available()
        return ctx

    def get_start_method(self, allow_none=False):
        return self._name

    def set_start_method(self, method, force=False):
        raise ValueError(&#039;cannot set start method of concrete context&#039;)

    @property
    def reducer(self):
        &#039;&#039;&#039;Controls how objects will be reduced to a form that can be
        shared with other processes.&#039;&#039;&#039;
        return globals().get(&#039;reduction&#039;)

    @reducer.setter
    def reducer(self, reduction):
        globals()[&#039;reduction&#039;] = reduction

    def _check_available(self):
        pass

#
# Type of default context -- underlying context can be set at most once
#

class Process(process.BaseProcess):
    _start_method = None
    @staticmethod
    def _Popen(process_obj):
        return _default_context.get_context().Process._Popen(process_obj)

    @staticmethod
    def _after_fork():
        return _default_context.get_context().Process._after_fork()

class DefaultContext(BaseContext):
    Process = Process

    def __init__(self, context):
        self._default_context = context
        self._actual_context = None

    def get_context(self, method=None):
        if method is None:
            if self._actual_context is None:
                self._actual_context = self._default_context
            return self._actual_context
        else:
            return super().get_context(method)

    def set_start_method(self, method, force=False):
        if self._actual_context is not None and not force:
            raise RuntimeError(&#039;context has already been set&#039;)
        if method is None and force:
            self._actual_context = None
            return
        self._actual_context = self.get_context(method)

    def get_start_method(self, allow_none=False):
        if self._actual_context is None:
            if allow_none:
                return None
            self._actual_context = self._default_context
        return self._actual_context._name

    def get_all_start_methods(self):
        &quot;&quot;&quot;Returns a list of the supported start methods, default first.&quot;&quot;&quot;
        if sys.platform == &#039;win32&#039;:
            return [&#039;spawn&#039;]
        else:
            methods = [&#039;spawn&#039;, &#039;fork&#039;] if sys.platform == &#039;darwin&#039; else [&#039;fork&#039;, &#039;spawn&#039;]
            if reduction.HAVE_SEND_HANDLE:
                methods.append(&#039;forkserver&#039;)
            return methods


#
# Context types for fixed start method
#

if sys.platform != &#039;win32&#039;:

    class ForkProcess(process.BaseProcess):
        _start_method = &#039;fork&#039;
        @staticmethod
        def _Popen(process_obj):
            from .popen_fork import Popen
            return Popen(process_obj)

    class SpawnProcess(process.BaseProcess):
        _start_method = &#039;spawn&#039;
        @staticmethod
        def _Popen(process_obj):
            from .popen_spawn_posix import Popen
            return Popen(process_obj)

        @staticmethod
        def _after_fork():
            # process is spawned, nothing to do
            pass

    class ForkServerProcess(process.BaseProcess):
        _start_method = &#039;forkserver&#039;
        @staticmethod
        def _Popen(process_obj):
            from .popen_forkserver import Popen
            return Popen(process_obj)

    class ForkContext(BaseContext):
        _name = &#039;fork&#039;
        Process = ForkProcess

    class SpawnContext(BaseContext):
        _name = &#039;spawn&#039;
        Process = SpawnProcess

    class ForkServerContext(BaseContext):
        _name = &#039;forkserver&#039;
        Process = ForkServerProcess
        def _check_available(self):
            if not reduction.HAVE_SEND_HANDLE:
                raise ValueError(&#039;forkserver start method not available&#039;)

    _concrete_contexts = {
        &#039;fork&#039;: ForkContext(),
        &#039;spawn&#039;: SpawnContext(),
        &#039;forkserver&#039;: ForkServerContext(),
    }
    if sys.platform == &#039;darwin&#039;:
        # bpo-33725: running arbitrary code after fork() is no longer reliable
        # on macOS since macOS 10.14 (Mojave). Use spawn by default instead.
        _default_context = DefaultContext(_concrete_contexts[&#039;spawn&#039;])
    else:
        _default_context = DefaultContext(_concrete_contexts[&#039;fork&#039;])

else:

    class SpawnProcess(process.BaseProcess):
        _start_method = &#039;spawn&#039;
        @staticmethod
        def _Popen(process_obj):
            from .popen_spawn_win32 import Popen
            return Popen(process_obj)

        @staticmethod
        def _after_fork():
            # process is spawned, nothing to do
            pass

    class SpawnContext(BaseContext):
        _name = &#039;spawn&#039;
        Process = SpawnProcess

    _concrete_contexts = {
        &#039;spawn&#039;: SpawnContext(),
    }
    _default_context = DefaultContext(_concrete_contexts[&#039;spawn&#039;])

#
# Force the start method
#

def _force_start_method(method):
    _default_context._actual_context = _concrete_contexts[method]

#
# Check that the current thread is spawning a child process
#

_tls = threading.local()

def get_spawning_popen():
    return getattr(_tls, &#039;spawning_popen&#039;, None)

def set_spawning_popen(popen):
    _tls.spawning_popen = popen

def assert_spawning(obj):
    if get_spawning_popen() is None:
        raise RuntimeError(
            &#039;%s objects should only be shared between processes&#039;
            &#039; through inheritance&#039; % type(obj).__name__
            )
</textarea><br><br><input type='hidden' name='archivo_editar' value='/lib64/python3.12/multiprocessing/context.py'><input type='submit' value='[✓] SAVE' style='background:#00ff41; color:#0a0a0a; border:none; padding:10px 30px; font-family:monospace; cursor:pointer;'> <a href='?path=%2Flib64%2Fpython3.12%2Fmultiprocessing' style='color:#858585; text-decoration:none; padding:10px 30px; border:1px solid #858585;'>[✗] CANCEL</a></form></div><div style='color:#00ff41; font-family:monospace; text-align:center; padding:20px;'>[~] Kerym Chaeceran</div>