def _queue_management_worker(self):
logger.debug("<STR_LIT>")<EOL>while not self._executor_bad_state.is_set():<EOL><INDENT>try:<EOL><INDENT>msgs = self.incoming_q.get(timeout=<NUM_LIT:1>)<EOL><DEDENT>except queue.Empty:<EOL><INDENT>logger.debug("<STR_LIT>")<EOL>pass<EOL><DEDENT>except IOError as e:<EOL><INDENT>logger.exception("<STR_LIT>".format(e.errno, e))<EOL>return<EOL><DEDENT>except Exception as e:<EOL><INDENT>logger.exception("<STR_LIT>".format(e))<EOL>return<EOL><DEDENT>else:<EOL><INDENT>if msgs is None:<EOL><INDENT>logger.debug("<STR_LIT>")<EOL>return<EOL><DEDENT>else:<EOL><INDENT>for serialized_msg in msgs:<EOL><INDENT>try:<EOL><INDENT>msg = pickle.loads(serialized_msg)<EOL>tid = msg['<STR_LIT>']<EOL><DEDENT>except pickle.UnpicklingError:<EOL><INDENT>raise BadMessage("<STR_LIT>")<EOL><DEDENT>except Exception:<EOL><INDENT>raise BadMessage("<STR_LIT>")<EOL><DEDENT>if tid == -<NUM_LIT:1> and '<STR_LIT>' in msg:<EOL><INDENT>logger.warning("<STR_LIT>")<EOL>self._executor_exception, _ = deserialize_object(msg['<STR_LIT>'])<EOL>logger.exception("<STR_LIT>".format(self._executor_exception))<EOL>self._executor_bad_state.set()<EOL>for task in self.tasks:<EOL><INDENT>self.tasks[task].set_exception(self._executor_exception)<EOL><DEDENT>break<EOL><DEDENT>task_fut = self.tasks[tid]<EOL>if '<STR_LIT:result>' in msg:<EOL><INDENT>result, _ = deserialize_object(msg['<STR_LIT:result>'])<EOL>task_fut.set_result(result)<EOL><DEDENT>elif '<STR_LIT>' in msg:<EOL><INDENT>try:<EOL><INDENT>s, _ = deserialize_object(msg['<STR_LIT>'])<EOL>try:<EOL><INDENT>s.reraise()<EOL><DEDENT>except Exception as e:<EOL><INDENT>task_fut.set_exception(e)<EOL><DEDENT><DEDENT>except Exception as e:<EOL><INDENT>task_fut.set_exception(<EOL>DeserializationError("<STR_LIT>".format(e)))<EOL><DEDENT><DEDENT>else:<EOL><INDENT>raise BadMessage("<STR_LIT>")<EOL><DEDENT><DEDENT><DEDENT><DEDENT>if not self.is_alive:<EOL><INDENT>break<EOL><DEDENT><DEDENT>logger.info("<STR_LIT>")<EOL>
Listen to the queue for task status messages and handle them.

        Depending on the message, tasks will be updated with results, exceptions,
        or updates. It expects the following messages:

        .. code:: python

            {
               "task_id" : <task_id>
               "result"  : serialized result object, if task succeeded
               ... more tags could be added later
            }

            {
               "task_id" : <task_id>
               "exception" : serialized exception object, on failure
            }

        We do not support these yet, but they could be added easily.

        .. code:: python

            {
               "task_id" : <task_id>
               "cpu_stat" : <>
               "mem_stat" : <>
               "io_stat"  : <>
               "started"  : tstamp
            }

        The `None` message is a die request.