#
# Module providing the `SyncManager` class for dealing
# with shared objects
#
# processing/managers.py
#
# Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt
#

__all__ = [ 'BaseManager', 'SyncManager', 'BaseProxy',
            'CreatorMethod', 'Token' ]

#
# Imports
#

import os
import sys
import socket
import weakref
import threading
import traceback
import array
import copy_reg
import cPickle

from processing.connection import Listener, Client, Pipe, AuthenticationError
from processing.connection import deliverChallenge, answerChallenge
from processing.process import Process, currentProcess
from processing.process import activeChildren, _registerAfterFork
from processing.logger import debug, info, subWarning
from processing.finalize import Finalize, _runFinalizers
from processing.forking import exit

#
# Register `array.array` for pickling
#

def reduceArray(a):
    return array.array, (a.typecode, a.tostring())

copy_reg.pickle(array.array, reduceArray)

#
# Exception class
#

class RemoteError(Exception):
    '''
    Exception type raised by managers
    '''
    def __init__(self, *args):
        if args:
            self.args = args
        else:
            info = sys.exc_info()
            self.args = (info[1], ''.join(traceback.format_exception(*info)))

    def __str__(self):
        return ('\n' + '-'*75 + '\nRemote ' + self.args[1] + '-'*75)

#
# Type for identifying shared objects
#

class Token(object):
    '''
    Type to uniquely indentify a shared object
    '''
    def __init__(self, typeid, address, id):
        self.typeid = typeid
        self.address = address
        self.id = id

    def __repr__(self):
        return 'Token(typeid=%r, address=%r, id=%r)' % \
               (self.typeid, self.address, self.id)

#
# Functions for communication with a manager's server process
#

def dispatch(c, id, methodname, args=(), kwds={}):
    '''
    Send a message to manager using connection `c` and return response
    '''
    c.send((id, methodname, args, kwds))
    kind, result = c.recv()
    if kind == '#RETURN':
        return result
    elif kind == '#ERROR':
        raise result
    else:
        raise ValueError

def transact(address, authkey, methodname, args=(), kwds={}):
    '''
    Create connection then send a message to manager and return response
    '''
    conn = Client(address, authkey=authkey)
    try:
        return dispatch(conn, None, methodname, args, kwds)
    finally:
        conn.close()

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

def allMethods(obj):
    '''
    Return a list of names of methods of `obj`
    '''
    temp = []
    for name in dir(obj):
        func = getattr(obj, name)
        if hasattr(func, '__call__'):
            temp.append(name)
    return temp

def publicMethods(obj):
    '''
    Return a list of names of methods of `obj` which do not start with '_'
    '''
    return filter(lambda name: name[0] != '_', allMethods(obj))

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

class Server(object):
    '''
    Server class which runs in a process controlled by a manager object
    '''
    public = ['shutdown', 'create', 'acceptConnection',
              'getMethods', 'debugInfo', 'dummy', 'incref', 'decref']

    def __init__(self, registry, address, authkey):
        assert type(authkey) is str
        self.registry = registry
        self.authkey = authkey

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

        self.address = self.listener.address
        if type(self.address) is tuple:
            self.address = (socket.getfqdn(self.address[0]), self.address[1])

        self.id_to_obj = {0: (None, ())}
        self.id_to_refcount = {}
        self.mutex = threading.RLock()
        self.stop = 0

    def serveForever(self):
        '''
        Run the server forever
        '''
        try:
            try:
                while 1:
                    c = self.listener.accept()
                    t = threading.Thread(target=self.handleRequest, args=(c,))
                    t.setDaemon(True)
                    t.start()
            except (KeyboardInterrupt, SystemExit):
                return
        finally:
            self.stop = 999
            self.listener.close()

    def handleRequest(self, c):
        '''
        Handle a new connection
        '''
        funcname = result = request = None
        try:
            deliverChallenge(c, self.authkey)
            answerChallenge(c, self.authkey)
            request = c.recv()
            ignore, funcname, args, kwds = request
            assert funcname in self.public, '%r unrecognized' % funcname
            func = getattr(self, funcname)
        except (SystemExit, KeyboardInterrupt):
            raise
        except Exception:
            msg = ('#ERROR', RemoteError())
        else:
            try:
                result = func(c, *args, **kwds)
                msg = ('#RETURN', result)
            except (SystemExit, KeyboardInterrupt):
                raise
            except Exception:
                msg = ('#ERROR', RemoteError())

        try:
            c.send(msg)
        except (SystemExit, KeyboardInterrupt):
            raise
        except Exception, e:
            if msg[0] == '#ERROR':
                subWarning('Failure to send exception: %r', msg[1])
            else:
                subWarning('Failure to send result: %r', msg[1])
            subWarning(' ... request was %r', request)
            subWarning(' ... exception was %r', e)

        c.close()

    def serveClient(self, connection):
        '''
        Handle requests from the proxies in a particular process/thread
        '''
        debug('starting server thread to service %r',
              threading.currentThread().getName())

        recv = connection.recv
        send = connection.send
        id_to_obj = self.id_to_obj

        while not self.stop:

            try:
                methodname = obj = None
                request = recv()
                ident, methodname, args, kwds = request
                obj, exposed = id_to_obj[ident]

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

                function = getattr(obj, methodname)

                try:
                    result = function(*args, **kwds)
                    msg = ('#RETURN', result)
                except (SystemExit, KeyboardInterrupt):
                    raise
                except Exception, e:
                    msg = ('#ERROR', e)

            except AttributeError, e:
                if methodname is None:
                    msg = ('#ERROR', RemoteError())
                else:
                    try:
                        fallback_func = self.fallback_mapping[methodname]
                        result = fallback_func(
                            self, connection, ident, obj, *args, **kwds
                            )
                        msg = ('#RETURN', result)
                    except (SystemExit, KeyboardInterrupt):
                        raise
                    except Exception:
                        msg = ('#ERROR', RemoteError())

            except EOFError:
                debug('got EOF -- exiting thread serving %r',
                      threading.currentThread().getName())
                sys.exit(0)

            except (SystemExit, KeyboardInterrupt):
                raise

            except:
                msg = ('#ERROR', RemoteError())

            try:
                try:
                    send(msg)
                except cPickle.PicklingError:
                    result = msg[1]
                    if hasattr(result, '__iter__') and hasattr(result, 'next'):
                        try:
                            # send a proxy for this iterator
                            res_ident, _ = self.create(
                                connection, 'iter', result
                                )
                            res_obj, res_exposed = self.id_to_obj[res_ident]
                            token = Token('iter', self.address, res_ident)
                            result = IteratorProxy(token, incref=False)
                            msg = ('#RETURN', result)
                        except (SystemExit, KeyboardInterrupt):
                            raise
                        except Exception:
                            msg = ('#ERROR', RemoteError())
                    else:
                        msg = ('#ERROR', RemoteError())
                    send(msg)
            except (SystemExit, KeyboardInterrupt):
                raise
            except Exception, e:
                subWarning('exception in thread serving %r',
                        threading.currentThread().getName())
                subWarning(' ... message was %r', msg)
                subWarning(' ... exception was %r', e)
                connection.close()
                sys.exit(1)

    def fallbackGetValue(self, connection, ident, obj):
        return obj

    def fallbackStr(self, connection, ident, obj):
        return str(obj)

    def fallbackRepr(self, connection, ident, obj):
        return repr(obj)

    def fallbackCmp(self, connection, ident, obj, *args):
        return cmp(obj, *args)

    fallback_mapping = {
        '__str__':fallbackStr, '__repr__':fallbackRepr,
        '__cmp__':fallbackCmp, '#GETVALUE':fallbackGetValue
        }

    def dummy(self, c):
        pass

    def debugInfo(self, c):
        '''
        Return some info --- useful to spot problems with refcounting
        '''
        self.mutex.acquire()
        try:
            result = []
            keys = self.id_to_obj.keys()
            keys.sort()
            for ident in keys:
                if ident != 0:
                    result.append('  %s:       refcount=%s\n    %s' %
                                  (hex(ident), self.id_to_refcount[ident],
                                   str(self.id_to_obj[ident][0])[:75]))
            return '\n'.join(result)
        finally:
            self.mutex.release()

    def shutdown(self, c):
        '''
        Shutdown this process
        '''
        c.send(('#RETURN', None))
        info('manager received shutdown message')

        # do some cleaning up
        _runFinalizers(0)
        for p in activeChildren():
            debug('terminating a child process of manager')
            p.terminate()
        for p in activeChildren():
            debug('terminating a child process of manager')
            p.join()
        _runFinalizers()
        info('manager exiting with exitcode 0')

        # now exit without waiting for other threads to finish
        exit(0)
            
    def create(self, c, typeid, *args, **kwds):
        '''
        Create a new shared object and return its id
        '''
        self.mutex.acquire()
        try:
            callable, exposed = self.registry[typeid]
            obj = callable(*args, **kwds)

            if exposed is None:
                exposed = publicMethods(obj)

            ident = id(obj)

            debug('have created %r object with id %r', typeid, ident)

            self.id_to_obj[ident] = (obj, set(exposed))
            if ident not in self.id_to_refcount:
                self.id_to_refcount[ident] = None
            return ident, tuple(exposed)
        finally:
            self.mutex.release()

    def getMethods(self, c, token):
        '''
        Return the methods of the shared object indicated by token
        '''
        return tuple(self.id_to_obj[token.id][1])

    def acceptConnection(self, c, name):
        '''
        Spawn a new thread to serve this connection
        '''
        threading.currentThread().setName(name)
        c.send(('#RETURN', None))
        self.serveClient(c)

    def incref(self, c, ident):
        self.mutex.acquire()
        try:
            try:
                self.id_to_refcount[ident] += 1
            except TypeError:
                assert self.id_to_refcount[ident] is None
                self.id_to_refcount[ident] = 1
        finally:
            self.mutex.release()

    def decref(self, c, ident):
        self.mutex.acquire()
        try:
            assert self.id_to_refcount[ident] >= 1
            self.id_to_refcount[ident] -= 1
            if self.id_to_refcount[ident] == 0:
                del self.id_to_obj[ident], self.id_to_refcount[ident]
                debug('disposing of obj with id %d', ident)
        finally:
            self.mutex.release()

#
# Definition of BaseManager
#

class BaseManager(object):
    '''
    Base class for managers
    '''
    def __init__(self, address=None, authkey=None):
        '''
        `address`:
            The address on which manager should listen for new
            connections.  If `address` is None then an arbitrary one
            is chosen (which will be available as `self.address`).

        `authkey`:
            Only connections from clients which are using `authkey` as an
            authentication key will be accepted.  If `authkey` is `None`
            then `currentProcess().getAuthKey()` is used.
        '''
        self._address = address     # XXX not necessarily the final address
        if authkey is None:
            self._authkey = authkey = currentProcess().getAuthKey()
        else:
            self._authkey = authkey
        assert type(authkey) is str
        self._started = False

    def start(self):
        '''
        Spawn a server process for this manager object
        '''
        assert not self._started
        self._started = True

        self._registry, _ = BaseManager._getRegistryCreators(self)

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

        # spawn process which runs a server
        self._process = Process(
            target=self._runServer,
            args=(self._registry, self._address, self._authkey, writer),
            )
        ident = ':'.join(map(str, self._process._identity))
        self._process.setName(type(self).__name__  + '-' + ident)
        self._process.setAuthKey(self._authkey)
        self._process.start()

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

        # register a finalizer
        self.shutdown = Finalize(
            self, BaseManager._finalizeManager,
            args=(self._process, self._address, self._authkey),
            exitpriority=0
            )

    @classmethod
    def _runServer(cls, registry, address, authkey, writer):
        '''
        Create a server, report its address and run it
        '''
        # create server
        server = Server(registry, address, authkey)
        currentProcess()._server = server

        # inform parent process of the server's address
        writer.send(server.address)
        writer.close()

        # run the manager
        info('manager serving at %r', server.address)
        server.serveForever()

    def serveForever(self, verbose=True):
        '''
        Start server in the current process
        '''
        assert not self._started
        self._started = True

        registry, _ = BaseManager._getRegistryCreators(self)
        server = Server(registry, self._address, self._authkey)
        currentProcess()._server = server
        if verbose:
            print >>sys.stderr, '%s serving at address %s' % \
                  (type(self).__name__, server.address)
        server.serveForever()

    @classmethod
    def fromAddress(cls, address, authkey):
        '''
        Create a new manager object for a pre-existing server process
        '''
        manager = cls(address, authkey)
        transact(address, authkey, 'dummy')
        manager._started = True
        return manager

    def _create(self, typeid, *args, **kwds):
        '''
        Create a new shared object; return the token and exposed tuple
        '''
        assert self._started
        id, exposed = transact(
            self._address, self._authkey, 'create', (typeid,) + args, kwds
            )
        return Token(typeid, self._address, id), exposed

    def join(self, timeout=None):
        '''
        Join the manager process (if it has been spawned)
        '''
        self._process.join(timeout)

    def _debugInfo(self):
        '''
        Return some info about the servers shared objects and connections
        '''
        return transact(self._address, self._authkey, 'debugInfo')

    def _proxyFromToken(self, token):
        '''
        Create a proxy for a token
        '''
        assert token.address == self.address
        _, creators = BaseManager._getRegistryCreators(self)
        proxytype = creators[token.typeid]._proxytype
        return proxytype(token, authkey=self._authkey)

    @staticmethod
    def _getRegistryCreators(self_or_cls):
        registry = {}
        creators = {}
        for name in dir(self_or_cls):
            obj = getattr(self_or_cls, name)
            info = getattr(obj, '_manager_info', None)
            if info is not None and hasattr(obj, '__call__'):
                creators[name] = obj
                typeid, callable, exposed = info
                assert typeid not in registry, 'typeids must be unique'
                registry[typeid] = (callable, exposed)
        return registry, creators

    def __enter__(self):
        return self

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

    @staticmethod
    def _finalizeManager(process, address, authkey):
        '''
        Shutdown the manager process; will be registered as a finalizer
        '''
        if process.isAlive():
            info('sending shutdown message to manager')
            try:
                transact(address, authkey, 'shutdown')
            except (SystemExit, KeyboardInterrupt):
                raise
            except Exception:
                pass

            process.join(timeout=0.2)
            if process.isAlive():
                info('manager still alive')
                if hasattr(process, 'terminate'):
                    info('trying to `terminate()` manager process')
                    process.terminate()
                    process.join(timeout=0.1)
                    if process.isAlive():
                        info('manager still alive after terminate')

        try:
            del BaseProxy._address_to_local[address]
        except KeyError:
            pass
        
    address = property(lambda self: self._address)

    # deprecated
    from_address = fromAddress
    serve_forever = serveForever

#
# Function for adding methods to managers
#

def CreatorMethod(callable=None, proxytype=None, exposed=None, typeid=None):
    '''
    Returns a method for a manager class which will create
    a shared object using `callable` and return a proxy for it.
    '''
    if exposed is None and hasattr(callable, '__exposed__'):
        exposed = callable.__exposed__

    if proxytype is None:
        proxytype = MakeAutoProxy

    typeid = typeid or _uniqueLabel(callable.__name__)

    def temp(self, *args, **kwds):
        debug('requesting creation of a shared %r object', typeid)
        token, exp = self._create(typeid, *args, **kwds)
        proxy = proxytype(
            token, manager=self, authkey=self._authkey, exposed=exp
            )
        return proxy

    try:
        temp.__name__ = typeid
    except TypeError:
        pass
    temp._manager_info = (typeid, callable, exposed)
    temp._proxytype = proxytype
    return temp

def _uniqueLabel(prefix, _count={}):
    '''
    Return a string beginning with 'prefix' which has not already been used.
    '''
    try:
        _count[prefix] += 1
        return prefix + '-' + str(_count[prefix])
    except KeyError:
        _count[prefix] = 0
        return prefix

#
# Subclasses of threading.local and set which get cleared after a fork
#

class ProcessLocalSet(set):
    def __init__(self):
        _registerAfterFork(self, set.clear)
    def __reduce__(self):
        return type(self), ()

class ThreadLocalStorage(threading.local):
    def __init__(self):
        _registerAfterFork(self, _clearNamespace)
    def __reduce__(self):
        return type(self), ()

def _clearNamespace(obj):
    obj.__dict__.clear()

#
# Definition of BaseProxy
#

class BaseProxy(object):
    '''
    A base for proxies of shared objects
    '''
    _address_to_local = {}
    _mutex = threading.Lock()

    def __init__(self, token, manager=None,
                 authkey=None, exposed=None, incref=True):
        BaseProxy._mutex.acquire()
        try:
            tls_idset = BaseProxy._address_to_local.get(token.address, None)
            if tls_idset is None:
                tls_idset = ThreadLocalStorage(), ProcessLocalSet()
                BaseProxy._address_to_local[token.address] = tls_idset
        finally:
            BaseProxy._mutex.release()

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

        # self._idset is 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._idset = tls_idset[1]

        self._token = token
        self._id = self._token.id
        self._manager = manager

        if authkey:
            self._authkey = authkey
        elif self._manager:
            self._authkey = self._manager
        else:
            self._authkey = currentProcess().getAuthKey()
        
        if incref:
            self._incref()
            
        _registerAfterFork(self, BaseProxy._afterFork)
        
    def _connect(self):
        debug('making connection to manager')
        name = currentProcess().getName()
        if threading.currentThread().getName() != 'MainThread':
            name += '|' + threading.currentThread().getName()
        connection = Client(self._token.address, authkey=self._authkey)
        dispatch(connection, None, 'acceptConnection', (name,))
        self._tls.connection = connection
        
    def _callMethod(self, methodname, args=(), kwds={}):
        '''
        Try to call a method of the referrent and return a copy of the result
        '''
        try:
            conn = self._tls.connection
        except AttributeError:
            debug('thread %r does not own a connection',
                 threading.currentThread().getName())
            self._connect()
            conn = self._tls.connection

        conn.send((self._id, methodname, args, kwds))
        kind, result = conn.recv()
        if kind == '#RETURN':
            return result
        elif kind == '#ERROR':
            raise result
        else:
            raise ValueError

    def _getValue(self):
        '''
        Get a copy of the value of the referent
        '''
        return self._callMethod('#GETVALUE')

    def _incref(self):
        connection = Client(self._token.address, authkey=self._authkey)
        dispatch(connection, None, 'incref', (self._id,))
        debug('INCREF %r', self._token.id)

        assert self._id not in self._idset    
        self._idset.add(self._id)

        shutdown = getattr(self._manager, 'shutdown', None)

        self._close = Finalize(
            self, BaseProxy._decref,
            args=(self._token, self._authkey, shutdown,
                  self._tls, self._idset),
            exitpriority=10
            )

    @staticmethod
    def _decref(token, authkey, shutdown, tls, idset):
        idset.remove(token.id)

        # check whether manager is still alive
        manager_still_alive = shutdown is None or shutdown.stillActive()
        if manager_still_alive:
            # tell manager this process no longer cares about referent
            try:
                debug('DECREF %r', token.id)
                connection = Client(token.address, authkey=authkey)
                dispatch(connection, None, 'decref', (token.id,))
            except (SystemExit, KeyboardInterrupt):
                raise
            except Exception, e:
                debug('... decref failed %s', e)

        else:
            debug('DECREF %r -- manager already shutdown',
                  token.id)

        # check whether we can close this thread's connection because
        # the process owns no more references to objects for this manager
        if not idset and hasattr(tls, 'connection'):
            debug('thread %r has no more proxies so closing conn',
                  threading.currentThread().getName())
            tls.connection.close()
            del tls.connection
            
    def _afterFork(self):
        self._manager = None
        self._incref()

    def __reduce__(self):
        if hasattr(self, '_exposed'):
            return (RebuildProxy, (MakeAutoProxy, self._token,
                                   {'exposed': self._exposed}))
        else:
            return (RebuildProxy, (type(self), self._token, {}))

    def __deepcopy__(self, memo):
        return self._getValue()
    
    def __hash__(self):
        raise NotImplementedError, 'proxies are unhashable'
    
    def __repr__(self):
        return '<Proxy[%s] object at %s>' % (self._token.typeid,
                                             '0x%x' % id(self))

    def __str__(self):
        '''
        Return representation of the referent (or a fall-back if that fails)
        '''
        try:
            return self._callMethod('__repr__')
        except (SystemExit, KeyboardInterrupt):
            raise
        except Exception:
            return repr(self)[:-1] + "; '__str__()' failed>"

    # deprecated
    _callmethod = _callMethod
    _getvalue = _getValue

#
# Since BaseProxy._mutex might be locked at time of fork we reset it
#

def _resetMutex(obj):
    obj._mutex = threading.Lock()
_registerAfterFork(BaseProxy, _resetMutex)

#
# Function used for unpickling
#

def RebuildProxy(func, token, kwds={}):
    '''
    Function used for unpickling proxy objects.

    If possible the shared object is returned, or otherwise a proxy for it.
    '''
    server = getattr(currentProcess(), '_server', None)
    
    if server and server.address == token.address:
        return server.id_to_obj[token.id][0]
    else:
        incref = (
            kwds.pop('incref', True) and 
            not getattr(currentProcess(), '_inheriting', False)
            )
        try:
            return func(token, manager=None, authkey=None,
                        incref=incref, **kwds)
        except AuthenticationError:
            raise AuthenticationError, 'cannot rebuild proxy without authkey'

#
# Functions to create proxies and proxy types
#

def MakeAutoProxyType(exposed, typeid='unnamed', _cache={}):
    '''
    Return an auto-proxy type whose methods are given by `exposed`
    '''
    exposed = tuple(exposed)
    try:
        return _cache[(typeid, exposed)]
    except KeyError:
        pass

    dic = {}

    for name in exposed:
        exec '''def %s(self, *args, **kwds):
        return self._callMethod(%r, args, kwds)''' % (name, name) in dic

    ProxyType = type('AutoProxy[%s]' % typeid, (BaseProxy,), dic)
    ProxyType._exposed = exposed
    _cache[(typeid, exposed)] = ProxyType
    return ProxyType


def MakeAutoProxy(token, manager=None, authkey=None,
                  exposed=None, incref=True):
    '''
    Return an auto-proxy for `token`
    '''
    if exposed is None:
        exposed = transact(token.address, authkey, 'getMethods', (token,))
    ProxyType = MakeAutoProxyType(exposed, token.typeid)
    proxy = ProxyType(token, manager=manager, authkey=authkey, incref=incref)
    return proxy

#
# Types (or functions) which we will register with SyncManager
#

from threading import BoundedSemaphore, Condition, Event, \
     Lock, RLock, Semaphore
from Queue import Queue

class Namespace(object):
    '''
    Instances of this class can be used as namespaces.

    A namespace object has no public methods but does have writable
    attributes.  Its represention shows the values of its attributes.
    '''
    __exposed__ = ('__getattribute__', '__setattr__', '__delattr__')

    def __repr__(self):
        items = self.__dict__.items()
        temp = []
        for name, value in items:
            if not name.startswith('_'):
                temp.append('%s=%r' % (name, value))
        temp.sort()
        return 'Namespace(%s)' % str.join(', ', temp)

class Value(object):
    '''
    Instances have a settable 'value' property
    '''
    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 '%s(%r, %r)'%(type(self).__name__, self._typecode, self._value)

    value = property(get, set)

#
# Proxy type used by BaseManager
#

class IteratorProxy(BaseProxy):
    '''
    Proxy type for iterators
    '''
    def __iter__(self):
        return self
    def next(self):
        return self._callMethod('next')

BaseManager._Iter = CreatorMethod(iter, IteratorProxy, ('next', '__iter__'))

#
# Proxy types used by SyncManager
#

class AcquirerProxy(BaseProxy):
    '''
    Base class for proxies which have acquire and release methods
    '''
    def acquire(self, blocking=1):
        return self._callMethod('acquire', (blocking,))
    def release(self):
        return self._callMethod('release')
    def __enter__(self):
        self._callMethod('acquire')
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        return self._callMethod('release')


class ConditionProxy(AcquirerProxy):
    def wait(self, timeout=None):
        return self._callMethod('wait', (timeout,))
    def notify(self):
        return self._callMethod('notify')
    def notifyAll(self):
        return self._callMethod('notifyAll')


class NamespaceProxy(BaseProxy):
    '''
    Proxy type for Namespace objects.

    Note that attributes beginning with '_' will "belong" to the proxy,
    while other attributes "belong" to the referent.
    '''
    def __getattr__(self, key):
        if key[0] == '_':
            return object.__getattribute__(self, key)
        callmethod = object.__getattribute__(self, '_callMethod')
        return callmethod('__getattribute__', (key,))
    
    def __setattr__(self, key, value):
        if key[0] == '_':
            return object.__setattr__(self, key, value)
        callmethod = object.__getattribute__(self, '_callMethod')
        return callmethod('__setattr__', (key, value))
    
    def __delattr__(self, key):
        if key[0] == '_':
            return object.__delattr__(self, key)
        callmethod = object.__getattribute__(self, '_callMethod')
        return callmethod('__delattr__', (key,))
    

_list_exposed = (
    '__add__', '__contains__', '__delitem__', '__delslice__',
    '__cmp__', '__getitem__', '__getslice__', '__iter__', '__imul__',
    '__len__', '__mul__', '__reversed__', '__rmul__', '__setitem__',
    '__setslice__', 'append', 'count', 'extend', 'index', 'insert',
    'pop', 'remove', 'reverse', 'sort'
    )

BaseListProxy = MakeAutoProxyType(_list_exposed, 'BaseListProxy')

class ListProxy(BaseListProxy):
    # augmented assignment functions must return self
    def __iadd__(self, value):
        self._callMethod('extend', (value,))
        return self
    def __imul__(self, value):
        # Inefficient since a copy of the target is transferred and discarded
        self._callMethod('__imul__', (value,))
        return self


_dict_exposed=(
    '__cmp__', '__contains__', '__delitem__', '__getitem__',
    '__iter__', '__len__', '__setitem__', 'clear', 'copy', 'get',
    'has_key', 'items', 'iteritems', 'iterkeys', 'itervalues',
    'keys', 'pop', 'popitem', 'setdefault', 'update', 'values'
    )


class ValueProxy(BaseProxy):
    def get(self):
        return self._callMethod('get')
    def set(self, value):
        return self._callMethod('set', (value,))
    value = property(get, set)

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

_arr_exposed = (
    '__len__', '__iter__', '__getitem__', '__setitem__',
    '__getslice__', '__setslice__'
    )

#
# Definition of SyncManager
#

class SyncManager(BaseManager):
    '''
    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 `processing.Manager` function creates instances of this class.
    '''
    Event = CreatorMethod(Event)
    Queue = CreatorMethod(Queue)
    Lock = CreatorMethod(Lock, AcquirerProxy)
    RLock = CreatorMethod(RLock, AcquirerProxy)
    Semaphore = CreatorMethod(Semaphore, AcquirerProxy)
    BoundedSemaphore = CreatorMethod(BoundedSemaphore, AcquirerProxy)
    Condition = CreatorMethod(Condition, ConditionProxy)
    Namespace = CreatorMethod(Namespace, NamespaceProxy)
    list = CreatorMethod(list, ListProxy, exposed=_list_exposed)
    dict = CreatorMethod(dict, exposed=_dict_exposed)
    Value = CreatorMethod(Value, ValueProxy)
    Array = CreatorMethod(Array, exposed=_arr_exposed)