async def send_manager_command(cmd, expect_reply=True, extra_fields={}):
    """Send a properly formatted command to the manager.

    :param cmd: The command to send (:class:`str`).
    :param expect_reply: If ``True``, wait for the manager to reply
        with an acknowledgement packet.
    :param extra_fields: A dictionary of extra information that's
        merged into the packet body (i.e. not under an extra key).
    """
    packet = {
        ExecutorProtocol.DATA_ID: DATA['id'],
        ExecutorProtocol.COMMAND: cmd,
    }
    packet.update(extra_fields)

    logger.debug("Sending command to listener: {}".format(json.dumps(packet)))

    # TODO what happens here if the push fails? we don't have any realistic recourse,
    # so just let it explode and stop processing
    queue_channel = EXECUTOR_SETTINGS['REDIS_CHANNEL_PAIR'][0]
    try:
        await redis_conn.rpush(queue_channel, json.dumps(packet))
    except Exception:
        logger.error("Error sending command to manager:\n\n{}".format(traceback.format_exc()))
        raise

    if not expect_reply:
        return

    for _ in range(_REDIS_RETRIES):
        response = await redis_conn.blpop(QUEUE_RESPONSE_CHANNEL, timeout=1)
        if response:
            break
    else:
        # NOTE: If there's still no response after a few seconds, the system is broken
        # enough that it makes sense to give up; we're isolated here, so if the manager
        # doesn't respond, we can't really do much more than just crash
        raise RuntimeError("No response from the manager after {} retries.".format(_REDIS_RETRIES))

    _, item = response
    result = json.loads(item.decode('utf-8'))[ExecutorProtocol.RESULT]
    assert result in [ExecutorProtocol.RESULT_OK, ExecutorProtocol.RESULT_ERROR]

    if result == ExecutorProtocol.RESULT_OK:
        return True

    return False
Send a properly formatted command to the manager.

    :param cmd: The command to send (:class:`str`).
    :param expect_reply: If ``True``, wait for the manager to reply
        with an acknowledgement packet.
    :param extra_fields: A dictionary of extra information that's
        merged into the packet body (i.e. not under an extra key).