/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.brooklyn.util.core.task;

import groovy.lang.Closure;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.brooklyn.api.mgmt.HasTaskChildren;
import org.apache.brooklyn.api.mgmt.Task;
import org.apache.brooklyn.api.mgmt.TaskQueueingContext;
import org.apache.brooklyn.core.mgmt.BrooklynTaskTags;
import org.apache.brooklyn.core.mgmt.internal.ManagementContextInternal;
import org.apache.brooklyn.util.collections.MutableMap;
import org.apache.brooklyn.util.exceptions.Exceptions;
import org.apache.brooklyn.util.time.CountdownTimer;
import org.apache.brooklyn.util.time.Duration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.annotations.Beta;
import com.google.common.collect.ImmutableList;

/** Represents a task whose run() method can create other tasks
 * which are run sequentially, but that sequence runs in parallel to this task
 * <p>
 * There is an optional primary job run with this task, along with multiple secondary children.
 * If any secondary task fails (assuming it isn't {@link Tasks#markInessential()} then by default
 * subsequent tasks are not submitted and the primary task fails (but no tasks are cancelled or interrupted).
 * You can change the behavior of this task with fields in {@link FailureHandlingConfig},
 * or the convenience {@link TaskQueueingContext#swallowChildrenFailures()}
 * (and {@link DynamicTasks#swallowChildrenFailures()} if you are inside the task).
 * <p>
 * This synchronizes on secondary tasks when submitting them, in case they may be manually submitted
 * and the submitter wishes to ensure it is only submitted once.
 * <p>
 * Improvements which would be nice to have:
 * <li> unqueued tasks not visible in api; would like that
 * <li> uses an extra thread (submitted as background task) to monitor the secondary jobs; would be nice to remove this,
 *      and rely on {@link BasicExecutionManager} to run the jobs sequentially (combined with fix to item above)
 * <li> would be nice to have cancel, resume, and possibly skipQueue available as operations (ideally in the REST API and GUI)   
 **/
public class DynamicSequentialTask<T> extends BasicTask<T> implements HasTaskChildren, TaskQueueingContext {

    private static final Logger log = LoggerFactory.getLogger(CompoundTask.class);
                
    protected final Queue<Task<?>> secondaryJobsAll = new ConcurrentLinkedQueue<Task<?>>();
    protected final Queue<Task<?>> secondaryJobsRemaining = new ConcurrentLinkedQueue<Task<?>>();
    protected final Object jobTransitionLock = new Object();
    protected volatile boolean primaryStarted = false;
    protected volatile boolean primaryFinished = false;
    protected volatile boolean secondaryQueueAborted = false;
    protected Thread primaryThread;
    protected DstJob dstJob;
    protected FailureHandlingConfig failureHandlingConfig = FailureHandlingConfig.DEFAULT;

    // default values for how to handle the various failures
    @Beta
    public static class FailureHandlingConfig {
        /** secondary queue runs independently of primary task (submitting and blocking on each secondary task in order), 
         * but can set it up not to submit any more tasks if the primary fails */
        public final boolean abortSecondaryQueueOnPrimaryFailure;
        /** as {@link #abortSecondaryQueueOnPrimaryFailure} but controls cancelling of secondary queue*/
        public final boolean cancelSecondariesOnPrimaryFailure;
        /** secondary queue can continue submitting+blocking tasks even if a secondary task fails (unusual;
         * typically handled by {@link TaskTags#markInessential(Task)} on the secondary tasks, in which case
         * the secondary queue is never aborted */
        public final boolean abortSecondaryQueueOnSecondaryFailure;
        /** unsubmitted secondary tasks (ie those further in the queue) can be cancelled if a secondary task fails */
        public final boolean cancelSecondariesOnSecondaryFailure;
        /** whether to issue cancel against primary task if a secondary task fails */
        public final boolean cancelPrimaryOnSecondaryFailure;
        /** whether to fail this task if a secondary task fails */
        public final boolean failParentOnSecondaryFailure;
        
        @Beta
        public FailureHandlingConfig(
                boolean abortSecondaryQueueOnPrimaryFailure, boolean cancelSecondariesOnPrimaryFailure,
                boolean abortSecondaryQueueOnSecondaryFailure, boolean cancelSecondariesOnSecondaryFailure,
                boolean cancelPrimaryOnSecondaryFailure, boolean failParentOnSecondaryFailure) {
            this.abortSecondaryQueueOnPrimaryFailure = abortSecondaryQueueOnPrimaryFailure;
            this.cancelSecondariesOnPrimaryFailure = cancelSecondariesOnPrimaryFailure;
            this.abortSecondaryQueueOnSecondaryFailure = abortSecondaryQueueOnSecondaryFailure;
            this.cancelSecondariesOnSecondaryFailure = cancelSecondariesOnSecondaryFailure;
            this.cancelPrimaryOnSecondaryFailure = cancelPrimaryOnSecondaryFailure;
            this.failParentOnSecondaryFailure = failParentOnSecondaryFailure;
        }
        
        public static final FailureHandlingConfig DEFAULT = new FailureHandlingConfig(false, false, true, false, false, true);
        public static final FailureHandlingConfig SWALLOWING_CHILDREN_FAILURES = new FailureHandlingConfig(false, false, false, false, false, false);
    }
    
    public static class QueueAbortedException extends IllegalStateException {
        private static final long serialVersionUID = -7569362887826818524L;
        
        public QueueAbortedException(String msg) {
            super(msg);
        }
        public QueueAbortedException(String msg, Throwable cause) {
            super(msg, cause);
        }
    }

    /**
     * Constructs a new compound task containing the specified units of work.
     * 
     * @param jobs  A potentially heterogeneous mixture of {@link Runnable}, {@link Callable}, {@link Closure} and {@link Task} can be provided. 
     * @throws IllegalArgumentException if any of the passed child jobs is not one of the above types 
     */
    public DynamicSequentialTask() {
        this(null);
    }
    
    public DynamicSequentialTask(Callable<T> mainJob) {
        this(MutableMap.of("tag", "compound"), mainJob);
    }
    
    public DynamicSequentialTask(Map<?,?> flags, Callable<T> mainJob) {
        super(flags);
        this.job = dstJob = new DstJob(mainJob);
    }
    
    @Override
    public void queue(Task<?> t) {
        synchronized (jobTransitionLock) {
            if (primaryFinished)
                throw new IllegalStateException("Cannot add a task to "+this+" which is already finished (trying to add "+t+")");
            if (secondaryQueueAborted)
                throw new QueueAbortedException("Cannot add a task to "+this+" whose queue has been aborted (trying to add "+t+")");
            secondaryJobsAll.add(t);
            secondaryJobsRemaining.add(t);
            BrooklynTaskTags.addTagsDynamically(t, ManagementContextInternal.SUB_TASK_TAG);
            ((TaskInternal<?>)t).markQueued();
            jobTransitionLock.notifyAll();
        }
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        return cancel(mayInterruptIfRunning, mayInterruptIfRunning, true);
    }
    public boolean cancel(boolean mayInterruptTask, boolean interruptPrimaryThread, boolean alsoCancelChildren) {
        if (isDone()) return false;
        if (log.isTraceEnabled()) log.trace("cancelling {}", this);
        boolean cancel = super.cancel(mayInterruptTask);
        if (alsoCancelChildren) {
            for (Task<?> t: secondaryJobsAll)
                cancel |= t.cancel(mayInterruptTask);
        }
        synchronized (jobTransitionLock) {
            if (primaryThread!=null) {
                if (interruptPrimaryThread) {
                    if (log.isTraceEnabled()) log.trace("cancelling {} - interrupting", this);
                    primaryThread.interrupt();
                }
                cancel = true;
            }
        }
        return cancel;
    }
    
    @Override
    public synchronized boolean uncancel() {
        secondaryQueueAborted = false;
        return super.uncancel();
    }

    @Override
    public Iterable<Task<?>> getChildren() {
        return Collections.unmodifiableCollection(secondaryJobsAll);
    }
    
    /** submits the indicated task for execution in the current execution context, and returns immediately */
    protected void submitBackgroundInheritingContext(Task<?> task) {
        BasicExecutionContext ec = BasicExecutionContext.getCurrentExecutionContext();
        if (log.isTraceEnabled()) {
            log.trace("task {} - submitting background task {} ({})", new Object[] { Tasks.current(), task, ec });
        }
        if (ec==null) {
            String message = Tasks.current()!=null ?
                    // user forgot ExecContext:
                        "Task "+this+" submitting background task requires an ExecutionContext (an ExecutionManager is not enough): submitting "+task+" in "+Tasks.current()
                    : // should not happen:
                        "Cannot submit tasks inside DST when not in a task : submitting "+task+" in "+this;
            log.warn(message+" (rethrowing)");
            throw new IllegalStateException(message);
        }
        synchronized (task) {
            if (task.isSubmitted()) {
                if (log.isTraceEnabled()) {
                    log.trace("DST "+this+" skipping submission of child "+task+" because it is already submitted");
                }
            } else {
                try {
                    ec.submit(task);
                } catch (Exception e) {
                    Exceptions.propagateIfFatal(e);
                    // Give some context when the submit fails (happens when the target is already unmanaged)
                    throw new IllegalStateException("Failure submitting task "+task+" in "+this+": "+e.getMessage(), e);
                }
            }
        }
    }

    public void setFailureHandlingConfig(FailureHandlingConfig failureHandlingConfig) {
        this.failureHandlingConfig = failureHandlingConfig;
    }
    @Override
    public void swallowChildrenFailures() {
        setFailureHandlingConfig(FailureHandlingConfig.SWALLOWING_CHILDREN_FAILURES);
    }
    
    protected class DstJob implements Callable<T> {
        protected Callable<T> primaryJob;
        /** currently executing (or just completed) secondary task, or null if none;
         * with jobTransitionLock notified on change and completion */
        protected volatile Task<?> currentSecondary = null;
        protected volatile boolean finishedSecondaries = false;
        
        public DstJob(Callable<T> mainJob) {
            this.primaryJob = mainJob;
        }

        @SuppressWarnings("unchecked")
        @Override
        public T call() throws Exception {

            synchronized (jobTransitionLock) {
                primaryStarted = true;
                primaryThread = Thread.currentThread();
                for (Task<?> t: secondaryJobsAll)
                    ((TaskInternal<?>)t).markQueued();
            }
            // TODO overkill having a thread/task for this, but it works
            // optimisation would either use newTaskEndCallback property on task to submit
            // or use some kind of single threaded executor for the queued tasks
            Task<List<Object>> secondaryJobMaster = Tasks.<List<Object>>builder().dynamic(false)
                    .displayName("DST manager (internal)")
                    // TODO marking it transient helps it be GC'd sooner, 
                    // but ideally we wouldn't have this,
                    // or else it would be a child
                    .tag(BrooklynTaskTags.TRANSIENT_TASK_TAG)
                    .body(new Callable<List<Object>>() {

                @Override
                public List<Object> call() throws Exception {
                    List<Object> result = new ArrayList<Object>();
                    try { 
                        while (!secondaryQueueAborted && (!primaryFinished || !secondaryJobsRemaining.isEmpty())) {
                            synchronized (jobTransitionLock) {
                                if (!primaryFinished && secondaryJobsRemaining.isEmpty()) {
                                    currentSecondary = null;
                                    jobTransitionLock.wait(1000);
                                }
                            }
                            @SuppressWarnings("rawtypes")
                            Task secondaryJob = secondaryJobsRemaining.poll();
                            if (secondaryJob != null) {
                                synchronized (jobTransitionLock) {
                                    currentSecondary = secondaryJob;
                                    submitBackgroundInheritingContext(secondaryJob);
                                    jobTransitionLock.notifyAll();
                                }
                                try {
                                    result.add(secondaryJob.get());
                                } catch (Exception e) {
                                    if (TaskTags.isInessential(secondaryJob)) {
                                        result.add(Tasks.getError(secondaryJob));
                                        if (log.isDebugEnabled())
                                            log.debug("Secondary job queue for "+DynamicSequentialTask.this+" ignoring error in inessential task "+secondaryJob+": "+e);
                                    } else {
                                        if (failureHandlingConfig.cancelSecondariesOnSecondaryFailure) {
                                            if (log.isDebugEnabled())
                                                log.debug("Secondary job queue for "+DynamicSequentialTask.this+" cancelling "+secondaryJobsRemaining.size()+" remaining, due to error in task "+secondaryJob+": "+e);
                                            synchronized (jobTransitionLock) {
                                                for (Task<?> t: secondaryJobsRemaining)
                                                    t.cancel(true);
                                                jobTransitionLock.notifyAll();
                                            }
                                        }
                                        
                                        if (failureHandlingConfig.abortSecondaryQueueOnSecondaryFailure) {
                                            if (log.isDebugEnabled())
                                                log.debug("Aborting secondary job queue for "+DynamicSequentialTask.this+" due to error in child task "+secondaryJob+" ("+e+", being rethrown)");
                                            secondaryQueueAborted = true;
                                            throw e;
                                        }

                                        if (!primaryFinished && failureHandlingConfig.cancelPrimaryOnSecondaryFailure) {
                                            cancel(true, false, false);
                                        }
                                        
                                        result.add(Tasks.getError(secondaryJob));
                                        if (log.isDebugEnabled())
                                            log.debug("Secondary job queue for "+DynamicSequentialTask.this+" continuing in presence of error in child task "+secondaryJob+" ("+e+", being remembered)");
                                    }
                                }
                            }
                        }
                    } finally {
                        synchronized (jobTransitionLock) {
                            currentSecondary = null;
                            finishedSecondaries = true;
                            jobTransitionLock.notifyAll();
                        }
                    }
                    return result;
                }
            }).build();
            ((BasicTask<?>)secondaryJobMaster).proxyTargetTask = DynamicSequentialTask.this;
            
            submitBackgroundInheritingContext(secondaryJobMaster);
            
            T result = null;
            Throwable error = null;
            Throwable uninterestingSelfError = null;
            boolean errorIsFromChild = false;
            try {
                if (log.isTraceEnabled()) log.trace("calling primary job for {}", this);
                if (primaryJob!=null) result = primaryJob.call();
            } catch (Throwable selfException) {
                Exceptions.propagateIfFatal(selfException);
                if (Exceptions.getFirstThrowableOfType(selfException, QueueAbortedException.class) != null) {
                    // Error was caused by the task already having failed, and this thread calling queue() to try
                    // to queue more work. The underlying cause will be much more interesting.
                    // Without this special catch, we record error = "Cannot add a task to ... whose queue has been aborted",
                    // which gets propagated instead of the more interesting child exception.
                    uninterestingSelfError = selfException;
                } else {
                    error = selfException;
                    errorIsFromChild = false;
                }
                if (failureHandlingConfig.abortSecondaryQueueOnPrimaryFailure) {
                    if (log.isDebugEnabled())
                        log.debug("Secondary job queue for "+DynamicSequentialTask.this+" aborting with "+secondaryJobsRemaining.size()+" remaining, due to error in primary task: "+selfException);
                    secondaryQueueAborted = true;
                }
                if (failureHandlingConfig.cancelSecondariesOnPrimaryFailure) {
                    if (log.isDebugEnabled())
                        log.debug(DynamicSequentialTask.this+" cancelling "+secondaryJobsRemaining.size()+" remaining, due to error in primary task: "+selfException);
                    synchronized (jobTransitionLock) {
                        for (Task<?> t: secondaryJobsRemaining)
                            t.cancel(true);
                        // do this early to prevent additions; and note we notify very soon below, so not notify is help off until below
                        primaryThread = null;
                        primaryFinished = true;
                    }
                }
            } finally {
                try {
                    if (log.isTraceEnabled()) log.trace("cleaning up for {}", this);
                    synchronized (jobTransitionLock) {
                        // semaphore might be nicer here (aled notes as it is this is a little hard to read)
                        primaryThread = null;
                        primaryFinished = true;
                        jobTransitionLock.notifyAll();
                    }
                    if (!isCancelled() && !Thread.currentThread().isInterrupted()) {
                        if (log.isTraceEnabled()) log.trace("waiting for secondaries for {}", this);
                        // wait on tasks sequentially so that blocking information is more interesting
                        DynamicTasks.waitForLast();
                        List<Object> result2 = secondaryJobMaster.get();
                        try {
                            if (primaryJob==null) result = (T)result2;
                        } catch (ClassCastException e) { /* ignore class cast exception; leave the result as null */ }
                    }
                } catch (Throwable childException) {
                    Exceptions.propagateIfFatal(childException);
                    if (error==null) {
                        error = childException;
                        errorIsFromChild = true;
                    } else {
                        if (log.isDebugEnabled()) log.debug("Parent task "+this+" ignoring child error ("+childException+") in presence of our own error ("+error+")");
                    }
                }
            }
            if (error!=null) {
                handleException(error, errorIsFromChild);
            }
            if (uninterestingSelfError != null) {
                handleException(uninterestingSelfError, false);
            }
            return result;
        }
        
        @Override
        public String toString() {
            return "DstJob:"+DynamicSequentialTask.this.getId();
        }

        /** waits for this job to complete, or the given time to elapse */
        public void join(boolean includePrimary, Duration optionalTimeout) throws InterruptedException {
            CountdownTimer timeLeft = optionalTimeout!=null ? CountdownTimer.newInstanceStarted(optionalTimeout) : null;
            while (true) {
                Task<?> cs;
                Duration remaining;
                synchronized (jobTransitionLock) {
                    cs = currentSecondary;
                    if (finishedSecondaries) return;
                    remaining = timeLeft==null ? Duration.ONE_SECOND : timeLeft.getDurationRemaining();
                    if (!remaining.isPositive()) return;
                    if (cs==null) {
                        if (!includePrimary && secondaryJobsRemaining.isEmpty()) return;
                        // parent still running, no children though
                        Tasks.setBlockingTask(DynamicSequentialTask.this);
                        jobTransitionLock.wait(remaining.toMilliseconds());
                        Tasks.resetBlockingDetails();
                    }
                }
                if (cs!=null) {
                    Tasks.setBlockingTask(cs);
                    cs.blockUntilEnded(remaining);
                    Tasks.resetBlockingDetails();
                }
            }
        }
    }

    @Override
    public List<Task<?>> getQueue() {
        return ImmutableList.copyOf(secondaryJobsAll);
    }

    public void handleException(Throwable throwable, boolean fromChild) throws Exception {
        Exceptions.propagateIfFatal(throwable);
        if (fromChild && !failureHandlingConfig.failParentOnSecondaryFailure) {
            log.debug("Parent task "+this+" swallowing child error: "+throwable);
            return;
        }
        handleException(throwable);
    }
    public void handleException(Throwable throwable) throws Exception { 
        Exceptions.propagateIfFatal(throwable);
        if (throwable instanceof Exception) {
            // allow checked exceptions to be passed through
            throw (Exception)throwable;
        }
        throw Exceptions.propagate(throwable);
    }

    @Override
    public void drain(Duration optionalTimeout, boolean includePrimary, boolean throwFirstError) {
        try {
            dstJob.join(includePrimary, optionalTimeout);
        } catch (InterruptedException e) {
            throw Exceptions.propagate(e);
        }
        if (throwFirstError) {
            if (isError()) 
                getUnchecked();
            for (Task<?> t: getQueue())
                if (t.isError() && !TaskTags.isInessential(t))
                    t.getUnchecked();
        }
    }

}