def resubmit(self, indices_or_msg_ids=None, subheader=None, block=None):
        """Resubmit one or more tasks.

        in-flight tasks may not be resubmitted.

        Parameters
        ----------

        indices_or_msg_ids : integer history index, str msg_id, or list of either
            The indices or msg_ids of indices to be retrieved

        block : bool
            Whether to wait for the result to be done

        Returns
        -------

        AsyncHubResult
            A subclass of AsyncResult that retrieves results from the Hub

        """
        block = self.block if block is None else block
        if indices_or_msg_ids is None:
            indices_or_msg_ids = -1

        if not isinstance(indices_or_msg_ids, (list,tuple)):
            indices_or_msg_ids = [indices_or_msg_ids]

        theids = []
        for id in indices_or_msg_ids:
            if isinstance(id, int):
                id = self.history[id]
            if not isinstance(id, str):
                raise TypeError("indices must be str or int, not %r"%id)
            theids.append(id)

        for msg_id in theids:
            self.outstanding.discard(msg_id)
            if msg_id in self.history:
                self.history.remove(msg_id)
            self.results.pop(msg_id, None)
            self.metadata.pop(msg_id, None)
        content = dict(msg_ids = theids)

        self.session.send(self._query_socket, 'resubmit_request', content)

        zmq.select([self._query_socket], [], [])
        idents,msg = self.session.recv(self._query_socket, zmq.NOBLOCK)
        if self.debug:
            pprint(msg)
        content = msg['content']
        if content['status'] != 'ok':
            raise self._unwrap_exception(content)

        ar = AsyncHubResult(self, msg_ids=theids)

        if block:
            ar.wait()

        return ar

 def test_timeout(self):
     """make sure select timeout has the right units (seconds)."""
     s1, s2 = self.create_bound_pair(zmq.PAIR, zmq.PAIR)
     tic = time.time()
     r,w,x = zmq.select([s1,s2],[],[],.005)
     toc = time.time()
     self.assertTrue(toc-tic < 1)
     self.assertTrue(toc-tic > 0.001)
     tic = time.time()
     r,w,x = zmq.select([s1,s2],[],[],.25)
     toc = time.time()
     self.assertTrue(toc-tic < 1)
     self.assertTrue(toc-tic > 0.1)

 def _runloop(self):
     """A single run-through of all sockets handled by this backend."""
     inputs = [self.pull_socket, sys.stdin]
     outputs = [] if self.pub_queue.empty() else [self.pub_socket]
     exceptionals = inputs + [self.pub_socket]
     read, write, error = zmq.select(inputs, outputs, exceptionals)
     for fileno in read:
         if fileno == self.pull_socket:
             self.route_message(self.pull_socket.recv_json())
         elif fileno == sys.stdin.fileno():
             for _ in sys.stdin:
                 pass
             sys.exit(0)
     for fileno in write:
         if fileno == self.pub_socket:
             try:
                 self.pub_socket.send_json(self.pub_queue.get_nowait())
             except Queue.Empty:
                 pass
     for fileno in error:
         if fileno == self.pub_socket:
             raise Exception('PUB socket in exceptional state')
         elif fileno == self.pull_socket:
             raise Exception('PULL socket in exceptional state')
         elif fileno == sys.stdin.fileno():
             raise Exception('stdin in exceptional state')

 def backend_send(self, *args, backend='echo'):
     sock = getattr(self, backend)
     self.assertEqual(([], [sock], []),
         zmq.select([], [sock], [], timeout=self.timeout))
     sock.send_multipart([
         a if isinstance(a, bytes) else a.encode('utf-8')
         for a in args], zmq.NOBLOCK)

 def backend_recv(self, backend='echo'):
     sock = getattr(self, backend)
     if (([sock], [], []) !=
         zmq.select([sock], [], [], timeout=self.timeout)):
         raise TimeoutError()
     val = sock.recv_multipart(zmq.NOBLOCK)
     return val

 def run(self):
     context = zmq.Context().instance()
     sub_socket = context.socket(zmq.SUB)
     sub_socket.connect("tcp://" + self.server_address + ":34234")
     sub_socket.setsockopt(zmq.SUBSCRIBE, "")
     
     listener = context.socket(zmq.PAIR)
     print(str(id(self)))
     listener.bind("inproc://" + str(id(self)))
     
     readable = []
     while listener not in readable:
         readable, _, _ = zmq.select([sub_socket, listener], [], [])
         if sub_socket in readable:
             published = json.loads(sub_socket.recv())
             if published.get("uuid") == self.uuid:
                 if not self.failed and published.get("status_name") == "failed":
                     self.failed = True
                     self.status_handler(self.alive, self.failed)
                 elif not self.alive and published.get("peers") and "mx" in published['peers']:
                     self.alive = True
                     self.status_handler(self.alive, self.failed)
     listener.recv()
     listener.close()
     sub_socket.close()

    def last_msg(self):
        r = [self.s]
        msg = None
        while r:
            r, w, x = zmq.select([self.s], [], [], 0.0)
            if r:
                msg = self.s.recv()

        r, w, x = zmq.select([self.s], [], [], 0.05)
        if r:
            msg = self.s.recv()

        if msg is not None:
            self._last = decode_utf8_json(msg)

        return self._last

def read_write_nontty(socket):
    EOF_reached = False
    try:
        r, w, x  = select.select([sys.stdin], [sys.stdout], [], 0.0)
    except select.error:
        pass
    zr,zw,zx = zmq.select([socket],[socket],[], timeout = 0.0)

    if (sys.stdin in r) and (socket in zw):
        x = sys.stdin.readline()
        if not EOF_reached:
            socket.send_json({'p':x})
        if x == '':
            EOF_reached = True

    if (socket in zr) and (sys.stdout in w):
        x = socket.recv_json()
        try:
            plain = x['p']
            sys.stdout.write(plain)
            while True:
                r, w, x  = select.select([], [sys.stdout], [], 0.0)
                if sys.stdout in w:
                    sys.stdout.flush()
                    break
        except KeyError:
            if 'ctrl' in x:
                ctrlmsg = x['ctrl']
                if 'terminated' in ctrlmsg:
                    sys.stdout.write('\r\nexiting... \r\n')
                    socket.send_json({'ctrl':'terminated'})
                    return 1
    return 0

def mdp_request(socket, service, msg, timeout=None):
    """Synchronous MDP request.

    This function sends a request to the given service and
    waits for a reply.

    If timeout is set and no reply received in the given time
    the function will return `None`.

    :param socket:    zmq REQ socket to use.
    :type socket:     zmq.Socket
    :param service:   service id to send the msg to.
    :type service:    str
    :param msg:       list of message parts to send.
    :type msg:        list of str
    :param timeout:   time to wait for answer in seconds.
    :type timeout:    float

    :rtype list of str:
    """
    if not timeout or timeout < 0.0:
        timeout = None
    if type(msg) in (bytes, unicode):
        msg = [msg]
    to_send = [PROTO_VERSION, service]
    to_send.extend(msg)
    socket.send_multipart(to_send)
    ret = None
    rlist, _, _ = select([socket], [], [], timeout)
    if rlist and rlist[0] == socket:
        ret = socket.recv_multipart()
        ret.pop(0) # remove service from reply
    return ret

def chat_room(kernel,output_box,socket_address):
    context = zmq.Context()
    subscribe_socket = context.socket(zmq.SUB)
    subscribe_socket.connect(socket_address)
    subscribe_socket.setsockopt(zmq.SUBSCRIBE, '')
    print '>>> welcome to the zmq chatroom'
    while True:
        # print 'loop'
        zr,zw,zx = zmq.select([subscribe_socket],[],[],timeout = 0.0)
        if subscribe_socket in zr:
            # print 'ok got message'
            message = subscribe_socket.recv_json()
            if 'plain_message' in message:
                plain_message = message['plain_message']
                output_box.value += '{}: {}\n'.format(plain_message['nickname'],plain_message['message'])
            elif 'object_transfer' in message:
                # print 'aqcuiring lock on shared state'
                ipythonchat_state.shared_state_lock.acquire()
                #ok now we no, nobody else is manipulating the shared_state object
                sender = message['object_transfer']['sender']
                import pickle
                from_sender = pickle.loads(message['object_transfer']['payload'])
                output_box.value += '<< got data object from {}>>\n'.format(sender)
                ipythonchat_state.shared_state = from_sender
                ipythonchat_state.shared_state_lock.release()
                # print 'released lock, shared state is: ', ipythonchat_state.shared_state
                kernel.do_one_iteration()
        time.sleep(0.01)

    def send_json(self, msg):
        """
        Send a JSON-encoded message. If the connection is not established yet,
        establish it before sending.

        @raises ZMQReqConnection.NoConnection:
            if the connection could not be established.
        """
        with self.__zmq_lock:
            if self._socket is None:
                self.__connect()

            if self._socket is not None:
                sockets = [self._socket]
                # Is the socket immediately available for writing?
                rlist, wlist, xlist = zmq.select(rlist=sockets,
                                                 wlist=sockets,
                                                 xlist=sockets,
                                                 timeout=0)

                try:
                    if self._socket in wlist:
                        self._socket.send_json(msg)
                    else:
                        raise ZMQReqConnection.NoConnection()
                except (zmq.ZMQError, ZMQReqConnection.NoConnection) as e:
                    logger.error('ZMQ error on send: %s', e)
                    self.__disconnect()
                    raise ZMQReqConnection.NoConnection(
                              'No connection to Trusted Host')

def run_loop():
    while True:
        http_loop()

        rlist, _, _ = zmq.select([httpd, zmq_socket], [], [])

        if zmq_socket in rlist:
            new_subscriber()

def trySend(sock, message):
    reply = None
    for _ in range(3):
        sock.send_json(message)
        ready = zmq.select([sock],[],[],10.0)
        if ready:
            return sock.recv_json()
        print "Awaiting response from server"
    return reply

    def _connect(self, sshserver, ssh_kwargs, timeout):
        """setup all our socket connections to the cluster. This is called from
        __init__."""
        
        # Maybe allow reconnecting?
        if self._connected:
            return
        self._connected=True

        def connect_socket(s, url):
            url = util.disambiguate_url(url, self._config['location'])
            if self._ssh:
                return tunnel.tunnel_connection(s, url, sshserver, **ssh_kwargs)
            else:
                return s.connect(url)
            
        self.session.send(self._query_socket, 'connection_request')
        r,w,x = zmq.select([self._query_socket],[],[], timeout)
        if not r:
            raise error.TimeoutError("Hub connection request timed out")
        idents,msg = self.session.recv(self._query_socket,mode=0)
        if self.debug:
            pprint(msg)
        msg = ss.Message(msg)
        content = msg.content
        self._config['registration'] = dict(content)
        if content.status == 'ok':
            if content.mux:
                self._mux_socket = self._context.socket(zmq.XREQ)
                self._mux_socket.setsockopt(zmq.IDENTITY, self.session.session)
                connect_socket(self._mux_socket, content.mux)
            if content.task:
                self._task_scheme, task_addr = content.task
                self._task_socket = self._context.socket(zmq.XREQ)
                self._task_socket.setsockopt(zmq.IDENTITY, self.session.session)
                connect_socket(self._task_socket, task_addr)
            if content.notification:
                self._notification_socket = self._context.socket(zmq.SUB)
                connect_socket(self._notification_socket, content.notification)
                self._notification_socket.setsockopt(zmq.SUBSCRIBE, b'')
            # if content.query:
            #     self._query_socket = self._context.socket(zmq.XREQ)
            #     self._query_socket.setsockopt(zmq.IDENTITY, self.session.session)
            #     connect_socket(self._query_socket, content.query)
            if content.control:
                self._control_socket = self._context.socket(zmq.XREQ)
                self._control_socket.setsockopt(zmq.IDENTITY, self.session.session)
                connect_socket(self._control_socket, content.control)
            if content.iopub:
                self._iopub_socket = self._context.socket(zmq.SUB)
                self._iopub_socket.setsockopt(zmq.SUBSCRIBE, b'')
                self._iopub_socket.setsockopt(zmq.IDENTITY, self.session.session)
                connect_socket(self._iopub_socket, content.iopub)
            self._update_engines(dict(content.engines))
        else:
            self._connected = False
            raise Exception("Failed to connect!")

def handle_tty(cmd,cid,socket):
    print 'handling tty docker session'

    import pty
    import shlex
    master, slave = pty.openpty()

    p = subprocess.Popen(shlex.split(cmd), stdin = slave, stdout = slave, stderr = slave)
    print 'started container with pid: {}'.format(p.pid)

    term_size = socket.recv_json()['ctrl']['term_size']
    set_winsize(master,term_size['rows'],term_size['cols'],p.pid)

    while True:
        #print "polling"
        r, w, x = select.select([master],[master],[master], 0.0)
        #print "master poll: r: {} w: {} x: {}".format(r,w,x)

        zr,zw,zx = zmq.select([socket], [socket],[socket], timeout = 0.0)
        #print "ZMQ poll: r: {} w: {} x: {}".format(zr,zw,zx)

        procpoll = p.poll()
        #print 'process: {}'.format(procpoll)

        if (procpoll is not None) and (socket in zw):
	    print "ending session because process ended"
            socket.send_json({'ctrl':'terminated'})
            print 'wait for ack from client'    
	    ack = socket.recv_json()
            print "return"
            return
        
        if (master in r) and (socket in zw):
            #print "reading!"
            fromprocess = os.read(master,1024)
            #print 'sending {}'.format(fromprocess)
            socket.send_json({'p':fromprocess})

        if (master in w) and (socket in zr):
            #  Wait for next request from client
            #print "recv"
            message = socket.recv_json()
            #print("Received request: {} length: {}".format(message,len(message)))
	    try:
                os.write(master,message['p'])
            except KeyError:
		if 'ctrl' in message:
		    ctrlmsg = message['ctrl']
        	    if 'term_size' in ctrlmsg:
		        set_winsize(master,ctrlmsg['term_size']['rows'],ctrlmsg['term_size']['cols'],p.pid)
		    if 'signal' in ctrlmsg:
 		        print 'got signal: {}'.format(ctrlmsg['signal']) 
                        os.kill(p.pid,ctrlmsg['signal'])
			if ctrlmsg['signal'] in [signal.SIGHUP,signal.SIGTERM,signal.SIGKILL]:
                            stop_container(get_container_id(cid))
                            return

 def control(self, *args):
     sock = self.zmq.socket(zmq.REQ)
     try:
         sock.connect(CONTROL_ADDR)
         sock.send_multipart([a.encode('utf-8') for a in args])
         self.assertEqual(([sock], [], []),
             zmq.select([sock], [], [], timeout=self.timeout))
         return sock.recv_multipart()
     finally:
         sock.close()

 def recv(self, timeout=0.0):
     r, w, x = zmq.select([self.s], [], [], timeout)
     if r:
         try:
             self._last = decode_utf8_json(self.s.recv())
             return self._last
         except ValueError:
             raise NoMessagesException
     else:
         raise NoMessagesException

 def backend_recv(self, backend=None):
     if backend is None:
         sock = self.chatfw
     else:
         raise NotImplementedError(backend)
     if ([sock], [], []) != zmq.select([sock], [], [], timeout=self.timeout):
         raise TimeoutError()
     val = sock.recv_multipart()
     if val[1] == b"heartbeat":
         return self.backend_recv(backend=backend)
     return val

def select(rlist, wlist, xlist, timeout):
    i, rlist = _mkindex(rlist)
    wi, wlist = _mkindex(wlist)
    xi, xlist = _mkindex(xlist)

    i.update(wi)
    i.update(xi)

    r, w, x = zmq.select(rlist, wlist, xlist, timeout)

    return _useindex(r, i),  _useindex(w, i), _useindex(x, i)

    def test_pair(self):
        s1, s2 = self.create_bound_pair(zmq.PAIR, zmq.PAIR)

        # Sleep to allow sockets to connect.
        wait()

        rlist, wlist, xlist = zmq.select([s1, s2], [s1, s2], [s1, s2])
        self.assert_(s1 in wlist)
        self.assert_(s2 in wlist)
        self.assert_(s1 not in rlist)
        self.assert_(s2 not in rlist)