package be.iminds.ilabt.jfed.highlevel.controller;

import be.iminds.ilabt.jfed.util.IOUtils;
import org.slf4j.LoggerFactory;
import org.slf4j.Marker;
import org.slf4j.MarkerFactory;

import javax.annotation.Nonnull;
import javax.inject.Inject;
import javax.inject.Singleton;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

import static java.util.concurrent.TimeUnit.MILLISECONDS;

/**
 * TaskThread: manages the thread on which Tasks (typically Sfa commands) are called
 * <p/>
 * Uses ScheduledExecutorService and adds names and dependencies to Callable
 */
@SuppressWarnings("SynchronizationOnLocalVariableOrMethodParameter")
@Singleton
public class TaskThread {
    private static final Marker TASKTHREAD_MARKER = MarkerFactory.getMarker("TASKTHREAD");
    private static final Marker ADD_TASK_MARKER = MarkerFactory.getMarker("ADD_TASK");
    private static final Marker ADD_FUTURE_TASK_MARKER = MarkerFactory.getMarker("ADD_FUTURE_TASK");
    private static final Marker TASKTHREAD_LOOP_MARKER = MarkerFactory.getMarker("TASKTHREAD_LOOP");
    private static final Marker PROCESS_NEW_TASK_EXECUTION_MARKER = MarkerFactory.getMarker("processNewTaskExecution");

    private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(TaskThread.class);
    private final List<TaskThreadThread> taskThreads = new ArrayList<>();
    /**
     * This boolean is used to check whether the taskthread should stop
     */
    private final AtomicBoolean stopRequested = new AtomicBoolean(false);
    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
    /**
     * 1 simultaneous assign future task possible
     * calls ready to be executed
     */
    private final BlockingDeque<TaskExecution> queuedTasks = new LinkedBlockingDeque<>();

    /**
     * only uses BlockingDeque when that queue is added to from some thread and waited on in other thread.
     * example: a running queue is never waited for anywhere, the thread that runs would add AND remove from that queue
     * calls waiting for dependencies
     */
    private final BlockingDeque<TaskExecution> blockedTasks = new LinkedBlockingDeque<>();
    /**
     * all active, future, and historic Tasks. only gets added to, never removed from
     * can be added to from multiple threads. Can be read from multiple threads.
     */
    final List<TaskExecution> allTasks = Collections.synchronizedList(new ArrayList<>());


    private final TaskExecutionFactory taskExecutionFactory;

    /**
     * Simple Thread extension that logs who interrupts the thread.
     * This is useful for debugging.
     */
    public class TaskThreadThread extends Thread {
        private final int threadNum;
        private final AtomicBoolean stopRequested;

        private TaskThreadThread(int threadNum, AtomicBoolean stopRequested) {
            this.threadNum = threadNum;
            this.stopRequested = stopRequested;
        }

        public void run() {
            try {
                while (!stopRequested.get())
                    runTask();
            } catch (Exception | AssertionError t) {
                LOG.error("Fatal error in jFed Experimenter Task Thread nr" + threadNum + ": " + t.getMessage(), t);
                throw t;
            } finally {
                LOG.info("jFed Experimenter Task Thread nr" + threadNum + " is shutting down.");
            }
        }

        public final List<String> interruptStacktraces = new ArrayList<>(); // Temporary field for debugging purpose.

        @Override
        public void interrupt() {
            String interruptStacktrace = IOUtils.currentStackToStacktraceString();
            interruptStacktraces.add(interruptStacktrace);
            LOG.warn("TaskThread was interrupted by: " + interruptStacktrace);

            super.interrupt();
        }


        /**
         * Execute a single call. (or execute no call in some cases) This blocks.
         */
        void runTask() {
            TaskExecution currentlyRunningTaskExecution;

            LOG.trace(TASKTHREAD_LOOP_MARKER, "runTask() Fetching task from queue");
            try {
                currentlyRunningTaskExecution = queuedTasks.takeFirst();
            } catch (InterruptedException e) {
                //someone is trying to cancel this thread?
                LOG.warn("runTask() for InterruptedException: stopping");
                return;
            }

            synchronized (currentlyRunningTaskExecution) {
                //last check of cancel flag (doesn't have to be thread safe. If it is too late, it is too late)
                if (currentlyRunningTaskExecution.isCanceledByUser()) {
//                currentlyRunningTaskExecution.setState(TaskExecution.TaskState.CANCELLED);
                    setStateAndDoCallback(currentlyRunningTaskExecution, TaskExecution.TaskState.CANCELLED);
                    return; //ignore task
                }

                currentlyRunningTaskExecution.setState(TaskExecution.TaskState.RUNNING);

            }
            currentlyRunningTaskExecution.setThread(Thread.currentThread());
            Thread.currentThread().setName("TaskThread-" + threadNum + "-" + currentlyRunningTaskExecution.getName());

            TaskExecution.TaskState completedState = TaskExecution.TaskState.FAILED;
            try {
                LOG.trace(TASKTHREAD_LOOP_MARKER, "runTask() Starting: " + currentlyRunningTaskExecution);
                currentlyRunningTaskExecution.registerRunStart();
                currentlyRunningTaskExecution.task.doTask(currentlyRunningTaskExecution);
                currentlyRunningTaskExecution.registerRunStop();
                LOG.trace(TASKTHREAD_LOOP_MARKER, "runTask() Run Successfully: " + currentlyRunningTaskExecution);
                completedState = TaskExecution.TaskState.SUCCESS;
            } catch (InterruptedException e) {
                completedState = TaskExecution.TaskState.CANCELLED;
                currentlyRunningTaskExecution.exception = e;
                LOG.warn("Run cancelled: " + currentlyRunningTaskExecution + "  -> Exception=\"" + e + "\"", e);
            } catch (Exception | AssertionError t) {
                completedState = TaskExecution.TaskState.FAILED;
                currentlyRunningTaskExecution.exception = t;
                LOG.error("Run failed: " + currentlyRunningTaskExecution + "  -> Exception=\"" + t + "\"", t);
            } finally {
                setStateAndDoCallback(currentlyRunningTaskExecution, completedState);
                Thread.currentThread().setName("TaskThread-" + threadNum);
            }

            currentlyRunningTaskExecution = null;

            updateBlocking();
        }

    }

    @Inject
    TaskThread(TaskExecutionFactory taskExecutionFactory) {
        this.taskExecutionFactory = taskExecutionFactory;

        for (int i = 0; i < 3; i++) {
            final int threadNum = i;
            TaskThreadThread t = new TaskThreadThread(i, stopRequested);
            t.setDaemon(true);
            t.setName("TaskThread-" + i);
            taskThreads.add(t);
            t.start();
        }
    }

    public void requestStop() {
        stopRequested.set(true);
        scheduledExecutorService.shutdownNow();
    }

    public boolean isStopRequested() {
        return stopRequested.get();
    }

    public void checkStateCorrectness(TaskExecution taskExecution) {
        if (taskExecution == null)
            throw new NullPointerException("taskExecution cannot be null");

        synchronized (taskExecution) {
            try {
                TaskExecution.TaskState state = taskExecution.getState();
                if (state == null)
                    throw new NullPointerException("taskExecution state cannot be null");

                if (state == TaskExecution.TaskState.BLOCKED)
                    assert blockedTasks.contains(taskExecution);
                else
                    assert !blockedTasks.contains(taskExecution);


                if (state == TaskExecution.TaskState.QUEUED) {
                /*we might catch the task as it is just being deque-ed for running, so we can't do this check*/
//                    assert queuedTasks.contains(taskExecution);
                } else
                    assert !queuedTasks.contains(taskExecution);
            } catch (Exception | AssertionError e) {
                LOG.error("checkStateCorrectness failed for taskExecution \"" + taskExecution + "\"", e);
                LOG.error(" taskExecution.getState()=" + taskExecution.getState());
                LOG.error("      queuedTasks=" + queuedTasks);
                LOG.error("     blockedTasks=" + blockedTasks);
                LOG.error("  THIS POTENTIALLY FATAL ERROR WILL BE IGNORED!");
            }
        }
    }

    public void cancel(List<? extends Task> tasks) {
        for (Task t : tasks)
            cancel(t);
    }

    public void cancel(Task task) {
        for (TaskExecution st : task.getCurrentActiveTaskExecutionsCopy())
            cancel(st);
        for (FutureTask ft : task.getFuturesCopy())
            ft.tryCancel();
    }

    /**
     * Try to deque or stop execution of a task. Note that this task may have started or completed during this method,
     * so success cannot be guaranteed. (Except that when this method returns, the task is not running and will
     * never start running)
     */
    public void cancel(TaskExecution task) {
        synchronized (task) {
            if (task.state == TaskExecution.TaskState.RUNNING) {
                task.setCanceledByUser(true);
                if (task.getThread() != null) {
                    //Note: interrupt will not force stop. It might be too late to cancel.
                    task.getThread().interrupt();
                }
                return;
            }

            if (task.state == TaskExecution.TaskState.BLOCKED)
                blockedTasks.remove(task);

            if (task.state == TaskExecution.TaskState.QUEUED)
                queuedTasks.remove(task);

            task.setCanceledByUser(true);
            task.state =
                    TaskExecution.TaskState.CANCELLED; //TODO: for consistency, this should call the callback unless it was running (in that case callback is done)! But we don't know if all callbacks expect that...
            task.task.removeActiveTaskExecution(task);
        }
        checkStateCorrectness(task);
        updateBlocking();
    }

    void setStateAndDoCallback(TaskExecution currentlyRunningTaskExecution, TaskExecution.TaskState completedState) {
        Collection<TaskFinishedCallback> taskFinishedCallbackList;
        synchronized (currentlyRunningTaskExecution) {
            taskFinishedCallbackList = new ArrayList<>();
            taskFinishedCallbackList.addAll(currentlyRunningTaskExecution.task.getCallbacks());
            taskFinishedCallbackList.addAll(currentlyRunningTaskExecution.getTaskExecutionFinishedCallbacks());

            currentlyRunningTaskExecution.setState(completedState);
            currentlyRunningTaskExecution.setThread(null);
            currentlyRunningTaskExecution.task.setLastCompletedTaskExecution(currentlyRunningTaskExecution);
            currentlyRunningTaskExecution.task.removeActiveTaskExecution(currentlyRunningTaskExecution);

            if (completedState == TaskExecution.TaskState.SUCCESS &&
                    currentlyRunningTaskExecution.task.newExecutionNeeded()) //NOTE: newExecutionNeeded() uses lastCompletedTaskExecution
                LOG.error("The task " + currentlyRunningTaskExecution.getName() + " just ran successfully, " +
                        "however, it still says that it is needed! That means something has gone wrong." +
                        " This is most likely a bug in the Task code.");
        }

        for (TaskFinishedCallback callback : taskFinishedCallbackList) {
            try {
                callback.onTaskExecutionFinished(currentlyRunningTaskExecution.task, currentlyRunningTaskExecution, completedState);
            } catch (Exception | AssertionError t) {
                LOG.error(TASKTHREAD_LOOP_MARKER, "runTask() Callback failed: " + t.getMessage(), t);
            }
        }
    }

    /**
     * check if all blocked tasks are still blocked
     */
    public void updateBlocking() {
        List<TaskExecution> toCheck = new ArrayList<>();

        boolean keepChecking =
                true; //might need to check more than once, since a failure can cause a cascade of failures.

        while (keepChecking) {
            toCheck.clear();
            blockedTasks.drainTo(toCheck);
            //blockedTasks is now empty!

            int prevBlockedCount = toCheck.size();

            if (!toCheck.isEmpty()) {
                LOG.debug(TASKTHREAD_LOOP_MARKER, "updateBlocking() Assigning blocked tasks count=" + toCheck.size());
                for (TaskExecution blockedCall : toCheck) {
                    processTaskExection(blockedCall);
                }
            }

            keepChecking = blockedTasks.size() != prevBlockedCount;
        }

        LOG.debug(TASKTHREAD_LOOP_MARKER, "updateBlocking() updateAfterTaskrun done, will unlock"
                + "\n\t" + queuedTasks.size() + " queuedTasks= " + queuedTasks
                + "\n\t" + blockedTasks.size() + " blockedTasks=" + blockedTasks);
    }

    /**
     * put calls either in the queued calls (can run now) or in the blocked calls (waiting for at least 1 dependency)
     */
    private synchronized void processNewTaskExecution(TaskExecution origTaskExecution) {
        if (origTaskExecution == null) return;

        //if deps are not added, add them now
        List<TaskExecution> taskExecutionsToAssign = new ArrayList<>();
        Queue<TaskExecution> newTaskExecutionsToProcess = new LinkedList<>();
        newTaskExecutionsToProcess.offer(origTaskExecution);
        taskExecutionsToAssign.add(origTaskExecution);

        LOG.debug(PROCESS_NEW_TASK_EXECUTION_MARKER, "processNewTaskExecution() origTaskExecution=" + origTaskExecution + " (id=" + origTaskExecution.id + ")"); //don't use {} -> that is an empty list!

        while (!newTaskExecutionsToProcess.isEmpty()) {
            LOG.trace("assigning calls: " + newTaskExecutionsToProcess);

            TaskExecution taskExecution = newTaskExecutionsToProcess.poll();

            Task task = taskExecution.task;

            LOG.debug(PROCESS_NEW_TASK_EXECUTION_MARKER, "processNewTaskExecution()   PROCESS task=" + task + " (id=" + task.getId() + ")  taskExecution=" + taskExecution + " (id=" + taskExecution.id + ")");

            /*
             * DependsOn-tasks only have to be executed once.
             * If it was executed in the past, we can use that result of that call.
             */
            for (Task dep : task.getDependsOn()) {
                assert (dep != null) : "Got a null DependsOn-dependency for task " + dep.getName();
                LOG.debug(PROCESS_NEW_TASK_EXECUTION_MARKER, "processNewTaskExecution()     DEP=" + dep + " (id=" + dep.getId() + ")");
//                    checkStateCorrectness(dep);

                TaskExecution<?> depTaskExecution = null;
                //Use any running task that is the same Task
                for (TaskExecution activeTaskExecution : dep.getCurrentActiveTaskExecutionsCopy()) {
                    LOG.debug(PROCESS_NEW_TASK_EXECUTION_MARKER, "processNewTaskExecution()         found in currentActiveSingleTasks activeTaskExecution=" + activeTaskExecution + "");
                    if (activeTaskExecution.getState() != TaskExecution.TaskState.FUTURE)
                        depTaskExecution = activeTaskExecution;
                }

                //Use any newly created unsubmitted task that is the same Task
                for (TaskExecution newlyCreatedTask : taskExecutionsToAssign)
                    if (newlyCreatedTask.task == dep) {
                        LOG.debug(PROCESS_NEW_TASK_EXECUTION_MARKER, "processNewTaskExecution()         found in taskExecutionsToAssign newlyCreatedTask=" + newlyCreatedTask + "");
                        depTaskExecution = newlyCreatedTask;
                    }

                //Use any finished task that is the same Task, unless newExecutionNeeded, because otherwise, you get bugs like FEDIBBTDEV-278,
                //but prefer tasks already found above if they existed (since it is being redone, we'll wait for it)
                //  this also eliminates the need for always calling newExecutionNeeded()
                if (depTaskExecution == null && !dep.newExecutionNeeded()) {
                    LOG.debug(PROCESS_NEW_TASK_EXECUTION_MARKER, "processNewTaskExecution()         found in dep.lastCompletedSingleTask=" + dep.getLastCompletedTaskExecution() + "");
                    depTaskExecution = dep.getLastCompletedTaskExecution();
                }

                //Create a new FXTaskExecution if no taskExecution to use found
                if (depTaskExecution == null) {
                    depTaskExecution = taskExecutionFactory.createTaskExecution(dep);
                    newTaskExecutionsToProcess.offer(depTaskExecution);
                    taskExecutionsToAssign.add(depTaskExecution);
                    depTaskExecution.task.addActiveTaskExecution(depTaskExecution);
                    LOG.debug(PROCESS_NEW_TASK_EXECUTION_MARKER, "processNewTaskExecution()         creating new: " + depTaskExecution + "");
                }

                assert depTaskExecution != null;

                //always register as dep
                taskExecution.getDependsOn().add(depTaskExecution);
                depTaskExecution.getDependingOnThis().add(taskExecution);
            }

            /*
             * AlwaysDependsOn-tasks should always be (re-)executed before doing this task.
              * This is necessary to detect changing statusses, etc
             */
            for (Task dep : task.getAlwaysDependsOn()) {
                assert (dep != null) : "Got a null AlwaysDependsOn-dependency for task " + dep.getName();
                LOG.debug(PROCESS_NEW_TASK_EXECUTION_MARKER, "processNewTaskExecution()         ALWAYSDEP=" + dep + " (id=" + dep.getId() + ")");
//                    checkStateCorrectness(dep);

                TaskExecution depTaskExecution = null;
                //Use any running task that is the same Task
                LOG.debug(PROCESS_NEW_TASK_EXECUTION_MARKER, "processNewTaskExecution()         dep.currentActiveSingleTasks=" + dep.getCurrentActiveTaskExecutionsCopy() + "");
                for (TaskExecution activeTaskExecution : dep.getCurrentActiveTaskExecutionsCopy()) {
                    LOG.debug(PROCESS_NEW_TASK_EXECUTION_MARKER, "processNewTaskExecution()             found in currentActiveSingleTasks activeTaskExecution=" + activeTaskExecution + "");
                    if (activeTaskExecution.getState() != TaskExecution.TaskState.FUTURE)
                        depTaskExecution = activeTaskExecution;
                }

                //Use any newly created unsubmitted task that is the same Task
                for (TaskExecution newlyCreatedTask : taskExecutionsToAssign)
                    if (newlyCreatedTask.task == dep) {
                        LOG.debug(PROCESS_NEW_TASK_EXECUTION_MARKER, "processNewTaskExecution()             found in taskExecutionsToAssign newlyCreatedTask=" + newlyCreatedTask + "");
                        depTaskExecution = newlyCreatedTask;
                    }

                //Create a new FXTaskExecution if no taskExecution to use found
                if (depTaskExecution == null) {
                    depTaskExecution = taskExecutionFactory.createTaskExecution(dep);
                    newTaskExecutionsToProcess.offer(depTaskExecution);
                    taskExecutionsToAssign.add(depTaskExecution);
                    depTaskExecution.task.addActiveTaskExecution(depTaskExecution);
                    LOG.debug(PROCESS_NEW_TASK_EXECUTION_MARKER, "processNewTaskExecution()             creating new: " + depTaskExecution + "");
                }

                assert depTaskExecution != null;

                //always register as dep
                taskExecution.getDependsOn().add(depTaskExecution);
                depTaskExecution.getDependingOnThis().add(taskExecution);
            }

            assert !allTasks.contains(taskExecution) : "task was already added: " + taskExecution.getName();
            allTasks.add(taskExecution);

        }

        LOG.debug(PROCESS_NEW_TASK_EXECUTION_MARKER, "processNewTaskExecution() finished recursivly adding deps to for new FXTaskExecution. Will now call processTaskExection() for all new TaskExecutions");

        while (!taskExecutionsToAssign.isEmpty()) {
            List<TaskExecution> assignedTasks = new ArrayList<>();
            for (TaskExecution newTaskExecution : taskExecutionsToAssign) {
                boolean assigned = processTaskExection(newTaskExecution);
                if (assigned) assignedTasks.add(newTaskExecution);
            }
            taskExecutionsToAssign.removeAll(assignedTasks);
            if (assignedTasks.isEmpty())
                throw new RuntimeException("Infinite loop while assigning: " + taskExecutionsToAssign); //was an assert, but it's best to always check
        }
    }

    /**
     * receives a TaskExecution with all dependencies filled in correctly. (and recursively filled in correctly for deps of deps etc...)
     * <p/>
     * Will submit unsubmitted TaskExecution.
     * Will queue blocked singleRuns if they are not blocked anymore
     * will fail (and move them to history) calls if their deps have failed.
     * <p/>
     * Also checks correctness of call status VS lists they are in
     * <p/>
     * Tasks sent here will not be in queuedTasks or blockedTasks, so checkStateCorrectness is incorrect at first!
     * <p/>
     * returns true if assign is successful (call is no longer unsubmitted)
     * in case any dependency is unsubmitted, this returns false and does not assign.
     */
    public boolean processTaskExection(TaskExecution<?> taskExecution) {
        synchronized (taskExecution) {
//            checkStateCorrectness(taskExecution); //not guaranteed

            //this is only supposed to be used in the following cases:
            assert taskExecution.getState() == TaskExecution.TaskState.BLOCKED ||
                    taskExecution.getState() == TaskExecution.TaskState.QUEUED ||
                    taskExecution.getState() == TaskExecution.TaskState.UNSUBMITTED :
                    "did not expect task given as argument to processTaskExection to be in state: " + taskExecution.getState();

            LOG.debug(TASKTHREAD_MARKER, "processTaskExection(" + taskExecution + ")");

            boolean allDepsOk = true;
            for (TaskExecution<?> taskExecutionDep : taskExecution.getDependsOn()) {
                synchronized (taskExecutionDep) {
//                    checkStateCorrectness(taskExecutionDep); //not guaranteed (deps might not yet have been assigned)

                    //fail this FXTaskExecution when a dependency failed or was cancelled
                    if (taskExecutionDep.getState() == TaskExecution.TaskState.FAILED ||
                            taskExecutionDep.getState() == TaskExecution.TaskState.CANCELLED) {
//                        taskExecution.setState(taskExecutionDep.getState());
                        setStateAndDoCallback(taskExecution, taskExecutionDep.getState());
                        taskExecution.task.onCancelledDueToDependencyFailed();

                        checkStateCorrectness(taskExecution);
                        LOG.debug(TASKTHREAD_MARKER, "processTaskExection taskExecutionDep(" + taskExecutionDep + ")  DEP FAILED! cancelling task!");
                        return true;
                    }

                    //this cannot happen because we synchronize on taskExecutionDep,
                    // and state would be CANCELLED if isCanceledByUser() gives true
                    assert (!taskExecutionDep.isCanceledByUser()) : "Inconsistent canceledByUser-state";

                    if (taskExecutionDep.getState() == TaskExecution.TaskState.UNSUBMITTED) {
                        LOG.debug(TASKTHREAD_MARKER, "processTaskExection taskExecutionDep(" + taskExecutionDep + ")  DEP unsubmitted.");

                        // Are there any other situations than another thread that is already processing this TaskExecution?
                        // If not, then we can state that it will be assigned in the other thread?

                        return false; //cannot continue until it is submitted
                    }


                    //***IMPORTANT*** This stuff is the cause of subtle bugs! see FEDIBBTDEV-329 and FEDIBBTDEV-278 and ask wvdemeer before touching this!
                    if (taskExecutionDep.task.newExecutionNeeded()) {
                        LOG.debug(TASKTHREAD_MARKER, "processTaskExection taskExecutionDep(" + taskExecutionDep + ").task.needed=" + taskExecutionDep.task.newExecutionNeeded() + " state=" + taskExecutionDep.getState() + ", so allDepsOk = false");

                        if (taskExecutionDep.state == TaskExecution.TaskState.SUCCESS) {
                            //task succesfull, BUT something went wrong that makes it needed again! This is bad!
                            throw new RuntimeException("BUG in Task dependencies! Task \"" + taskExecutionDep.getName() + "\" success, even though newExecutionNeeded() says true! (might hang task " + taskExecution.getName() + ")");

                            //if you ignore this and set allDepsOk = false, you'll get subtle hang bugs!
                            //if you ignore this and leave allDepsOk unchanged, you'll get subtle bugs like FEDIBBTDEV-278, where calls are not done again, even though they need to be redone.

                            //the solution is to never get in this situation, by calling newExecutionNeeded and if that is true, prevent reusing a FXTaskExecution
                        }

                        allDepsOk = false;
                    }
                }
            }

            LOG.debug(TASKTHREAD_MARKER, "processTaskExection allDepsOk=" + allDepsOk);

            try {
                if (allDepsOk) {
                    synchronized (taskExecution) {
                        taskExecution.setState(TaskExecution.TaskState.QUEUED);
                        queuedTasks.putFirst(taskExecution);
                    }
                    LOG.debug(TASKTHREAD_MARKER, "processTaskExection registered as QUEUED");
                } else {
                    synchronized (taskExecution) {
                        taskExecution.setState(TaskExecution.TaskState.BLOCKED);
                        blockedTasks.putFirst(taskExecution);
                    }
                    LOG.debug(TASKTHREAD_MARKER, "processTaskExection registered as BLOCKED");
                }
            } catch (InterruptedException e) {
                LOG.error("Did not expect InterruptedException could occur in processTaskExecution. This is possibly a problem. It will be ignored.", e);
            }
        }
        checkStateCorrectness(taskExecution);
        return true;
    }


    public <T extends Task> TaskExecution<T> addTask(final T task) {
        return addTask(task, Collections.emptySet());
    }

    public <T extends Task> TaskExecution<T> addTask(final T task, @Nonnull TaskExecutionFinishedCallback<T> callback) {
        return addTask(task, Collections.singleton(callback));
    }

    public <T extends Task> TaskExecution<T> addTask(final T task, @Nonnull Collection<TaskExecutionFinishedCallback<T>> callbacks) {
        LOG.debug(ADD_TASK_MARKER, "addTask() Adding task: \"" + task + "\"");

        if (task.isActive()) {
            LOG.error("trying to add active task: " + task.getCurrentActiveTaskExecutionsCopy() + " -> task will not be added.");
            @SuppressWarnings("unchecked")
            final TaskExecution<T> lastTaskExecution =
                    task.getCurrentActiveTaskExecutionsCopy().get(task.getCurrentActiveTaskExecutionsCopy().size() - 1);
            lastTaskExecution.getTaskExecutionFinishedCallbacks().addAll(callbacks);
            return lastTaskExecution;
        }

        LOG.debug(ADD_TASK_MARKER, "addTask() {} queuedTasks= {}", queuedTasks.size(), queuedTasks);
        LOG.debug(ADD_TASK_MARKER, "addTask() {} blockedTasks= {}", blockedTasks.size(), blockedTasks);

        TaskExecution<T> taskExecution = taskExecutionFactory.createTaskExecution(task, callbacks);
        task.addActiveTaskExecution(taskExecution);
        processNewTaskExecution(taskExecution);

        LOG.debug(ADD_TASK_MARKER, "addTask() Added task: \"{}\"", task);
        LOG.debug(ADD_TASK_MARKER, "addTask() {} queuedTasks= {}", queuedTasks.size(), queuedTasks);
        LOG.debug(ADD_TASK_MARKER, "addTask() {} blockedTasks= {}", blockedTasks.size(), blockedTasks);

        return taskExecution;
    }


    public <T extends Task> TaskExecution scheduleTask(final T task, final long delayMs) {
        return scheduleTask(task, delayMs, Collections.emptySet());
    }
    public <T extends Task> TaskExecution scheduleTask(final T task, final long delayMs,@Nonnull TaskExecutionFinishedCallback<T> callback) {
        return scheduleTask(task, delayMs, Collections.singleton(callback));
    }

    public <T extends Task> TaskExecution scheduleTask(final T task, final long delayMs, @Nonnull Collection<TaskExecutionFinishedCallback<T>> callbacks) {
        LOG.debug("Scheduled task '{}' in {} ms", task.getName(), delayMs);
        synchronized (task) {
            TaskExecution taskExecution = taskExecutionFactory.createTaskExecution(task, callbacks);
            taskExecution.setState(TaskExecution.TaskState.FUTURE);
            task.addActiveTaskExecution(taskExecution);

            FutureTask futureTask = new FutureTask(taskExecution);
            task.addFuture(futureTask);
            ScheduledFuture scheduledFuture =
                    scheduledExecutorService.schedule(futureTask, delayMs, MILLISECONDS); //Note: synchronized (task) because this may not be done before previous line (which used to be next line)
            futureTask.setScheduledFuture(scheduledFuture);

            return taskExecution;
        }
    }

    public class FutureTask implements Runnable {

        private final TaskExecution task;
        JavaFXTaskThread.FutureTask javafx;
        private ScheduledFuture scheduledFuture;
        private boolean cancel = false;

        FutureTask(TaskExecution task) {
            this.task = task;
        }

        public TaskExecution getTask() {
            return task;
        }

        public ScheduledFuture getScheduledFuture() {
            return scheduledFuture;
        }

        void setScheduledFuture(ScheduledFuture scheduledFuture) {
            this.scheduledFuture = scheduledFuture;
        }

        public void tryCancel() {
            cancel = true;
            if (scheduledFuture != null) {
                boolean wasCancelled = scheduledFuture.cancel(false);
                if (wasCancelled) {
//                    task.setState(TaskExecution.TaskState.CANCELLED);
                    setStateAndDoCallback(task, TaskExecution.TaskState.CANCELLED);
                }
            }
        }

        public long getTimeLeftMs() {
            if (scheduledFuture != null)
                return scheduledFuture.getDelay(MILLISECONDS);
            else
                return -1;
        }

        @Override
        public void run() {
            if (cancel) {
//                task.setState(TaskExecution.TaskState.CANCELLED);
                setStateAndDoCallback(task, TaskExecution.TaskState.CANCELLED);
                return;
            }
            try {
                LOG.debug("Adding future task: \"" + task + "\"");

                synchronized (task.task) {
                    boolean wasRemoved = task.task.removeFuture(this);
                    assert wasRemoved;
                }

                task.setState(TaskExecution.TaskState.UNSUBMITTED);

                processNewTaskExecution(task);
                LOG.debug(ADD_FUTURE_TASK_MARKER, "Added future task: \"" + task + "\"");
                LOG.debug(ADD_FUTURE_TASK_MARKER, "{} queuedTasks= {}", queuedTasks.size(), queuedTasks);
                LOG.debug(ADD_FUTURE_TASK_MARKER, "{} blockedTasks= {}", blockedTasks.size(), blockedTasks);
            } catch (Exception | AssertionError e) {
                LOG.error(ADD_FUTURE_TASK_MARKER, "Exception while running addCall on future task. This is a bug. " + e.getMessage(), e);
            }
        }
    }

}
