def rebuild_ctype(type_, wrapper, length):
    if length is not None:
        type_ = type_ * length
    ForkingPickler.register(type_, reduce_ctype)
    obj = type_.from_address(wrapper.get_address())
    obj._wrapper = wrapper
    return obj

def rebuild_ctype(type_, wrapper, length):
    if length is not None:
        type_ = type_ * length
    ForkingPickler.register(type_, reduce_ctype)
    buf = wrapper.create_memoryview()
    obj = type_.from_buffer(buf)
    obj._wrapper = wrapper
    return obj

    def _feed(buffer, notempty, send_bytes, 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

                        # serialize the data before acquiring the lock
                        obj = ForkingPickler.dumps(obj)
                        if wacquire is None:
                            send_bytes(obj)
                        else:
                            wacquire()
                            try:
                                send_bytes(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 put(self, obj):
     # serialize the data before acquiring the lock
     obj = ForkingPickler.dumps(obj)
     if self._wlock is None:
         # writes to a message oriented win32 pipe are atomic
         self._writer.send_bytes(obj)
     else:
         with self._wlock:
             self._writer.send_bytes(obj)

 def get(self, block=True, timeout=None):
     if block and timeout is None:
         with self._rlock:
             res = self._recv_bytes()
         self._sem.release()
     else:
         if block:
             deadline = time.time() + timeout
         if not self._rlock.acquire(block, timeout):
             raise Empty
         try:
             if block:
                 timeout = deadline - time.time()
                 if timeout < 0 or not self._poll(timeout):
                     raise Empty
             elif not self._poll():
                 raise Empty
             res = self._recv_bytes()
             self._sem.release()
         finally:
             self._rlock.release()
     # unserialize the data after having released the lock
     return ForkingPickler.loads(res)

    new_handle = recv_handle(conn)
    conn.close()
    return new_handle


def reduce_connection(conn):
    rh = reduce_handle(conn.fileno())
    return (rebuild_connection, (rh, conn.readable, conn.writable))


def rebuild_connection(reduced_handle, readable, writable):
    handle = rebuild_handle(reduced_handle)
    return _multiprocessing.Connection(handle, readable=readable, writable=writable)


ForkingPickler.register(_multiprocessing.Connection, reduce_connection)

def fromfd(fd, family, type_, proto = 0):
    s = socket.fromfd(fd, family, type_, proto)
    if s.__class__ is not socket.socket:
        s = socket.socket(_sock=s)
    return s


def reduce_socket(s):
    reduced_handle = reduce_handle(s.fileno())
    return (rebuild_socket, (reduced_handle,
      s.family,
      s.type,
      s.proto))

            if timeout <= 0:
                return select.select(object_list, [], [], 0)[0]
            else:
                deadline = time.time() + timeout
        while True:
            try:
                return select.select(object_list, [], [], timeout)[0]
            except OSError as e:
                if e.errno != errno.EINTR:
                    raise
            if timeout is not None:
                timeout = deadline - time.time()

#
# Make connection and socket objects sharable if possible
#

if sys.platform == 'win32':
    from . import reduction
    ForkingPickler.register(socket.socket, reduction.reduce_socket)
    ForkingPickler.register(Connection, reduction.reduce_connection)
    ForkingPickler.register(PipeConnection, reduction.reduce_pipe_connection)
else:
    try:
        from . import reduction
    except ImportError:
        pass
    else:
        ForkingPickler.register(socket.socket, reduction.reduce_socket)
        ForkingPickler.register(Connection, reduction.reduce_connection)

 def recv(self):
     """Receive a (picklable) object"""
     self._check_closed()
     self._check_readable()
     buf = self._recv_bytes()
     return ForkingPickler.loads(buf.getbuffer())

 def send(self, obj):
     """Send a (picklable) object"""
     self._check_closed()
     self._check_writable()
     self._send_bytes(ForkingPickler.dumps(obj))