/*
 * 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.uima.ducc.sm;

import java.io.File;
import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Timer;
import java.util.TimerTask;
import java.util.TreeMap;

import javax.naming.ServiceUnavailableException;

import org.apache.uima.UIMAFramework;
import org.apache.uima.ducc.cli.IUiOptions.UiOption;
import org.apache.uima.ducc.common.IServiceStatistics;
import org.apache.uima.ducc.common.TcpStreamHandler;
import org.apache.uima.ducc.common.persistence.services.IStateServices;
import org.apache.uima.ducc.common.utils.DuccLogger;
import org.apache.uima.ducc.common.utils.DuccProperties;
import org.apache.uima.ducc.common.utils.DuccSchedulerClasses;
import org.apache.uima.ducc.common.utils.SystemPropertyResolver;
import org.apache.uima.ducc.common.utils.id.DuccId;
import org.apache.uima.ducc.transport.event.common.DuccWorkJob;
import org.apache.uima.ducc.transport.event.common.IDuccCompletionType.JobCompletionType;
import org.apache.uima.ducc.transport.event.common.IDuccState.JobState;
import org.apache.uima.ducc.transport.event.sm.IService.ServiceClass;
import org.apache.uima.ducc.transport.event.sm.IService.ServiceState;
import org.apache.uima.ducc.transport.event.sm.IService.ServiceType;
import org.apache.uima.ducc.transport.event.sm.IServiceDescription;
import org.apache.uima.ducc.transport.event.sm.ServiceDescription;
import org.apache.uima.util.Level;

/**
 * Represents the collection of process, jobs, and such that implement a given service.
 */

public class ServiceSet
	implements SmConstants
{
    /**
	 * 
	 */

	private DuccLogger logger = DuccLogger.getLogger(this.getClass().getName(), COMPONENT_NAME);	
    private ServiceHandler handler;
    private IStateServices stateHandler;

    // key is unique id of descriptor.  The descriptor inherits key from a Job's DuccId, or from
    // a unique-to SM key for implicit references.

    Map<Long, ServiceInstance> implementors = new HashMap<Long, ServiceInstance>();
    TreeMap<Integer, Integer>  available_instance_ids = new TreeMap<Integer, Integer>();  // UIMA-4258
    Map<Long, Integer>         pending_instances = new HashMap<Long, Integer>();          // For hot bounce, restore the instance ids UIMA-4258

    // List<ServiceInstance> pendingStarts = new LinkedList<ServiceInstance>();           // UIMA-4258 not used anywhere

    // key is job/service id, value is same.  it's a map for fast existence check
    Map<DuccId, DuccId> references = new HashMap<DuccId, DuccId>();

    // For a registered service, here is my registered id
    DuccId id;
    // UIMA-5244 HashMap<Long, DuccId> friendly_ids = new HashMap<Long, DuccId>();
    String history_key = IStateServices.SvcMetaProps.work_instances.pname();
    String implementors_key = IStateServices.SvcMetaProps.implementors.pname();

    // incoming nodes, for dup checking
    List<ServiceSet> predecessors = new ArrayList<ServiceSet>();
    List<ServiceSet> successors   = new ArrayList<ServiceSet>();

    // for UIMA-AS this is the endpoint as the unique identifier of this service
    String key;

    // UIMA-AS endpoint information
    String endpoint;
    String broker;
    String broker_host;
    int broker_port;
    int broker_jmx_port = 1099;
    String[] independentServices = null;

    // Registered services, the submitter
    String user;

    // Automatically start at boot, and keep implementors alive
    boolean autostart = false;
    // We've been stopped, which is used to override autostart
    // boolean stopped   = false;     // TODO Must get rid of this entirely
    boolean enabled   = true;

    // We've been started, so we know to enforce instance count even if not autostarted
    boolean started   = false;
    // Remember if was started by reference only so we can stop when refs die
    boolean reference_start = false;
    // is it ping-only?
    boolean ping_only = false;
    // debug specified in the registration?
    boolean process_debug = false;

    // Date of last known use of the service.  0 means "I don't know"
    long last_use = 0;

    // Date of last known successful ping of the service.  0 means never.  UIMA-4309
    long last_ping = 0;

    boolean notPinging = false;
    String notPingingReason = null;
    
    // Date of last known time any instance made it to Running state.  0 means never. UIMA-4309
    long last_runnable = 0;

    // The number of instances to maintain live.
    // Pinger or manual start/stop may make the number of live instances differ from the registered number
    int instances = 1;
    int registered_instances;

    // Service monitor / pinger 
    IServiceMeta serviceMeta = null;

    // registered services state files
    private DuccProperties job_props  = null;
    private DuccProperties meta_props = null;

    boolean deregistered = false;

    ServiceType  service_type  = ServiceType.Undefined;
    ServiceClass service_class = ServiceClass.Undefined;
    ServiceState service_state = ServiceState.Stopped;;

    // structures to manage service linger after it exits
    Timer timer = null;
    LingerTask linger = null;
    long linger_time = 60000;

    int init_failure_max = ServiceManagerComponent.init_failure_max;
    int init_failures = 0;                   // max allowed consecutive failures, current failure count
    
    int ping_failure_max = ServiceManagerComponent.failure_max;
    int ping_failures = 0;                   // for ping-only services, if the external pinger throws errors we
                                             // need to govern it

    int run_failures = 0;
    boolean excessiveRunFailures = false;       // signaled by monitor / pinger if we have too many

    boolean inShutdown = false;

    String[] coOwners = null;

    //  Swapped these 2 values  UIMA-5244
    String archive_key = IStateServices.SvcMetaProps.is_archived.columnName();
    String archive_flag  = "true";
    
    private String warning = "";  // May hold class-change msg

    //
    // Constructor for a registered service
    //
    public ServiceSet(ServiceHandler handler, IStateServices stateHandler, DuccId id, DuccProperties props, DuccProperties meta)
    {
        this.handler = handler;
        this.stateHandler = stateHandler;
        this.job_props = props;
        this.meta_props = meta;
        this.id = id;

        // Check for valid scheduling class here (was in the CLI)
        validateSchedulingClass(props);
        
        this.service_state = ServiceState.Stopped;
        this.linger_time = props.getLongProperty(UiOption.ServiceLinger.pname(), linger_time);
        this.key = meta.getProperty(IStateServices.SvcMetaProps.endpoint.pname());

        parseEndpoint(key);

        this.user = meta.getProperty(IStateServices.SvcMetaProps.user.pname());
        this.instances = meta.getIntProperty(IStateServices.SvcMetaProps.instances.pname(), 1);
        this.registered_instances = this.instances;
        this.autostart = meta.getBooleanProperty(IStateServices.SvcMetaProps.autostart.pname(), false);
        this.ping_only = meta.getBooleanProperty(IStateServices.SvcMetaProps.ping_only.pname(), false);
        this.enabled   = meta.getBooleanProperty(IStateServices.SvcMetaProps.enabled.pname(), enabled);
        this.service_class = ServiceClass.Registered;
        this.init_failure_max = props.getIntProperty(IStateServices.SvcRegProps.instance_init_failures_limit.pname(), init_failure_max);
        this.reference_start = meta.getBooleanProperty(IStateServices.SvcMetaProps.reference.pname(), this.reference_start);

        
        // Check if key has a value vs. empty or missing
        if ( props.containsKey(UiOption.ProcessDebug.pname()) &&
                ((String) props.get(UiOption.ProcessDebug.pname())).length() > 0)  {
            this.process_debug = true;
        }

        if ( props.containsKey(UiOption.Administrators.pname()) ) {
            String adm = props.getProperty(UiOption.Administrators.pname());
            if ( adm != null ) {
                coOwners = adm.split("\\s+");
            }
        }

        parseIndependentServices();

        meta_props.put(IStateServices.SvcMetaProps.references.pname(), "");         // Will get refreshed in upcoming OR state messages
        meta_props.remove(IStateServices.SvcMetaProps.stopped.pname());             // obsolete flag, clean out of older registrations

        meta_props.put(IStateServices.SvcMetaProps.service_class.pname(), ""+service_class.decode());
        meta_props.put(IStateServices.SvcMetaProps.service_type.pname(), ""+service_type.decode());
        meta_props.put(IStateServices.SvcMetaProps.enabled.pname(), "" + enabled);         // may not have been there in the first place
        meta_props.put(IStateServices.SvcMetaProps.service_state.pname(), ""+getState());
        meta_props.put(IStateServices.SvcMetaProps.ping_active.pname(), "false");
        meta_props.put(IStateServices.SvcMetaProps.service_alive.pname(),      "false");
        meta_props.put(IStateServices.SvcMetaProps.service_healthy.pname(),    "false");
        meta_props.put(IStateServices.SvcMetaProps.service_statistics.pname(), "N/A");
        setReferenced(this.reference_start);

        setLastUse(meta_props.getLongProperty(IStateServices.SvcMetaProps.last_use.pname(), 0L));
        setLastPing(meta_props.getLongProperty(IStateServices.SvcMetaProps.last_ping.pname(), 0L));
        setLastRunnable(meta_props.getLongProperty(IStateServices.SvcMetaProps.last_runnable.pname(), 0L));

        if ( (!job_props.containsKey(UiOption.ProcessExecutable.pname())) && (service_type != ServiceType.UimaAs) ) {
            meta_props.put(IStateServices.SvcMetaProps.ping_only.pname(), "true");
            this.ping_only = true;
        } else {
            meta_props.put(IStateServices.SvcMetaProps.ping_only.pname(), "false");
            this.ping_only = false;
        }

        savePendingInstanceIds();         // UIMA-4258
        // caller will save the meta props, **if** the rest of registration is ok.

        //UIMAFramework.getLogger(BaseUIMAAsynchronousEngineCommon_impl.class).setLevel(Level.OFF);
        //UIMAFramework.getLogger(BaseUIMAAsynchronousEngine_impl.class).setLevel(Level.OFF);
        // there are a couple junky messages that slip by the above configurations.  turn the whole danged thing off.
        UIMAFramework.getLogger().setLevel(Level.OFF);
    }

    DuccId getId()
    {
        return id;
    }

    String getEndpoint() {
    	return key;
    }
    
    String getWarning() {
      return warning;
    }
    
    // UIMA-4258
    // Get potentially pending instances from meta and stash them away for a bit
    // Used in hot-start to remap instance ids to ducc ids
    void savePendingInstanceIds()
    {
        String ids = meta_props.getProperty(implementors_key);
        if ( ids == null ) return;

        // UIMA-4258 Conversion: if no . then there is no instance, and it is an old service format service.
        //           Must remove the implementors from the meta and return.
        //
        if ( ids.indexOf(".") <= 0 ) {
            meta_props.put(implementors_key, "");
            return;
        }

        String[] tmp = ids.split("\\s+");
        for (String s : tmp) {
            String[] id_inst = s.split("\\.");
            pending_instances.put(Long.parseLong(id_inst[0]), Integer.parseInt(id_inst[1]));
        }
    }

    void parseEndpoint(String ep)
    {
        if ( ep.startsWith(ServiceType.UimaAs.decode()) ) {
            int ndx = ep.indexOf(":");
            ep = ep.substring(ndx+1);
            ndx = ep.indexOf(":");

            this.endpoint = ep.substring(0, ndx).trim();
            this.broker = ep.substring(ndx+1).trim();
            this.service_type = ServiceType.UimaAs;

            URL url = null;
  			try {                
				url = new URL(null, broker, new TcpStreamHandler());
			} catch (MalformedURLException e) {
                throw new IllegalArgumentException("Invalid broker URL: " + broker);
			}
            this.broker_host = url.getHost();
            this.broker_port = url.getPort();

            if ( this.endpoint.equals("") || this.broker.equals("") ) {
                throw new IllegalArgumentException("The endpoint cannot be parsed.  Expecting UIMA-AS:Endpoint:Broker, received " + key);
            }
        } else {
            this.service_type = ServiceType.Custom;
            int ndx = ep.indexOf(":");
            this.endpoint = ep.substring(ndx+1);
        }

    }

    // Check if a valid class name & if must be changed to a fixed one
    private void validateSchedulingClass(DuccProperties properties) {
      DuccSchedulerClasses duccSchedulerClasses = DuccSchedulerClasses.getInstance();
      String key = UiOption.SchedulingClass.pname();
      String schedulingClass = properties.getProperty(key);
      try {
        if (schedulingClass == null) {
          properties.setProperty(key, duccSchedulerClasses.getDebugClassDefaultName());
          return;
        }
        //if (!duccSchedulerClasses.getClasses().containsKey(schedulingClass)) {
        //  throw new IllegalArgumentException("Unknown scheduling_class: " + schedulingClass);
        //}
        if (duccSchedulerClasses.isPreemptable(schedulingClass)) {
          String fixedClass = duccSchedulerClasses.getDebugClassSpecificName(schedulingClass);
          if (fixedClass == null) {
            throw new IllegalArgumentException("Invalid class configuration - all classes must have a debug (fixed) entry");
          }
          properties.setProperty(key, fixedClass);
          warning = " (changed preemptable " + key + " to " + fixedClass + ")";
        }
      } catch (IllegalArgumentException e) {
        throw e;
      } catch (Exception e) {
        throw new IllegalArgumentException(e);
      }
    }

    synchronized Long[] getImplementors()
    {
        return implementors.keySet().toArray(new Long[implementors.size()]);
    }

    synchronized String getHostFor(Long implid)
    {
        return implementors.get(implid).getHost();
    }

    synchronized long getShareFor(Long implid)
    {
        return implementors.get(implid).getShareId();
    }

    synchronized DuccId[] getReferences()
    {
        return references.keySet().toArray(new DuccId[references.size()]);
    }

    void setIncoming(ServiceSet sset)
    {
        predecessors.add(sset);
    }

    void clearEdges()
    {
        predecessors.clear();
        successors.clear();
    }

    boolean hasPredecessor()
    {
        return predecessors.size() != 0;
    }

    List<ServiceSet> getPredecessors()
    {
        return predecessors;
    }

    void removePredecessor(ServiceSet pred)
    {
        predecessors.remove(pred);
    }

    void setOutgoing(ServiceSet sset)
    {
        this.successors.add(sset);
    }

    List<ServiceSet> getSuccessors()
    {
        return new ArrayList<ServiceSet>(this.successors);
    }

    void removeSuccessor(ServiceSet succ)
    {
        successors.remove(succ);
    }

    boolean hasSuccessor()
    {
        return successors.size() != 0;
    }

    String[] getIndependentServices()
    {
        return independentServices;
    }

    // just for testing!
    void setIndependentServices(String[] ind)
    {
        this.independentServices = ind;
    }

    void deleteJobProperty(String k) {
        job_props.put(k, "");
    }

    void setJobProperty(String k, String v)
    {
        job_props.put(k, v);
    }

    void setMetaProperty(String k, String v)
    {
        meta_props.put(k, v);
    }

    boolean isDebug()
    {
        return process_debug;
    }

    /**
     * Is 'user' a registered co-owner?
     */
    boolean isAuthorized(String user)
    {
        if ( coOwners == null ) return false;
        for ( String s : coOwners ) {
            if ( s.equals(user) ) return true;
        }
        return false;
    }

    void parseAdministrators(String admins)
    {
        if ( admins != null ) {
            coOwners = admins.split("\\s+");
        }
    }

    private void parseIndependentServices()
    {
        String depstr = job_props.getProperty(UiOption.ServiceDependency.pname());
        String[] result = null;

        if ( depstr != null ) {
            result = depstr.split("\\s+");
            for ( int i = 0; i < result.length; i++ ) {
                result[i] = result[i].trim();
            }
        }
        independentServices = result;
    }

    /**
     * At boot only ... synchronize my state with published OR state.
     * 
     * We do this in the first phase of boot, then bootComplete is called to synchronize
     * history and update the physical meta properties file.
     */
    Map<Long, ServiceInstance> pendingImplementors = new HashMap<Long, ServiceInstance>();
    void bootImplementor(DuccId id, JobState state)
    {
    	String methodName = "bootImplementor";
        ServiceInstance si = new ServiceInstance(this);

        if ( ! pending_instances.containsKey(id.getFriendly()) ) {
            logger.warn(methodName, id, "Incoming Orchestrator state indicates active service instance but it is not in my meta data.");
            logger.warn(methodName, id, "Instance ignored.  This is usally caused by system or database failure.");
            return;
        }

        si.setState(state);
        si.setId(id.getFriendly());
        si.setStopped(false);
        si.setUser(this.user);
        si.setInstanceId(pending_instances.get(id.getFriendly())); // UIMA-4258

        handler.addInstance(this, si);
        pendingImplementors.put(id.getFriendly(), si);        // remember which instances we hear about in current OR publication
    }

    /**
     * Second phase, update history, and physical metaprops.
     */
    void bootComplete()
        throws Exception
    {
        //String methodName = "bootComplete";
        //
        // During boot, inactive implementors are removed.  Here we cull the implementors list to
        // remove stuff that didn't come in.
        //

        if ( isPingOnly() && enabled() ) {
            start();   // nothing to recover but we need the pseudo service to run
            return;   
        }
        implementors = pendingImplementors;   // only the ones that check in.  others are toast

        // 
        // must update history against stuff we used to have and don't any more
        //
        // TODO: update the history record in the meta
        //

        // UIMA-4258 restore instance ID if this is a hot restart
        if ( pending_instances.size() != 0 ) {
            TreeMap<Integer, Integer> nst = new TreeMap<Integer, Integer>();
            for (int i : pending_instances.values()) {
                nst.put(i, i);
            }
            int ndx = 0;
            while ( nst.size() > 0 ) {
                if ( nst.containsKey(ndx) ) {
                    nst.remove(ndx);
                } else {
                    available_instance_ids.put(ndx, ndx);
                }
                ndx++;
            }
        }
        pending_instances = null;

        // on restart, if we think we were ref started when we crashed, but there are no
        // implementors, we can't actually be ref started, so clean that up.
        if ( isReferencedStart() && (countImplementors() == 0 ) ) {
            this.reference_start = false;
        }
        updateMetaProperties();
    }

    
    /**
     *
     */
    synchronized void enforceAutostart()
    {
        String methodName = "enforceAutostart";
        if ( ! autostart ) return;                           // not doing auto, nothing to do
        if ( ! enabled() ) return;                           // doing auto, but we are disabled
        if ( init_failures >= init_failure_max ) return;     // too many init failures, no more enforcement
        if ( ping_failures >= ping_failure_max ) return;     // not pinging, let's not start more stuff
        
        // could have more implementors than instances if some were started dynamically but the count not persisted via registration
        int needed = Math.max(0, instances - countImplementors());
        if ( needed > 0 ) {
            logger.info(methodName, id, "Autostarting", needed, "instance" + ((needed > 1) ? "s" : ""), "already have", countImplementors());
            start();
        }
    }
    

    boolean isUimaAs()
    {
        return (service_type == ServiceType.UimaAs);
    }

    boolean isCustom()
    {
        return (service_type == ServiceType.Custom);
    }

    DuccProperties getJobProperties()
    {
        return job_props;
    }

    DuccProperties getMetaProperties()
    {
        return meta_props;
    }

    boolean isPingOnly()
    {
        return ping_only;
    }

    synchronized long getLastUse()
    {
        return last_use;
    }

    // UIMA-4309
    synchronized long getLastPing()
    {
        return last_ping;
    }

    // UIMA-4309
    synchronized long getLastRunnable()
    {
        return last_runnable;
    }

    synchronized void setLastUse(long lu)
    {
        this.last_use = lu;
        meta_props.put(IStateServices.SvcMetaProps.last_use.pname(), Long.toString(lu));
        if ( last_use == 0 ) {
            meta_props.put(IStateServices.SvcMetaProps.last_use_readable.pname(), "Unknown");
        } else {
            meta_props.put(IStateServices.SvcMetaProps.last_use_readable.pname(), (new Date(lu)).toString());
        }
    }

    // UIMA-4309
    synchronized void setLastPing(long lp)
    {
        this.last_ping = lp;
        meta_props.put(IStateServices.SvcMetaProps.last_ping.pname(), Long.toString(lp));
        if ( last_ping == 0 ) {
            meta_props.put(IStateServices.SvcMetaProps.last_ping_readable.pname(), "Unknown");
        } else {
            meta_props.put(IStateServices.SvcMetaProps.last_ping_readable.pname(), (new Date(lp)).toString());
        }
    }

    // UIMA-4309
    synchronized void setLastRunnable(long lr)
    {
        this.last_runnable = lr;
        meta_props.put(IStateServices.SvcMetaProps.last_runnable.pname(), Long.toString(lr));
        if ( last_runnable == 0 ) {
            meta_props.put(IStateServices.SvcMetaProps.last_runnable_readable.pname(), "Unknown");
        } else {
            meta_props.put(IStateServices.SvcMetaProps.last_runnable_readable.pname(), (new Date(lr)).toString());
        }
    }

    synchronized void resetRuntimeErrors()
    {
        run_failures = 0;
        ping_failures = 0;
        init_failures = 0;
        // Can't just remove as DB is updated from entries in the map
        meta_props.put(IStateServices.SvcMetaProps.submit_error.pname(), "");
        excessiveRunFailures = false;
    }

    synchronized void setAutostart(boolean auto)
    {
        if (!this.autostart && auto) {   // UIMA-5390 Restrict these resets to only if autostart was off but is now on 
            // turning this on gives benefit of the doubt on failure management
            // by definition, an autostarted services is NOT reference started
            cancelLinger();
            setReferenced(false);
            init_failures = 0;
            resetRuntimeErrors();
        }
        meta_props.setProperty(IStateServices.SvcMetaProps.autostart.pname(), auto ? "true" : "false");
        this.autostart = auto;
    }

    synchronized void restartPinger()
    {
        stopPingThread(); 
        resetRuntimeErrors();
    }

    /**
     * Manual start: turn off manual stop
     *               override reference_start
     *               remember manual start was done
     */
    synchronized void setStarted()
    {
        started = true;
        init_failures = 0;
    }

    /**
     * Manual stop: override reference_start and manual start.
     *              remember 'stopped' so enforceAutostart doesn't restart
     */
    // synchronized void setStopped()
    // {
    //     started = false;
    //     stopped = true;
    // }

    /**
     * Start by reference: if autostarted or already manually started, don't change anything
     *                     else remember we're ref started and not stopped
     */
    // synchronized void xsetReferencedStart(boolean is_start)
    // {
    //     if ( is_start ) {
    //         if ( isAutostart() || isStarted() ) return;
    //         this.stopped = false;
    //         this.reference_start = true;
    //         init_failures = 0;
    //         resetRuntimeErrors();
    //     } else {
    //         this.reference_start = false;
    //     }
    // }

    /**
     * Is the service stopped or about to stop?
     */
    synchronized boolean isStopped()
    {
        switch ( service_state ) {
            case Stopping:
            case Stopped:
                return true;
            default:
                return false;
        }
    }

    synchronized void ignoreReferences()
    {
        setReferenced(false);
        cancelLinger();
    }

    synchronized void observeReferences()
    {
        setReferenced(true);
        if ( countReferences() == 0 ) {
            lingeringStop();
        }
    }

    synchronized void disable(String reason)
    {
        meta_props.put(IStateServices.SvcMetaProps.disable_reason.pname(), reason);
        this.enabled = false;
    }

    synchronized void enable()
    {
        // Can't just remove as DB is updated from entries in the map
        meta_props.put(IStateServices.SvcMetaProps.disable_reason.pname(), "");
        resetRuntimeErrors();
        this.enabled = true;
    }

    synchronized boolean enabled()
    {
        return this.enabled;
    }

    synchronized String getDisableReason()
    {
        return meta_props.getStringProperty(IStateServices.SvcMetaProps.disable_reason.pname(), "Unknown");
    }

    /**
     * "Manually" started.
     */
    synchronized boolean isStarted()
    {
        return this.started;
    }

    /**
     * Started "by reference"
     */
    synchronized boolean isReferencedStart()
    {
        return this.reference_start;
    }

    synchronized boolean isAutostart()
    {
        return this.autostart;
    }

    String getUser()
    {
        return user;
    }

    boolean isDeregistered()
    {
        return deregistered;
    }

    void deregister()
    {
        deregistered = true;
    }

    // /**
    //  * Returns the number of currently running instances
    //  */
    // synchronized int getNInstances()
    // {
    //     return instances;
    // }

    /**
     * Returns the number of registered instances.
     */
    synchronized int getNInstancesRegistered()
    {
        return registered_instances;
    }

    /**
     * Service is unregistered, remove props from main DB, write them into the history DB.
     */
    synchronized void deleteProperties()
        throws Exception
    {
    	String methodName = "deleteProperties";

        String ak = meta_props.getProperty(archive_key);
        if ( (ak != null ) && ak.equals(archive_flag) ) { // (Migth not be set, that's ok)
            // Because of races and stuff we can get called more than once but should only
            //  archive once.
            logger.info(methodName, id, "Bypassing move to history; already moved.");
            return;
        }

        /*   UIMA-5244 Can remove this as friendly_ids is always empty
        String history = meta_props.getStringProperty(history_key, "");
        for ( Long id : friendly_ids.keySet() ) {
            history = history + " " + id.toString();
        }
        meta_props.put(history_key, history);
        */
        meta_props.put(archive_key, archive_flag);

        try {
            stateHandler.moveToHistory(id, job_props, meta_props);
        } catch ( Exception e ) {
            logger.error(methodName, id, "Could not move properties files to history: ", e);
        }
    }


    /**
     * Save both properties in a single transaction.
     */
    synchronized void storeProperties(boolean isRecovered)
        throws Exception
    {
    	//String methodName = "storeProperties";

        // no, don't store if it gets deregistered this fast
        if ( isDeregistered() ) return;

        //Long strid = id.getFriendly();
        prepareMetaProperties();     // these always need house cleaning before storing or syncing

        if ( ! isRecovered ) {       // if not recovery, no need to mess with the record
            stateHandler.storeProperties(id, job_props, meta_props);
        } else {                
            stateHandler.updateJobProperties(id, (Properties) job_props);
            stateHandler.updateMetaProperties(id, meta_props);
        }
    }

    synchronized void updateSvcProperties()
        throws Exception
    {
        // no more changes
        if ( isDeregistered() ) return;

        stateHandler.updateJobProperties(id, (Properties) job_props);
    }

    synchronized void updateMetaProperties()
    	throws Exception
    {
    	// String methodName = "saveMetaProperties";
        if ( isDeregistered() ) return;                // we may have deleted the properties but stuff
                                                       // lingers out of our control.  no more updates
                                                       // which can leave junk in the registry directory
                                                       // for file-based registry.

        prepareMetaProperties();
        stateHandler.updateMetaProperties(id, meta_props);
    }
    
    private long pingStabilityDefault = 10;
    private long pingRateDefault = 60 * 1000;
    
    private long pingStability = -1;
	private long pingRate = -1;
    
	private void configPingStability() {
		pingStability = SystemPropertyResolver.getLongProperty("ducc.sm.meta.ping.stability", pingStabilityDefault);
	}
	
	private void configPingRate() {
		pingRate = SystemPropertyResolver.getLongProperty("ducc.sm.meta.ping.rate", pingRateDefault);
	}
	
	private void configPing() {
		configPingRate();
		configPingStability();
	}
	
	/**
	 * If the Service state is Available but the pinger data has not been updated
	 * beyond the expiry time then determine that the pinger data is stale
	 * 
	 * The expiry time is calculated as pingStability * pingRate, nominally 10 * 60000.
	 */
    private void determinePingerStatus() {
    	String location = "determinePingerStatus";
    	switch(getState()) {
    	case Available:
    	case Waiting:
    		if(serviceMeta == null) {
    			notPinging = true;
    			notPingingReason = "pinger is not running";
    		}
    		else {
    			configPing();
    			long pingExpiry = pingStability * pingRate;
        		long now = System.currentTimeMillis();
        		long pingElapsed = now - last_ping;
        		if (pingElapsed > pingExpiry && last_ping != 0) {    // Don't treat first ping as stale
        			notPinging = true;
        			notPingingReason = "pinger data is stale";
        		}
        		else {
        			notPinging = false;
        			notPingingReason = "N/A";
        		}
        	}
    		break;
    	default:
    		notPinging = false;
			notPingingReason = "N/A";
    		break;
    	}
    	if(notPinging) {
    		logger.info(location, id, notPingingReason);
    		switch(service_state) {
    		case Available:
    			logger.info(location, id, service_state+" => "+ServiceState.Waiting);
    			service_state = ServiceState.Waiting;
    		}
    	}
    	return;
    }
    
    void prepareMetaProperties()
    {
        // String methodName = "saveMetaProperties";
        
        // try {
        //     throw new IllegalStateException("Saving meta properties");
        // } catch ( Throwable t) {
        //     t.printStackTrace();
        // }
        
        // UIMA-4587 Why bypass, as state can still dribble in.
        // if ( isDeregistered() ) return;

        if ( implementors.size() == 0 ) {
            meta_props.put(implementors_key, "");
        } else {
            StringBuffer sb_ducc_id = new StringBuffer();
            for ( Long l : implementors.keySet() ) {
                // UIMA-4258 Add instance id to ducc id when saving
                ServiceInstance inst = implementors.get(l);
                sb_ducc_id.append(Long.toString(l));
                sb_ducc_id.append(".");
                sb_ducc_id.append(Integer.toString(inst.getInstanceId()));
                sb_ducc_id.append(" ");
            }
            String s = sb_ducc_id.toString().trim();
            meta_props.setProperty(implementors_key, s);
        }

        determinePingerStatus();
        
        meta_props.put(IStateServices.SvcMetaProps.reference.pname(), isReferencedStart() ? "true" : "false");
        meta_props.put(IStateServices.SvcMetaProps.autostart.pname(), isAutostart()       ? "true" : "false");

        meta_props.put(IStateServices.SvcMetaProps.enabled.pname(), ""+enabled);
        meta_props.put(IStateServices.SvcMetaProps.service_state.pname(), ""+ getState());
        meta_props.put(IStateServices.SvcMetaProps.ping_active.pname(), "" + !notPinging);
        meta_props.put(IStateServices.SvcMetaProps.service_alive.pname(),      "false");
        meta_props.put(IStateServices.SvcMetaProps.service_healthy.pname(),    "false");

        if ( excessiveFailures() ) {
        	String msg = init_failures >= init_failure_max ? "initialization failures [" + init_failures + "]" 
        			                                       : "runtime failures [" + run_failures + "]";
        	meta_props.put(IStateServices.SvcMetaProps.submit_error.pname(), "Service disabled by excessive " + msg);
        } else {
            meta_props.put(IStateServices.SvcMetaProps.service_statistics.pname(), "N/A");
        }
        
        if ( serviceMeta != null ) {
            IServiceStatistics ss = serviceMeta.getServiceStatistics();
            if ( ss != null ) {
                meta_props.put(IStateServices.SvcMetaProps.service_alive.pname(),      "" + ss.isAlive());
                meta_props.put(IStateServices.SvcMetaProps.service_healthy.pname(),    "" + ss.isHealthy());
                meta_props.put(IStateServices.SvcMetaProps.service_statistics.pname(), "" + ss.getInfo());

                if ( ss.isAlive() ) {                    // UIMA-4309
                    setLastPing(serviceMeta.getServiceStatisticsTimestamp());
                }
            }
        }
        
        return;
    }

    synchronized void updateInstance(long iid, long share_id, String host)
    {
    	String methodName = "updateInstance";
        ServiceInstance inst = implementors.get(iid);
        if ( inst == null ) {
            logger.warn(methodName, id, "Cannot find instance", iid, "for update:", host + ":" + share_id);
            return;
        }
        inst.update(share_id, host);
    }

    /**
     * 
     * @param n     the new value for the register instance count
     * 
     * Called when the registration is updated via the CLI
     * Does NOT update the desired number of running instances as this can be set by the pinger.
     * Also modifications to the registration should not affect running instances.
     */
    synchronized void updateRegisteredInstances(int n)
    {
        meta_props.setProperty(IStateServices.SvcMetaProps.instances.pname(), Integer.toString(n));
        registered_instances = n;
    }

    /**
     * @param n      is the target number of instances we want running
     * 
     * This param dropped??
     * @param update indicates whether to match registration to the target
     * 
     * Called by doStart & doStop so may be making the running instances differ from the registered number
     */
    synchronized void updateInstances(int n)
    {
        if ( n >= 0 ) {
     
            instances = n;
            
            int running    = countImplementors();
            int diff       = n - running;
                
            if ( diff > 0 ) {
                start();
            } else if ( diff < 0 ) {
                stop(-diff); // TODO: no good, fix when changeTo is ready
            }
        }
    }

    synchronized void updateDebug(String val)
    {
        if ( val.equals("off") ) {
            job_props.put(UiOption.ProcessDebug.pname(), "");
            this.process_debug = false;
        } else {
            job_props.put(UiOption.ProcessDebug.pname(), val);
            this.process_debug = true;
        }
    }

    synchronized void updateLinger(String val)
    {
    	String methodName = "updateLinger";
        try {
            this.linger_time = Long.parseLong(val);
        } catch( NumberFormatException e ) {
            logger.error(methodName, id, "Cannot update linger, not numeric:", val);
        }
    }

    synchronized void updateInitFailureLimit(String val)
    {
    	String methodName = "updateInitFailureLimit";
        try {
            this.init_failure_max = Integer.parseInt(val);
        } catch( NumberFormatException e ) {
            logger.error(methodName, id, "Cannot update init failure max, not numeric:", val);
        }
    }


    synchronized void persistReferences()
    {
        String methodName = "persistReferences";

        if ( references.size() == 0 ) {
            meta_props.put(IStateServices.SvcMetaProps.references.pname(), "");
        } else {
            StringBuffer sb = new StringBuffer();
            for ( DuccId id : references.keySet() ) {
                sb.append(id.toString());
                sb.append(" ");
            }
            String s = sb.toString().trim();
            meta_props.setProperty(IStateServices.SvcMetaProps.references.pname(), s);
        }
        try {
            updateMetaProperties();
        } catch ( Exception e ) {
            logger.warn(methodName, id, "Cannot update meta properties:", e);
        }
    }

    void clearQueue()
    {
    	String methodName = "clearQueue";
    	
        if ( !deregistered ) {
            logger.info(methodName, id, "Not clearing queue because service is still registered.");
            return;
        }

        if ( implementors.size() != 0 ) {
            logger.info(methodName, id, "Not clearing queue because", implementors.size(), "implementors are still alive (", key, ").");
            return;
        }

        handler.removeService(this);
        try {
			deleteProperties();
		} catch (Exception e1) {
			logger.error(methodName, id, "Cannot complete deferred delete of properties:", e1);
		}

        if ( service_type != ServiceType.UimaAs ) {
            logger.info(methodName, id, "Deleting unregistered service; not clearing queue because this is not a UIMA-AS service:", key);
            return;
        }

        if ( isPingOnly() ) {
            logger.info(methodName, id, "Deleting unregistered service; not clearing queue for ping-only service", key);
            return;
        }

        String pingclass = job_props.getStringProperty(UiOption.ServicePingClass.pname(), UimaAsPing.class.getName());
        if ( !pingclass.equals(UimaAsPing.class.getName()) ) {
            logger.info(methodName, id, "Deleting unregistered service: not clearing queue because not using the default UIMA-AS pinger:", pingclass, "(", key, ")");
            return;
        }

        // Only do this if using the default pinger.  It's the pinger's job otherwise.
        UimaAsServiceMonitor monitor = new UimaAsServiceMonitor(endpoint, broker_host, broker_jmx_port);
        logger.info(methodName, id, "Deleting unregistered service and clearing queues for", key, "at [" + broker_host + ":" + broker_jmx_port + "]");
        try {
            monitor.init(null);
            monitor.clearQueues();
            monitor.stop(	);
        } catch (IOException e) {
            Throwable t = e.getCause();
            // ServiceUnavailbleException means it's gone already and we're processing some sort of stale state
            // that's still floating around.
            if ( ! (t instanceof ServiceUnavailableException) ) {
                logger.info(methodName, id, e);
            }
        } catch (Throwable e) {
            logger.info(methodName, id, e.toString());
        }
    }

    public synchronized int countImplementors()
    {
        return implementors.size();
    }

    public synchronized int countReferences()
    {
        return references.size();
    }

    public synchronized Long[] getActiveInstances()
    {
        ArrayList<Long> instIds = new ArrayList<Long>();
        for ( ServiceInstance inst : implementors.values() ) {
            if ( inst.isRunning() ) {
                instIds.add(inst.getId());
            }
        }
        return instIds.toArray(new Long[instIds.size()]);
    }

    synchronized void cancelLinger()
    {
    	String methodName = "cancelLinger";
        if ( linger != null ) {
            logger.debug(methodName, this.id, " ---------------- Canceling linger task");
            linger.cancel();
            linger = null;
        }
    }

    public void setErrorString(String s)
    	throws Exception
    {
        meta_props.put(IStateServices.SvcMetaProps.submit_error.pname(), s);
        updateMetaProperties();
    }

    public String getErrorString()
    {
        return meta_props.getProperty(IStateServices.SvcMetaProps.submit_error.pname()); 
    }

    void setReferenced(boolean r)
    {
        this.reference_start = r;
        meta_props.put(IStateServices.SvcMetaProps.reference.pname(), Boolean.toString(this.reference_start));
    }

    public synchronized void reference(DuccId id)
    {
        String methodName = "reference";

        logger.info(methodName, this.id, "Reference start requested by ", id);

        if ( ! enabled() ) {
             logger.warn(methodName, this.id, "Not reference starting new service instances because service is disabled.");
             return;
        }

        if ( excessiveFailures() ) {
            logger.warn(methodName, this.id, "Reference start fails, excessive failures: init[" + init_failures + "], run[" + run_failures + "]");
            return;
        }

        cancelLinger();
        references.put(id, id);
        logger.info(methodName, this.id, " References job/service", id, "count[" + references.size() + "] implementors [" + implementors.size() + "]");

        boolean idle = true;
        for (ServiceInstance si : implementors.values() ) {     // see if anything is running
            logger.debug(methodName, this.id, "Implementor", si.getId(), "state:", si.getState());
            if ( si.isRunning() ) {                             // and if so, no need to start anything
                idle = false;
                break;
            }
        }

        // Nothing running, so we do referenced start.
        if ( idle ) {
            logger.info(methodName, this.id, "Reference starting new service instances.");
            init_failures = 0;
            resetRuntimeErrors();
            setReferenced(true);
            start();
        } 

        persistReferences();
    }

    public synchronized void dereference(DuccId id)
    {
        String methodName = "dereference";
        if ( references.remove(id) == null ) {
            logger.error(methodName, this.id, "Dereference job/service",  id,  "not found in map for", getKey());
            return;
        }

        // stop the pinger if no longer needed
        if ( (references.size() == 0) && isReferencedStart() ) {                         // nothing left
            lingeringStop();
        }

        logger.info(methodName, this.id, " Dereferences job/service", id, "count[" + references.size() + "]");
        persistReferences();
    }

//     public synchronized int countReferences()
//     {
//         // note that this could change as soon as you get it so don't count on it being correct
//         // this is intended only for messages that don't have to be too accurate
//         return references.size();
//     }

    boolean containsImplementor(DuccId id)
    {
        return implementors.containsKey(id.getFriendly());
    }
    
    /**
     * Called by the PingDriver to return ping/monitor results, and to act on the results.
     *
     * @param nadditions           This is the number of new instances to start.
     * @param deletions            These are the specific instances to stop.
     * @param ndeleteions          This is the number of instances to stop.  This may well be smaller than
     *                             the size of the 'deletions' array because PingDriver caps deletions to
     *                             prevent over-aggressive or buggy monitors from killing a service.
     * @param isExcessiveFailuress This is set to 'true' if the ping/monitor decides there have been
     *                             too many instance failures and SM should stop trying to restart them.
     */
    synchronized void signalRebalance(int nadditions, Long[] deletions, int ndeletions, boolean isExcessiveFailures)
    {
        String methodName = "signalRebalance";
        logger.info(methodName, id, 
                    "PING: Additions:", nadditions, 
                    "deletions:", ndeletions, 
                    "excessive failures:", isExcessiveFailures, 
                    "implementors", countImplementors(),
                    "references", countReferences()
                    );

        ping_failures = 0;   
        
        this.excessiveRunFailures = isExcessiveFailures;
        
        // Note that nadditions could == ndeletions.  This is ok, because the monitor may want
        // to 'reboot' an instance by killing a specific one and also starting up a new one.
        
        if ( nadditions > 0) {
            start();
        }
        
        for ( int i = 0; i < ndeletions; i++ ) {
            instances -= stop(deletions[i]); // stop() may return 0 or 1
        }
        // keep services with autostart=true or with active references 
        // running with at least one instance on a rebalance
        // UIMA-4995 May happen when pinger says both delete 1 and add 1
        if ((this.isAutostart() || (this.countReferences() > 0) ) && instances == 0) {
          instances = 1;
        }

        try {
            updateMetaProperties();
        } catch ( Exception e ) {
            logger.warn(methodName, id, "Cannot update meta properties:", e);
        }
    }

    /**
     * Based on the state of the dwj, can we delete this instance from the records?
     */
    boolean canDeleteInstance(DuccWorkJob dwj)
    {
        // These are the job states
		// Received,				// Job has been vetted, persisted, and assigned unique Id
		// WaitingForDriver,		// Process Manager is launching Job Driver
		// WaitingForServices,		// Service Manager is checking/starting services for Job
		// WaitingForResources,	// Scheduler is assigning resources to Job
        // Assigned                 // passed basic tests, dispatched, not yet started to initialize
		// Initializing,			// Process Agents are initializing pipelines
		// Running,				// At least one Process Agent has reported process initialization complete
		// Completing,				// Job processing is completing
		// Completed,				// Job processing is completed
		// Undefined				// None of the above

        switch ( dwj.getJobState() ) {
            case Completing:
            case Completed:
                return true;
            default:
                return false;
        }
    }

    /**
     * We want to be sure the most-recently started instance get resources before 
     * allowing a new start.
     */
    boolean needNextStart(JobState old, JobState current)
    {
        // UIMA-4587 & UIMA-5244
    	String methodName="needNextStart";
    	// Can't do this before we handle the OR publication of "defunct" instances that were running when DUCC was shutdown
    	// They should not be counted as errors when the SM restarts.
 /*       if ( isDeregistered() || !enabled() ) {
            logger.info(methodName, id, "Bypassing instance start because service is unregistered or disabled.");*/
        if ( isDeregistered() ) {
            logger.info(methodName, id, "Bypassing instance start because service is unregistered.");
            return false;
        }

        switch ( old ) {
            case Received:
            case WaitingForDriver:
            case WaitingForServices:
            case WaitingForResources:
                switch (current) {
                    case Assigned:
                    case Initializing:
                    case Running:
                        return true;
                    default:
                        break;
                }
            default:
                break;
        }
        return false;
    }

    void removeImplementor(ServiceInstance si)
    {
    	String methodName = "removeImplementor";
        logger.info(methodName, id, "Removing implementor", si.getId());
        implementors.remove(si.getId());
        // Note, we don't save the instance id because this is only for ping-only services that have no instid
    }

    /**
     * This is one of my service instances.  Update its state and maybe kick the
     * state machine as well.
     TODO: proof this carefully
     */
    synchronized void signalUpdate(DuccWorkJob dwj)
    {
    	String methodName = "signalUpdate";
        ServiceInstance inst = implementors.get(dwj.getDuccId().getFriendly());
        
        if ( inst == null ) {            // he's gone and we don't care any more
            logger.warn(methodName, id, "Process", dwj.getDuccId(), "is no longer an implementor.  Perhaps it exited earlier.");
            return; 
        }

        JobState old_state = inst.getState();
        JobState state = dwj.getJobState();
        DuccId inst_id = dwj.getDuccId();
        long fid = inst_id.getFriendly();

        if ( state == JobState.Running && old_state != JobState.Running ) {
            // running, and wasn't before, we can reset the error counter
            logger.info(methodName, id, "Resetting init error counter from", init_failures, "to 0 on transition from", old_state, "to", state);
            init_failures = 0;
        }

        boolean save_meta = false;

        if ( needNextStart(old_state, state) ) {
            // sequnced startup
            start();
        }

        if ( canDeleteInstance(dwj) ) {
            // State Completed or Completing
            JobCompletionType jct = dwj.getCompletionType();
            ServiceInstance stoppedInstance = null;
            
            logger.info(methodName, this.id, "Removing implementor", fid, "(", key, ") completion", jct);
            stoppedInstance = implementors.remove(fid);          // won't fail, was checked for on entry
            conditionally_stash_instance_id(stoppedInstance.getInstanceId()); // UIMA-4258
            
            // TODO: put history into a better place
            String history = meta_props.getStringProperty(history_key, "");
            history = history + " " + fid;
            meta_props.put(history_key, history);
            save_meta = true;

            logger.info(methodName, id, "Removing stopped instance",  inst_id, "from maps: state[", state, "] completion[", jct, "] service-enabled", enabled());

            clearQueue();        // this won't do anything if it looks like the service is still active somehow

            if ( instances > countImplementors() ) {          // have we fallen short of the nInstances we have to maintain?
                
                // You can stop an instance with the ducc_services CLI, in which case this counts as a manual stop and not
                // an error.  Or the thing can go away for no clear reason, in which case it does as an error, even if somebody
                 // use the DuccServiceCancel API to stop it.
                //
                // TODO: Update the ducc_services CLI to allow stop and restart of specific instances without counting failure.
                if ( stoppedInstance.isStopped() ) {
                    logger.info(methodName, id, "Instance", inst_id, "stopped by SM.  Not restarting.");
                } else {
                    // An instance stopped and we (SM) didn't ask it to - by definition this is failure no matter how it exits.
 
                  // If the RM purges the instance on a node failure the service should not be penalized UIMA-6111
                  if (jct == JobCompletionType.ResourcesUnavailable) {
                    logger.info(methodName, id, "Instance purged by RM (node died?) - prior state[", old_state, 
                            "] current state[", state, "] completion[", jct, "]");
                  } else {
                    switch ( old_state ) {
                        case WaitingForServices:
                        case WaitingForResources:
                        case Initializing:
                        case Assigned:
                            init_failures++;
                            logger.info(methodName, id, "Tally initialization failure:", init_failures);
                            break;
                        case Running:
                            run_failures++;
                            logger.info(methodName, id, "Tally runtime failure", run_failures);
                            break;
                        default:
                            // other states we blow off - we can enter this place a bunch of time a things wind down
                            logger.info(methodName, id, "Instance stopped unexpectedly: prior state[", old_state, " completion[", jct, "]");
                            break;
                    }
                  }

                    if ( excessiveFailures() ) { 
                        String disable_reason = null;
                        if ( excessiveRunFailures ) {
                            logger.warn(methodName, id, "Instance", inst_id, "Monitor signals excessive terminations. Not restarting.");
                            disable_reason = "Excessive runtime errors";
                        } else {
                            logger.warn(methodName, id, "Instance", inst_id,
                                        "Excessive initialization failures. Total failures[" + init_failures + "]",
                                        "allowed [" + init_failure_max + "], not restarting.");
                            disable_reason = "Excessive initialization errors";
                        }
                        disable(disable_reason);
                        save_meta = true;
                    } else {
                        logger.warn(methodName, id, "Instance", inst_id + ": Unsolicited termination, not yet excessive.  Restarting instance.");
                        start();
                        return;         // don't use termination to set state - start will signal the state machine
                    }
                }
            }
        } 

        try {
            if ( save_meta ) updateMetaProperties();
        } catch ( Exception e ) {
            logger.warn(methodName, id, "Error updating meta properties:", e);
        }
        inst.setState(state);            
        signal(inst);
    }

    private ServiceState translateJobState(JobState js)
    {        
        switch ( js ) {
		    case Received:				// Job has been vetted, persisted, and assigned unique Id
		    case WaitingForDriver:		// Process Manager is launching Job Driver
		    case WaitingForServices:	// Service Manager is checking/starting services for Job
		    case WaitingForResources:	// Scheduler is assigning resources to Job
            case Assigned:
                return ServiceState.Starting;
		    case Initializing:			// Process Agents are initializing pipelines
                return ServiceState.Initializing;
		    case Running:				// At least one Process Agent has reported process initialization complete
                return ServiceState.Available;
		    case Completing:			// Job processing is completing
                return ServiceState.Stopping;
		    case Completed:				// Job processing is completed
                return ServiceState.Stopped;
		    default:
                return ServiceState.NotAvailable;  // Should not ever get here. It's a noop if we do.
        }
    }

    /**
     * The MAX of the states of the implementors.
     *   case Available:    return 8;
     *   case Waiting:      return 7;
     *   case Initializing: return 6;
     *   case Starting:     return 5;
     *   case Stopping:     return 4;
     *   case Stopped:      return 3;
     *   case NotAvailable: return 2;
     *   case Undefined:    return 1;
     *   default:           return 0;
     */
    private ServiceState cumulativeJobState()
    {
    	String methodName = "cumulativeJobState";
        ServiceState response = ServiceState.Stopped;

        for ( ServiceInstance si : implementors.values() ) {
            JobState js = si.getState();
            ServiceState translated = translateJobState(js);
            if (  translated.ordinality() > response.ordinality() ) response = translated;
        }
                
        // If there is a pinger, and it isn't pinging, we must not advance beyond the pinger's state.
        // If there is no pinger, we may never advance beyond Waiting
        if ( serviceMeta == null ) {
            response = (response.ordinality() < ServiceState.Waiting.ordinality()) ? response : ServiceState.Waiting;
        } else if ( serviceMeta != null ) {
            logger.trace(methodName, id, "Cumulative before checking monitor/pinger:", response, ".  Monitor state:", serviceMeta.getServiceState());
            if ( serviceMeta.getServiceState().ordinality() <= response.ordinality() ) response = serviceMeta.getServiceState();
        }

        return response;
    }

    synchronized ServiceState getState()
    {
        return service_state;
    }

    synchronized void setState(ServiceState ss)
    {
    	String methodName = "setState";
    	logger.debug(methodName, id, service_state, "==>", ss);
        service_state = ss;
    }
    
    synchronized void setState(ServiceState req_new_state, ServiceState req_cumulative, ServiceInstance si)
    {
        String methodName = "setState";

        String tail = "";
        if ( si == null ) {
            tail = "none/none";            
        } else {
            tail = si.getId() + "/" + si.getState();
        }

        ServiceState prev = this.service_state;
        switch(prev) {
        case Dispossessed:
        	logger.debug(methodName, id, prev);
        	return;
        }
        ServiceState new_state = req_new_state;
        ServiceState cumulative = req_cumulative;
        
        /**
         * If pinger is stale and state is Available then force state to be Waiting
         */
        determinePingerStatus();
        if(notPinging) {
        	switch(new_state) {
        	case Available:
        	case Waiting:	
        		new_state = ServiceState.Waiting;
        		cumulative = new_state;
        		logger.debug(methodName, id, "NotPinging[1]: "+req_new_state+" => "+new_state+"; "+req_cumulative+" => "+cumulative);
        		break;
        	default:
        		switch(cumulative) {
        		case Available:
        			new_state = ServiceState.Waiting;
            		cumulative = new_state;
            		logger.debug(methodName, id, "NotPinging[2]: "+req_new_state+" => "+new_state+"; "+req_cumulative+" => "+cumulative);
        			break;
        		default:
        			logger.debug(methodName, id, "NotPinging[3]: "+req_new_state+" => "+new_state+"; "+req_cumulative+" => "+cumulative);
        			break;
        		}
        		break;
        	}
        }
        else {
        	logger.debug(methodName, id, "Pinging: "+req_new_state+" => "+new_state+"; "+req_cumulative+" => "+cumulative);
        }
        
        this.service_state = new_state;
        if ( prev != new_state ) {
            logger.info(methodName, id, "State update from[" + prev + "] to[" + new_state + "] via[" + cumulative + "] Inst[" + tail + "]" );
            try {
                updateMetaProperties();
            } catch ( Exception e ) {
                logger.warn(methodName, id, "Error updating meta properties:", e);
            }

        }

        // Execute actions that must always occur based on the new state
        // These are all idempotent actions, call them as often as you want and no harm.
        switch(new_state) {            
            case Available: 
                setLastRunnable(System.currentTimeMillis());
                startPingThread();
                break;
            case Initializing:
                break;
            case Starting:
                break;
            case Waiting:
                setLastRunnable(System.currentTimeMillis());
                startPingThread();
                break;
            case Stopping:
                stopPingThread();
                break;
            case Stopped:
                setReferenced(false);
                stopPingThread();
                break;
            default:
                setReferenced(false);
                stopPingThread();
                break;
        }
    }

    public synchronized void signal(ServiceInstance si)
    {
        String methodName = "signal";

        if ( true ) {
            ServiceState cumulative = cumulativeJobState();
            //
            // Note on the CUMULATIVE state: this is the cumulative state as determined by service processes.  If they
            // should all die at once through some temporary glitch the state could go to Unavailable even though the
            // SM would now be in active retry - the states below avoid regression state if CUMULATIVE goes to
            // Unavailable but the retry count indicates retry is still in progress.
            //

            //
            // The ping state is pretty much always the right state.  But if we're
            // not yet pinging we need to see if any of the implementors states 
            // indicates we should be pinging, in which case, start the pinger.
            //
            logger.trace(methodName, id, "serviceState", getState(), "cumulativeState", cumulative);
            switch ( getState() ) {
                // If I'm brand new and something is initting then I can be too.  If something is
                // actually running then I can start a pinger which will set my state.

            	case Dispossessed:
            		return;
                case Available:
                    switch ( cumulative ) {
                        case Starting:
                            logger.warn(methodName, id, "STATE REGRESSION:", getState(), "->", cumulative); // can't do anything about it but complain
                            setState(ServiceState.Starting, cumulative, si);
                            break;

                        case Initializing:
                            // Not immediately clear what would cause this other than an error but let's not crash.
                            logger.warn(methodName, id, "STATE REGRESSION:", getState(), "->", cumulative); // can't do anything about it but complain
                            setState(ServiceState.Initializing, cumulative, si);
                            break;

                        case Available:
                            setState(ServiceState.Available, cumulative, si);
                            break;

                        case Stopping:
                            setState(ServiceState.Stopping, cumulative, si);
                            break;

                        case Stopped:
                            setState(ServiceState.Stopped, cumulative, si);
                            break;

                        case Waiting:
                            setState(ServiceState.Waiting, cumulative, si);
                            break;

                        default:
                            stopPingThread();
                            logger.warn(methodName, id, "ILLEGAL STATE TRANSITION:", getState(), "->", cumulative); 
                            break;

                    }

                    break;

                    // If I'm initting and now something is running we can start a pinger
                case Initializing:
                    switch ( cumulative ) { 
                        case Starting:
                            logger.warn(methodName, id, "STATE REGRESSION:", getState(), "->", cumulative); // can't do anything about it but complain
                            setState(ServiceState.Starting, cumulative, si);
                            break;

                        case Initializing:
                            setState(ServiceState.Initializing, cumulative, si);
                            break;

                        case Available:
                            logger.warn(methodName, id, "UNEXPECTED STATE TRANSITION:", getState(), "->", cumulative); 
                            setState(ServiceState.Waiting, cumulative, si);
                            break;

                        case Stopping:
                            setState(ServiceState.Stopping, cumulative, si);
                            break;

                        case Stopped:
                            setState(ServiceState.Stopped, cumulative, si);
                          break;

                        case Waiting:
                            setState(ServiceState.Waiting, cumulative, si);
                            break;

                        default:
                            logger.warn(methodName, id, "ILLEGAL STATE TRANSITION:", getState(), "->", cumulative); 
                            break;
                    }
                    break;

                    // If I'm initting and now something is running we can start a pinger
                case Starting:
                    switch ( cumulative ) { 
                        case Starting:
                            setState(ServiceState.Starting, cumulative, si);
                            break;

                        case Initializing:
                            setState(ServiceState.Initializing, cumulative, si);
                            break;

                        case Available:
                            setState(ServiceState.Waiting, cumulative, si);
                            break;

                        case Stopping:
                            logger.info(methodName, id, "RETRY RETRY RETRY prevents state regression from Initializing");
                            break;

                        case Stopped:
                            setState(ServiceState.Stopped, cumulative, si);
                          break;

                        case Waiting:
                            logger.warn(methodName, id, "ILLEGAL STATE TRANSITION:", getState(), "->", cumulative); 
                            break;

                    }
                    break;

                case Waiting:                    
                    switch ( cumulative ) {
                        case Starting:
                            logger.warn(methodName, id, "STATE REGRESSION:", getState(), "->", cumulative); // can't do anything about it but complain
                            setState(ServiceState.Starting, cumulative, si);
                            break;

                        case Initializing:
                            logger.warn(methodName, id, "STATE REGRESSION:", getState(), "->", cumulative); // can't do anything about it but complain
                            setState(ServiceState.Initializing, cumulative, si);
                            break;

                        case Available:
                            setState(ServiceState.Available, cumulative, si);
                            break;

                        case Stopping:
                            setState(ServiceState.Stopping, cumulative, si);
                            break;

                        case Stopped:   
                            setState(ServiceState.Stopped, cumulative, si);
                            break;

                        case Waiting:
                            setState(ServiceState.Waiting, cumulative, si);
                            break;

                        default:
                            logger.warn(methodName, id, "ILLEGAL STATE TRANSITION:", getState(), "->", cumulative); 

                            break;

                    }
                    break;

                case Stopping:
                    switch ( cumulative ) {

                        case Starting:
                            setState(ServiceState.Starting, cumulative, si);
                            break;

                        case Initializing:
                            setState(ServiceState.Initializing, cumulative, si);
                            break;


                        case Available:                
                            setState(ServiceState.Available, cumulative, si);
                            break;

                        case Stopped:                
                            setState(ServiceState.Stopped, cumulative, si);
                            break;

                        case Stopping:                
                            setState(ServiceState.Stopping, cumulative, si);
                            break;

                        default:
                            logger.warn(methodName, id, "ILLEGAL STATE TRANSITION:", getState(), "->", cumulative); 
                            break;
                    }
                    break;                    

                case Stopped:
                    // OK
                    // Every transition can happen here because of hot-start of SM
                    switch ( cumulative ) { 
                        case Starting:
                            setState(ServiceState.Starting, cumulative, si);
                            break;

                        case Initializing:
                            setState(ServiceState.Initializing, cumulative, si);
                            break;

                        case Available:
                            setState(ServiceState.Waiting, cumulative, si);
                            break;

                        case Waiting:
                            setState(ServiceState.Waiting, cumulative, si);
                            break;
                        case Stopped:
                            // Trailing OR publications cause this.  Just record it for the log.
                            setState(ServiceState.Stopped, cumulative, si);
                            break;

                        case Stopping:
                            setState(ServiceState.Stopping, cumulative, si);
                            logger.warn(methodName, id, "UNEXPECTED STATE:", getState(), "->", cumulative); 
                            break;

                        case NotAvailable:
                            // junk.  just ignore it
                            logger.warn(methodName, id, "UNEXPECTED STATE:", getState(), "->", cumulative); 
                            break;
                    }
                    break;

                case NotAvailable:
                case Undefined:
                    // OK
                    logger.warn(methodName, id, "Illiegal state", getState(), "Ignored.");
                    break;
            }
        }

    }

    synchronized String getKey()
    {
        return key;
    }

    synchronized int getRunFailures()
    {
        return run_failures;
    }

    /**
     * Analyze failures - either too  many init failures, or the pinger says too many run failures.
     */
    synchronized boolean excessiveFailures()
    {
        String methodName = "excessiveFailures";
        if ( init_failures >= init_failure_max ) {
            logger.trace(methodName, id, "INIT FAILURES EXCEEDED");
            return true;
        } 
        
        if ( excessiveRunFailures ) {
            logger.trace(methodName, id, "EXCESSIVE RUN FAILURES SIGNALLED FROM SERVICE MONITOR.");
            return true;
        }

        return false;
    }

    private void startPingThread()
    {
    	String methodName = "startPingThread";
        if ( serviceMeta != null ) return;         // don't start multiple times.
        if ( inShutdown ) return;              // in shutdown, don't restart

        if ( ping_failures > ping_failure_max ) {
            String msg = "Service stopped as pinger failed to start " + ping_failures + " times.";
            logger.warn(methodName, id, msg);
            meta_props.put(IStateServices.SvcMetaProps.submit_error.pname(), msg);
            disableAndStop(msg);
            return;
        }

        try {
            logger.info(methodName, id, "Starting service monitor.");
            serviceMeta = new PingDriver(this);
        } catch ( Throwable t ) {
            logger.error(methodName, id, "Cannot instantiate service pinger.", t);
            return;
        }

        //setState(ServiceState.Waiting);
        Thread t = new Thread(serviceMeta);
        t.setName("Pinger-" + id.getFriendly());
        t.start();
    }

    synchronized void pingExited(int rc, PingDriver which_meta)
    {
        String methodName = "pingExited";

        logger.info(methodName, id, "Service Monitor/Pinger exits, rc", rc);
        if ( which_meta == serviceMeta ) {
            serviceMeta = null;
        } // otherwise, it was already removed by some intrepid unit

        if ( rc != 0 ) {
            ++ping_failures;
            logger.warn(methodName, id, "Ping exited with failure, total failures:", ping_failures);

            if ( isPingOnly() && (ping_failures > ping_failure_max) ) {
                logger.warn(methodName, id, "Stopping ping-only service due to excessive falutes:", ping_failure_max);
                meta_props.put(IStateServices.SvcMetaProps.submit_error.pname(), "Stopping ping-only service due to excessive falutes: " + ping_failure_max);

                stop(-1L);        // must be -lL Long to get the right overload
                implementors.remove(-1L);
            } 
        }

    }

    synchronized void stopMonitor()
    {
        String methodName = "stopMonitor";
        logger.info(methodName, id, "Stopping pinger due to shutdown");
        inShutdown = true;
        stopPingThread();
    }

    public synchronized void stopPingThread()
    {
        String methodName = "stopPingThread";

        if ( serviceMeta != null ) {
            logger.info(methodName, id, "Stopping monitor/ping thread for", key);
            serviceMeta.stop();
            serviceMeta = null;
        }

        try {
            updateMetaProperties();
        } catch ( Exception e ) {
            logger.warn(methodName, id, "Error updating meta properties:", e);
        }
    }

    void log_text(String logdir, String text)
    {
    	String methodName = "log_text";
        String[] args = {
            System.getProperty("ducc.agent.launcher.ducc_spawn_path"),
            "-u",
            user,
            "-f",
            logdir + "/service.err.log",
            "-a",
            "--",
            text
        };

        ProcessBuilder pb = new ProcessBuilder(args);
        pb.redirectOutput(new File("/dev/null"));
        pb.redirectError(new File("/dev/null"));
        try {
            Process p = pb.start();
            int rc = p.waitFor();
            if ( rc != 0 ) {
                logger.warn(methodName, id, "Attempt to update user's service.err.log returns with rc ", rc);
            }
        } catch (Throwable t) {
            logger.warn(methodName, id, "Cannot update user's service.err.log:", t);
        } 

    }

    void log_errors(List<String> outlines, List<String> errlines)
    {

        Date date= new Date();
        String ts = (new Timestamp(date.getTime())).toString();

        String logdir = job_props.getProperty(UiOption.LogDirectory.pname());

        StringBuffer buf = new StringBuffer();

        // header
        buf.append("==========");
        buf.append(" Instance Startup Failure (stdout) ");
        buf.append(ts);
        buf.append(" ========================================\n");

        // stdout
        for ( String s : outlines ) {
            buf.append(s);
            buf.append("\n");
        }
        log_text(logdir, buf.toString());
        
        buf = new StringBuffer();
        buf.append("----------");
        buf.append("(stderr) ");
        buf.append(ts);
        buf.append(" ----------------------------------------\n");
        for ( String s : errlines ) {
            buf.append(s);
            buf.append("\n");
        }
        buf.append("==========");
        buf.append(" End Startup Failure ");
        buf.append(ts);
        buf.append(" ========================================\n");
        log_text(logdir, buf.toString());
    }

    /**
     * See if there is an instance ID to reuse - if so, we need the lowest one.  
     * If not, assign the next on in sequence.
     *
     * This maintains the property that all instances sequential from 0 to max are either 
     * already assigned, or on the available_instance_ids tree.  Thus, if it's not on the tree,
     * we can find the next one by taking the lenght of the implementors structure.
     *
     * The reason we need to remember this is because pingers are allowed to stop specific
     * instances.  As well, specific instances may croak.  We always want to restart with the
     * lowest available instance if we have to reuse ids.
     */
    synchronized int find_next_instance()
    {
        int ret = implementors.size();
        if ( available_instance_ids.size() > 0 ) {
            ret = available_instance_ids.firstKey();
            available_instance_ids.remove(ret);
        }
        return ret;
    }

    /**
     * Save the id for possible reuse.
     *
     * It's an error, albeit non-fatal, if the instance is already stashed.
     * Can happen if an instance dies while  being stopped explicitly
     * UIMA-4258
     */
    synchronized void stash_instance_id(int instid)
    {
        /* UIMA-5244 No need to warn as is expected.
    	String methodName = "stash_intance_id";
        if ( available_instance_ids.containsKey(instid) ) {
            try {
                // put a scary marker in the log
                throw new Exception("Duplicate instance id found: " + instid);
            } catch (Exception e) {
                logger.warn(methodName, id, e);
            }
            return;
        }
        */

        available_instance_ids.put(instid, instid);
    }

    /**
     * Save the id for possible reuse, if it hasn't already been saved.
     * This is called when we see a state change that indicates a service has exited.  Usually we
     * hope this is because it was stopped, in which case we already stashed the id.  But if it
     * crashed it may not be stashed yet, so we do it here.
     * UIMA-4258
     */
    synchronized void conditionally_stash_instance_id(int instid)
    {
        if ( available_instance_ids.containsKey(instid) ) {           // Is this necessary? Could simply let the put replace
            return;
        }
        stash_instance_id(instid);        
    }

    synchronized void start()
    {
    	String methodName = "start";

        // UIMA-4587 & UIMA-5244
    	// Can't do this yet
/*        if ( isDeregistered() || !enabled()) {
            logger.info(methodName, id, "Bypass start because service is unregistered or disabled.");*/
    	if ( isDeregistered()) {
    	    logger.info(methodName, id, "Bypass start because service is unregistered.");
            return;
        }

        if ( countImplementors() >= instances ) {
            return;
        }

        if ( isPingOnly() ) {
            if ( implementors.containsKey(-1l) ) {
                logger.info(methodName, id, "PING_ONLY: already started.");
                return;
            }

            ServiceInstance si = new PingOnlyServiceInstance(this);
            si.setId(-1L);
            si.setUser(this.user);
            implementors.put(-1l, si);
            handler.addInstance(this, si);
            si.start(null, null);
            signal(si);
        } else {

            if ( isDebug() ) {
                if ( countImplementors() > 0 ) {
                    logger.warn(methodName, id, "Ignoring start of additional instances because process_debug is set.");
                    return;         // only one, in debug
                }
            }

            ServiceInstance si = new ServiceInstance(this);
            si.setInstanceId(find_next_instance());
            long inst_ducc_id = -1L;
            logger.info(methodName, id, "Starting instance. Current count", countImplementors(), "needed", instances);
            if ( (inst_ducc_id = si.start(job_props, meta_props)) >= 0 ) {
                implementors.put(inst_ducc_id, si);
                handler.addInstance(this, si);
                signal(si);
                logger.info(methodName, id, "Instance[", countImplementors(), "] ducc_id ", inst_ducc_id);
            } else {
                logger.info(methodName, id, "Instance[", countImplementors(), "] ducc_id ", inst_ducc_id, "Failed to start.");
                disable("Cannot submit service process");
                signal(si);
            }
        }    

        try {
            updateMetaProperties();
        } catch ( Exception e ) {
            logger.warn(methodName, id, "Error updating meta properties:", e);
        }        
    }

    /**
     * Stop a specific instance.
     */
    synchronized int stop(Long iid)
    {
        String methodName = "stop(id)";

        logger.info(methodName, id, "Stopping specific instance", iid);

        ServiceInstance si = implementors.get(iid);
        if ( si == null ) {
            logger.warn(methodName, id, "Can't find instance", iid, ", perhaps it's already gone.");
            return 0;
        } else {
            si.stop();
            stash_instance_id(si.getInstanceId());         // UIMA-4258
            signal(si);
            return 1;
        }
    }

    /**
     * Stop 'count' services.
     */
    synchronized void stop(int count)
    {
        String methodName = "stop(count)";

        logger.info(methodName, id, "Stopping", count, "implementors");

        Long[] keys = implementors.keySet().toArray(new Long[implementors.size()]);
        Arrays.sort(keys);
        for ( int i = 0, j = keys.length-1; i < count; i++, j-- ) {
            Stopper s = new Stopper(implementors.get(keys[j]));
            new Thread(s).start();
        }
    }

    synchronized void stopAll()
    {
        stop(implementors.size());
    }

    /**
     * Make the thing stop and not restart.
     */
    synchronized void disableAndStop(String reason)
    {
        disable(reason);
        stopAll();
        // instances = 0;     // Is this needed to ensure no more are started? UIMA-5244
    }

    // /**
    //  * Stop everything
    //  */
    // synchronized void stop()
    // {
    //     // TODO
    //     // change state to Stopping and spawn stop threads for all implementors
    //     for ( ServiceInstance si : implementors.values() ) {
    //         Stopper s = new Stopper(si);
    //         new Thread(s).start();
    //     }
    // }

    private class LingerTask
        extends TimerTask
    {
        //ServiceSet sset;
        //LingerTask(ServiceSet sset)
        LingerTask()
        {        
            String methodName = "LingerTask.init";
            logger.debug(methodName, id, "Linger starts", linger_time);
            //this.sset = sset;
        }

        public void run()
        {
            String methodName = "LingerTask.run";
            logger.info(methodName, id, "Linger time reached ... stopping all instances.");
            // doesn't matter how its started i think, we have to set this flag off when we stop
            linger = null;
            setReferenced(false);
            stopAll();
        }
    }

    void lingeringStop()
    {
        if ( timer == null ) {
            timer = new Timer();
        }
        //linger = new LingerTask(this);
        linger = new LingerTask();
        timer.schedule(linger, linger_time);
    }

    IServiceDescription query()
    {
        IServiceDescription sd = new ServiceDescription();
        
        ArrayList<Long> impls = new ArrayList<Long>();
        ArrayList<Integer> instids = new ArrayList<Integer>();
        for ( Long id : implementors.keySet() ) {
            // UIMA-4258 Add instance id to ducc id when saving
            ServiceInstance inst = implementors.get(id);            
            impls.add(id);
            instids.add(inst.getInstanceId());
        }
        sd.setImplementors(impls, instids);

        ArrayList<Long> ref = new ArrayList<Long>();
        ref.clear();
        for ( DuccId id : references.keySet() ) {
            ref.add(id.getFriendly());
        }
        sd.setReferences(ref);

        sd.setInstances(getNInstancesRegistered());

        sd.setType(service_type);
        sd.setSubclass(service_class);
        sd.setEndpoint(endpoint);
        sd.setBroker(broker);
        sd.setServiceState(getState());
        sd.setActive(serviceMeta != null);
        sd.setEnabled(enabled());
        sd.setAutostart(isAutostart());
        sd.setLinger(linger_time);
        sd.setId(id.getFriendly());
        sd.setUser(user);
        sd.setDisableReason(meta_props.getStringProperty(IStateServices.SvcMetaProps.disable_reason.pname(), null));
        sd.setLastUse(last_use);
        sd.setLastPing(last_ping);            // UIMA-4309
        sd.setLastRunnable(last_runnable);    // UIMA-4309
        sd.setRegistrationDate(meta_props.getStringProperty(IStateServices.SvcMetaProps.registration_date.pname(), ""));
        sd.setReferenceStart(reference_start);
        sd.setErrorString(meta_props.getStringProperty(IStateServices.SvcMetaProps.submit_error.pname(), null));

        if ( serviceMeta != null ) {
            sd.setQueueStatistics(serviceMeta.getServiceStatistics());
        }

        return sd;
    }


    /**
     * For debugging, so it's easier to identify this guy in the eclipse debugger.
     */
    public String toString()
    {
        return endpoint;
    }

    class Stopper
        implements Runnable
    {
        ServiceInstance si;
        Stopper(ServiceInstance si)
        {
            this.si = si;
        }

        public void run() {
            si.stop();
            stash_instance_id(si.getInstanceId());         // UIMA-4258
        }
    }


}

/**
    For reference

	public enum JobState {
		Received,				// Job has been vetted, persisted, and assigned unique Id
		WaitingForDriver,		// Process Manager is launching Job Driver
		WaitingForServices,		// Service Manager is checking/starting services for Job
		WaitingForResources,	// Scheduler is assigning resources to Job
        Assigned,               // Resources assgned, job not yet started.
		Initializing,			// Process Agents are initializing pipelines
		Running,				// At least one Process Agent has reported process initialization complete
		Completing,				// Job processing is completing
		Completed,				// Job processing is completed
		Undefined				// None of the above
	};
*/