import time

from contextlib import contextmanager
import threading
try:
    from Queue import Queue, Empty
except ImportError:
    from queue import Queue, Empty

from parallels.core.logging import get_logger
from parallels.core.logging import subscription_context
from parallels.core import messages, APIError, MigrationStopError, MigratorInterruptException
from parallels.core.registry import Registry
from parallels.core.utils.migrator_utils import normalize_domain_name
from parallels.core.utils.stop_mark import StopMark
from parallels.core.utils.subscription_operations import command_name_to_operation
from parallels.core.workflow.runner.base import Runner
from parallels.core import MigrationError, MigrationNoContextError
from parallels.core.actions.base.action_pointer import ActionPointer
from parallels.core.actions.base.compound_action import CompoundAction
from parallels.core.actions.base.condition_action_pointer import ConditionActionPointer
from parallels.core.actions.base.subscription_action import SubscriptionAction
from parallels.core.migrator_config import MultithreadingStatus
from parallels.core.utils.migration_progress import SubscriptionMigrationStatus
from parallels.core.utils.steps_profiler import get_default_steps_profiler
from parallels.core.utils.common import ilen, all_equal, safe_format, default

logger = get_logger(__name__)

# Whether to print low-level debug messages, which could be interesting only for action runner debugging,
# but most probably are not interesting for the other cases
log_low_level_debug = False


class ActionRunnerBySubscription(Runner):
    """Executes workflow actions and entry points by subscription

    "By subscription" means that:
    for the 1st subscription all actions are executed
    for the 2nd subscription all actions are executed
    and so on.

    As we have actions that are executed for all subscriptions at bulk,
    we can't have true "by subscription" runner, so this runner works by subscriptions
    only when possible, for such bulk actions it works by layers.
    """
    def __init__(self, multithreading=None, actions_to_skip=None):
        """Class constructor

        :type multithreading: parallels.core.migrator_config.MultithreadingParams | None
        :param actions_to_skip: dict[str|unicode, list[str|unicode]|None]
            dictionary where:
              - key is action identifier
              - value is
                * None if action should be completely skipped
                * list of normalized subscription names if action should be skipped for specific subscriptions
        """
        super(ActionRunnerBySubscription, self).__init__(multithreading)
        self._action_to_skip = default(actions_to_skip, {})
        self._executed_common_paths = list()
        self._keyboard_interrupt = False
        self._stop_request = False

    def run(self, action, action_id=None, handle_stop_mark=True):
        """Run specified action

        :type action: parallels.core.actions.base.base_action.BaseAction
        :type action_id: str | unicode
        :type handle_stop_mark: bool
        :rtype: None
        """
        # 1) Make flat actions list out of actions tree
        flat_actions = self._make_flat_action_list(action_id, action)
        # 2) Group actions in a list into 2 kind of blocks: subscriptions blocks and common blocks
        blocks = self._split_actions_by_blocks(flat_actions)

        for i, block in enumerate(blocks):
            # 3) For each block create a tree, consisting only of block items
            actions_tree = self._build_tree(block)

            # 4) Run block with corresponding run function
            if isinstance(block, ByLayerActionFlatBlock):
                if log_low_level_debug:
                    logger.debug(messages.DEBUG_ENTER_COMMON_BLOCK)
                self._run_common_actions_tree(actions_tree, handle_stop_mark=handle_stop_mark)
                if log_low_level_debug:
                    logger.debug(messages.DEBUG_EXIT_COMMON_BLOCK)
            elif isinstance(block, BySubscriptionActionFlatBlock):
                if log_low_level_debug:
                    logger.debug(messages.ENTER_SUBSCRIPTION_ACTION_BLOCK)
                self._run_subscription_actions(actions_tree, None, [], handle_stop_mark=handle_stop_mark)
                if log_low_level_debug:
                    logger.debug(messages.EXIT_SUBSCRIPTION_ACTION_BLOCK)
            else:
                assert False

    def _run_common_actions_tree(self, actions_tree, path=None, handle_stop_mark=True):
        """
        :type actions_tree: parallels.core.actions.base.compound_action.CompoundAction
        :type path: Iterable[str | unicode] | None
        :rtype: None
        """
        if path is None:
            path = []

        for action_id, action in actions_tree.get_all_actions():
            if handle_stop_mark and StopMark.is_set():
                raise MigratorInterruptException()

            if self._is_skip_whole_action(action_id):
                logger.fdebug(messages.SKIP_ACTION, action_id=action_id)
                continue

            with self._profile_action(action_id, action):
                with get_default_steps_profiler().measure_time_gap('Common', action.get_description()):
                    action_path = path + [action_id]
                    if isinstance(action, CompoundAction):
                        with self._log_action_start(action):
                            if action.filter_action(self._context):
                                self._run_common_actions_tree(action, action_path, handle_stop_mark=handle_stop_mark)
                            self._log_action_end(action)
                    elif isinstance(action, SubscriptionAction):
                        subscriptions = list(self._context.iter_all_subscriptions())
                        for num, subscription in enumerate(subscriptions, start=1):
                            if self._is_skip_action_for_subscription(action_id, subscription.name):
                                logger.fdebug(
                                    messages.SKIP_ACTION_FOR_SUBSCRIPTION,
                                    action_id=action_id, subscription=subscription.name
                                )
                                continue

                            with subscription_context(subscription.name):
                                self._run_subscription_action(subscription, action, action_path)
                    else:
                        self._run_common_action_plain(action, action_path)

                    self._mark_common_path_as_executed(action_path)

    def _run_common_action_plain(self, action, action_path):
        """
        :type action: parallels.core.actions.base.common_action.CommonAction
        :type action_path: list[str | unicode]
        :rtype: None
        """
        action_failure_message = action.get_failure_message(self._context)

        def run():
            if log_low_level_debug:
                logger.debug(messages.CHECKING_WHETHER_IT_IS_REQUIRED_EXECUTE)
            if not action.filter_action(self._context):
                if log_low_level_debug:
                    logger.debug(messages.NO_NEED_EXECUTE_ACTION_CONTINUE_NEXT)
                return
            with self._log_action_start(action):
                with self._context.progress.overall_action(action.get_description()):
                    action.run(self._context)
                state_vars = Registry.get_instance().get_migration_state().get_vars(action='/'.join(action_path))
                state_vars['executed'] = True
                self._log_action_end(action)

        def run_multi_attempts():
            """Run the action multiple times according to repeat properties"""
            repeat_props = action.get_repeat_properties(self._context)

            for attempt in range(repeat_props.max_repeat_attempts):
                try:
                    run()
                except MigratorInterruptException:
                    # action execution interrupted by customer's request => return from that function with no errors
                    return
                except Exception as err:
                    logger.exception()
                    if attempt == repeat_props.max_repeat_attempts - 1:  # the last attempt
                        # all attempts finished, and the last one finished with error => re-raise the error
                        try:
                            action.on_exception(self._context, err)
                        except:
                            logger.exception()

                        raise err
                    else:
                        if repeat_props.repeat_message:
                            logger.warn(repeat_props.repeat_message)
                        else:
                            logger.warn(action_failure_message)

                        if repeat_props.repeat_interval > 0:
                            time.sleep(repeat_props.repeat_interval)
                else:
                    if attempt > 0:  # successfully executed after several attempts
                        if repeat_props.success_message:
                            logger.info(repeat_props.success_message)

                    # successfully executed the action => return from that function
                    return

        if action.is_critical():
            try:
                run_multi_attempts()
            except MigrationNoContextError:
                # "no context" errors should be displayed "as is", with not information
                # about action, stopped migration, etc
                raise
            except MigrationStopError:
                # "stop" errors should be displayed "as is", with not information
                # about action, stopped migration, etc
                raise
            except APIError:
                # API errors are handled at upper level for better logging
                raise
            except Exception as e:
                logger.exception()
                full_failure_message = messages.STOP_BY_CRITICAL_ERROR % (
                    action_failure_message, e
                )
                raise MigrationError(full_failure_message)
        else:
            with self._context.safe.try_general(action_failure_message, ''):
                run_multi_attempts()

    def _mark_common_path_as_executed(self, path):
        """Mark action as executed, for correct detection of necessary shutdown actions

        :type path: Iterable[str | unicode]
        :rtype: None
        """
        self._executed_common_paths.append("/".join(path))

    def _log_action(self, action, level=0):
        """
        :type action: parallels.core.actions.base.base_action.BaseAction
        :type level: int
        :rtype: None
        """
        if action.get_description() is not None:
            description = action.get_description()
            logging_function = self._get_log_function(action)
            logging_function(("    " * level) + description)

    def _log_action_end(self, action):
        """
        :type action: parallels.core.actions.base.base_action.BaseAction
        :rtype: None
        """
        if action.get_description() is None:
            return

        if action.get_logging_properties().compound:
            log_function = self._get_log_function(action)
            log_function(u"FINISH: %s", action.get_description())

    @contextmanager
    def _log_action_start(self, action):
        """
        :type action: parallels.core.actions.base.base_action.BaseAction
        :rtype: None
        """
        if action.get_description() is None:
            yield
        else:
            logging_props = action.get_logging_properties()
            if logging_props.info_log:
                start_level_function = logger.info_level_context
            else:
                if not logging_props.top_level:
                    start_level_function = logger.debug_level_context
                else:
                    @contextmanager
                    def info_no_context(*args, **kwargs):
                        logger.fdebug(*args, **kwargs)
                        yield

                    start_level_function = info_no_context

            if logging_props.compound:
                with start_level_function(messages.START_ACTION, description=action.get_description()):
                    yield
            else:
                with start_level_function(action.get_description()):
                    yield

    @staticmethod
    def _get_log_function(action):
        """
        :type action: parallels.core.actions.base.base_action.BaseAction
        :rtype: (str | unicode, *list[str | unicode]) -> None
        """
        if action.get_logging_properties().info_log:
            log_function = logger.info
        else:
            log_function = logger.debug
        return log_function

    def _flat_subscription_actions(self, action):
        """
        :type action: parallels.core.actions.base.base_action.BaseAction
        :rtype: list[parallels.core.actions.base.subscription_action.SubscriptionAction]
        """
        if isinstance(action, CompoundAction):
            result = []
            for _, child_action in action.get_all_actions():
                result += self._flat_subscription_actions(child_action)
            return result
        else:
            return [action]

    def _run_subscription_actions(self, action, action_id, action_path, handle_stop_mark=True):
        """
        :type action: parallels.core.actions.base.base_action.BaseAction
        :type action_id: str | unicode | None
        :type action_path: list[str | unicode]
        :rtype: None
        """
        leaf_actions = self._flat_subscription_actions(action)

        locks = {}

        for leaf_action in leaf_actions:
            if not leaf_action.get_multithreading_properties().can_use_threads:
                locks[leaf_action] = threading.Lock()

        subscription_queue = Queue()

        subscription_count = ilen(self._context.iter_all_subscriptions_unfiltered())

        for num, subscription in enumerate(self._context.iter_all_subscriptions_unfiltered(), start=1):
            self._context.progress.get_subscription(subscription.name)
            subscription_queue.put((num, subscription))

        full_multithreading = self._multithreading.status == MultithreadingStatus.FULL
        if full_multithreading or self._is_multithreading_enabled_for_actions(leaf_actions):
            num_workers = self._multithreading.num_workers
        else:
            num_workers = 1

        if log_low_level_debug:
            logger.debug(messages.RUN_GROUP_ACTIONS_WITH_WORKERS % num_workers)
        plain_log = num_workers > 1

        if plain_log:
            common_parents = self._list_common_parents(action, action_id)
        else:
            common_parents = []

        threads = []

        if num_workers > 1:
            with self._profile_action_item_list(common_parents):
                for worker_num in range(0, num_workers):
                    thread = threading.Thread(
                        target=lambda: self._run_subscription_queue_worker(
                            subscription_queue, action, action_id, action_path, locks, subscription_count, plain_log,
                        ),
                        name='SubscriptionThread%s' % (worker_num + 1)
                    )
                    # It is not good practice to have daemon threads, but that is required to force
                    # interruption of a thread, as there is no safe way to do that (actually there is no way at all).
                    thread.daemon = True
                    thread.start()
                    threads.append(thread)

        def wait_for_threads_to_finish():
            while any(t.is_alive() for t in threads):
                for t in threads:
                    if t.is_alive():
                        # wait not more than for 1 second, if no timeout is specified
                        # then you can't get KeyboardInterrupt (Ctrl+C)
                        t.join(1)

                        if handle_stop_mark and StopMark.is_set():
                            self._stop_request = True
                            raise MigratorInterruptException()

        try:
            if num_workers > 1:
                wait_for_threads_to_finish()
            else:
                self._run_subscription_queue_worker(
                    subscription_queue, action, action_id, action_path, locks, subscription_count, plain_log,
                )
        except KeyboardInterrupt:
            self._keyboard_interrupt = True

            logger.info(messages.USER_INTERRUPT_REQUESTED_TRYING_STOP_MIGRATION)

            try:
                # wait for workers to finish once more
                # they should process '_keyboard_interrupt' flag - stop on next action
                if num_workers > 1:
                    wait_for_threads_to_finish()
                self._context.progress.stop_progress_thread()
                if self._context.progress.progress_thread is not None:
                    self._context.progress.progress_thread.join()
            except KeyboardInterrupt:
                logger.info(messages.USER_INTERRUPT_REQUESTED_FORCING_STOP_MIGRATION)
                raise

            raise

    def _run_subscription_queue_worker(
        self, subscription_queue, action, action_id, action_path, locks, subscription_count, plain_log=False
    ):
        """
        :type subscription_queue: Queue.Queue
        :type action: parallels.core.actions.base.base_action.BaseAction
        :type action_id: str | unicode
        :type action_path: list[str | unicode]
        :type locks: dict[parallels.core.actions.base.base_action.BaseAction, threading.Lock]
        :type subscription_count: int
        :type plain_log: bool
        :rtype: None
        """
        while True:
            try:
                subscription_num, subscription = subscription_queue.get(False)
            except Empty:
                return  # no more subscriptions in a queue

            with subscription_context(subscription.name):
                try:
                    subscription_progress = self._context.progress.get_subscription(
                        subscription.name
                    )
                    if subscription_progress.status not in SubscriptionMigrationStatus.finished():
                        subscription_progress.status = SubscriptionMigrationStatus.IN_PROGRESS
                    leaf_actions = self._flat_subscription_actions(action)
                    has_info_log_count = len([
                        leaf_action for leaf_action in leaf_actions if action.get_logging_properties().info_log
                    ])
                    if has_info_log_count:
                        overall_logging_function = logger.info
                        overall_logging_context = logger.info_level_context
                    else:
                        overall_logging_function = logger.debug
                        overall_logging_context = logger.debug_level_context

                    if not plain_log:
                        overall_logging_function("")
                        overall_logging_function(
                            messages.LOG_START_SUBSCRIPTION,
                            subscription.name, subscription_num, subscription_count
                        )

                    with overall_logging_context(
                        messages.LOG_LEVEL_PROCESS_SUBSCRIPTION, subscription=subscription.name
                    ):
                        self._run_action_tree(
                            subscription, action, action_id, action_path, locks, plain_log=plain_log,
                            subscription_num=subscription_num, subscription_count=subscription_count
                        )

                    if not plain_log:
                        overall_logging_function(messages.END_PROCESSING_SUBSCRIPTION, subscription.name)
                except Exception as e:
                    error_message = safe_format(
                        messages.INTERNAL_ERROR_IN_SUBSCRIPTION_THREAD,
                        subscription=subscription.name, reason=e
                    )
                    self._context.safe.fail_subscription(subscription.name, error_message)
                    logger.debug(messages.EXCEPTION_IN_SUBSCRIPTION_THREAD, exc_info=True)
                finally:
                    subscription_queue.task_done()

    def _run_subscription_action(
        self, subscription, action, action_path, lock=None, level=0, plain_log=False,
    ):
        """
        :type subscription: parallels.core.migrated_subscription.MigratedSubscription
        :type action: parallels.core.actions.base.subscription_action.SubscriptionAction
        :type action_path: list[str | unicode]
        :type lock: threading.Lock
        :type level: int
        :type plain_log: bool
        :rtype: None
        """
        if (
            self._context.safe.is_failed_subscription(subscription.name) and
            not action.is_execute_for_failed_subscriptions()
        ):
            # subscription is completely failed, do not execute the action
            return

        command_name = Registry.get_instance().get_command_name()
        operation = command_name_to_operation(command_name)
        subscription_progress = self._context.progress.get_subscription(
            subscription.name
        )

        if (
            self._context.subscriptions_status.is_stop(subscription.name)
        ):
            if subscription_progress.status not in SubscriptionMigrationStatus.finished():
                self._context.subscriptions_status.set_operation_status(
                    subscription.name, operation, SubscriptionMigrationStatus.CANCELLED
                )

            # migration of subscription is stopped by user request, do not execute the action
            return

        if action.filter_subscription(self._context, subscription):
            with self._log_action_start(action):
                subscription_progress.action = action.get_description()

                if subscription_progress.status not in SubscriptionMigrationStatus.finished():
                    subscription_progress.status = SubscriptionMigrationStatus.IN_PROGRESS
                    if operation is not None:
                        self._context.subscriptions_status.set_operation_status(
                            subscription.name, operation, SubscriptionMigrationStatus.IN_PROGRESS
                        )

                if lock is not None:
                    lock.acquire()
                try:
                    safe = self._context.safe
                    action_failure_message = action.get_failure_message(
                        self._context, subscription
                    )
                    full_failure_message = action_failure_message
                    if action.is_critical():
                        full_failure_message += "\n" + messages.MIGRATION_FOR_SUBSCRIPTION_FAILED

                    with (
                        get_default_steps_profiler().measure_time_gap(
                            messages.PROFILE_SUBSCRIPTION_TITLE % subscription.name,
                            action.get_description()
                        )
                    ):
                        executed = safe.try_subscription_with_rerun(
                            lambda: action.run(self._context, subscription),
                            subscription.name,
                            error_message=full_failure_message,
                            repeat_error=messages.S_TRY_REPEAT_OPERATION_ONCE_MORE % action_failure_message,
                            repeat_count=action.get_repeat_count(),
                            is_critical=action.is_critical(),
                            use_log_context=False
                        )
                        state_vars = Registry.get_instance().get_migration_state().get_vars(
                            action='/'.join(action_path), subscription=subscription.name
                        )
                        state_vars['executed'] = executed
                finally:
                    subscription_progress.action = None
                    if subscription_progress.status not in SubscriptionMigrationStatus.finished():
                        subscription_progress.status = SubscriptionMigrationStatus.ON_HOLD
                        if operation is not None:
                            self._context.subscriptions_status.set_operation_status(
                                subscription.name, operation, SubscriptionMigrationStatus.ON_HOLD
                            )
                    if lock is not None:
                        lock.release()

    def _run_action_tree(
        self, subscription, action_tree, action_id, action_path, locks, level=0, plain_log=False,
        subscription_num=None, subscription_count=None
    ):
        """
        :type subscription: parallels.core.migrated_subscription.MigratedSubscription
        :type action_tree: parallels.core.actions.base.base_action.BaseAction
        :type action_id: str | unicode
        :type action_path: list[str | unicode]
        :type locks: dict[parallels.core.actions.base.base_action.BaseAction, threading.Lock]
        :type level: int
        :type plain_log: bool
        :type subscription_num: int
        :type subscription_count: int
        :rtype: None
        """
        if self._keyboard_interrupt or self._stop_request:
            # graceful subscription migration interrupt by user request
            # (either from keyboard by Ctrl+C or from GUI by "stop-migration" CLI command)
            return

        if self._is_skip_action_for_subscription(action_id, subscription.name):
            logger.fdebug(
                messages.SKIP_ACTION_FOR_SUBSCRIPTION,
                action_id=action_id, subscription=subscription.name
            )
            return

        with self._profile_action(action_id, action_tree, not plain_log):
            if isinstance(action_tree, CompoundAction):
                if not plain_log:
                    self._log_action(action_tree, level)
                for child_action_id, child_action in action_tree.get_all_actions():
                    child_action_path = action_path + [child_action_id]
                    self._run_action_tree(
                        subscription, child_action, child_action_id, child_action_path, locks, level + 1,
                        plain_log=plain_log, subscription_num=subscription_num, subscription_count=subscription_count
                    )
            elif isinstance(action_tree, SubscriptionAction):
                lock = locks.get(action_tree)
                self._run_subscription_action(
                    subscription, action_tree, action_path, lock, level, plain_log=plain_log,
                )

    def _make_flat_action_list(self, action_id, action, parent_actions=None):
        """Make list of "flat" actions out of actions tree.

        By "flat" we mean plain list of leaf actions that should be executed
        (comparing to tree with compound actions).

        Each item of returned list contains:
        - Leaf action ID.
        - Leaf action object.
        - Ordered list of all parent actions, each item has
            - Parent action ID.
            - Parent action object.

        :type action_id: str | unicode
        :type action: parallels.core.actions.base.base_action.BaseAction
        :type parent_actions: list[parallels.core.workflow.runner.by_subscription.ActionItem] | None
        :rtype: list[parallels.core.workflow.runner.by_subscription.FlatAction]
        """
        if parent_actions is None:
            parent_actions = []

        action = self._resolve_action_pointers(action)

        if isinstance(action, CompoundAction):
            # compound action with child actions
            return sum([
                self._make_flat_action_list(
                    child_action_id, child_action, parent_actions + [ActionItem(action_id, action)]
                )
                for child_action_id, child_action in action.get_all_actions()
            ], [])
        else:
            # leaf action
            return [FlatAction(ActionItem(action_id, action), parent_actions)]

    @staticmethod
    def _split_actions_by_blocks(flat_actions):
        """Split flat actions list by 'subscription' and 'common' action groups

        'subscription' action group consists of actions that are executed per-subscription
        (parallels.core.actions.base.subscription_action.SubscriptionAction)
        'common' action group consists of actions that are executed overall, per-migration
        (parallels.core.actions.base.common_action.CommonAction)

        Returns ordered (according to workflow execution sequence) list of action groups.

        :type flat_actions: list[parallels.core.workflow.runner.by_subscription.FlatAction]
        :rtype: list[
            parallels.core.workflow.runner.by_subscription.SubscriptionActionFlatBlock|
            parallels.core.workflow.runner.by_subscription.CommonActionFlatBlock
        ]
        """
        # either SubscriptionActionFlatBlock.BLOCK_TYPE or CommonActionFlatBlock.BLOCK_TYPE
        previous_action_type = None

        blocks = []
        current_block = []

        for flat_action in flat_actions:
            is_by_subscription_action = (
                isinstance(flat_action.action_item.action, SubscriptionAction) and
                all([parent.action.run_by_subscription for parent in flat_action.parent_action_items])
            )
            if is_by_subscription_action:
                action_type = BySubscriptionActionFlatBlock.BLOCK_TYPE
            else:
                action_type = ByLayerActionFlatBlock.BLOCK_TYPE

            if action_type == previous_action_type:
                # type is the same, add to already created block
                current_block.flat_actions.append(flat_action)
            else:
                # start new block
                if is_by_subscription_action:
                    current_block = BySubscriptionActionFlatBlock()
                else:
                    current_block = ByLayerActionFlatBlock()
                current_block.flat_actions.append(flat_action)
                blocks.append(current_block)

            previous_action_type = action_type

        return blocks

    @staticmethod
    def _build_tree(flat_block):
        """Build action tree out of flat actions block

        :type flat_block: parallels.core.workflow.runner.by_subscription.BaseActionFlatBlock
        :rtype parallels.core.actions.base.compound_action.CompoundAction
        """
        root_action = CompoundAction()

        all_parent_actions = set([
            parent_action_item.action
            for flat_action in flat_block.flat_actions
            for parent_action_item in flat_action.parent_action_items
        ])
        old_action_to_new_action = dict()

        # foreach compound action in parents list - create new compound action with same description and properties,
        # but empty list of child actions
        for parent_action in all_parent_actions:
            old_action_to_new_action[parent_action] = parent_action.clone_without_children()

        restored_relations = set([])
        for flat_action in flat_block.flat_actions:
            current_new_parent_action = root_action
            current_parent_action = None

            # 1) Restore tree structure of compound actions
            for parent_action_item in flat_action.parent_action_items[1:]:
                old_action = parent_action_item.action
                new_action = old_action_to_new_action[old_action]

                if (current_parent_action, old_action) not in restored_relations:
                    current_new_parent_action.insert_action(parent_action_item.action_id, new_action)
                    restored_relations.add((current_parent_action, old_action))

                current_parent_action = old_action
                current_new_parent_action = new_action

            # 2) Insert leaf actions into the new tree
            current_new_parent_action.insert_action(
                flat_action.action_item.action_id, flat_action.action_item.action
            )

        return root_action

    def _list_common_parents(self, action, action_id):
        """Take an action tree and find out the maximum common parents list

        Return list of parent action items (each action item is action id plus action object)
        from the most outer to the inner ones. Each of an item is a parent of
        each leaf action in the tree.

        :type action: parallels.core.actions.base.base_action.BaseAction
        :type action_id: str | unicode
        :rtype: list[parallels.core.workflow.runner.by_subscription.ActionItem]
        """
        common_parents = []

        flat_actions_list = self._make_flat_action_list(action_id, action)
        if len(flat_actions_list) > 0:
            min_parents_list_length = min([
                len(flat_action.parent_action_items)
                for flat_action in flat_actions_list
            ])
            for i in range(1, min_parents_list_length):
                parent_action_items = [
                    flat_action.parent_action_items[i]
                    for flat_action in flat_actions_list
                ]
                if all_equal(parent_action_items):
                    common_parents.append(parent_action_items[0])
                else:
                    break

        return common_parents

    def _resolve_action_pointers(self, action):
        """If specified action is action pointer - resolve it, otherwise return "as-is"

        :type action: parallels.core.actions.base.base_action.BaseAction
        :rtype: parallels.core.actions.base.base_action.BaseAction
        """
        if isinstance(action, ActionPointer):
            action = action.resolve()
        elif isinstance(action, ConditionActionPointer):
            action = action.resolve(self._context)
        return action

    def _is_multithreading_enabled_for_actions(self, actions):
        """Check if multithreading could be used for a group of subscription actions

        Multithreading is enabled for a group, if it is enabled for all actions in a group

        :type actions: list[parallels.core.actions.base.subscription_action.SubscriptionAction]
        :rtype: bool
        """
        for action in actions:
            if not self._is_multithreading_enabled(action):
                return False

        return True

    def _is_multithreading_enabled(self, action):
        """Check if multithreading could be used for a single subscription action

        :type action: parallels.core.actions.base.subscription_action.SubscriptionAction
        :rtype: bool
        """
        action_threading_props = action.get_multithreading_properties()

        if self._multithreading.status == MultithreadingStatus.DISABLED:
            multithreading_enabled = False
        elif self._multithreading.status == MultithreadingStatus.DEFAULT:
            multithreading_enabled = action_threading_props.use_threads_by_default
        elif self._multithreading.status == MultithreadingStatus.FULL:
            multithreading_enabled = action_threading_props.can_use_threads
        else:
            assert False

        return multithreading_enabled

    @contextmanager
    def _profile_action_item_list(self, action_item_list):
        """Measure time of an action passed as list parent actions

        Action item list is a list of all parent action, plus an action
        for which we measure time itself. The list is ordered - the outer actions are at the beginning,
        the inner are at the end

        :type action_item_list: list[parallels.core.workflow.runner.by_subscription.ActionItem]
        """
        if len(action_item_list) == 0:
            yield
        else:
            first_action_item = action_item_list[0]
            action_items_list_tail = action_item_list[1:]
            action_description = first_action_item.action.get_description()
            if action_description is not None:
                with get_default_steps_profiler().measure_time(
                    first_action_item.action_id,
                ):
                    with self._profile_action_item_list(action_items_list_tail):
                        yield
            else:
                with self._profile_action_item_list(action_items_list_tail):
                    yield

    def _shutdown(self, entry_point):
        """Run shutdown actions for specified entry point, considering already executed actions

        :type entry_point: parallels.core.actions.base.entry_point_action.EntryPointAction
        :rtype: None
        """
        # keep shutdown actions ordered, don't use unordered structures like built-in dict() or set()
        shutdown_paths = []

        def add_shutdown_path(new_shutdown_path):
            if (
                # schedule run each shutdown action only once
                new_shutdown_path not in shutdown_paths and
                # run it only if it was not executed before
                new_shutdown_path not in self._executed_common_paths
            ):
                shutdown_paths.append(new_shutdown_path)

        # add shutdown actions for each set-up action
        all_shutdown_actions = entry_point.get_shutdown_actions()
        for executed_path in self._executed_common_paths:
            for shutdown_path in all_shutdown_actions.get(executed_path, []):
                add_shutdown_path(shutdown_path)

        # add overall shutdown actions
        for shutdown_path in all_shutdown_actions.get(None, []):
            add_shutdown_path(shutdown_path)

        # execute selected shutdown actions
        for shutdown_path in shutdown_paths:
            if log_low_level_debug:
                logger.debug(messages.DEBUG_EXECUTE_SHUTDOWN_ACTION, shutdown_path)
            self.run(entry_point.get_path(shutdown_path), shutdown_path, handle_stop_mark=False)

    def _is_skip_whole_action(self, action_id):
        """Whether to skip action for all subscriptions (list of actions to skip is specified by config.ini option)

        :type action_id: str | unicode
        :rtype: bool
        """
        return action_id in self._action_to_skip and self._action_to_skip[action_id] is None

    def _is_skip_action_for_subscription(self, action_id, subscription_name):
        """Whether to skip action for specific subscription (list of actions to skip is specified by config.ini option)

        :type action_id: str | unicode
        :type subscription_name: str | unicode
        :rtype: bool
        """
        return action_id in self._action_to_skip and (
            self._action_to_skip[action_id] is None or
            (
                isinstance(self._action_to_skip[action_id], list) and
                normalize_domain_name(subscription_name) in self._action_to_skip[action_id]
            )
        )


class BaseActionFlatBlock(object):
    def __init__(self, flat_actions=None):
        if flat_actions is None:
            flat_actions = []
        self.flat_actions = flat_actions

    def __eq__(self, other):
        return other.flat_actions == self.flat_actions


class BySubscriptionActionFlatBlock(BaseActionFlatBlock):
    BLOCK_TYPE = 'subscription'


class ByLayerActionFlatBlock(BaseActionFlatBlock):
    BLOCK_TYPE = 'common'


class FlatAction(object):
    def __init__(self, action_item, parent_action_items):
        self.action_item = action_item
        self.parent_action_items = parent_action_items


class ActionItem(object):
    def __init__(self, action_id, action):
        self.action_id = action_id
        self.action = action

    def __eq__(self, other):
        return other.action == self.action and other.action_id == self.action_id
