def SocketClient(address):
    '''
    Return a connection object connected to the socket given by `address`
    '''
    family = address_type(address)
    with socket.socket( getattr(socket, family) ) as s:
        s.setblocking(True)
        t = _init_timeout()

        while 1:
            try:
                s.connect(address)
            except socket.error as e:
                if e.args[0] != errno.ECONNREFUSED or _check_timeout(t):
                    debug('failed to connect to address %s', address)
                    raise
                time.sleep(0.01)
            else:
                break
        else:
            raise

        fd = duplicate(s.fileno())
    conn = _multiprocessing.Connection(fd)
    return conn

    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:
            util.debug('thread %r does not own a connection',
                       threading.current_thread().name)
            self._connect()
            conn = self._tls.connection

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

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

def worker(inqueue, outqueue, initializer=None, initargs=()):
    put = outqueue.put
    get = inqueue.get
    if hasattr(inqueue, '_writer'):
        inqueue._writer.close()
        outqueue._reader.close()

    if initializer is not None:
        initializer(*initargs)

    while 1:
        try:
            task = get()
        except (EOFError, IOError):
            debug('worker got EOFError or IOError -- exiting')
            break

        if task is None:
            debug('worker got sentinel -- exiting')
            break

        job, i, func, args, kwds = task
        try:
            result = (True, func(*args, **kwds))
        except Exception, e:
            result = (False, e)
        put((job, i, result))

 def join(self):
     debug('joining pool')
     assert self._state in (CLOSE, TERMINATE)
     self._task_handler.join()
     self._result_handler.join()
     for p in self._pool:
         p.join()

 def cancel_join_thread(self):
     debug('Queue.cancel_join_thread()')
     self._joincancelled = True
     try:
         self._jointhread.cancel()
     except AttributeError:
         pass

 def _help_stuff_finish(inqueue, task_handler, size):
     # task_handler may be blocked trying to put items on inqueue
     debug('removing tasks from inqueue until task handler finished')
     inqueue._rlock.acquire()
     while task_handler.is_alive() and inqueue._reader.poll():
         inqueue._reader.recv()
         time.sleep(0)

 def _finalize_close(buffer, notempty):
     debug('telling queue thread to quit')
     notempty.acquire()
     try:
         buffer.append(_sentinel)
         notempty.notify()
     finally:
         notempty.release()

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

    def __init__(self, kind, value, maxvalue):
        sl = self._semlock = _multiprocessing.SemLock(kind, value, maxvalue)
        debug('created semlock with handle %s' % sl.handle)
        self._make_methods()

        if sys.platform != 'win32':
            def _after_fork(obj):
                obj._semlock._after_fork()
            register_after_fork(self, _after_fork)

    def _feed(buffer, notempty, send, writelock, close, ignore_epipe):
        debug('starting thread to feed data to pipe')
        from .util import is_exiting

        nacquire = notempty.acquire
        nrelease = notempty.release
        nwait = notempty.wait
        bpopleft = buffer.popleft
        sentinel = _sentinel
        if sys.platform != 'win32':
            wacquire = writelock.acquire
            wrelease = writelock.release
        else:
            wacquire = None

        try:
            while 1:
                nacquire()
                try:
                    if not buffer:
                        nwait()
                finally:
                    nrelease()
                try:
                    while 1:
                        obj = bpopleft()
                        if obj is sentinel:
                            debug('feeder thread got sentinel -- exiting')
                            close()
                            return

                        if wacquire is None:
                            send(obj)
                        else:
                            wacquire()
                            try:
                                send(obj)
                            finally:
                                wrelease()
                except IndexError:
                    pass
        except Exception as e:
            if ignore_epipe and getattr(e, 'errno', 0) == errno.EPIPE:
                return
            # Since this runs in a daemon thread the resources it uses
            # may be become unusable while the process is cleaning up.
            # We ignore errors which happen after the process has
            # started to cleanup.
            try:
                if is_exiting():
                    info('error in queue thread: %s', e)
                else:
                    import traceback
                    traceback.print_exc()
            except Exception:
                pass

 def _start(self):
     from .connection import Listener
     assert self._listener is None
     debug('starting listener and thread for sending handles')
     self._listener = Listener(authkey=current_process().authkey)
     self._address = self._listener.address
     t = threading.Thread(target=self._serve)
     t.daemon = True
     t.start()
     self._thread = t

 def temp(self, *args, **kwds):
     util.debug('requesting creation of a shared %r object', 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, 'decref', (token.id,))
     return proxy

 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]
             util.debug('disposing of obj with id %r', ident)
     finally:
         self.mutex.release()

    def _handle_workers(pool):
        thread = threading.current_thread()

        # Keep maintaining workers until the cache gets drained, unless the pool
        # is terminated.
        while thread._state == RUN or (pool._cache and thread._state != TERMINATE):
            pool._maintain_pool()
            time.sleep(0.1)
        # send sentinel to stop workers
        pool._taskqueue.put(None)
        debug('worker handler exiting')

    def _decref(token, authkey, state, tls, idset, _Client):
        idset.discard(token.id)

        # 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('DECREF %r', token.id)
                conn = _Client(token.address, authkey=authkey)
                dispatch(conn, None, 'decref', (token.id,))
            except Exception, e:
                util.debug('... decref failed %s', e)

 def _after_fork(self):
     debug('Queue._after_fork()')
     self._notempty = threading.Condition(threading.Lock())
     self._buffer = collections.deque()
     self._thread = None
     self._jointhread = None
     self._joincancelled = False
     self._closed = False
     self._close = None
     self._send = self._writer.send
     self._recv = self._reader.recv
     self._poll = self._reader.poll

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

 def _join_exited_workers(self):
     """Cleanup after any worker processes which have exited due to reaching
     their specified lifetime.  Returns True if any workers were cleaned up.
     """
     cleaned = False
     for i in reversed(range(len(self._pool))):
         worker = self._pool[i]
         if worker.exitcode is not None:
             # worker exited
             debug('cleaning up worker %d' % i)
             worker.join()
             cleaned = True
             del self._pool[i]
     return cleaned

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

                if sys.stdout != sys.__stdout__:
                    util.debug('resetting stdout, stderr')
                    sys.stdout = sys.__stdout__
                    sys.stderr = sys.__stderr__

                util._run_finalizers(0)

                for p in active_children():
                    util.debug('terminating a child process of manager')
                    p.terminate()

                for p in active_children():
                    util.debug('terminating a child process of manager')
                    p.join()

                util._run_finalizers()
                util.info('manager exiting with exitcode 0')
            except:
                import traceback
                traceback.print_exc()
        finally:
            exit(0)

 def _repopulate_pool(self):
     """Bring the number of pool processes up to the specified number,
     for use after reaping workers which have exited.
     """
     for i in range(self._processes - len(self._pool)):
         w = self.Process(target=worker,
                          args=(self._inqueue, self._outqueue,
                                self._initializer,
                                self._initargs, self._maxtasksperchild)
                         )
         self._pool.append(w)
         w.name = w.name.replace('Process', 'PoolWorker')
         w.daemon = True
         w.start()
         debug('added worker')