import os
import shutil
import time
from contextlib import contextmanager
from threading import Lock, Thread

from parallels.core import messages, get_logger
from parallels.core.task_manager.task import Task
from parallels.core.utils.common import mkdir_p, open_no_inherit
from parallels.core.utils.locks.file_lock import FileLock
from parallels.core.utils.locks.file_lock_exception import FileLockException


logger = get_logger(__name__)


class TaskManager(object):
    def __init__(self, session_path):
        """
        :type session_path: str | unicode 
        """
        self._session_path = session_path
        self._queue_lock = FileLock(self._session_path, 'queue.lock')
        self._queue_thread_lock = Lock()
        self._runner_lock = FileLock(self._session_path, 'runner.lock')

    @contextmanager
    def task_operations(self):
        with self._queue_thread_lock:
            self._queue_lock.acquire_block()
            try:
                yield TaskOperations(self._session_path)
            finally:
                self._queue_lock.release()

    def run_tasks_sequential(self, task_runner, task_selector):
        """
        :type task_runner: parallels.core.task_manager.task_runner.TaskRunner 
        :type task_selector: parallels.core.task_manager.task_selector.TaskSelector 
        :rtype: None 
        """
        try:
            # Note: do not release the lock. It will be released automatically once the process is stopped.
            self._runner_lock.acquire()
        except FileLockException:
            logger.finfo(messages.ANOTHER_INSTANCE_OF_RUN_QUEUE_RUNNING)
            return

        while True:
            with self.task_operations() as task_operations:
                assert isinstance(task_operations, TaskOperations)
                all_tasks = task_operations.get_tasks_to_run()
                selected_task = task_selector.select(all_tasks)

                if selected_task is not None:
                    task_operations.set_task_status(selected_task, Task.STATUS_RUNNING)

            if selected_task is None:
                break

            logger.finfo(messages.RUN_TASK, task_id=selected_task.task_id)
            try:
                task_runner.run(selected_task)
            except Exception as e:
                logger.exception()
                logger.ferror(messages.FAILED_TASK, task_id=selected_task.task_id, reason=e)
                status = Task.STATUS_FAILED
            else:
                logger.finfo(messages.FINISHED_TASK, task_id=selected_task.task_id)
                status = Task.STATUS_FINISHED

            with self.task_operations() as task_operations:
                assert isinstance(task_operations, TaskOperations)
                if task_operations.get_task_status(selected_task) == Task.STATUS_RUNNING:
                    task_operations.set_task_status(selected_task, status)

    def run_tasks_parallel(self, task_runner):
        """
        :type task_runner: parallels.core.task_manager.task_runner.TaskRunner
        :rtype: None
        """
        try:
            # Note: do not release the lock. It will be released automatically once the process is stopped.
            self._runner_lock.acquire()
        except FileLockException:
            logger.finfo(messages.ANOTHER_INSTANCE_OF_RUN_QUEUE_RUNNING)
            return

        threads = []
        executed_task_ids = set()

        while True:
            with self.task_operations() as task_operations:
                assert isinstance(task_operations, TaskOperations)
                all_tasks = task_operations.get_tasks_to_run()

                for task in all_tasks:
                    if task.task_id not in executed_task_ids:
                        task_operations.set_task_status(task, Task.STATUS_RUNNING)

                threads = [t for t in threads if t.is_alive()]

            if len(all_tasks) == 0 and len(threads) == 0:
                break

            for task in all_tasks:
                if task.task_id not in executed_task_ids:
                    executed_task_ids.add(task.task_id)
                    thread = Thread(target=lambda task_arg=task: self._run_task(task_runner, task_arg))
                    thread.start()
                    threads.append(thread)

            time.sleep(1)

    def _run_task(self, task_runner, task):
        """
        :type task_runner: parallels.core.task_manager.task_runner.TaskRunner
        :type task: parallels.core.task_manager.task.Task
        :return:
        """
        logger.finfo(messages.RUN_TASK, task_id=task.task_id)
        try:
            task_runner.run(task)
        except Exception as e:
            logger.exception()
            logger.ferror(messages.FAILED_TASK, task_id=task.task_id, reason=e)
            status = Task.STATUS_FAILED
        else:
            logger.finfo(messages.FINISHED_TASK, task_id=task.task_id)
            status = Task.STATUS_FINISHED

        with self.task_operations() as task_operations:
            assert isinstance(task_operations, TaskOperations)
            if task_operations.get_task_status(task) == Task.STATUS_RUNNING:
                task_operations.set_task_status(task, status)

    def wait_for_shutdown(self):
        """Wait once the task runner running in another process is stopped.
        
        Attention: do not call inside of "task_operations" context. Task manager won't be able to stop
        if we are in "task_operations" context as it calls itself "task_operations" to mark finished task
        as finished.
        
        :rtype: None 
        """
        self._runner_lock.acquire_block()


class TaskOperations(object):
    def __init__(self, session_path):
        self._session_path = session_path

    def create_task(self, task_id=None):
        """
        :type task_id: str | unicode | None 
        :rtype: parallels.core.task_manager.task.Task
        """
        mkdir_p(self._session_path)

        if task_id is None:
            task_num = 0

            while True:
                task_num += 1
                task_id = "%s-%s" % (time.strftime('%Y-%m-%d-%H-%M-%S'), task_num)
                task_dir = os.path.join(self._session_path, task_id)
                if not os.path.exists(task_dir):
                    break

        task_dir = os.path.join(self._session_path, task_id)
        status_file = os.path.join(task_dir, 'status')

        mkdir_p(task_dir)

        with open_no_inherit(status_file, 'wb') as fp:
            fp.write(Task.STATUS_PREPARE)

        return Task(task_id, task_dir)

    def get_tasks_to_run(self):
        """
        :rtype: list[parallels.core.task_manager.task.Task] 
        """
        all_tasks = self.get_all_tasks()

        ready_to_run_tasks = [
            task for task in all_tasks
            if self.get_task_status(task) in [
                Task.STATUS_READY,
                # consider tasks which are marked as running, as ready to run, to handle situation
                # when task runner crashed or terminated abnormally and
                # left status of unfinished task as "running"
                Task.STATUS_RUNNING
            ]
        ]

        return ready_to_run_tasks

    def get_all_tasks(self):
        """
        :rtype: list[parallels.core.task_manager.task.Task]
        """
        task_ids = [
            item for item in os.listdir(self._session_path)
            if os.path.isdir(os.path.join(self._session_path, item))
        ]

        return [
            Task(task_id, os.path.join(self._session_path, task_id))
            for task_id in task_ids
        ]

    @staticmethod
    def get_task_status(task):
        """
        :type task: parallels.core.task_manager.task.Task 
        :rtype: str | unicode 
        """
        status_file = os.path.join(task.task_dir, 'status')

        try:
            with open_no_inherit(status_file) as fp:
                status = fp.read()
                if status in Task.all_statuses:
                    return status
                else:
                    return Task.STATUS_PREPARE
        except IOError:
            return Task.STATUS_PREPARE

    @staticmethod
    def set_task_status(task, status):
        """
        :type task: parallels.core.task_manager.task.Task 
        :type status: str | unicode 
        :rtype: None 
        """
        status_file = os.path.join(task.task_dir, 'status')

        with open_no_inherit(status_file, 'wb') as fp:
            fp.write(status)

    def schedule_task(self, task):
        """
        :type task: parallels.core.task_manager.task.Task 
        :rtype: None 
        """
        self.set_task_status(task, Task.STATUS_READY)

    @staticmethod
    def remove_task(task):
        """
        :type task: parallels.core.task_manager.task.Task 
        :rtype: None 
        """
        shutil.rmtree(task.task_dir)
