

@spin_first
def result_status(self, msg_ids, status_only=True):
    "Check on the status of the result(s) of the apply request with `msg_ids`.\n        \n        If status_only is False, then the actual results will be retrieved, else\n        only the status of the results will be checked.\n        \n        Parameters\n        ----------\n        \n        msg_ids : list of msg_ids\n            if int:\n                Passed as index to self.history for convenience.\n        status_only : bool (default: True)\n            if False:\n                Retrieve the actual results of completed tasks.\n        \n        Returns\n        -------\n        \n        results : dict\n            There will always be the keys 'pending' and 'completed', which will\n            be lists of msg_ids that are incomplete or complete. If `status_only`\n            is False, then completed results will be keyed by their `msg_id`.\n        "
    if (not isinstance(msg_ids, (list, tuple))):
        msg_ids = [msg_ids]
    theids = []
    for msg_id in msg_ids:
        if isinstance(msg_id, int):
            msg_id = self.history[msg_id]
        if (not isinstance(msg_id, str)):
            raise TypeError(('msg_ids must be str, not %r' % msg_id))
        theids.append(msg_id)
    completed = []
    local_results = {
        
    }
    for msg_id in theids:
        if (msg_id in self.results):
            completed.append(msg_id)
            local_results[msg_id] = self.results[msg_id]
            theids.remove(msg_id)
    if theids:
        content = dict(msg_ids=theids, status_only=status_only)
        msg = self.session.send(self._query_socket, 'result_request', content=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)
        buffers = msg['buffers']
    else:
        content = dict(completed=[], pending=[])
    content['completed'].extend(completed)
    if status_only:
        return content
    failures = []
    content.update(local_results)
    for msg_id in sorted(theids):
        if (msg_id in content['completed']):
            rec = content[msg_id]
            parent = rec['header']
            header = rec['result_header']
            rcontent = rec['result_content']
            iodict = rec['io']
            if isinstance(rcontent, str):
                rcontent = self.session.unpack(rcontent)
            md = self.metadata[msg_id]
            msg_id.update(self._extract_metadata(header, parent, rcontent))
            md.update(iodict)
            if (rcontent['status'] == 'ok'):
                (res, buffers) = util.unserialize_object(buffers)
            else:
                print(rcontent)
                res = self._unwrap_exception(rcontent)
                failures.append(res)
            self.results[msg_id] = res
            content[msg_id] = res
    if ((len(theids) == 1) and failures):
        raise failures[0]
    error.collect_exceptions(failures, 'result_status')
    return content

dataset/ETHPy150Open ipython/ipython-py3k/IPython/parallel/client/client.py Client.result_status/VarMisuse@430/869 `md`->`msg_id`