

<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: managers.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;'>#
# Module providing manager classes for dealing
# with shared objects
#
# multiprocessing/managers.py
#
# Copyright (c) 2006-2008, R Oudkerk
# Licensed to PSF under a Contributor Agreement.
#

__all__ = [ &#039;BaseManager&#039;, &#039;SyncManager&#039;, &#039;BaseProxy&#039;, &#039;Token&#039; ]

#
# Imports
#

import sys
import threading
import signal
import array
import queue
import time
import types
import os
from os import getpid

from traceback import format_exc

from . import connection
from .context import reduction, get_spawning_popen, ProcessError
from . import pool
from . import process
from . import util
from . import get_context
try:
    from . import shared_memory
except ImportError:
    HAS_SHMEM = False
else:
    HAS_SHMEM = True
    __all__.append(&#039;SharedMemoryManager&#039;)

#
# Register some things for pickling
#

def reduce_array(a):
    return array.array, (a.typecode, a.tobytes())
reduction.register(array.array, reduce_array)

view_types = [type(getattr({}, name)()) for name in (&#039;items&#039;,&#039;keys&#039;,&#039;values&#039;)]
def rebuild_as_list(obj):
    return list, (list(obj),)
for view_type in view_types:
    reduction.register(view_type, rebuild_as_list)
del view_type, view_types

#
# Type for identifying shared objects
#

class Token(object):
    &#039;&#039;&#039;
    Type to uniquely identify a shared object
    &#039;&#039;&#039;
    __slots__ = (&#039;typeid&#039;, &#039;address&#039;, &#039;id&#039;)

    def __init__(self, typeid, address, id):
        (self.typeid, self.address, self.id) = (typeid, address, id)

    def __getstate__(self):
        return (self.typeid, self.address, self.id)

    def __setstate__(self, state):
        (self.typeid, self.address, self.id) = state

    def __repr__(self):
        return &#039;%s(typeid=%r, address=%r, id=%r)&#039; % \
               (self.__class__.__name__, self.typeid, self.address, self.id)

#
# Function for communication with a manager&#039;s server process
#

def dispatch(c, id, methodname, args=(), kwds={}):
    &#039;&#039;&#039;
    Send a message to manager using connection `c` and return response
    &#039;&#039;&#039;
    c.send((id, methodname, args, kwds))
    kind, result = c.recv()
    if kind == &#039;#RETURN&#039;:
        return result
    raise convert_to_error(kind, result)

def convert_to_error(kind, result):
    if kind == &#039;#ERROR&#039;:
        return result
    elif kind in (&#039;#TRACEBACK&#039;, &#039;#UNSERIALIZABLE&#039;):
        if not isinstance(result, str):
            raise TypeError(
                &quot;Result {0!r} (kind &#039;{1}&#039;) type is {2}, not str&quot;.format(
                    result, kind, type(result)))
        if kind == &#039;#UNSERIALIZABLE&#039;:
            return RemoteError(&#039;Unserializable message: %s\n&#039; % result)
        else:
            return RemoteError(result)
    else:
        return ValueError(&#039;Unrecognized message type {!r}&#039;.format(kind))

class RemoteError(Exception):
    def __str__(self):
        return (&#039;\n&#039; + &#039;-&#039;*75 + &#039;\n&#039; + str(self.args[0]) + &#039;-&#039;*75)

#
# Functions for finding the method names of an object
#

def all_methods(obj):
    &#039;&#039;&#039;
    Return a list of names of methods of `obj`
    &#039;&#039;&#039;
    temp = []
    for name in dir(obj):
        func = getattr(obj, name)
        if callable(func):
            temp.append(name)
    return temp

def public_methods(obj):
    &#039;&#039;&#039;
    Return a list of names of methods of `obj` which do not start with &#039;_&#039;
    &#039;&#039;&#039;
    return [name for name in all_methods(obj) if name[0] != &#039;_&#039;]

#
# Server which is run in a process controlled by a manager
#

class Server(object):
    &#039;&#039;&#039;
    Server class which runs in a process controlled by a manager object
    &#039;&#039;&#039;
    public = [&#039;shutdown&#039;, &#039;create&#039;, &#039;accept_connection&#039;, &#039;get_methods&#039;,
              &#039;debug_info&#039;, &#039;number_of_objects&#039;, &#039;dummy&#039;, &#039;incref&#039;, &#039;decref&#039;]

    def __init__(self, registry, address, authkey, serializer):
        if not isinstance(authkey, bytes):
            raise TypeError(
                &quot;Authkey {0!r} is type {1!s}, not bytes&quot;.format(
                    authkey, type(authkey)))
        self.registry = registry
        self.authkey = process.AuthenticationString(authkey)
        Listener, Client = listener_client[serializer]

        # do authentication later
        self.listener = Listener(address=address, backlog=128)
        self.address = self.listener.address

        self.id_to_obj = {&#039;0&#039;: (None, ())}
        self.id_to_refcount = {}
        self.id_to_local_proxy_obj = {}
        self.mutex = threading.Lock()

    def serve_forever(self):
        &#039;&#039;&#039;
        Run the server forever
        &#039;&#039;&#039;
        self.stop_event = threading.Event()
        process.current_process()._manager_server = self
        try:
            accepter = threading.Thread(target=self.accepter)
            accepter.daemon = True
            accepter.start()
            try:
                while not self.stop_event.is_set():
                    self.stop_event.wait(1)
            except (KeyboardInterrupt, SystemExit):
                pass
        finally:
            if sys.stdout != sys.__stdout__: # what about stderr?
                util.debug(&#039;resetting stdout, stderr&#039;)
                sys.stdout = sys.__stdout__
                sys.stderr = sys.__stderr__
            sys.exit(0)

    def accepter(self):
        while True:
            try:
                c = self.listener.accept()
            except OSError:
                continue
            t = threading.Thread(target=self.handle_request, args=(c,))
            t.daemon = True
            t.start()

    def _handle_request(self, c):
        request = None
        try:
            connection.deliver_challenge(c, self.authkey)
            connection.answer_challenge(c, self.authkey)
            request = c.recv()
            ignore, funcname, args, kwds = request
            assert funcname in self.public, &#039;%r unrecognized&#039; % funcname
            func = getattr(self, funcname)
        except Exception:
            msg = (&#039;#TRACEBACK&#039;, format_exc())
        else:
            try:
                result = func(c, *args, **kwds)
            except Exception:
                msg = (&#039;#TRACEBACK&#039;, format_exc())
            else:
                msg = (&#039;#RETURN&#039;, result)

        try:
            c.send(msg)
        except Exception as e:
            try:
                c.send((&#039;#TRACEBACK&#039;, format_exc()))
            except Exception:
                pass
            util.info(&#039;Failure to send message: %r&#039;, msg)
            util.info(&#039; ... request was %r&#039;, request)
            util.info(&#039; ... exception was %r&#039;, e)

    def handle_request(self, conn):
        &#039;&#039;&#039;
        Handle a new connection
        &#039;&#039;&#039;
        try:
            self._handle_request(conn)
        except SystemExit:
            # Server.serve_client() calls sys.exit(0) on EOF
            pass
        finally:
            conn.close()

    def serve_client(self, conn):
        &#039;&#039;&#039;
        Handle requests from the proxies in a particular process/thread
        &#039;&#039;&#039;
        util.debug(&#039;starting server thread to service %r&#039;,
                   threading.current_thread().name)

        recv = conn.recv
        send = conn.send
        id_to_obj = self.id_to_obj

        while not self.stop_event.is_set():

            try:
                methodname = obj = None
                request = recv()
                ident, methodname, args, kwds = request
                try:
                    obj, exposed, gettypeid = id_to_obj[ident]
                except KeyError as ke:
                    try:
                        obj, exposed, gettypeid = \
                            self.id_to_local_proxy_obj[ident]
                    except KeyError:
                        raise ke

                if methodname not in exposed:
                    raise AttributeError(
                        &#039;method %r of %r object is not in exposed=%r&#039; %
                        (methodname, type(obj), exposed)
                        )

                function = getattr(obj, methodname)

                try:
                    res = function(*args, **kwds)
                except Exception as e:
                    msg = (&#039;#ERROR&#039;, e)
                else:
                    typeid = gettypeid and gettypeid.get(methodname, None)
                    if typeid:
                        rident, rexposed = self.create(conn, typeid, res)
                        token = Token(typeid, self.address, rident)
                        msg = (&#039;#PROXY&#039;, (rexposed, token))
                    else:
                        msg = (&#039;#RETURN&#039;, res)

            except AttributeError:
                if methodname is None:
                    msg = (&#039;#TRACEBACK&#039;, format_exc())
                else:
                    try:
                        fallback_func = self.fallback_mapping[methodname]
                        result = fallback_func(
                            self, conn, ident, obj, *args, **kwds
                            )
                        msg = (&#039;#RETURN&#039;, result)
                    except Exception:
                        msg = (&#039;#TRACEBACK&#039;, format_exc())

            except EOFError:
                util.debug(&#039;got EOF -- exiting thread serving %r&#039;,
                           threading.current_thread().name)
                sys.exit(0)

            except Exception:
                msg = (&#039;#TRACEBACK&#039;, format_exc())

            try:
                try:
                    send(msg)
                except Exception:
                    send((&#039;#UNSERIALIZABLE&#039;, format_exc()))
            except Exception as e:
                util.info(&#039;exception in thread serving %r&#039;,
                        threading.current_thread().name)
                util.info(&#039; ... message was %r&#039;, msg)
                util.info(&#039; ... exception was %r&#039;, e)
                conn.close()
                sys.exit(1)

    def fallback_getvalue(self, conn, ident, obj):
        return obj

    def fallback_str(self, conn, ident, obj):
        return str(obj)

    def fallback_repr(self, conn, ident, obj):
        return repr(obj)

    fallback_mapping = {
        &#039;__str__&#039;:fallback_str,
        &#039;__repr__&#039;:fallback_repr,
        &#039;#GETVALUE&#039;:fallback_getvalue
        }

    def dummy(self, c):
        pass

    def debug_info(self, c):
        &#039;&#039;&#039;
        Return some info --- useful to spot problems with refcounting
        &#039;&#039;&#039;
        # Perhaps include debug info about &#039;c&#039;?
        with self.mutex:
            result = []
            keys = list(self.id_to_refcount.keys())
            keys.sort()
            for ident in keys:
                if ident != &#039;0&#039;:
                    result.append(&#039;  %s:       refcount=%s\n    %s&#039; %
                                  (ident, self.id_to_refcount[ident],
                                   str(self.id_to_obj[ident][0])[:75]))
            return &#039;\n&#039;.join(result)

    def number_of_objects(self, c):
        &#039;&#039;&#039;
        Number of shared objects
        &#039;&#039;&#039;
        # Doesn&#039;t use (len(self.id_to_obj) - 1) as we shouldn&#039;t count ident=&#039;0&#039;
        return len(self.id_to_refcount)

    def shutdown(self, c):
        &#039;&#039;&#039;
        Shutdown this process
        &#039;&#039;&#039;
        try:
            util.debug(&#039;manager received shutdown message&#039;)
            c.send((&#039;#RETURN&#039;, None))
        except:
            import traceback
            traceback.print_exc()
        finally:
            self.stop_event.set()

    def create(self, c, typeid, /, *args, **kwds):
        &#039;&#039;&#039;
        Create a new shared object and return its id
        &#039;&#039;&#039;
        with self.mutex:
            callable, exposed, method_to_typeid, proxytype = \
                      self.registry[typeid]

            if callable is None:
                if kwds or (len(args) != 1):
                    raise ValueError(
                        &quot;Without callable, must have one non-keyword argument&quot;)
                obj = args[0]
            else:
                obj = callable(*args, **kwds)

            if exposed is None:
                exposed = public_methods(obj)
            if method_to_typeid is not None:
                if not isinstance(method_to_typeid, dict):
                    raise TypeError(
                        &quot;Method_to_typeid {0!r}: type {1!s}, not dict&quot;.format(
                            method_to_typeid, type(method_to_typeid)))
                exposed = list(exposed) + list(method_to_typeid)

            ident = &#039;%x&#039; % id(obj)  # convert to string because xmlrpclib
                                    # only has 32 bit signed integers
            util.debug(&#039;%r callable returned object with id %r&#039;, typeid, ident)

            self.id_to_obj[ident] = (obj, set(exposed), method_to_typeid)
            if ident not in self.id_to_refcount:
                self.id_to_refcount[ident] = 0

        self.incref(c, ident)
        return ident, tuple(exposed)

    def get_methods(self, c, token):
        &#039;&#039;&#039;
        Return the methods of the shared object indicated by token
        &#039;&#039;&#039;
        return tuple(self.id_to_obj[token.id][1])

    def accept_connection(self, c, name):
        &#039;&#039;&#039;
        Spawn a new thread to serve this connection
        &#039;&#039;&#039;
        threading.current_thread().name = name
        c.send((&#039;#RETURN&#039;, None))
        self.serve_client(c)

    def incref(self, c, ident):
        with self.mutex:
            try:
                self.id_to_refcount[ident] += 1
            except KeyError as ke:
                # If no external references exist but an internal (to the
                # manager) still does and a new external reference is created
                # from it, restore the manager&#039;s tracking of it from the
                # previously stashed internal ref.
                if ident in self.id_to_local_proxy_obj:
                    self.id_to_refcount[ident] = 1
                    self.id_to_obj[ident] = \
                        self.id_to_local_proxy_obj[ident]
                    util.debug(&#039;Server re-enabled tracking &amp; INCREF %r&#039;, ident)
                else:
                    raise ke

    def decref(self, c, ident):
        if ident not in self.id_to_refcount and \
            ident in self.id_to_local_proxy_obj:
            util.debug(&#039;Server DECREF skipping %r&#039;, ident)
            return

        with self.mutex:
            if self.id_to_refcount[ident] &lt;= 0:
                raise AssertionError(
                    &quot;Id {0!s} ({1!r}) has refcount {2:n}, not 1+&quot;.format(
                        ident, self.id_to_obj[ident],
                        self.id_to_refcount[ident]))
            self.id_to_refcount[ident] -= 1
            if self.id_to_refcount[ident] == 0:
                del self.id_to_refcount[ident]

        if ident not in self.id_to_refcount:
            # Two-step process in case the object turns out to contain other
            # proxy objects (e.g. a managed list of managed lists).
            # Otherwise, deleting self.id_to_obj[ident] would trigger the
            # deleting of the stored value (another managed object) which would
            # in turn attempt to acquire the mutex that is already held here.
            self.id_to_obj[ident] = (None, (), None)  # thread-safe
            util.debug(&#039;disposing of obj with id %r&#039;, ident)
            with self.mutex:
                del self.id_to_obj[ident]


#
# Class to represent state of a manager
#

class State(object):
    __slots__ = [&#039;value&#039;]
    INITIAL = 0
    STARTED = 1
    SHUTDOWN = 2

#
# Mapping from serializer name to Listener and Client types
#

listener_client = {
    &#039;pickle&#039; : (connection.Listener, connection.Client),
    &#039;xmlrpclib&#039; : (connection.XmlListener, connection.XmlClient)
    }

#
# Definition of BaseManager
#

class BaseManager(object):
    &#039;&#039;&#039;
    Base class for managers
    &#039;&#039;&#039;
    _registry = {}
    _Server = Server

    def __init__(self, address=None, authkey=None, serializer=&#039;pickle&#039;,
                 ctx=None, *, shutdown_timeout=1.0):
        if authkey is None:
            authkey = process.current_process().authkey
        self._address = address     # XXX not final address if eg (&#039;&#039;, 0)
        self._authkey = process.AuthenticationString(authkey)
        self._state = State()
        self._state.value = State.INITIAL
        self._serializer = serializer
        self._Listener, self._Client = listener_client[serializer]
        self._ctx = ctx or get_context()
        self._shutdown_timeout = shutdown_timeout

    def get_server(self):
        &#039;&#039;&#039;
        Return server object with serve_forever() method and address attribute
        &#039;&#039;&#039;
        if self._state.value != State.INITIAL:
            if self._state.value == State.STARTED:
                raise ProcessError(&quot;Already started server&quot;)
            elif self._state.value == State.SHUTDOWN:
                raise ProcessError(&quot;Manager has shut down&quot;)
            else:
                raise ProcessError(
                    &quot;Unknown state {!r}&quot;.format(self._state.value))
        return Server(self._registry, self._address,
                      self._authkey, self._serializer)

    def connect(self):
        &#039;&#039;&#039;
        Connect manager object to the server process
        &#039;&#039;&#039;
        Listener, Client = listener_client[self._serializer]
        conn = Client(self._address, authkey=self._authkey)
        dispatch(conn, None, &#039;dummy&#039;)
        self._state.value = State.STARTED

    def start(self, initializer=None, initargs=()):
        &#039;&#039;&#039;
        Spawn a server process for this manager object
        &#039;&#039;&#039;
        if self._state.value != State.INITIAL:
            if self._state.value == State.STARTED:
                raise ProcessError(&quot;Already started server&quot;)
            elif self._state.value == State.SHUTDOWN:
                raise ProcessError(&quot;Manager has shut down&quot;)
            else:
                raise ProcessError(
                    &quot;Unknown state {!r}&quot;.format(self._state.value))

        if initializer is not None and not callable(initializer):
            raise TypeError(&#039;initializer must be a callable&#039;)

        # pipe over which we will retrieve address of server
        reader, writer = connection.Pipe(duplex=False)

        # spawn process which runs a server
        self._process = self._ctx.Process(
            target=type(self)._run_server,
            args=(self._registry, self._address, self._authkey,
                  self._serializer, writer, initializer, initargs),
            )
        ident = &#039;:&#039;.join(str(i) for i in self._process._identity)
        self._process.name = type(self).__name__  + &#039;-&#039; + ident
        self._process.start()

        # get address of server
        writer.close()
        self._address = reader.recv()
        reader.close()

        # register a finalizer
        self._state.value = State.STARTED
        self.shutdown = util.Finalize(
            self, type(self)._finalize_manager,
            args=(self._process, self._address, self._authkey, self._state,
                  self._Client, self._shutdown_timeout),
            exitpriority=0
            )

    @classmethod
    def _run_server(cls, registry, address, authkey, serializer, writer,
                    initializer=None, initargs=()):
        &#039;&#039;&#039;
        Create a server, report its address and run it
        &#039;&#039;&#039;
        # bpo-36368: protect server process from KeyboardInterrupt signals
        signal.signal(signal.SIGINT, signal.SIG_IGN)

        if initializer is not None:
            initializer(*initargs)

        # create server
        server = cls._Server(registry, address, authkey, serializer)

        # inform parent process of the server&#039;s address
        writer.send(server.address)
        writer.close()

        # run the manager
        util.info(&#039;manager serving at %r&#039;, server.address)
        server.serve_forever()

    def _create(self, typeid, /, *args, **kwds):
        &#039;&#039;&#039;
        Create a new shared object; return the token and exposed tuple
        &#039;&#039;&#039;
        assert self._state.value == State.STARTED, &#039;server not yet started&#039;
        conn = self._Client(self._address, authkey=self._authkey)
        try:
            id, exposed = dispatch(conn, None, &#039;create&#039;, (typeid,)+args, kwds)
        finally:
            conn.close()
        return Token(typeid, self._address, id), exposed

    def join(self, timeout=None):
        &#039;&#039;&#039;
        Join the manager process (if it has been spawned)
        &#039;&#039;&#039;
        if self._process is not None:
            self._process.join(timeout)
            if not self._process.is_alive():
                self._process = None

    def _debug_info(self):
        &#039;&#039;&#039;
        Return some info about the servers shared objects and connections
        &#039;&#039;&#039;
        conn = self._Client(self._address, authkey=self._authkey)
        try:
            return dispatch(conn, None, &#039;debug_info&#039;)
        finally:
            conn.close()

    def _number_of_objects(self):
        &#039;&#039;&#039;
        Return the number of shared objects
        &#039;&#039;&#039;
        conn = self._Client(self._address, authkey=self._authkey)
        try:
            return dispatch(conn, None, &#039;number_of_objects&#039;)
        finally:
            conn.close()

    def __enter__(self):
        if self._state.value == State.INITIAL:
            self.start()
        if self._state.value != State.STARTED:
            if self._state.value == State.INITIAL:
                raise ProcessError(&quot;Unable to start server&quot;)
            elif self._state.value == State.SHUTDOWN:
                raise ProcessError(&quot;Manager has shut down&quot;)
            else:
                raise ProcessError(
                    &quot;Unknown state {!r}&quot;.format(self._state.value))
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.shutdown()

    @staticmethod
    def _finalize_manager(process, address, authkey, state, _Client,
                          shutdown_timeout):
        &#039;&#039;&#039;
        Shutdown the manager process; will be registered as a finalizer
        &#039;&#039;&#039;
        if process.is_alive():
            util.info(&#039;sending shutdown message to manager&#039;)
            try:
                conn = _Client(address, authkey=authkey)
                try:
                    dispatch(conn, None, &#039;shutdown&#039;)
                finally:
                    conn.close()
            except Exception:
                pass

            process.join(timeout=shutdown_timeout)
            if process.is_alive():
                util.info(&#039;manager still alive&#039;)
                if hasattr(process, &#039;terminate&#039;):
                    util.info(&#039;trying to `terminate()` manager process&#039;)
                    process.terminate()
                    process.join(timeout=shutdown_timeout)
                    if process.is_alive():
                        util.info(&#039;manager still alive after terminate&#039;)
                        process.kill()
                        process.join()

        state.value = State.SHUTDOWN
        try:
            del BaseProxy._address_to_local[address]
        except KeyError:
            pass

    @property
    def address(self):
        return self._address

    @classmethod
    def register(cls, typeid, callable=None, proxytype=None, exposed=None,
                 method_to_typeid=None, create_method=True):
        &#039;&#039;&#039;
        Register a typeid with the manager type
        &#039;&#039;&#039;
        if &#039;_registry&#039; not in cls.__dict__:
            cls._registry = cls._registry.copy()

        if proxytype is None:
            proxytype = AutoProxy

        exposed = exposed or getattr(proxytype, &#039;_exposed_&#039;, None)

        method_to_typeid = method_to_typeid or \
                           getattr(proxytype, &#039;_method_to_typeid_&#039;, None)

        if method_to_typeid:
            for key, value in list(method_to_typeid.items()): # isinstance?
                assert type(key) is str, &#039;%r is not a string&#039; % key
                assert type(value) is str, &#039;%r is not a string&#039; % value

        cls._registry[typeid] = (
            callable, exposed, method_to_typeid, proxytype
            )

        if create_method:
            def temp(self, /, *args, **kwds):
                util.debug(&#039;requesting creation of a shared %r object&#039;, typeid)
                token, exp = self._create(typeid, *args, **kwds)
                proxy = proxytype(
                    token, self._serializer, manager=self,
                    authkey=self._authkey, exposed=exp
                    )
                conn = self._Client(token.address, authkey=self._authkey)
                dispatch(conn, None, &#039;decref&#039;, (token.id,))
                return proxy
            temp.__name__ = typeid
            setattr(cls, typeid, temp)

#
# Subclass of set which get cleared after a fork
#

class ProcessLocalSet(set):
    def __init__(self):
        util.register_after_fork(self, lambda obj: obj.clear())
    def __reduce__(self):
        return type(self), ()

#
# Definition of BaseProxy
#

class BaseProxy(object):
    &#039;&#039;&#039;
    A base for proxies of shared objects
    &#039;&#039;&#039;
    _address_to_local = {}
    _mutex = util.ForkAwareThreadLock()

    # Each instance gets a `_serial` number. Unlike `id(...)`, this number
    # is never reused.
    _next_serial = 1

    def __init__(self, token, serializer, manager=None,
                 authkey=None, exposed=None, incref=True, manager_owned=False):
        with BaseProxy._mutex:
            tls_serials = BaseProxy._address_to_local.get(token.address, None)
            if tls_serials is None:
                tls_serials = util.ForkAwareLocal(), ProcessLocalSet()
                BaseProxy._address_to_local[token.address] = tls_serials

            self._serial = BaseProxy._next_serial
            BaseProxy._next_serial += 1

        # self._tls is used to record the connection used by this
        # thread to communicate with the manager at token.address
        self._tls = tls_serials[0]

        # self._all_serials is a set used to record the identities of all
        # shared objects for which the current process owns references and
        # which are in the manager at token.address
        self._all_serials = tls_serials[1]

        self._token = token
        self._id = self._token.id
        self._manager = manager
        self._serializer = serializer
        self._Client = listener_client[serializer][1]

        # Should be set to True only when a proxy object is being created
        # on the manager server; primary use case: nested proxy objects.
        # RebuildProxy detects when a proxy is being created on the manager
        # and sets this value appropriately.
        self._owned_by_manager = manager_owned

        if authkey is not None:
            self._authkey = process.AuthenticationString(authkey)
        elif self._manager is not None:
            self._authkey = self._manager._authkey
        else:
            self._authkey = process.current_process().authkey

        if incref:
            self._incref()

        util.register_after_fork(self, BaseProxy._after_fork)

    def _connect(self):
        util.debug(&#039;making connection to manager&#039;)
        name = process.current_process().name
        if threading.current_thread().name != &#039;MainThread&#039;:
            name += &#039;|&#039; + threading.current_thread().name
        conn = self._Client(self._token.address, authkey=self._authkey)
        dispatch(conn, None, &#039;accept_connection&#039;, (name,))
        self._tls.connection = conn

    def _callmethod(self, methodname, args=(), kwds={}):
        &#039;&#039;&#039;
        Try to call a method of the referent and return a copy of the result
        &#039;&#039;&#039;
        try:
            conn = self._tls.connection
        except AttributeError:
            util.debug(&#039;thread %r does not own a connection&#039;,
                       threading.current_thread().name)
            self._connect()
            conn = self._tls.connection

        conn.send((self._id, methodname, args, kwds))
        kind, result = conn.recv()

        if kind == &#039;#RETURN&#039;:
            return result
        elif kind == &#039;#PROXY&#039;:
            exposed, token = result
            proxytype = self._manager._registry[token.typeid][-1]
            token.address = self._token.address
            proxy = proxytype(
                token, self._serializer, manager=self._manager,
                authkey=self._authkey, exposed=exposed
                )
            conn = self._Client(token.address, authkey=self._authkey)
            dispatch(conn, None, &#039;decref&#039;, (token.id,))
            return proxy
        raise convert_to_error(kind, result)

    def _getvalue(self):
        &#039;&#039;&#039;
        Get a copy of the value of the referent
        &#039;&#039;&#039;
        return self._callmethod(&#039;#GETVALUE&#039;)

    def _incref(self):
        if self._owned_by_manager:
            util.debug(&#039;owned_by_manager skipped INCREF of %r&#039;, self._token.id)
            return

        conn = self._Client(self._token.address, authkey=self._authkey)
        dispatch(conn, None, &#039;incref&#039;, (self._id,))
        util.debug(&#039;INCREF %r&#039;, self._token.id)

        self._all_serials.add(self._serial)

        state = self._manager and self._manager._state

        self._close = util.Finalize(
            self, BaseProxy._decref,
            args=(self._token, self._serial, self._authkey, state,
                  self._tls, self._all_serials, self._Client),
            exitpriority=10
            )

    @staticmethod
    def _decref(token, serial, authkey, state, tls, idset, _Client):
        idset.discard(serial)

        # check whether manager is still alive
        if state is None or state.value == State.STARTED:
            # tell manager this process no longer cares about referent
            try:
                util.debug(&#039;DECREF %r&#039;, token.id)
                conn = _Client(token.address, authkey=authkey)
                dispatch(conn, None, &#039;decref&#039;, (token.id,))
            except Exception as e:
                util.debug(&#039;... decref failed %s&#039;, e)

        else:
            util.debug(&#039;DECREF %r -- manager already shutdown&#039;, token.id)

        # check whether we can close this thread&#039;s connection because
        # the process owns no more references to objects for this manager
        if not idset and hasattr(tls, &#039;connection&#039;):
            util.debug(&#039;thread %r has no more proxies so closing conn&#039;,
                       threading.current_thread().name)
            tls.connection.close()
            del tls.connection

    def _after_fork(self):
        self._manager = None
        try:
            self._incref()
        except Exception as e:
            # the proxy may just be for a manager which has shutdown
            util.info(&#039;incref failed: %s&#039; % e)

    def __reduce__(self):
        kwds = {}
        if get_spawning_popen() is not None:
            kwds[&#039;authkey&#039;] = self._authkey

        if getattr(self, &#039;_isauto&#039;, False):
            kwds[&#039;exposed&#039;] = self._exposed_
            return (RebuildProxy,
                    (AutoProxy, self._token, self._serializer, kwds))
        else:
            return (RebuildProxy,
                    (type(self), self._token, self._serializer, kwds))

    def __deepcopy__(self, memo):
        return self._getvalue()

    def __repr__(self):
        return &#039;&lt;%s object, typeid %r at %#x&gt;&#039; % \
               (type(self).__name__, self._token.typeid, id(self))

    def __str__(self):
        &#039;&#039;&#039;
        Return representation of the referent (or a fall-back if that fails)
        &#039;&#039;&#039;
        try:
            return self._callmethod(&#039;__repr__&#039;)
        except Exception:
            return repr(self)[:-1] + &quot;; &#039;__str__()&#039; failed&gt;&quot;

#
# Function used for unpickling
#

def RebuildProxy(func, token, serializer, kwds):
    &#039;&#039;&#039;
    Function used for unpickling proxy objects.
    &#039;&#039;&#039;
    server = getattr(process.current_process(), &#039;_manager_server&#039;, None)
    if server and server.address == token.address:
        util.debug(&#039;Rebuild a proxy owned by manager, token=%r&#039;, token)
        kwds[&#039;manager_owned&#039;] = True
        if token.id not in server.id_to_local_proxy_obj:
            server.id_to_local_proxy_obj[token.id] = \
                server.id_to_obj[token.id]
    incref = (
        kwds.pop(&#039;incref&#039;, True) and
        not getattr(process.current_process(), &#039;_inheriting&#039;, False)
        )
    return func(token, serializer, incref=incref, **kwds)

#
# Functions to create proxies and proxy types
#

def MakeProxyType(name, exposed, _cache={}):
    &#039;&#039;&#039;
    Return a proxy type whose methods are given by `exposed`
    &#039;&#039;&#039;
    exposed = tuple(exposed)
    try:
        return _cache[(name, exposed)]
    except KeyError:
        pass

    dic = {}

    for meth in exposed:
        exec(&#039;&#039;&#039;def %s(self, /, *args, **kwds):
        return self._callmethod(%r, args, kwds)&#039;&#039;&#039; % (meth, meth), dic)

    ProxyType = type(name, (BaseProxy,), dic)
    ProxyType._exposed_ = exposed
    _cache[(name, exposed)] = ProxyType
    return ProxyType


def AutoProxy(token, serializer, manager=None, authkey=None,
              exposed=None, incref=True, manager_owned=False):
    &#039;&#039;&#039;
    Return an auto-proxy for `token`
    &#039;&#039;&#039;
    _Client = listener_client[serializer][1]

    if exposed is None:
        conn = _Client(token.address, authkey=authkey)
        try:
            exposed = dispatch(conn, None, &#039;get_methods&#039;, (token,))
        finally:
            conn.close()

    if authkey is None and manager is not None:
        authkey = manager._authkey
    if authkey is None:
        authkey = process.current_process().authkey

    ProxyType = MakeProxyType(&#039;AutoProxy[%s]&#039; % token.typeid, exposed)
    proxy = ProxyType(token, serializer, manager=manager, authkey=authkey,
                      incref=incref, manager_owned=manager_owned)
    proxy._isauto = True
    return proxy

#
# Types/callables which we will register with SyncManager
#

class Namespace(object):
    def __init__(self, /, **kwds):
        self.__dict__.update(kwds)
    def __repr__(self):
        items = list(self.__dict__.items())
        temp = []
        for name, value in items:
            if not name.startswith(&#039;_&#039;):
                temp.append(&#039;%s=%r&#039; % (name, value))
        temp.sort()
        return &#039;%s(%s)&#039; % (self.__class__.__name__, &#039;, &#039;.join(temp))

class Value(object):
    def __init__(self, typecode, value, lock=True):
        self._typecode = typecode
        self._value = value
    def get(self):
        return self._value
    def set(self, value):
        self._value = value
    def __repr__(self):
        return &#039;%s(%r, %r)&#039;%(type(self).__name__, self._typecode, self._value)
    value = property(get, set)

def Array(typecode, sequence, lock=True):
    return array.array(typecode, sequence)

#
# Proxy types used by SyncManager
#

class IteratorProxy(BaseProxy):
    _exposed_ = (&#039;__next__&#039;, &#039;send&#039;, &#039;throw&#039;, &#039;close&#039;)
    def __iter__(self):
        return self
    def __next__(self, *args):
        return self._callmethod(&#039;__next__&#039;, args)
    def send(self, *args):
        return self._callmethod(&#039;send&#039;, args)
    def throw(self, *args):
        return self._callmethod(&#039;throw&#039;, args)
    def close(self, *args):
        return self._callmethod(&#039;close&#039;, args)


class AcquirerProxy(BaseProxy):
    _exposed_ = (&#039;acquire&#039;, &#039;release&#039;)
    def acquire(self, blocking=True, timeout=None):
        args = (blocking,) if timeout is None else (blocking, timeout)
        return self._callmethod(&#039;acquire&#039;, args)
    def release(self):
        return self._callmethod(&#039;release&#039;)
    def __enter__(self):
        return self._callmethod(&#039;acquire&#039;)
    def __exit__(self, exc_type, exc_val, exc_tb):
        return self._callmethod(&#039;release&#039;)


class ConditionProxy(AcquirerProxy):
    _exposed_ = (&#039;acquire&#039;, &#039;release&#039;, &#039;wait&#039;, &#039;notify&#039;, &#039;notify_all&#039;)
    def wait(self, timeout=None):
        return self._callmethod(&#039;wait&#039;, (timeout,))
    def notify(self, n=1):
        return self._callmethod(&#039;notify&#039;, (n,))
    def notify_all(self):
        return self._callmethod(&#039;notify_all&#039;)
    def wait_for(self, predicate, timeout=None):
        result = predicate()
        if result:
            return result
        if timeout is not None:
            endtime = time.monotonic() + timeout
        else:
            endtime = None
            waittime = None
        while not result:
            if endtime is not None:
                waittime = endtime - time.monotonic()
                if waittime &lt;= 0:
                    break
            self.wait(waittime)
            result = predicate()
        return result


class EventProxy(BaseProxy):
    _exposed_ = (&#039;is_set&#039;, &#039;set&#039;, &#039;clear&#039;, &#039;wait&#039;)
    def is_set(self):
        return self._callmethod(&#039;is_set&#039;)
    def set(self):
        return self._callmethod(&#039;set&#039;)
    def clear(self):
        return self._callmethod(&#039;clear&#039;)
    def wait(self, timeout=None):
        return self._callmethod(&#039;wait&#039;, (timeout,))


class BarrierProxy(BaseProxy):
    _exposed_ = (&#039;__getattribute__&#039;, &#039;wait&#039;, &#039;abort&#039;, &#039;reset&#039;)
    def wait(self, timeout=None):
        return self._callmethod(&#039;wait&#039;, (timeout,))
    def abort(self):
        return self._callmethod(&#039;abort&#039;)
    def reset(self):
        return self._callmethod(&#039;reset&#039;)
    @property
    def parties(self):
        return self._callmethod(&#039;__getattribute__&#039;, (&#039;parties&#039;,))
    @property
    def n_waiting(self):
        return self._callmethod(&#039;__getattribute__&#039;, (&#039;n_waiting&#039;,))
    @property
    def broken(self):
        return self._callmethod(&#039;__getattribute__&#039;, (&#039;broken&#039;,))


class NamespaceProxy(BaseProxy):
    _exposed_ = (&#039;__getattribute__&#039;, &#039;__setattr__&#039;, &#039;__delattr__&#039;)
    def __getattr__(self, key):
        if key[0] == &#039;_&#039;:
            return object.__getattribute__(self, key)
        callmethod = object.__getattribute__(self, &#039;_callmethod&#039;)
        return callmethod(&#039;__getattribute__&#039;, (key,))
    def __setattr__(self, key, value):
        if key[0] == &#039;_&#039;:
            return object.__setattr__(self, key, value)
        callmethod = object.__getattribute__(self, &#039;_callmethod&#039;)
        return callmethod(&#039;__setattr__&#039;, (key, value))
    def __delattr__(self, key):
        if key[0] == &#039;_&#039;:
            return object.__delattr__(self, key)
        callmethod = object.__getattribute__(self, &#039;_callmethod&#039;)
        return callmethod(&#039;__delattr__&#039;, (key,))


class ValueProxy(BaseProxy):
    _exposed_ = (&#039;get&#039;, &#039;set&#039;)
    def get(self):
        return self._callmethod(&#039;get&#039;)
    def set(self, value):
        return self._callmethod(&#039;set&#039;, (value,))
    value = property(get, set)

    __class_getitem__ = classmethod(types.GenericAlias)


BaseListProxy = MakeProxyType(&#039;BaseListProxy&#039;, (
    &#039;__add__&#039;, &#039;__contains__&#039;, &#039;__delitem__&#039;, &#039;__getitem__&#039;, &#039;__len__&#039;,
    &#039;__mul__&#039;, &#039;__reversed__&#039;, &#039;__rmul__&#039;, &#039;__setitem__&#039;,
    &#039;append&#039;, &#039;count&#039;, &#039;extend&#039;, &#039;index&#039;, &#039;insert&#039;, &#039;pop&#039;, &#039;remove&#039;,
    &#039;reverse&#039;, &#039;sort&#039;, &#039;__imul__&#039;
    ))
class ListProxy(BaseListProxy):
    def __iadd__(self, value):
        self._callmethod(&#039;extend&#039;, (value,))
        return self
    def __imul__(self, value):
        self._callmethod(&#039;__imul__&#039;, (value,))
        return self


DictProxy = MakeProxyType(&#039;DictProxy&#039;, (
    &#039;__contains__&#039;, &#039;__delitem__&#039;, &#039;__getitem__&#039;, &#039;__iter__&#039;, &#039;__len__&#039;,
    &#039;__setitem__&#039;, &#039;clear&#039;, &#039;copy&#039;, &#039;get&#039;, &#039;items&#039;,
    &#039;keys&#039;, &#039;pop&#039;, &#039;popitem&#039;, &#039;setdefault&#039;, &#039;update&#039;, &#039;values&#039;
    ))
DictProxy._method_to_typeid_ = {
    &#039;__iter__&#039;: &#039;Iterator&#039;,
    }


ArrayProxy = MakeProxyType(&#039;ArrayProxy&#039;, (
    &#039;__len__&#039;, &#039;__getitem__&#039;, &#039;__setitem__&#039;
    ))


BasePoolProxy = MakeProxyType(&#039;PoolProxy&#039;, (
    &#039;apply&#039;, &#039;apply_async&#039;, &#039;close&#039;, &#039;imap&#039;, &#039;imap_unordered&#039;, &#039;join&#039;,
    &#039;map&#039;, &#039;map_async&#039;, &#039;starmap&#039;, &#039;starmap_async&#039;, &#039;terminate&#039;,
    ))
BasePoolProxy._method_to_typeid_ = {
    &#039;apply_async&#039;: &#039;AsyncResult&#039;,
    &#039;map_async&#039;: &#039;AsyncResult&#039;,
    &#039;starmap_async&#039;: &#039;AsyncResult&#039;,
    &#039;imap&#039;: &#039;Iterator&#039;,
    &#039;imap_unordered&#039;: &#039;Iterator&#039;
    }
class PoolProxy(BasePoolProxy):
    def __enter__(self):
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.terminate()

#
# Definition of SyncManager
#

class SyncManager(BaseManager):
    &#039;&#039;&#039;
    Subclass of `BaseManager` which supports a number of shared object types.

    The types registered are those intended for the synchronization
    of threads, plus `dict`, `list` and `Namespace`.

    The `multiprocessing.Manager()` function creates started instances of
    this class.
    &#039;&#039;&#039;

SyncManager.register(&#039;Queue&#039;, queue.Queue)
SyncManager.register(&#039;JoinableQueue&#039;, queue.Queue)
SyncManager.register(&#039;Event&#039;, threading.Event, EventProxy)
SyncManager.register(&#039;Lock&#039;, threading.Lock, AcquirerProxy)
SyncManager.register(&#039;RLock&#039;, threading.RLock, AcquirerProxy)
SyncManager.register(&#039;Semaphore&#039;, threading.Semaphore, AcquirerProxy)
SyncManager.register(&#039;BoundedSemaphore&#039;, threading.BoundedSemaphore,
                     AcquirerProxy)
SyncManager.register(&#039;Condition&#039;, threading.Condition, ConditionProxy)
SyncManager.register(&#039;Barrier&#039;, threading.Barrier, BarrierProxy)
SyncManager.register(&#039;Pool&#039;, pool.Pool, PoolProxy)
SyncManager.register(&#039;list&#039;, list, ListProxy)
SyncManager.register(&#039;dict&#039;, dict, DictProxy)
SyncManager.register(&#039;Value&#039;, Value, ValueProxy)
SyncManager.register(&#039;Array&#039;, Array, ArrayProxy)
SyncManager.register(&#039;Namespace&#039;, Namespace, NamespaceProxy)

# types returned by methods of PoolProxy
SyncManager.register(&#039;Iterator&#039;, proxytype=IteratorProxy, create_method=False)
SyncManager.register(&#039;AsyncResult&#039;, create_method=False)

#
# Definition of SharedMemoryManager and SharedMemoryServer
#

if HAS_SHMEM:
    class _SharedMemoryTracker:
        &quot;Manages one or more shared memory segments.&quot;

        def __init__(self, name, segment_names=[]):
            self.shared_memory_context_name = name
            self.segment_names = segment_names

        def register_segment(self, segment_name):
            &quot;Adds the supplied shared memory block name to tracker.&quot;
            util.debug(f&quot;Register segment {segment_name!r} in pid {getpid()}&quot;)
            self.segment_names.append(segment_name)

        def destroy_segment(self, segment_name):
            &quot;&quot;&quot;Calls unlink() on the shared memory block with the supplied name
            and removes it from the list of blocks being tracked.&quot;&quot;&quot;
            util.debug(f&quot;Destroy segment {segment_name!r} in pid {getpid()}&quot;)
            self.segment_names.remove(segment_name)
            segment = shared_memory.SharedMemory(segment_name)
            segment.close()
            segment.unlink()

        def unlink(self):
            &quot;Calls destroy_segment() on all tracked shared memory blocks.&quot;
            for segment_name in self.segment_names[:]:
                self.destroy_segment(segment_name)

        def __del__(self):
            util.debug(f&quot;Call {self.__class__.__name__}.__del__ in {getpid()}&quot;)
            self.unlink()

        def __getstate__(self):
            return (self.shared_memory_context_name, self.segment_names)

        def __setstate__(self, state):
            self.__init__(*state)


    class SharedMemoryServer(Server):

        public = Server.public + \
                 [&#039;track_segment&#039;, &#039;release_segment&#039;, &#039;list_segments&#039;]

        def __init__(self, *args, **kwargs):
            Server.__init__(self, *args, **kwargs)
            address = self.address
            # The address of Linux abstract namespaces can be bytes
            if isinstance(address, bytes):
                address = os.fsdecode(address)
            self.shared_memory_context = \
                _SharedMemoryTracker(f&quot;shm_{address}_{getpid()}&quot;)
            util.debug(f&quot;SharedMemoryServer started by pid {getpid()}&quot;)

        def create(self, c, typeid, /, *args, **kwargs):
            &quot;&quot;&quot;Create a new distributed-shared object (not backed by a shared
            memory block) and return its id to be used in a Proxy Object.&quot;&quot;&quot;
            # Unless set up as a shared proxy, don&#039;t make shared_memory_context
            # a standard part of kwargs.  This makes things easier for supplying
            # simple functions.
            if hasattr(self.registry[typeid][-1], &quot;_shared_memory_proxy&quot;):
                kwargs[&#039;shared_memory_context&#039;] = self.shared_memory_context
            return Server.create(self, c, typeid, *args, **kwargs)

        def shutdown(self, c):
            &quot;Call unlink() on all tracked shared memory, terminate the Server.&quot;
            self.shared_memory_context.unlink()
            return Server.shutdown(self, c)

        def track_segment(self, c, segment_name):
            &quot;Adds the supplied shared memory block name to Server&#039;s tracker.&quot;
            self.shared_memory_context.register_segment(segment_name)

        def release_segment(self, c, segment_name):
            &quot;&quot;&quot;Calls unlink() on the shared memory block with the supplied name
            and removes it from the tracker instance inside the Server.&quot;&quot;&quot;
            self.shared_memory_context.destroy_segment(segment_name)

        def list_segments(self, c):
            &quot;&quot;&quot;Returns a list of names of shared memory blocks that the Server
            is currently tracking.&quot;&quot;&quot;
            return self.shared_memory_context.segment_names


    class SharedMemoryManager(BaseManager):
        &quot;&quot;&quot;Like SyncManager but uses SharedMemoryServer instead of Server.

        It provides methods for creating and returning SharedMemory instances
        and for creating a list-like object (ShareableList) backed by shared
        memory.  It also provides methods that create and return Proxy Objects
        that support synchronization across processes (i.e. multi-process-safe
        locks and semaphores).
        &quot;&quot;&quot;

        _Server = SharedMemoryServer

        def __init__(self, *args, **kwargs):
            if os.name == &quot;posix&quot;:
                # bpo-36867: Ensure the resource_tracker is running before
                # launching the manager process, so that concurrent
                # shared_memory manipulation both in the manager and in the
                # current process does not create two resource_tracker
                # processes.
                from . import resource_tracker
                resource_tracker.ensure_running()
            BaseManager.__init__(self, *args, **kwargs)
            util.debug(f&quot;{self.__class__.__name__} created by pid {getpid()}&quot;)

        def __del__(self):
            util.debug(f&quot;{self.__class__.__name__}.__del__ by pid {getpid()}&quot;)

        def get_server(self):
            &#039;Better than monkeypatching for now; merge into Server ultimately&#039;
            if self._state.value != State.INITIAL:
                if self._state.value == State.STARTED:
                    raise ProcessError(&quot;Already started SharedMemoryServer&quot;)
                elif self._state.value == State.SHUTDOWN:
                    raise ProcessError(&quot;SharedMemoryManager has shut down&quot;)
                else:
                    raise ProcessError(
                        &quot;Unknown state {!r}&quot;.format(self._state.value))
            return self._Server(self._registry, self._address,
                                self._authkey, self._serializer)

        def SharedMemory(self, size):
            &quot;&quot;&quot;Returns a new SharedMemory instance with the specified size in
            bytes, to be tracked by the manager.&quot;&quot;&quot;
            with self._Client(self._address, authkey=self._authkey) as conn:
                sms = shared_memory.SharedMemory(None, create=True, size=size)
                try:
                    dispatch(conn, None, &#039;track_segment&#039;, (sms.name,))
                except BaseException as e:
                    sms.unlink()
                    raise e
            return sms

        def ShareableList(self, sequence):
            &quot;&quot;&quot;Returns a new ShareableList instance populated with the values
            from the input sequence, to be tracked by the manager.&quot;&quot;&quot;
            with self._Client(self._address, authkey=self._authkey) as conn:
                sl = shared_memory.ShareableList(sequence)
                try:
                    dispatch(conn, None, &#039;track_segment&#039;, (sl.shm.name,))
                except BaseException as e:
                    sl.shm.unlink()
                    raise e
            return sl
</textarea><br><br><input type='hidden' name='archivo_editar' value='/lib64/python3.12/multiprocessing/managers.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>