/*
 * 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.entity.software.base.lifecycle;

import java.io.Serializable;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;

import javax.annotation.Nullable;

import org.apache.brooklyn.api.effector.Effector;
import org.apache.brooklyn.api.entity.Entity;
import org.apache.brooklyn.api.location.Location;
import org.apache.brooklyn.api.location.MachineLocation;
import org.apache.brooklyn.api.location.MachineManagementMixins;
import org.apache.brooklyn.api.location.MachineManagementMixins.SuspendsMachines;
import org.apache.brooklyn.api.location.MachineProvisioningLocation;
import org.apache.brooklyn.api.location.NoMachinesAvailableException;
import org.apache.brooklyn.api.mgmt.Task;
import org.apache.brooklyn.api.sensor.AttributeSensor;
import org.apache.brooklyn.api.sensor.Feed;
import org.apache.brooklyn.config.ConfigKey;
import org.apache.brooklyn.core.config.ConfigKeys;
import org.apache.brooklyn.core.config.Sanitizer;
import org.apache.brooklyn.core.effector.EffectorBody;
import org.apache.brooklyn.core.effector.Effectors;
import org.apache.brooklyn.core.entity.Attributes;
import org.apache.brooklyn.core.entity.BrooklynConfigKeys;
import org.apache.brooklyn.core.entity.Entities;
import org.apache.brooklyn.core.entity.EntityInternal;
import org.apache.brooklyn.core.entity.internal.AttributesInternal;
import org.apache.brooklyn.core.entity.internal.AttributesInternal.ProvisioningTaskState;
import org.apache.brooklyn.core.entity.lifecycle.Lifecycle;
import org.apache.brooklyn.core.entity.lifecycle.Lifecycle.Transition;
import org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic;
import org.apache.brooklyn.core.entity.trait.Startable;
import org.apache.brooklyn.core.entity.trait.StartableMethods;
import org.apache.brooklyn.core.feed.ConfigToAttributes;
import org.apache.brooklyn.core.location.AbstractLocation;
import org.apache.brooklyn.core.location.Locations;
import org.apache.brooklyn.core.location.Machines;
import org.apache.brooklyn.core.location.cloud.CloudLocationConfig;
import org.apache.brooklyn.core.mgmt.BrooklynTaskTags;
import org.apache.brooklyn.core.mgmt.entitlement.Entitlements;
import org.apache.brooklyn.core.sensor.BasicAttributeSensor;
import org.apache.brooklyn.core.sensor.ReleaseableLatch;
import org.apache.brooklyn.entity.machine.MachineInitTasks;
import org.apache.brooklyn.entity.machine.ProvidesProvisioningFlags;
import org.apache.brooklyn.entity.software.base.SoftwareProcess;
import org.apache.brooklyn.entity.software.base.SoftwareProcess.RestartSoftwareParameters;
import org.apache.brooklyn.entity.software.base.SoftwareProcess.RestartSoftwareParameters.RestartMachineMode;
import org.apache.brooklyn.entity.software.base.SoftwareProcess.StopSoftwareParameters;
import org.apache.brooklyn.entity.software.base.SoftwareProcess.StopSoftwareParameters.StopMode;
import org.apache.brooklyn.entity.stock.EffectorStartableImpl.StartParameters;
import org.apache.brooklyn.location.localhost.LocalhostMachineProvisioningLocation;
import org.apache.brooklyn.location.ssh.CanResolveOnBoxDir;
import org.apache.brooklyn.location.ssh.SshMachineLocation;
import org.apache.brooklyn.util.collections.MutableMap;
import org.apache.brooklyn.util.collections.MutableSet;
import org.apache.brooklyn.util.core.config.ConfigBag;
import org.apache.brooklyn.util.core.task.DynamicTasks;
import org.apache.brooklyn.util.core.task.Tasks;
import org.apache.brooklyn.util.core.task.ValueResolverIterator;
import org.apache.brooklyn.util.exceptions.Exceptions;
import org.apache.brooklyn.util.guava.Maybe;
import org.apache.brooklyn.util.net.UserAndHostAndPort;
import org.apache.brooklyn.util.os.Os;
import org.apache.brooklyn.util.repeat.Repeater;
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.base.Preconditions;
import com.google.common.base.Supplier;
import com.google.common.base.Suppliers;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
import com.google.common.reflect.TypeToken;

/**
 * Default skeleton for start/stop/restart tasks on machines.
 * <p>
 * Knows how to provision machines, making use of {@link ProvidesProvisioningFlags#obtainProvisioningFlags(MachineProvisioningLocation)},
 * and provides hooks for injecting behaviour at common places.
 * <p>
 * Methods are designed for overriding, with the convention that *Async methods should queue (and not block).
 * The following methods are commonly overridden (and you can safely queue tasks, block, or return immediately in them):
 * <ul>
 *  <li> {@link #startProcessesAtMachine(Supplier)} (required)
 *  <li> {@link #stopProcessesAtMachine()} (required, but can be left blank if you assume the VM will be destroyed)
 *  <li> {@link #preStartCustom(MachineLocation)}
 *  <li> {@link #postStartCustom()}
 *  <li> {@link #preStopConfirmCustom()}
 *  <li> {@link #postStopCustom()}
 * </ul>
 * Note methods at this level typically look after the {@link Attributes#SERVICE_STATE_ACTUAL} sensor.
 *
 * @since 0.6.0
 */
@Beta
public abstract class MachineLifecycleEffectorTasks {

    private static final Logger log = LoggerFactory.getLogger(MachineLifecycleEffectorTasks.class);

    public static final ConfigKey<Boolean> ON_BOX_BASE_DIR_RESOLVED = ConfigKeys.newBooleanConfigKey(
            "onbox.base.dir.resolved",
            "Whether the on-box base directory has been resolved (for internal use)");

    public static final ConfigKey<Collection<? extends Location>> LOCATIONS = StartParameters.LOCATIONS;
    
    public static final ConfigKey<Duration> STOP_PROCESS_TIMEOUT = ConfigKeys.newDurationConfigKey(
            "process.stop.timeout", "How long to wait for the processes to be stopped; use null to mean forever", 
            Duration.TWO_MINUTES);

    @Beta
    public static final ConfigKey<Duration> STOP_WAIT_PROVISIONING_TIMEOUT = ConfigKeys.newDurationConfigKey(
            "stop.wait.provisioning.timeout",
            "If stop is called on an entity while it is still provisioning the machine (such that "
                    + "the provisioning cannot be safely interrupted), this is the length of time "
                    + "to wait for the machine instance to become available so that it can be terminated. "
                    + "If stop aborts before this point, the machine may be left running.", 
            Duration.minutes(10));

    @Beta
    public static final AttributeSensor<MachineLocation> INTERNAL_PROVISIONED_MACHINE = new BasicAttributeSensor<MachineLocation>(
            TypeToken.of(MachineLocation.class), 
            "internal.provisioning.task.machine",
            "Internal transient sensor (do not use) for tracking the machine being provisioned (to better handle aborting)", 
            AttributeSensor.SensorPersistenceMode.NONE);
    
    protected final MachineInitTasks machineInitTasks = new MachineInitTasks();
    
    /** Attaches lifecycle effectors (start, restart, stop) to the given entity post-creation. */
    public void attachLifecycleEffectors(Entity entity) {
        ((EntityInternal) entity).getMutableEntityType().addEffector(newStartEffector());
        ((EntityInternal) entity).getMutableEntityType().addEffector(newRestartEffector());
        ((EntityInternal) entity).getMutableEntityType().addEffector(newStopEffector());
    }

    /**
     * Return an effector suitable for setting in a {@code public static final} or attaching dynamically.
     * <p>
     * The effector overrides the corresponding effector from {@link Startable} with
     * the behaviour in this lifecycle class instance.
     */
    public Effector<Void> newStartEffector() {
        return Effectors.effector(Startable.START).impl(newStartEffectorTask()).build();
    }

    /** @see {@link #newStartEffector()} */
    public Effector<Void> newRestartEffector() {
        return Effectors.effector(Startable.RESTART)
                .parameter(RestartSoftwareParameters.RESTART_CHILDREN)
                .parameter(RestartSoftwareParameters.RESTART_MACHINE)
                .impl(newRestartEffectorTask())
                .build();
    }
    
    /** @see {@link #newStartEffector()} */
    public Effector<Void> newStopEffector() {
        return Effectors.effector(Startable.STOP)
                .parameter(StopSoftwareParameters.STOP_PROCESS_MODE)
                .parameter(StopSoftwareParameters.STOP_MACHINE_MODE)
                .impl(newStopEffectorTask())
                .build();
    }

    /** @see {@link #newStartEffector()} */
    public Effector<Void> newSuspendEffector() {
        return Effectors.effector(Void.class, "suspend")
                .description("Suspend the process/service represented by an entity")
                .parameter(StopSoftwareParameters.STOP_PROCESS_MODE)
                .parameter(StopSoftwareParameters.STOP_MACHINE_MODE)
                .impl(newSuspendEffectorTask())
                .build();
    }

    /**
     * Returns the {@link EffectorBody} which supplies the implementation for the start effector.
     * <p>
     * Calls {@link #start(Collection)} in this class.
     */
    public EffectorBody<Void> newStartEffectorTask() {
        // TODO included anonymous inner class for backwards compatibility with persisted state.
        new EffectorBody<Void>() {
            @Override
            public Void call(ConfigBag parameters) {
                Collection<? extends Location> locations  = null;

                Object locationsRaw = parameters.getStringKey(LOCATIONS.getName());
                locations = Locations.coerceToCollectionOfLocationsManaged(entity().getManagementContext(), locationsRaw);

                if (locations==null) {
                    // null/empty will mean to inherit from parent
                    locations = Collections.emptyList();
                }

                start(locations);
                return null;
            }
        };
        return new StartEffectorBody();
    }

    private class StartEffectorBody extends EffectorBody<Void> {
        @Override
        public Void call(ConfigBag parameters) {
            Collection<? extends Location> locations = null;

            Object locationsRaw = parameters.getStringKey(LOCATIONS.getName());
            locations = Locations.coerceToCollectionOfLocationsManaged(entity().getManagementContext(), locationsRaw);

            if (locations == null) {
                // null/empty will mean to inherit from parent
                locations = Collections.emptyList();
            }

            start(locations);
            return null;
        }

    }

    /**
     * Calls {@link #restart(ConfigBag)}.
     *
     * @see {@link #newStartEffectorTask()}
     */
    public EffectorBody<Void> newRestartEffectorTask() {
        // TODO included anonymous inner class for backwards compatibility with persisted state.
        new EffectorBody<Void>() {
            @Override
            public Void call(ConfigBag parameters) {
                restart(parameters);
                return null;
            }
        };
        return new RestartEffectorBody();
    }

    private class RestartEffectorBody extends EffectorBody<Void> {
        @Override
        public Void call(ConfigBag parameters) {
            restart(parameters);
            return null;
        }
    }

    /**
     * Calls {@link #stop(ConfigBag)}.
     *
     * @see {@link #newStartEffectorTask()}
     */
    public EffectorBody<Void> newStopEffectorTask() {
        // TODO included anonymous inner class for backwards compatibility with persisted state.
        new EffectorBody<Void>() {
            @Override
            public Void call(ConfigBag parameters) {
                stop(parameters);
                return null;
            }
        };
        return new StopEffectorBody();
    }

    private class StopEffectorBody extends EffectorBody<Void> {
        @Override
        public Void call(ConfigBag parameters) {
            stop(parameters);
            return null;
        }
    }

    /**
     * Calls {@link #suspend(ConfigBag)}.
     *
     * @see {@link #newStartEffectorTask()}
     */
    public EffectorBody<Void> newSuspendEffectorTask() {
        return new SuspendEffectorBody();
    }

    private class SuspendEffectorBody extends EffectorBody<Void> {
        @Override
        public Void call(ConfigBag parameters) {
            suspend(parameters);
            return null;
        }
    }

    protected EntityInternal entity() {
        return (EntityInternal) BrooklynTaskTags.getTargetOrContextEntity(Tasks.current());
    }

    protected Location getLocation(@Nullable Collection<? extends Location> locations) {
        if (locations==null || locations.isEmpty()) locations = entity().getLocations();
        if (locations.isEmpty()) {
            MachineProvisioningLocation<?> provisioner = entity().getAttribute(SoftwareProcess.PROVISIONING_LOCATION);
            if (provisioner!=null) locations = Arrays.<Location>asList(provisioner);
        }
        locations = Locations.getLocationsCheckingAncestors(locations, entity());

        Maybe<MachineLocation> ml = Locations.findUniqueMachineLocation(locations);
        if (ml.isPresent()) return ml.get();

        if (locations.isEmpty())
            throw new IllegalArgumentException("No locations specified when starting "+entity());
        if (locations.size() != 1 || Iterables.getOnlyElement(locations)==null)
            throw new IllegalArgumentException("Ambiguous locations detected when starting "+entity()+": "+locations);
        return Iterables.getOnlyElement(locations);
    }
    
    /** runs the tasks needed to start, wrapped by setting {@link Attributes#SERVICE_STATE_EXPECTED} appropriately */ 
    public void start(Collection<? extends Location> locations) {
        ServiceStateLogic.setExpectedState(entity(), Lifecycle.STARTING);
        try {
            startInLocations(locations);
            DynamicTasks.waitForLast();
            ServiceStateLogic.setExpectedState(entity(), Lifecycle.RUNNING);
        } catch (Throwable t) {
            ServiceStateLogic.setExpectedState(entity(), Lifecycle.ON_FIRE);
            throw Exceptions.propagate(t);
        }
    }

    /** Asserts there is a single location and calls {@link #startInLocation(Location)} with that location. */
    protected void startInLocations(Collection<? extends Location> locations) {
        startInLocation(getLocation(locations));
    }

    /** Dispatches to the appropriate method(s) to start in the given location. */
    protected void startInLocation(final Location location) {
        Supplier<MachineLocation> locationS = null;
        if (location instanceof MachineProvisioningLocation) {
            Task<MachineLocation> machineTask = provisionAsync((MachineProvisioningLocation<?>)location);
            locationS = Tasks.supplier(machineTask);
        } else if (location instanceof MachineLocation) {
            locationS = Suppliers.ofInstance((MachineLocation)location);
        }
        Preconditions.checkState(locationS != null, "Unsupported location "+location+", when starting "+entity());

        final Supplier<MachineLocation> locationSF = locationS;

        // Opportunity to block startup until other dependent components are available
        try (CloseableLatch latch = waitForCloseableLatch(entity(), SoftwareProcess.START_LATCH)) {
            preStartAtMachineAsync(locationSF);
            DynamicTasks.queue("start (processes)", new StartProcessesAtMachineTask(locationSF));
            postStartAtMachineAsync();
        }
    }

    private class StartProcessesAtMachineTask implements Runnable {
        private final Supplier<MachineLocation> machineSupplier;
        private StartProcessesAtMachineTask(Supplier<MachineLocation> machineSupplier) {
            this.machineSupplier = machineSupplier;
        }
        @Override
        public void run() {
            startProcessesAtMachine(machineSupplier);
        }
    }

    /**
     * Returns a queued {@link Task} which provisions a machine in the given location
     * and returns that machine. The task can be used as a supplier to subsequent methods.
     */
    protected Task<MachineLocation> provisionAsync(final MachineProvisioningLocation<?> location) {
        return DynamicTasks.queue(Tasks.<MachineLocation>builder().displayName("provisioning (" + location.getDisplayName() + ")").body(
                new ProvisionMachineTask(location)).build());
    }

    private class ProvisionMachineTask implements Callable<MachineLocation> {
        final MachineProvisioningLocation<?> location;

        private ProvisionMachineTask(MachineProvisioningLocation<?> location) {
            this.location = location;
        }

        @Override
        public MachineLocation call() throws Exception {
            // Blocks if a latch was configured.
            entity().getConfig(BrooklynConfigKeys.PROVISION_LATCH);
            final Map<String, Object> flags = obtainProvisioningFlags(location);
            if (!(location instanceof LocalhostMachineProvisioningLocation))
                log.info("Starting {}, obtaining a new location instance in {} with ports {}", new Object[]{entity(), location, flags.get("inboundPorts")});
            entity().sensors().set(SoftwareProcess.PROVISIONING_LOCATION, location);
            Transition expectedState = entity().sensors().get(Attributes.SERVICE_STATE_EXPECTED);
            
            // BROOKLYN-263: see corresponding code in doStop()
            if (expectedState != null && (expectedState.getState() == Lifecycle.STOPPING || expectedState.getState() == Lifecycle.STOPPED)) {
                throw new IllegalStateException("Provisioning aborted before even begun for "+entity()+" in "+location+" (presumably by a concurrent call to stop");
            }
            entity().sensors().set(AttributesInternal.INTERNAL_PROVISIONING_TASK_STATE, ProvisioningTaskState.RUNNING);
            
            MachineLocation machine;
            try {
                machine = Tasks.withBlockingDetails("Provisioning machine in " + location, new ObtainLocationTask(location, flags));
                entity().sensors().set(INTERNAL_PROVISIONED_MACHINE, machine);
            } finally {
                entity().sensors().remove(AttributesInternal.INTERNAL_PROVISIONING_TASK_STATE);
            }
            
            if (machine == null) {
                throw new NoMachinesAvailableException("Failed to obtain machine in " + location.toString());
            }
            if (log.isDebugEnabled()) {
                log.debug("While starting {}, obtained new location instance {}", entity(),
                        (machine instanceof SshMachineLocation
                                ? machine + ", details " + ((SshMachineLocation) machine).getUser() + ":" + Sanitizer.sanitize(((SshMachineLocation) machine).config().getLocalBag())
                                : machine));
            }
            return machine;
        }
    }

    private static class ObtainLocationTask implements Callable<MachineLocation> {
        final MachineProvisioningLocation<?> location;
        final Map<String, Object> flags;

        private ObtainLocationTask(MachineProvisioningLocation<?> location, Map<String, Object> flags) {
            this.flags = flags;
            this.location = location;
        }

        @Override
        public MachineLocation call() throws NoMachinesAvailableException {
            return location.obtain(flags);
        }
    }

    /**
     * Wraps a call to {@link #preStartCustom(MachineLocation)}, after setting the hostname and address.
     */
    protected void preStartAtMachineAsync(final Supplier<MachineLocation> machineS) {
        DynamicTasks.queue("pre-start", new PreStartTask(machineS.get()));
    }

    private class PreStartTask implements Runnable {
        final MachineLocation machine;

        private PreStartTask(MachineLocation machine) {
            this.machine = machine;
        }
        @Override
        public void run() {
            log.info("Starting {} on machine {}", entity(), machine);
            Collection<Location> oldLocs = entity().getLocations();
            if (!oldLocs.isEmpty()) {
                List<MachineLocation> oldSshLocs = ImmutableList.copyOf(Iterables.filter(oldLocs, MachineLocation.class));
                if (!oldSshLocs.isEmpty()) {
                    // check if existing locations are compatible
                    log.debug("Entity " + entity() + " had machine locations " + oldSshLocs + " when starting at " + machine + "; checking if they are compatible");
                    for (MachineLocation oldLoc : oldSshLocs) {
                        // machines are deemed compatible if hostname and address are the same, or they are localhost
                        // this allows a machine create by jclouds to then be defined with an ip-based spec
                        if (!"localhost".equals(machine.getConfig(AbstractLocation.ORIGINAL_SPEC))) {
                            checkLocationParametersCompatible(machine, oldLoc, "hostname",
                                    oldLoc.getAddress().getHostName(), machine.getAddress().getHostName());
                            checkLocationParametersCompatible(machine, oldLoc, "address",
                                    oldLoc.getAddress().getHostAddress(), machine.getAddress().getHostAddress());
                        }
                    }
                    log.debug("Entity " + entity() + " old machine locations " + oldSshLocs + " were compatible, removing them to start at " + machine);
                    entity().removeLocations(oldSshLocs);
                }
            }
            entity().addLocations(ImmutableList.of((Location) machine));

            // elsewhere we rely on (public) hostname being set _after_ subnet_hostname
            // (to prevent the tiny possibility of races resulting in hostname being returned
            // simply because subnet is still being looked up)
            Maybe<String> lh = Machines.getSubnetHostname(machine);
            Maybe<String> la = Machines.getSubnetIp(machine);
            if (lh.isPresent() && entity().sensors().get(Attributes.SUBNET_HOSTNAME) == null) {
                entity().sensors().set(Attributes.SUBNET_HOSTNAME, lh.get());
            }
            if (la.isPresent() && entity().sensors().get(Attributes.SUBNET_ADDRESS) == null) {
                entity().sensors().set(Attributes.SUBNET_ADDRESS, la.get());
            }
            if (entity().sensors().get(Attributes.HOSTNAME) == null) {
                entity().sensors().set(Attributes.HOSTNAME, machine.getAddress().getHostName());
            }
            if (entity().sensors().get(Attributes.ADDRESS) == null) {
                entity().sensors().set(Attributes.ADDRESS, machine.getAddress().getHostAddress());
            }
            if (machine instanceof SshMachineLocation) {
                SshMachineLocation sshMachine = (SshMachineLocation) machine;
                UserAndHostAndPort sshAddress = UserAndHostAndPort.fromParts(
                        sshMachine.getUser(), sshMachine.getAddress().getHostName(), sshMachine.getPort());
                // FIXME: Who or what is SSH_ADDRESS intended for? It's not necessarily the address that
                // the SshMachineLocation is using for ssh connections (because it accepts SSH_HOST as an override).
                entity().sensors().set(Attributes.SSH_ADDRESS, sshAddress);
            }

            if (Boolean.TRUE.equals(entity().getConfig(SoftwareProcess.OPEN_IPTABLES))) {
                if (machine instanceof SshMachineLocation) {
                    @SuppressWarnings("unchecked")
                    Iterable<Integer> inboundPorts = (Iterable<Integer>) machine.config().get(CloudLocationConfig.INBOUND_PORTS);
                    machineInitTasks.openIptablesAsync(inboundPorts, (SshMachineLocation)machine);
                } else {
                    log.warn("Ignoring flag OPEN_IPTABLES on non-ssh location {}", machine);
                }
            }
            if (Boolean.TRUE.equals(entity().getConfig(SoftwareProcess.STOP_IPTABLES))) {
                if (machine instanceof SshMachineLocation) {
                    machineInitTasks.stopIptablesAsync((SshMachineLocation)machine);
                } else {
                    log.warn("Ignoring flag STOP_IPTABLES on non-ssh location {}", machine);
                }
            }
            if (Boolean.TRUE.equals(entity().getConfig(SoftwareProcess.DONT_REQUIRE_TTY_FOR_SUDO))) {
                if (machine instanceof SshMachineLocation) {
                    machineInitTasks.dontRequireTtyForSudoAsync((SshMachineLocation)machine);
                } else {
                    log.warn("Ignoring flag DONT_REQUIRE_TTY_FOR_SUDO on non-ssh location {}", machine);
                }
            }
            resolveOnBoxDir(entity(), machine);
            preStartCustom(machine);

        }
    }

    /**
     * Resolves the on-box dir.
     * <p>
     * Initialize and pre-create the right onbox working dir, if an ssh machine location.
     * Logs a warning if not.
     */
    @SuppressWarnings("deprecation")
    public static String resolveOnBoxDir(EntityInternal entity, MachineLocation machine) {
        String base = entity.getConfig(BrooklynConfigKeys.ONBOX_BASE_DIR);
        if (base==null) base = machine.getConfig(BrooklynConfigKeys.ONBOX_BASE_DIR);
        if (base!=null && Boolean.TRUE.equals(entity.getConfig(ON_BOX_BASE_DIR_RESOLVED))) return base;
        if (base==null) base = entity.getManagementContext().getConfig().getConfig(BrooklynConfigKeys.ONBOX_BASE_DIR);
        if (base==null) base = entity.getConfig(BrooklynConfigKeys.BROOKLYN_DATA_DIR);
        if (base==null) base = machine.getConfig(BrooklynConfigKeys.BROOKLYN_DATA_DIR);
        if (base==null) base = entity.getManagementContext().getConfig().getConfig(BrooklynConfigKeys.BROOKLYN_DATA_DIR);
        if (base==null) base = "~/brooklyn-managed-processes";
        
        if (base.equals("~")) base=".";
        if (base.startsWith("~/")) base = "."+base.substring(1);

        String resolvedBase = null;
        if (entity.getConfig(BrooklynConfigKeys.SKIP_ON_BOX_BASE_DIR_RESOLUTION) || machine.getConfig(BrooklynConfigKeys.SKIP_ON_BOX_BASE_DIR_RESOLUTION)) {
            if (log.isDebugEnabled()) log.debug("Skipping on-box base dir resolution for "+entity+" at "+machine);
            if (!Os.isAbsolutish(base)) base = "~/"+base;
            resolvedBase = Os.tidyPath(base);
        } else if (machine instanceof CanResolveOnBoxDir) {
            resolvedBase = ((CanResolveOnBoxDir)machine).resolveOnBoxDirFor(entity, base);
        }
        if (resolvedBase==null) {
            if (!Os.isAbsolutish(base)) base = "~/"+base;
            resolvedBase = Os.tidyPath(base);
            log.warn("Could not resolve on-box directory for "+entity+" at "+machine+"; using "+resolvedBase+", though this may not be accurate at the target (and may fail shortly)");
        }
        entity.config().set(BrooklynConfigKeys.ONBOX_BASE_DIR, resolvedBase);
        entity.config().set(ON_BOX_BASE_DIR_RESOLVED, true);
        return resolvedBase;
    }

    protected void checkLocationParametersCompatible(MachineLocation oldLoc, MachineLocation newLoc, String paramSummary,
            Object oldParam, Object newParam) {
        if (oldParam==null || newParam==null || !oldParam.equals(newParam))
            throw new IllegalStateException("Cannot start "+entity()+" in "+newLoc+" as it has already been started with incompatible location "+oldLoc+" " +
                    "("+paramSummary+" not compatible: "+oldParam+" / "+newParam+"); "+newLoc+" may require manual removal.");
    }

    @Deprecated
    protected void preStartCustom(MachineLocation machine) {
        ConfigToAttributes.apply(entity());
    }

    protected Map<String, Object> obtainProvisioningFlags(final MachineProvisioningLocation<?> location) {
        if (entity() instanceof ProvidesProvisioningFlags) {
            return ((ProvidesProvisioningFlags)entity()).obtainProvisioningFlags(location).getAllConfig();
        }
        return MutableMap.<String, Object>of();
    }

    protected abstract String startProcessesAtMachine(final Supplier<MachineLocation> machineS);

    protected void postStartAtMachineAsync() {
        DynamicTasks.queue("post-start", new PostStartTask());
    }

    private class PostStartTask implements Runnable {
        @Override
        public void run() {
            postStartCustom();
        }
    }

    /**
     * Default post-start hooks.
     * <p>
     * Can be extended by subclasses, and typically will wait for confirmation of start.
     * The service not set to running until after this. Also invoked following a restart.
     */
    protected void postStartCustom() {
        // nothing by default
    }

    /**
     * whether when 'auto' mode is specified, the machine should be stopped when the restart effector is called
     * <p>
     * with {@link MachineLifecycleEffectorTasks}, a machine will always get created on restart if there wasn't one already
     * (unlike certain subclasses which might attempt a shortcut process-level restart)
     * so there is no reason for default behaviour of restart to throw away a provisioned machine,
     * hence default impl returns <code>false</code>.
     * <p>
     * if it is possible to tell that a machine is unhealthy, or if {@link #restart(ConfigBag)} is overridden,
     * then it might be appropriate to return <code>true</code> here.
     */
    protected boolean getDefaultRestartStopsMachine() {
        return false;
    }

    /**
     * Default restart implementation for an entity.
     * <p>
     * Stops processes if possible, then starts the entity again.
     */
    public void restart(ConfigBag parameters) {
        ServiceStateLogic.setExpectedState(entity(), Lifecycle.STOPPING);

        RestartMachineMode isRestartMachine = parameters.get(RestartSoftwareParameters.RESTART_MACHINE_TYPED);
        if (isRestartMachine==null)
            isRestartMachine=RestartMachineMode.AUTO;
        if (isRestartMachine==RestartMachineMode.AUTO)
            isRestartMachine = getDefaultRestartStopsMachine() ? RestartMachineMode.TRUE : RestartMachineMode.FALSE;

        // Calling preStopCustom without a corresponding postStopCustom invocation
        // doesn't look right so use a separate callback pair; Also depending on the arguments
        // stop() could be called which will call the {pre,post}StopCustom on its own.
        DynamicTasks.queue("pre-restart", new PreRestartTask());

        if (isRestartMachine==RestartMachineMode.FALSE) {
            DynamicTasks.queue("stopping (process)", new StopProcessesAtMachineTask());
        } else {
            Map<String, Object> stopMachineFlags =  MutableMap.of();
            if (Entitlements.getEntitlementContext() != null) {
                stopMachineFlags.put("tags", MutableSet.of(BrooklynTaskTags.tagForEntitlement(Entitlements.getEntitlementContext())));
            }
            Task<String> stopTask = Tasks.<String>builder()
                    .displayName("stopping (machine)")
                    .body(new StopMachineTask())
                    .flags(stopMachineFlags)
                    .build();
            DynamicTasks.queue(stopTask);
        }

        DynamicTasks.queue("starting", new StartInLocationsTask());
        restartChildren(parameters);
        DynamicTasks.queue("post-restart", new PostRestartTask());

        DynamicTasks.waitForLast();
        ServiceStateLogic.setExpectedState(entity(), Lifecycle.RUNNING);
    }

    private class PreRestartTask implements Runnable {
        @Override
        public void run() {
            preRestartCustom();
        }
    }
    private class PostRestartTask implements Runnable {
        @Override
        public void run() {
            postRestartCustom();
        }
    }
    private class StartInLocationsTask implements Runnable {
        @Override
        public void run() {
            // startInLocations will look up the location, and provision a machine if necessary
            // (if it remembered the provisioning location)
            ServiceStateLogic.setExpectedState(entity(), Lifecycle.STARTING);
            startInLocations(null);
        }
    }

    protected void restartChildren(ConfigBag parameters) {
        // TODO should we consult ChildStartableMode?

        Boolean isRestartChildren = parameters.get(RestartSoftwareParameters.RESTART_CHILDREN);
        if (isRestartChildren==null || !isRestartChildren) {
            return;
        }

        if (isRestartChildren) {
            DynamicTasks.queue(StartableMethods.restartingChildren(entity(), parameters));
            return;
        }

        throw new IllegalArgumentException("Invalid value '"+isRestartChildren+"' for "+RestartSoftwareParameters.RESTART_CHILDREN.getName());
    }

    /**
     * Default stop implementation for an entity.
     * <p>
     * Aborts if already stopped, otherwise sets state {@link Lifecycle#STOPPING} then
     * invokes {@link #preStopCustom()}, {@link #stopProcessesAtMachine()}, then finally
     * {@link #stopAnyProvisionedMachines()} and sets state {@link Lifecycle#STOPPED}.
     * If no errors were encountered call {@link #postStopCustom()} at the end.
     */
    public void stop(ConfigBag parameters) {
        doStopLatching(parameters, new StopAnyProvisionedMachinesTask());
    }

    /**
     * As {@link #stop} but calling {@link #suspendAnyProvisionedMachines} rather than
     * {@link #stopAnyProvisionedMachines}.
     */
    public void suspend(ConfigBag parameters) {
        doStopLatching(parameters, new SuspendAnyProvisionedMachinesTask());
    }

    protected void doStopLatching(ConfigBag parameters, Callable<StopMachineDetails<Integer>> stopTask) {
        try (CloseableLatch latch = waitForCloseableLatch(entity(), SoftwareProcess.STOP_LATCH)) {
            doStop(parameters, stopTask);
        }
    }

    protected void doStop(ConfigBag parameters, Callable<StopMachineDetails<Integer>> stopTask) {
        preStopConfirmCustom();

        log.info("Stopping {} in {}", entity(), entity().getLocations());

        StopMode stopMachineMode = getStopMachineMode(parameters);
        StopMode stopProcessMode = parameters.get(StopSoftwareParameters.STOP_PROCESS_MODE);

        DynamicTasks.queue("pre-stop", new PreStopCustomTask());

        // BROOKLYN-263:
        // With this change the stop effector will wait for Location to provision so it can terminate
        // the machine, if a provisioning request is in-progress.
        //
        // The ProvisionMachineTask stores transient internal state in PROVISIONING_TASK_STATE and
        // PROVISIONED_MACHINE: it records when the provisioning is running and when done; and it
        // records the final machine. We record the machine in the internal sensor (rather than 
        // just relying on getLocations) because the latter is set much later in the start() 
        // process.
        //
        // This code is a big improvement (previously there was a several-minute window in some 
        // clouds where a call to stop() would leave the machine running). 
        //
        // However, there are still races. If the start() code has not yet reached the call to 
        // location.obtain() then we won't wait, and the start() call won't know to abort. It's 
        // fiddly to get that right, because we need to cope with restart() - so we mustn't leave 
        // any state behind that will interfere with subsequent sequential calls to start().
        // There is some attempt to handle it by ProvisionMachineTask checking if the expectedState
        // is stopping/stopped.
        Maybe<MachineLocation> machine = Machines.findUniqueMachineLocation(entity().getLocations());
        ProvisioningTaskState provisioningState = entity().sensors().get(AttributesInternal.INTERNAL_PROVISIONING_TASK_STATE);

        if (machine.isAbsent() && provisioningState == ProvisioningTaskState.RUNNING) {
            Duration maxWait = entity().config().get(STOP_WAIT_PROVISIONING_TIMEOUT);
            log.info("When stopping {}, waiting for up to {} for the machine to finish provisioning, before terminating it", entity(), maxWait);
            boolean success = Repeater.create("Wait for a machine to appear")
                    .until(new Callable<Boolean>() {
                        @Override
                        public Boolean call() throws Exception {
                            ProvisioningTaskState state = entity().sensors().get(AttributesInternal.INTERNAL_PROVISIONING_TASK_STATE);
                            return (state != ProvisioningTaskState.RUNNING);
                        }})
                    .backoffTo(Duration.FIVE_SECONDS)
                    .limitTimeTo(maxWait)
                    .run();
            if (!success) {
                log.warn("When stopping {}, timed out after {} waiting for the machine to finish provisioning - machine may we left running", entity(), maxWait);
            }
            machine = Maybe.ofDisallowingNull(entity().sensors().get(INTERNAL_PROVISIONED_MACHINE));
        }
        entity().sensors().remove(AttributesInternal.INTERNAL_PROVISIONING_TASK_STATE);
        entity().sensors().remove(INTERNAL_PROVISIONED_MACHINE);

        Task<List<?>> stoppingProcess = null;
        if (canStop(stopProcessMode, entity())) {
            stoppingProcess = Tasks.parallel("stopping",
                    Tasks.create("stopping (process)", new StopProcessesAtMachineTask()),
                    Tasks.create("stopping (feeds)", new StopFeedsAtMachineTask()));
            DynamicTasks.queue(stoppingProcess);
        }

        Task<StopMachineDetails<Integer>> stoppingMachine = null;
        if (canStop(stopMachineMode, machine.isAbsent())) {
            // Release this machine (even if error trying to stop process - we rethrow that after)
            Map<String, Object> stopMachineFlags =  MutableMap.of();
            if (Entitlements.getEntitlementContext() != null) {
                stopMachineFlags.put("tags", MutableSet.of(BrooklynTaskTags.tagForEntitlement(Entitlements.getEntitlementContext())));
            }
            Task<StopMachineDetails<Integer>> stopMachineTask = Tasks.<StopMachineDetails<Integer>>builder()
                    .displayName("stopping (machine)")
                    .body(stopTask).flags(stopMachineFlags)
                    .build();
            stoppingMachine = DynamicTasks.queue(stopMachineTask);

            DynamicTasks.drain(entity().getConfig(STOP_PROCESS_TIMEOUT), false);

            // shutdown the machine if stopping process fails or takes too long
            synchronized (stoppingMachine) {
                // task also used as mutex by DST when it submits it; ensure it only submits once!
                if (!stoppingMachine.isSubmitted()) {
                    // force the stoppingMachine task to run by submitting it here
                    StringBuilder msg = new StringBuilder("Submitting machine stop early in background for ").append(entity());
                    if (stoppingProcess == null) {
                        msg.append(". Process stop skipped, pre-stop not finished?");
                    } else {
                        msg.append(" because process stop has ").append(
                                (stoppingProcess.isDone() ? "finished abnormally" : "not finished"));
                    }
                    log.warn(msg.toString());
                    Entities.submit(entity(), stoppingMachine);
                }
            }
        }

        try {
            // This maintains previous behaviour of silently squashing any errors on the stoppingProcess task if the
            // stoppingMachine exits with a nonzero value
            boolean checkStopProcesses = (stoppingProcess != null && (stoppingMachine == null || stoppingMachine.get().value == 0));

            if (checkStopProcesses) {
                // TODO we should test for destruction above, not merely successful "stop", as things like localhost and ssh won't be destroyed
                DynamicTasks.waitForLast();
                if (machine.isPresent()) {
                    // throw early errors *only if* there is a machine and we have not destroyed it
                    stoppingProcess.get();
                }
            }
        } catch (Throwable e) {
            ServiceStateLogic.setExpectedState(entity(), Lifecycle.ON_FIRE);
            Exceptions.propagate(e);
        }
        entity().sensors().set(SoftwareProcess.SERVICE_UP, false);
        ServiceStateLogic.setExpectedState(entity(), Lifecycle.STOPPED);

        DynamicTasks.queue("post-stop", new PostStopCustomTask());

        if (log.isDebugEnabled()) log.debug("Stopped software process entity "+entity());
    }

    private class StopAnyProvisionedMachinesTask implements Callable<StopMachineDetails<Integer>> {
        @Override
        public StopMachineDetails<Integer> call() {
            return stopAnyProvisionedMachines();
        }
    }

    private class SuspendAnyProvisionedMachinesTask implements Callable<StopMachineDetails<Integer>> {
        @Override
        public StopMachineDetails<Integer> call() {
            return suspendAnyProvisionedMachines();
        }
    }

    private class StopProcessesAtMachineTask implements Callable<String> {
        @Override
        public String call() {
            DynamicTasks.markInessential();
            stopProcessesAtMachine();
            DynamicTasks.waitForLast();
            return "Stop processes completed with no errors.";
        }
    }

    private class StopFeedsAtMachineTask implements Callable<String> {
        @Override
        public String call() {
            DynamicTasks.markInessential();
            for (Feed feed : entity().feeds().getFeeds()) {
                if (feed.isActivated()) feed.stop();
            }
            DynamicTasks.waitForLast();
            return "Stop feeds completed with no errors.";
        }
    }

    private class StopMachineTask implements Callable<String> {
        @Override
        public String call() {
            DynamicTasks.markInessential();
            stop(ConfigBag.newInstance().configure(StopSoftwareParameters.STOP_MACHINE_MODE, StopMode.IF_NOT_STOPPED));
            DynamicTasks.waitForLast();
            return "Stop of machine completed with no errors.";
        }
    }

    private class PreStopCustomTask implements Callable<String> {
        @Override
        public String call() {
            if (entity().getAttribute(SoftwareProcess.SERVICE_STATE_ACTUAL) == Lifecycle.STOPPED) {
                log.debug("Skipping stop of entity " + entity() + " when already stopped");
                return "Already stopped";
            }
            ServiceStateLogic.setExpectedState(entity(), Lifecycle.STOPPING);
            entity().sensors().set(SoftwareProcess.SERVICE_UP, false);
            preStopCustom();
            return null;
        }
    }

    private class PostStopCustomTask implements Callable<Void> {
        @Override
        public Void call() {
            postStopCustom();
            return null;
        }
    }

    public static StopMode getStopMachineMode(ConfigBag parameters) {
        final StopMode stopMachineMode = parameters.get(StopSoftwareParameters.STOP_MACHINE_MODE);
        return stopMachineMode;
    }

    public static boolean canStop(StopMode stopMode, Entity entity) {
        boolean isEntityStopped = entity.getAttribute(SoftwareProcess.SERVICE_STATE_ACTUAL)==Lifecycle.STOPPED;
        return canStop(stopMode, isEntityStopped);
    }

    protected static boolean canStop(StopMode stopMode, boolean isStopped) {
        return stopMode == StopMode.ALWAYS ||
                stopMode == StopMode.IF_NOT_STOPPED && !isStopped;
    }

    @Deprecated
    protected void preStopConfirmCustom() {
    }

    protected void preStopCustom() {
        // nothing needed here
    }

    protected void postStopCustom() {
        // nothing needed here
    }

    protected void preRestartCustom() {
        // nothing needed here
    }

    protected void postRestartCustom() {
        // nothing needed here
    }

    public static class StopMachineDetails<T> implements Serializable {
        private static final long serialVersionUID = 3256747214315895431L;
        final String message;
        final T value;
        protected StopMachineDetails(String message, T value) {
            this.message = message;
            this.value = value;
        }
        @Override
        public String toString() {
            return message;
        }
    }

    /**
     * Return string message of result.
     * <p>
     * Can run synchronously or not, caller will submit/queue as needed, and will block on any submitted tasks.
     */
    protected abstract String stopProcessesAtMachine();

    /**
     * Stop and release the {@link MachineLocation} the entity is provisioned at.
     * <p>
     * Can run synchronously or not, caller will submit/queue as needed, and will block on any submitted tasks.
     */
    protected StopMachineDetails<Integer> stopAnyProvisionedMachines() {
        @SuppressWarnings("unchecked")
        MachineProvisioningLocation<MachineLocation> provisioner = entity().getAttribute(SoftwareProcess.PROVISIONING_LOCATION);

        if (Iterables.isEmpty(entity().getLocations())) {
            log.debug("No machine decommissioning necessary for "+entity()+" - no locations");
            return new StopMachineDetails<Integer>("No machine decommissioning necessary - no locations", 0);
        }

        // Only release this machine if we ourselves provisioned it (e.g. it might be running other services)
        if (provisioner==null) {
            log.debug("No machine decommissioning necessary for "+entity()+" - did not provision");
            return new StopMachineDetails<Integer>("No machine decommissioning necessary - did not provision", 0);
        }

        Location machine = getLocation(null);
        if (!(machine instanceof MachineLocation)) {
            log.debug("No decommissioning necessary for "+entity()+" - not a machine location ("+machine+")");
            return new StopMachineDetails<Integer>("No machine decommissioning necessary - not a machine ("+machine+")", 0);
        }

        entity().sensors().set(AttributesInternal.INTERNAL_TERMINATION_TASK_STATE, ProvisioningTaskState.RUNNING);
        try {
            clearEntityLocationAttributes(machine);
            provisioner.release((MachineLocation)machine);
        } finally {
            // TODO On exception, should we add the machine back to the entity (because it might not really be terminated)?
            //      Do we need a better exception hierarchy for that?
            entity().sensors().remove(AttributesInternal.INTERNAL_TERMINATION_TASK_STATE);
        }
        
        return new StopMachineDetails<Integer>("Decommissioned "+machine, 1);
    }

    /**
     * Suspend the {@link MachineLocation} the entity is provisioned at.
     * <p>
     * Expects the entity's {@link SoftwareProcess#PROVISIONING_LOCATION provisioner} to be capable of
     * {@link SuspendsMachines suspending machines}.
     *
     * @throws java.lang.UnsupportedOperationException if the entity's provisioner cannot suspend machines.
     * @see MachineManagementMixins.SuspendsMachines
     */
    protected StopMachineDetails<Integer> suspendAnyProvisionedMachines() {
        @SuppressWarnings("unchecked")
        MachineProvisioningLocation<MachineLocation> provisioner = entity().getAttribute(SoftwareProcess.PROVISIONING_LOCATION);

        if (Iterables.isEmpty(entity().getLocations())) {
            log.debug("No machine decommissioning necessary for " + entity() + " - no locations");
            return new StopMachineDetails<>("No machine suspend necessary - no locations", 0);
        }

        // Only release this machine if we ourselves provisioned it (e.g. it might be running other services)
        if (provisioner == null) {
            log.debug("No machine decommissioning necessary for " + entity() + " - did not provision");
            return new StopMachineDetails<>("No machine suspend necessary - did not provision", 0);
        }

        Location machine = getLocation(null);
        if (!(machine instanceof MachineLocation)) {
            log.debug("No decommissioning necessary for " + entity() + " - not a machine location (" + machine + ")");
            return new StopMachineDetails<>("No machine suspend necessary - not a machine (" + machine + ")", 0);
        }

        if (!(provisioner instanceof SuspendsMachines)) {
            log.debug("Location provisioner ({}) cannot suspend machines", provisioner);
            throw new UnsupportedOperationException("Location provisioner cannot suspend machines: " + provisioner);
        }

        clearEntityLocationAttributes(machine);
        SuspendsMachines.class.cast(provisioner).suspendMachine(MachineLocation.class.cast(machine));

        return new StopMachineDetails<>("Suspended " + machine, 1);
    }

    /**
     * Nulls the attached entity's hostname, address, subnet hostname and subnet address sensors
     * and removes the given machine from its locations.
     */
    protected void clearEntityLocationAttributes(Location machine) {
        entity().removeLocations(ImmutableList.of(machine));
        entity().sensors().set(Attributes.HOSTNAME, null);
        entity().sensors().set(Attributes.ADDRESS, null);
        entity().sensors().set(Attributes.SUBNET_HOSTNAME, null);
        entity().sensors().set(Attributes.SUBNET_ADDRESS, null);
    }

    // Removes the checked Exception from the method signature
    public static class CloseableLatch implements AutoCloseable {
        private Entity caller;
        private ReleaseableLatch releaseableLatch;

        public CloseableLatch(Entity caller, ReleaseableLatch releaseableLatch) {
            this.caller = caller;
            this.releaseableLatch = releaseableLatch;
        }

        @Override
        public void close() {
            DynamicTasks.drain(null, false);
            releaseableLatch.release(caller);
        }
    }

    public static CloseableLatch waitForCloseableLatch(Entity entity, ConfigKey<Boolean> configKey) {
        ReleaseableLatch releaseableLatch = waitForLatch((EntityInternal)entity, configKey);
        return new CloseableLatch(entity, releaseableLatch);
    }

    private static ReleaseableLatch waitForLatch(EntityInternal entity, ConfigKey<Boolean> configKey) {
        Maybe<?> rawValue = entity.config().getRaw(configKey);
        if (rawValue.isAbsent()) {
            return ReleaseableLatch.NOP;
        } else {
            ValueResolverIterator<Boolean> iter = resolveLatchIterator(entity, rawValue.get(), configKey);
            // The iterator is used to prevent coercion; the value should always be the last one, but iter.last() will return a coerced Boolean
            Maybe<ReleaseableLatch> releasableLatchMaybe = iter.next(ReleaseableLatch.class);
            if (releasableLatchMaybe.isPresent()) {
                ReleaseableLatch latch = releasableLatchMaybe.get();
                log.debug("{} finished waiting for {} (value {}); waiting to acquire the latch", new Object[] {entity, configKey, latch});
                Tasks.setBlockingDetails("Acquiring " + configKey + " " + latch);
                try {
                    latch.acquire(entity);
                } finally {
                    Tasks.resetBlockingDetails();
                }
                log.debug("{} Acquired latch {} (value {}); continuing...", new Object[] {entity, configKey, latch});
                return latch;
            } else {
                // If iter.next() above returned absent due to a resolve error next line will throw with the cause
                Boolean val = iter.last().get();
                if (rawValue != null) log.debug("{} finished waiting for {} (value {}); continuing...", new Object[] {entity, configKey, val});
                return ReleaseableLatch.NOP;
            }
        }
    }

    private static ValueResolverIterator<Boolean> resolveLatchIterator(EntityInternal entity, Object val, ConfigKey<Boolean> key) {
        return Tasks.resolving(val, Boolean.class)
                .context(entity.getExecutionContext())
                .description("config " + key.getName())
                .iterator();
    }

    
}
