/*
 * 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.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import org.apache.uima.ducc.common.utils.DuccLogger;
import org.apache.uima.ducc.common.utils.DuccProperties;
import org.apache.uima.ducc.common.utils.id.DuccId;
import org.apache.uima.ducc.transport.event.ServiceModifyEvent;
import org.apache.uima.ducc.transport.event.ServiceQueryEvent;
import org.apache.uima.ducc.transport.event.ServiceQueryReplyEvent;
import org.apache.uima.ducc.transport.event.ServiceReplyEvent;
import org.apache.uima.ducc.transport.event.ServiceStartEvent;
import org.apache.uima.ducc.transport.event.ServiceStopEvent;
import org.apache.uima.ducc.transport.event.ServiceUnregisterEvent;
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.common.IDuccWork;
import org.apache.uima.ducc.transport.event.sm.IServiceDescription;
import org.apache.uima.ducc.transport.event.sm.ServiceDependency;
import org.apache.uima.ducc.transport.event.sm.ServiceMap;


public class ServiceHandler
    implements SmConstants, 
               Runnable
{
    /**
	 * 
	 */
	private static final long serialVersionUID = 1L;
	private DuccLogger logger = DuccLogger.getLogger(ServiceHandler.class.getName(), COMPONENT_NAME);	
    private IServiceManager serviceManager;

    private ServiceStateHandler serviceStateHandler = new ServiceStateHandler();
	private ServiceMap serviceMap = new ServiceMap();       // note this is the sync object for publish

    private Map<DuccId, IDuccWork> newJobs = new HashMap<DuccId, IDuccWork>();
    private Map<DuccId, IDuccWork> newServices = new HashMap<DuccId, IDuccWork>();
    
    private Map<DuccId, IDuccWork> deletedJobs = new HashMap<DuccId, IDuccWork>();
    private Map<DuccId, IDuccWork> deletedServices = new HashMap<DuccId, IDuccWork>();
    
    private Map<DuccId, IDuccWork> modifiedJobs = new HashMap<DuccId, IDuccWork>();
    private Map<DuccId, IDuccWork> modifiedServices = new HashMap<DuccId, IDuccWork>();

    private List<ApiHandler> pendingRequests = new LinkedList<ApiHandler>();
    private Object stateUpdateLock = new Object();

    public ServiceHandler(IServiceManager serviceManager)
    {
        this.serviceManager = serviceManager;        
    }

    public synchronized void run()
    {
    	String methodName = "run";
        while ( true ) {
            try {
				wait();
			} catch (InterruptedException e) {
				logger.error(methodName, null, e);
			}

            try {
                runCommands();           // enqueued orders that came in while I was away
                processUpdates();
            } catch (Throwable t) {
                logger.error(methodName, null, t);
            }
        }
    }

    /**
     * At boot only ... pass in the set of all known active services to each service so it can update
     * internal state with current published state.
     */
    void synchronizeImplementors(Map<DuccId, JobState> servicemap)
    {
        ArrayList<String> keys = serviceStateHandler.getServiceNames();
        for ( String k : keys ) {
            ServiceSet sset = serviceStateHandler.getServiceByName(k);
            sset.synchronizeImplementors(servicemap);
        }
    }

    void processUpdates()
    {
    	String methodName = "processUpdates";
        logger.info(methodName, null, "Processing updates.");
        Map<DuccId, IDuccWork> deletedJobsMap      = new HashMap<DuccId, IDuccWork>();
        Map<DuccId, IDuccWork> modifiedJobsMap     = new HashMap<DuccId, IDuccWork>();
        Map<DuccId, IDuccWork> newJobsMap          = new HashMap<DuccId, IDuccWork>();
        Map<DuccId, IDuccWork> deletedServicesMap  = new HashMap<DuccId, IDuccWork>();
        Map<DuccId, IDuccWork> modifiedServicesMap = new HashMap<DuccId, IDuccWork>();
        Map<DuccId, IDuccWork> newServicesMap      = new HashMap<DuccId, IDuccWork>();

        synchronized(stateUpdateLock) {
                deletedJobsMap.putAll(deletedJobs);
                deletedJobs.clear();
                        
                modifiedJobsMap.putAll(modifiedJobs);
                modifiedJobs.clear();
                       
                deletedServicesMap.putAll(deletedServices);
                deletedServices.clear();
                        
                modifiedServicesMap.putAll(modifiedServices);
                modifiedServices.clear();
            
                newServicesMap.putAll(newServices);
                newServices.clear();
                        
                newJobsMap.putAll(newJobs);
                newJobs.clear();
        }

        // We could potentially have several updates where a service or arrives, is modified, and then deleted, while
        // we are busy.  Need to handle them in the right order.  
        //
        // Jobs are dependent on services but not the other way around - I think we need to handle services first,
        // to avoid the case where something is dependent on something that will exist soon but doesn't currently.
        handleNewServices     (newServicesMap     );
        handleModifiedServices(modifiedServicesMap);
        handleDeletedServices (deletedServicesMap );
        handleImplicitServices(                   );

        handleNewJobs         (newJobsMap         );
        handleModifiedJobs    (modifiedJobsMap    );
        handleDeletedJobs     (deletedJobsMap     );

        serviceManager.publish(serviceMap);

        List<ServiceSet> regsvcs = serviceStateHandler.getRegisteredServices();
        for ( ServiceSet sset : regsvcs ) {
            sset.enforceAutostart();
        }
    }

    void signalUpdates( // This is the incoming or map, with work split into categories.
                                     // The incoming maps are volatile - must save contents before returning.
                                    HashMap<DuccId, IDuccWork> newJobs, 
                                    HashMap<DuccId, IDuccWork> newServices,                               
                                    HashMap<DuccId, IDuccWork> deletedJobs,
                                    HashMap<DuccId, IDuccWork> deletedServices,                                    
                                    HashMap<DuccId, IDuccWork> modifiedJobs,
                                    HashMap<DuccId, IDuccWork> modifiedServices
                                    )
    {

        synchronized(stateUpdateLock) {
            this.newJobs.putAll(newJobs);            
            this.newServices.putAll(newServices);            
            this.deletedJobs.putAll(deletedJobs);            
            this.deletedServices.putAll(deletedServices);
            this.modifiedJobs.putAll(modifiedJobs);
            this.modifiedServices.putAll(modifiedServices);
        }
        synchronized(this) {
            notify();
        }
    } 

    void runCommands()
    {
        String methodName = "runCommands";
        LinkedList<ApiHandler> tmp = new LinkedList<ApiHandler>();
        synchronized(pendingRequests) {
            tmp.addAll(pendingRequests);
            pendingRequests.clear();
        }
        logger.info(methodName, null, "Running", tmp.size(), "API Tasks.");
        for ( ApiHandler apih : tmp ) {
            apih.run();
        }
    }

    void addApiTask(ApiHandler apih)
    {
        synchronized(pendingRequests) {
            pendingRequests.add(apih);
        }
    }

    /**
     * Resolves state for the job in id based on the what it is dependent upon - the independent services
     */
    protected void resolveState(DuccId id, ServiceDependency dep)
    {        
        String methodName = "resolveState";
        Map<String, ServiceSet> services = serviceStateHandler.getServicesForJob(id);
        if ( services == null ) {
            dep.setState(ServiceState.NotAvailable);       // says that nothing i need is available
            return;
        }

        ServiceState state = ServiceState.Available;              
        //
        // Start with the most permissive state and reduce it as we walk the list
        // Running > Initializing > Waiting > NotAvailable
        //
        // This sets the state to the min(all dependent service states)
        //
        for ( ServiceSet sset : services.values() ) {
            if ( sset.getServiceState().ordinality() < state.ordinality() ) state = sset.getServiceState();
             dep.setIndividualState(sset.getKey(), sset.getServiceState());
             logger.debug(methodName, id, "Set individual state", sset.getServiceState());
        }
        dep.setState(state);
    }

    /**
     * This is called when an endpoint is referenced as a dependent service from a job or a service.
     * It is called only when a new job or service is first discovred in the OR map.
     */
    protected Map<String, ServiceSet> resolveDependencies(DuccWorkJob w, ServiceDependency s)
    {
    	String methodName = "resolveDependencies";
    	DuccId id = w.getDuccId();
        String[] deps = w.getServiceDependencies();

        // New services, if any are discovered
        boolean fatal = false;
        Map<String, ServiceSet> jobServices = new HashMap<String, ServiceSet>();
        for ( String dep : deps ) {
            
            // put it into the global map of known services if needed and up the ref count
            ServiceSet sset = serviceStateHandler.getServiceByName(dep);                            
            if ( sset == null ) {                              // first time, so it's by reference only
                try {
                    sset = new ServiceSet(dep, serviceManager.newId());
                    serviceStateHandler.putServiceByName(dep, sset);
                } catch ( Exception e ) {       // if 'dep' is invalid, or we can't get a duccid,  we throw
                    s.addMessage(dep, e.getMessage());
                    s.setState(ServiceState.NotAvailable);
                    fatal = true;
                    continue;
                }
            }

            if ( sset.isDeregistered() ) { 
                // Registerered services only - the service might even still be alive because it can
                // take a while to get rid of these guys - we need to be sure we don't attach any
                // new jobs to it.
                s.addMessage(dep, "Independent registered service [" + dep + "] has been deregistered and is terminating.");
                s.setState(ServiceState.NotAvailable);
                fatal = true;
                continue;
            }

            //
            // We try to vet all services so the message is complete.  If we've already had some fatal problems
            // we need to bypass any attempt to cope with registered services or updating the sset.
            //
            if ( ! fatal ) {
                if ( sset.isRegistered() && (sset.countImplementors() == 0) && sset.isStartable() ) {
                    // Registered but not alive, well, we can fix that!
                    int ninstances = sset.getNInstances();
                    logger.debug(methodName, sset.getId(), "Reference-starting registered service, instances =", ninstances);
                    if ( ! sset.isAutostart() ) {         // must avoid races ith autostart for referenced start
                        sset.setReferencedStart(true);
                    }
                    for ( int i = 0; i < ninstances; i++ ) {
                        if ( ! sset.start() ) {
                            s.addMessage(dep, "Can't start independent service.");
                            s.setState(ServiceState.NotAvailable);
                            break;
                        }
                    }
                }
                
                jobServices.put(dep, sset);
                sset.reference(id);
                serviceStateHandler.putServiceForJob(w.getDuccId(), sset);
                logger.debug(methodName, id, "Service init ok. Ref[", dep, "] incr to", sset.countReferences());
            }
        }

        if ( fatal ) {
            jobServices.clear();            
        }
        return jobServices;
    }
 
    protected void handleNewJobs(Map<DuccId, IDuccWork> work)
    { 
        String methodName = "handleNewJobs";

        // Map of updates to send to OR
        HashMap<DuccId, ServiceDependency> updates = new HashMap<DuccId, ServiceDependency>();

        for ( DuccId id : work.keySet() ) {
            DuccWorkJob w = (DuccWorkJob) work.get(id);

            if ( !w.isActive() ) {
                logger.info(methodName, id, "Bypassing inactive job, state =", w.getStateObject());
                continue;
            }

            ServiceDependency s = new ServiceDependency(); // for the OR
            updates.put(id, s);

            String[] deps = w.getServiceDependencies();
            if ( deps == null ) {   // no deps, just mark it running and move on
                s.setState(ServiceState.Available);
                logger.info(methodName, id, "Added to map, no service dependencies.");
                continue;
            }

            //
            // Get dependency references and fire up their state machines
            //
            Map<String, ServiceSet> jobServices = resolveDependencies(w, s);
            for ( ServiceSet sset : jobServices.values() ) {
                sset.establish();
            }
            resolveState(id, s);
            logger.info(methodName, id, "Added job to map, with service dependency state.", s.getState());
        }

        serviceMap.putAll(updates);
    }

    /**
     * A job or service has ended.  Here's common code to clean up the dependent services.
     * @param id - the id of the job or service that stopped
     * @param deps - the services that 'id' was dependent upon
     */
    protected void stopDependentServices(DuccId id)
    {
    	String methodName = "stopDependentServices";

        Map<String, ServiceSet> deps = serviceStateHandler.getServicesForJob(id);
        if ( deps == null ) {
            logger.debug(methodName, id, "No dependent services to stop, returning.");
            return;                                              // service already deleted, timing issue
        }

        //
        // Bop through all the things job 'id' is dependent upon, and update their refcounts. If
        // the refs go to 0 we stop the pinger and sometimes the independent service itself. 
        //
        for ( String dep : deps.keySet() ) {
            logger.debug(methodName, id, "Looking up service", dep);
            
            ServiceSet sset = deps.get(dep);
            if ( sset == null ) {
                throw new IllegalStateException("Null service for " + dep);      // sanity check, should never happen
            }
            
            int count = sset.dereference(id);                                    // also maybe stops the pinger
            logger.debug(methodName, id, "Ref count for", sset.getKey(), "goes down to", count);
            if ( count == 0 ) {
                if ( sset.isImplicit() ) {
                    logger.debug(methodName, id, "Removing unreferenced implicit service", dep, "refcount", count);
                    serviceStateHandler.removeService(dep);
                }
                if ( sset.isRegistered() && sset.isReferencedStart() ) {
                    if ( sset.isAutostart() ) {            // could have happened after the reference
                        sset.setReferencedStart(false);    // so we don't linger, just reset
                    } else {
                        logger.debug(methodName, id, "Stopping reference-started service", dep, "refcount", count);
                        sset.lingeringStop();
                    }
                }

            }
        }

        // last, indicate that job 'id' has nothing its dependent upon any more
        serviceStateHandler.removeServicesForJob(id);            
    }

    protected void handleDeletedJobs(Map<DuccId, IDuccWork> work)
    {
        String methodName = "handleCompletedJobs";

        for ( DuccId id : work.keySet() ) {
            DuccWorkJob w = (DuccWorkJob) work.get(id);
            
            String[] deps = w.getServiceDependencies();
            if ( deps == null ) {   // no deps, just mark it running and move on
                logger.info(methodName, id, "No service dependencies, no updates made.");
                continue;
            }

            stopDependentServices(id);

            logger.info(methodName, id, "Deleted job from map");
        }

        serviceMap.removeAll(work.keySet());
    }

    protected void handleModifiedJobs(Map<DuccId, IDuccWork> work)
    {
        String methodName = "handleModifiedJobs";

        //
        // Only look at active jobs.  The others will be going away soon and we use
        // that time as a grace period to keep the management machinery running in 
        // case more work comes in in the next few minutes.
        //
        // Everything is already in the service map so we just update the state.
        //
        for ( DuccId id : work.keySet() ) {
            
            DuccWorkJob j = (DuccWorkJob) work.get(id);
            String[] deps = j.getServiceDependencies();
            if ( deps == null ) {   // no deps, just mark it running and move on
                logger.info(methodName, id, "No service dependencies, no updates made.");
                continue;
            }

            ServiceDependency s = serviceMap.get(id);
            if ( j.isFinished() ) {
                stopDependentServices(id);
                s.setState(ServiceState.NotAvailable);
                s.clearMessages();
            } else  if ( j.isActive() ) {
                resolveState(id, s);
            } 
        }

    }

    protected void handleNewServices(Map<DuccId, IDuccWork> work)
    {
        String methodName = "handleNewServices";

        Map<DuccId, ServiceDependency>  updates = new HashMap<DuccId, ServiceDependency>();   // to be added to the service map sent to OR
        Map<String, ServiceSet>     newservices = new HashMap<String, ServiceSet>();          // to be added to our internal maps in serviceState
        for ( DuccId id : work.keySet() ) {
            DuccWorkJob w = (DuccWorkJob) work.get(id);

            //
            // On restart we sometimes get stale stuff that we just ignore.
            // What else? Is the the right thing to do?
            //  
            if ( !w.isActive() ) {
                logger.info(methodName, id, "Bypassing inactive service, state=", w.getStateObject());
                continue;
            }

            ServiceDependency s = new ServiceDependency();
            updates.put(id, s);

            String endpoint = w.getServiceEndpoint();
            if ( endpoint == null ) {                                     // the job is damaged if this happens
                String msg = "No service endpoint.  Service cannot be validated.";
                logger.warn(methodName, id, msg);
                s.addMessage("null", msg);                                // this is a fatal state always
                s.setState(ServiceState.NotAvailable);
                continue;
            }

            String[] deps = w.getServiceDependencies();                  // other services this svc depends on
            ServiceSet sset = serviceStateHandler.getServiceByName(endpoint);
            if ( sset == null ) {
                // submitted, we just track but not much else
                try {
                    sset = new ServiceSet(serviceManager.newId(), id, endpoint, deps);             // creates a "submitted" service
                    sset.addImplementor(id, w.getJobState());
                    serviceStateHandler.putServiceByName(endpoint, sset);
                } catch ( Exception e ) {
                    s.addMessage(endpoint, e.getMessage());
                    s.setState(ServiceState.NotAvailable);
                    continue;
                }
            } else if ( sset.isDeregistered() ) {
                s.addMessage(endpoint, "Duplicate endpoint: terminating deregistered service.");
                s.setState(ServiceState.NotAvailable);
                continue;
            } else if ( sset.matches(id) ) {
                // TODO: not clear we have to do anything here since establish() below will
                //       add to the implementors.  Be sure to update the check so the
                //       code in the following 'else' clause is executed correctly though.

                // and instance/implementor of our own registered services
                sset.addImplementor(id, w.getJobState());
            } else { 
                //
                // If the new service is not a registered service, and it is a duplicate of another service
                // which isn't registered, we allow it to join the party.
                //
                // When it joins, it needs to "propmote" the ServiceSet to "Submitted".
                //
                // a) in the case of "implicit" we don't know enough to many any moral judgements at all
                // b) in the case of "submitted" it could be the user is increasing the pool of servers by
                //    submitting more jobs.  Perhaps we would better handle this via modify but for the moment,
                //    just allow it.
                // c) in the case of "registered" we know and manage everything and don't allow it.  users must
                //    use the services modify api to increase or decrease instances.
                //

                if ( !sset.isRegistered() ) {
                    sset.addImplementor(id, w.getJobState());
                    sset.promote();          // we'll do this explicitly as a reminder that it's happening and
                                             // to insure we NEVER promote a registered service (which is actually
                                             // a demotion!).
                } else {
                    String msg = "Duplicate endpoint: Registered service.";
                    logger.warn(methodName, id, msg);
                    s.addMessage(endpoint, msg);
                    s.setState(ServiceState.NotAvailable);
                    continue;
                }
            }

            // The service is new and unique if we get this far

            //
            // No deps.  Put it in the map and move on.
            //
            if ( deps == null ) {                
                logger.info(methodName, id, "Added service to map, no service dependencies. ");
                s.setState(ServiceState.Available);                        // good to go in the OR (the state of things i'm dependent upon)
                sset.establish(id, w.getJobState());                       // sets my own state based entirely on state of w
                continue;
            }

            Map<String, ServiceSet> jobServices = resolveDependencies(w, s); // 
            for ( ServiceSet depset : jobServices.values() ) {
                depset.establish();
            }
            resolveState(id, s);
            sset.establish(id, w.getJobState());
            logger.info(methodName, id, "Added to map, with service dependencies,", s.getState());
        }

        serviceStateHandler.recordNewServices(newservices);
        serviceMap.putAll(updates);
    }

    //
    // We're here because we got OR state for the service that it has stopped running.
    // Must clean up.
    //
    protected void handleDeletedServices(Map<DuccId, IDuccWork> work)
    {
        String methodName = "handleDeletedServices";

        for ( DuccId id : work.keySet() ) {
        	DuccWorkJob w = (DuccWorkJob) work.get(id);
        	String endpoint = w.getServiceEndpoint();
            logger.info(methodName, id, "Deleted service:", endpoint);
            
            // 
            // Dereference and maybe stop the services I'm dependent upon
            //
            if ( w.getServiceDependencies() == null ) { 
                logger.info(methodName, id, "No service dependencies to update on removal.");
            } else {
                stopDependentServices(id);        // update references, remove implicit services if any
            }

            if (endpoint == null ) {              // probably impossible but lets not chance NPE
                logger.warn(methodName, id, "Missing service endpoint, ignoring.");
                continue;
            }
            ServiceSet sset = serviceStateHandler.getServiceByName(endpoint);            

            // may have been removed already if we saw it go to complet[ed/ing] and it lingered a while anyway, which is usual
            if ( sset != null ) {                
                sset.removeImplementor(id);                      // also stops the ping thread if it's the last one
            }
        }

        //serviceStateHandler.removeServicesForJobs(work.keySet());   // services we were dependent upon
        serviceMap.removeAll(work.keySet());                          // and finally the deleted services
                                                                      // from the published map
    }
    
    /**
     * The pinger may have died while we weren't looking.  Registered services take care
     * of themselves from handleModifiedServices, but we know very little about implicit
     * services so we walk them all and make their ServiceSet keep them clean.
     */
    protected void handleImplicitServices()
    {
        ArrayList<String> keys = serviceStateHandler.getServiceNames();
        for ( String k : keys ) {
            ServiceSet sset = serviceStateHandler.getServiceByName(k);
            if ( sset.isImplicit() ) {
                sset.establish();
            }
        }
    }

    protected void handleModifiedServices(Map<DuccId, IDuccWork> work)
    {
        String methodName = "handleModifiedServices";        
        
        //
        // This is a specific service process, but not necessarily the whole service.
        //
        for ( DuccId id : work.keySet() ) {
            DuccWorkJob w = (DuccWorkJob) work.get(id);
            String endpoint = w.getServiceEndpoint();

            if (endpoint == null ) {              // probably impossible but lets not chance NPE
                logger.info(methodName, id, "Missing service endpoint, ignoring.");
                continue;
            }

            ServiceSet sset = serviceStateHandler.getServiceByName(endpoint);
            if ( sset == null ) {
                // may have already died and this is just leftover OR publications.
                if ( w.isActive() ) {             // or maybe we just screwed up!
                    logger.info(methodName, id, "Got update for active service instance", id.toString(), "but no ServiceSet! Job state:", w.getJobState());
                    continue;
                }
                continue;
            }
            
            if ( !sset.containsImplementor(id) ) {
                logger.info(methodName, id, "Bypassing removed service instance for", endpoint);
                continue;
            }
        
            ServiceDependency s = serviceMap.get(id);
            if ( w.isFinished() ) {              // nothing more, just dereference and maybe stop stuff I'm dependent upon
                stopDependentServices(id);
                s.setState(ServiceState.NotAvailable);              // tell orchestrator
            } else  if ( w.getServiceDependencies() != null ) {     // update state from things I'm dependent upon
                resolveState(id, s);
            } 

            // See what happened to the instance ...
            if (  w.isActive() ) {
                // Hard to know for sure, if there are a bunch of instances, some working and some not, how to manage this.
                // But this is a state *change* of something, and the something is active, so probably the service is OK now
                // if it hadn't been before.
                
                // Need to be cautious here - this will get reset if ANYthing is running.  So we could have a bunch
                // of live instances and some new ones, where the live ones are ok but for some reason we can't start
                // new ones, in which case this gets set too often.
                //
                // This seems like it would be rare and since we aren't actually pounding restarts (only attempts every
                // SM cycle) maybe its ok.  The alternative is to track state changes which is added complexity - for
                // waht gain, we need to determine with experience.
                //
                // I suppose the ServiceManagerHandler could easily track the per-process state change - we'd have to
                // modify the thing in the map it passes in to show 'before' and 'after' states instead of just passing
                // in the DuccWork thing.
                //
                JobState          state = w.getJobState();
                if ( state == JobState.Running ) {         // only if we confirm it's alive
                    sset.resetRunFailures();
                }
            } else {
                JobState          state = w.getJobState();
                
                if ( state == JobState.Completed ) {
                    sset.removeImplementor(id);
                    JobCompletionType jct = w.getCompletionType();
                        
                    logger.info(methodName, id, "Removing stopped instance from maps: state[", state, "] completion[", jct, "]");
                    switch ( jct ) {
                        case EndOfJob:
                        case CanceledByUser:
                        case CanceledByAdministrator:
                        case Undefined:
                            break;
                        default:
                            logger.debug(methodName, id, "RECORDING FAILURE");
                            // all other cases are errors that contribute to the error count
                            if ( sset.excessiveRunFailures() ) {    // if true, the count is exceeeded, but reset
                                logger.warn(methodName, null, "Process Failure: " + jct + " Maximum consecutive failures[" + sset.failure_run + "] max [" + sset.failure_max + "]");
                            } else {
                                sset.start();
                            }
                            break;
                    }
                }
            }

            // Now factor in cumulative state of the implementors and manage the ping thread as needed
            sset.establish(id, w.getJobState());

            if ( (sset.getServiceState() == ServiceState.NotAvailable) && (sset.countReferences() == 0) && (sset.countImplementors() == 0) ) {
                // this service is now toast.  remove from our maps asap to avoid clashes if it gets
                // resubmitted before the OR can purge it.
                if ( ! sset.isRegistered() ) {
                    logger.debug(methodName, id, "Removing service", endpoint, "because it died and has no more references.");
                    serviceStateHandler.removeService(endpoint);
                }
                serviceStateHandler.removeServicesForJob(id);
            }
        }
        
    }

    /**
     * Add in the service dependencies to the query.
     */
    void updateServiceQuery(IServiceDescription sd, ServiceSet sset)
    {

        if ( sset.isRegistered() ) {
            // 
            // The thing may not be running yet / at-all.  Pull out the deps from the registration and
            // query them individually.
            //
            String[] deps = sset.getIndependentServices();
            if ( deps != null ) {
                for ( String dep : deps ) {
                    ServiceSet independent = serviceStateHandler.getServiceByName(dep);
                    if ( independent != null ) {
                        sd.addDependency(dep, independent.getServiceState().decode());
                    } else {
                        sd.addDependency(dep, ServiceState.NotAvailable.decode());
                    }
                }
            }
        } else {
            //
            // If it's not registered we have to look up all the dependencies of the implementors instead
            //
            Map<DuccId, JobState> implementors = sset.getImplementors();
            for ( DuccId id : implementors.keySet() ) {
                Map<String, ServiceSet> deps = serviceStateHandler.getServicesForJob(id); // all the stuff 'id' is dependent upon
                if ( deps != null ) {
                    for ( String s : deps.keySet() ) {
                        ServiceSet depsvc = deps.get(s);
                        sd.addDependency(s, depsvc.getServiceState().decode());
                    }
                }                
            }
        }
    }

    ServiceQueryReplyEvent query(ServiceQueryEvent ev)
    {
    	//String methodName = "query";
        long friendly = ev.getFriendly();
        String epname = ev.getEndpoint();
        ServiceQueryReplyEvent reply = new ServiceQueryReplyEvent();

        if (( friendly == -1) && ( epname == null )) {
            ArrayList<String> keys = serviceStateHandler.getServiceNames();
            for ( String k : keys ) {
                ServiceSet sset = serviceStateHandler.getServiceByName(k);
                if ( k == null ) continue;                    // the unlikely event it changed out from under us
                
                IServiceDescription sd = sset.query();
                updateServiceQuery(sd, sset);
                reply.addService(sd);
            }
        } else {
            ServiceSet sset = serviceStateHandler.getServiceForApi(friendly, epname);
            if ( sset == null ) {
                reply.setMessage("Unknown");
                reply.setEndpoint(epname);
                reply.setId(friendly);
                reply.setReturnCode(false);
            } else {
                IServiceDescription sd = sset.query();
                updateServiceQuery(sd, sset);
                reply.addService(sd);
            }
        } 

        return reply;
    }

    ServiceReplyEvent start(ServiceStartEvent ev)
    {
        //String methodName = "start";
        
        long friendly = ev.getFriendly();
        String epname = ev.getEndpoint();
        ServiceSet sset = serviceStateHandler.getServiceForApi(friendly, epname);
        if ( sset == null ) {
            return new ServiceReplyEvent(false, "Unknown", epname, friendly);
        }

        String userin = ev.getUser();
        String userout = sset.getUser();

        if ( !userin.equals(userout) && !serviceManager.isAdministrator(userin) ) {
            return new ServiceReplyEvent(false, "Owned by " + userout,  epname, friendly);
        }

        if ( sset.isRegistered() ) {
            int running = sset.countImplementors();
            int instances = ev.getInstances();
            int registered = sset.getNInstances();
            int wanted = 0;

            if ( instances == -1 ) {
                wanted = Math.max(0, registered - running);
            } else {
                wanted = instances;
            }
            if ( wanted == 0 ) {
                return new ServiceReplyEvent(true, 
                                             "Already has instances[" + running + "] - no additional instances started", 
                                             sset.getKey(), 
                                             sset.getId().getFriendly());
            }

            pendingRequests.add(new ApiHandler(ev, this));

//             // only start something if we don't have enought already going
//             ApiHandler  apih = new ApiHandler(ev, this);
//             Thread t = new Thread(apih);
//             t.start();

            return new ServiceReplyEvent(true, 
                                         "New instances[" + wanted + "]", 
                                         sset.getKey(), 
                                         sset.getId().getFriendly());
        } else {
            return new ServiceReplyEvent(false, "Not registered", sset.getKey(), sset.getId().getFriendly());      
        }
    }

    //
    // Everything to do this must be vetted before it is called
    //
    // Start with no instance says: start enough new processes to get up the registered amount
    // Start with some instances says: start exactly this many
    // If the --save option is included, also update the registration
    //
    void doStart(long friendly, String epname, int instances, boolean update)
    {
    	//String methodName = "doStart";
        ServiceSet sset = serviceStateHandler.getServiceForApi(friendly, epname);

        int running    = sset.countImplementors();
        int registered = sset.getNInstances();
        int wanted     = 0;

        if ( instances == -1 ) {
            wanted = Math.max(0, registered - running);
        } else {
            wanted = instances;
        }

        if ( update ) {
            sset.setNInstances(running + instances);
        }
                
        sset.resetRunFailures();                  // manual start overrides, if there's still a problem
                                                  // the service will be stopped soon anyway.
        for ( int i = 0; i < wanted; i++ ) {
            if ( !sset.start() ) break;
        } 

    }

    ServiceReplyEvent stop(ServiceStopEvent ev)
    {
        long friendly = ev.getFriendly();
        String epname = ev.getEndpoint();
        ServiceSet sset = serviceStateHandler.getServiceForApi(friendly, epname);
        if ( sset == null ) {
            return new ServiceReplyEvent(false, "Unknown", epname, friendly);
        }

        String userin = ev.getUser();
        String userout = sset.getUser();

        if ( !userin.equals(userout) && !serviceManager.isAdministrator(userin) ) {
            return new ServiceReplyEvent(false, "Owned by " + userout,  epname, friendly);
        }

        if ( sset.isRegistered() ) {
            if ( sset.isStopped() ) {
                return new ServiceReplyEvent(false, "Already stopped", sset.getKey(), sset.getId().getFriendly());
            }

            pendingRequests.add(new ApiHandler(ev, this));
            return new ServiceReplyEvent(true, "Stopping", sset.getKey(), sset.getId().getFriendly());
        } else {
            return new ServiceReplyEvent(false, "Not registered", sset.getKey(), sset.getId().getFriendly());         
        }

    }

    //
    // Everything to do this must be vetted before it is called
    //
    // If instances == 0 set stop the whole service
    // Otherwise we just stop the number asked for
    // If --save is insicated we update the registry
    //
    void doStop(long friendly, String epname, int instances, boolean update)
    {
        //String methodName = "doStop";

        ServiceSet sset = serviceStateHandler.getServiceForApi(friendly, epname);

        int running    = sset.countImplementors();
        int tolose;
        if ( instances == -1 ) {
            tolose = running;
        } else {
            tolose = Math.min(instances, running);
        }

        if ( update ) {
            sset.setNInstances(Math.max(0, running - instances)); // never persist < 0 registered instance
        }
        
        if ( tolose == running ) {
            sset.stop();                                  // blind stop
        } else {
            sset.stop(tolose);                            // selective stop 
        }
    }

    ServiceReplyEvent register(DuccId id, String props_filename, String meta_filename, DuccProperties props, DuccProperties meta)
    {
    	String methodName = "register";

        String error = null;
        boolean must_deregister = false;

        ServiceSet sset = null;
        try {
            sset = new ServiceSet(id, props_filename, meta_filename, props, meta);
        } catch (Throwable t) {
            error = t.getMessage();
            return new ServiceReplyEvent(false, t.getMessage(), "?", id.getFriendly());            
        }

        String key = sset.getKey();

        // Check if already registered
        ServiceSet sset0 = serviceStateHandler.getServiceByName(key);
        if ( sset0 != null ) {
            error = ("Duplicate owned by: " + sset0.getUser());
        } else {
            try {
                sset.saveServiceProperties();
            } catch ( Exception e ) {
                error = ("Internal error; unable to store service descriptor. " + key);
                logger.error(methodName, id, e);
                must_deregister = true;
            }
            
            try {
                if ( ! must_deregister ) {
                    sset.saveMetaProperties();
                }
            } catch ( Exception e ) {
                error = ("Internal error; unable to store service meta-descriptor. " + key);
                logger.error(methodName, id, e);
                must_deregister = true;
            }

            // must check for cycles or we can deadlock
            if ( ! must_deregister ) {
                CycleChecker cc = new CycleChecker(sset);
                if ( cc.hasCycle() ) {
                    error = ("Service dependencies contain a cycle with " + cc.getCycles());
                    logger.error(methodName, id, error);
                    must_deregister = true;
                }
            }
        }

        if ( error == null ) {
            serviceStateHandler.putServiceByName(sset.getKey(), sset);
            return new ServiceReplyEvent(true, "Registered", key, id.getFriendly());
        } else {
            File mf = new File(meta_filename);
            mf.delete();
            
            File pf = new File(props_filename);
            pf.delete();
            return new ServiceReplyEvent(false, error, key, id.getFriendly());
        }
    }

    public ServiceReplyEvent modify(ServiceModifyEvent ev)
    {
        long friendly       = ev.getFriendly();
        String epname = ev.getEndpoint();
    	ServiceSet sset = serviceStateHandler.getServiceForApi(friendly, epname);
        if ( sset == null ) {
            return new ServiceReplyEvent(false, "Unknown", epname, friendly);
        }

        String userin = ev.getUser();
        String userout = sset.getUser();

        if ( !userin.equals(userout) && !serviceManager.isAdministrator(userin) ) {
            return new ServiceReplyEvent(false, "Owned by " + userout,  epname, friendly);
        }
        
    	if ( sset.isRegistered() ) {            
            pendingRequests.add(new ApiHandler(ev, this));
//             ApiHandler  apih = new ApiHandler(ev, this);
//             Thread t = new Thread(apih);
//             t.start();
            return new ServiceReplyEvent(true, "Modifing", sset.getKey(), sset.getId().getFriendly());
        } else {
            return new ServiceReplyEvent(false, "Not registered", sset.getKey(), sset.getId().getFriendly());           
        }
    }

    void doModify(long friendly, String epname, int instances, Trinary autostart, boolean activate)
    {
        //String methodName = "doStop";

        ServiceSet sset = serviceStateHandler.getServiceForApi(friendly, epname);
        
        if ( instances > 0 ) {
            sset.setNInstances(instances);                // also persists instances
            if ( activate ) {
                int running    = sset.countImplementors();
                int diff       = instances - running;
                
                if ( diff > 0 ) {
                    while ( diff-- > 0 ) {
                        if ( !sset.start() ) break;
                    }
                } else if ( diff < 0 ) {
                    sset.stop(-diff);
                }
            }          
        }

        if ( autostart != Trinary.Unset ) {
            sset.setAutostart(autostart.decode());
            if ( activate ) {
                sset.enforceAutostart();
            }
        }

    }

    public ServiceReplyEvent unregister(ServiceUnregisterEvent ev)
    {
        long friendly = ev.getFriendly();
        String epname = ev.getEndpoint();
        ServiceSet sset = serviceStateHandler.getServiceForApi(friendly, epname);
        if ( sset == null ) {
            return new ServiceReplyEvent(false, "Unknown",  epname, friendly);
        }

        String userin = ev.getUser();
        String userout = sset.getUser();

        if ( !userin.equals(userout) && !serviceManager.isAdministrator(userin) ) {
            return new ServiceReplyEvent(false, "Owned by " + userout,  epname, friendly);
        }

        if ( sset.isRegistered() ) {            
            sset.deregister();          // just sets a flag so we know how to handle it when it starts to die
            pendingRequests.add(new ApiHandler(ev, this));
//             ApiHandler  apih = new ApiHandler(ev, this);
//             Thread t = new Thread(apih);
//             t.start();
            return new ServiceReplyEvent(true, "Shutting down implementors", sset.getKey(), sset.getId().getFriendly());
        } else {
            return new ServiceReplyEvent(false, "Not registered", sset.getKey(), sset.getId().getFriendly());            
        }
        
    }

    //
    // Everything to do this must be vetted before it is called. Run in a new thread to not hold up the API.
    //
    void doUnregister(long friendly, String epname)
    {
    	String methodName = "doUnregister";
        ServiceSet sset = serviceStateHandler.getServiceForApi(friendly, epname);                

        if ( sset.countImplementors() > 0 ) {
            logger.debug(methodName, sset.getId(), "Stopping implementors:", friendly, epname);
            sset.stop();
        } else if ( sset.isPingOnly() ) {
            logger.debug(methodName, sset.getId(), "Unregister ping-only setvice:", friendly, epname);
            sset.stop();
            serviceStateHandler.removeService(epname, friendly);
        } else {
            logger.debug(methodName, sset.getId(), "Removing from map:", friendly, epname);
            serviceStateHandler.removeService(epname, friendly);
        }

        sset.deleteProperties();
    }

    /**
     * From: http://en.wikipedia.org/wiki/Topological_sorting
     *
     * L �? Empty list that will contain the sorted elements
     * S �? Set of all nodes with no incoming edges
     * while S is non-empty do
     *     remove a node n from S
     *     insert n into L
     *     for each node m with an edge e from n to m do
     *         remove edge e from the graph
     *         if m has no other incoming edges then
     *             insert m into S
     * if graph has edges then
     *     return error (graph has at least one cycle)
     * else 
     *     return L (a topologically sorted order)
     */
    class CycleChecker
    {
        ServiceSet sset;
        int edges = 0;
        List<String> cycles = null;

        CycleChecker(ServiceSet sset)
        {
            this.sset = sset;
        }

        boolean hasCycle()
        {
            // Start by building the dependency graph
            // TODO: Maybe consider saving this.  Not clear there's much of a
            //       gain doing the extra bookeeping beause the graphs will always
            //       be small and will only need checking on registration or arrival
            //       of a submitted service.  So this cycle checking is always
            //       fast anyway.
            //
            //       Bookeeping could be a bit ugly because a submitted service could
            //       bop in and change some dependency graph.  We really only care
            //       for checking cycles, so we'll check the cycles as things change
            //       and then forget about it.
            //
            String[] deps = sset.getIndependentServices();
            if ( deps == null ) return false;          // man, that was fast!

            Map<String, ServiceSet> visited = new HashMap<String, ServiceSet>();     // all the nodes in the graph
            clearEdges(sset, visited);

            List<ServiceSet> nodes = new ArrayList<ServiceSet>();
            nodes.addAll(visited.values());
            buildGraph(nodes);
            
            List<ServiceSet>        sorted = new ArrayList<ServiceSet>();          // topo-sorted list of nodes
            List<ServiceSet>        current = new ArrayList<ServiceSet>();         // nodes with no incoming edges

            // Constant: current has all nodes with no incoming edges
            for ( ServiceSet node : nodes ) {
                if ( ! node.hasPredecessor() ) current.add(node);
            }

            while ( current.size() > 0 ) {
                ServiceSet next = current.remove(0);                            // remove a node n from S
                sorted.add(next);                                               // insert n int L
                List<ServiceSet> successors = next.getSuccessors();
                for ( ServiceSet succ : successors ) {                          // for each node m(pred) with an edge e from n to m do
                    next.removeSuccessor(succ);                                 // remove edge from graph
                    succ.removePredecessor(next);                               //    ...
                    edges--;
                    if ( !succ.hasPredecessor() ) current.add(succ);            // if m(pred) has no incoming edges insert m into S
                }
            }

            if ( edges == 0 ) return false;                                     // if graph has no edges, no cycles

            cycles = new ArrayList<String>();                                   // oops, and here they are
            for ( ServiceSet node : nodes ) {
                if ( node.hasSuccessor() ) {
                    for ( ServiceSet succ : node.getSuccessors() ) {
                        cycles.add(node.getKey() + " -> " + succ.getKey());
                    }
                }
            }
            return true;
        }

        String getCycles()
        {
            return cycles.toString();
        }
        
        //
        // Traveerse the graph and make sure all the nodes are "clean" 
        //
        void clearEdges(ServiceSet node, Map<String, ServiceSet> visited)
        {
            String key = node.getKey();
            node.clearEdges();
            if ( visited.containsKey(key) ) return;

            visited.put(node.getKey(), node);
            String[] deps = node.getIndependentServices();
            if ( deps == null ) return;
            
            for ( String dep : deps ) {
                ServiceSet sset = serviceStateHandler.getServiceByName(dep);
                if ( sset != null ) {
                	clearEdges(sset, visited);
                }
            }
        }
            
        void buildGraph(List<ServiceSet> nodes)
        {            
            for ( ServiceSet node : nodes ) {           
                String[] deps = node.getIndependentServices();           // never null if we get this far
                if ( deps != null ) {
                    for ( String d : deps ) {
                        ServiceSet outgoing = serviceStateHandler.getServiceByName(d);
                        if ( outgoing == null ) continue;
                        outgoing.setIncoming(node);
                        node.setOutgoing(outgoing);
                        edges++;
                    }
                }
            }
        }
    }

    class ServiceStateHandler
    {

        // Map of active service descriptors by endpoint.  For UIMA services, key is the endpoint.
        private Map<String,  ServiceSet>  servicesByName     = new HashMap<String,  ServiceSet>();
        private Map<Long,    ServiceSet>  servicesByFriendly = new HashMap<Long,    ServiceSet>();

        // For each job, the collection of services it is dependent upon
        // DUccId is a Job Id (or id for serice that has dependencies)
        private Map<DuccId, Map<String, ServiceSet>>  servicesByJob = new HashMap<DuccId, Map<String, ServiceSet>>();

        ServiceStateHandler()
        {
        }

        /**
         * Return a copy of the keys so we can fetch the services in an orderly manner.
         */
        synchronized ArrayList<String> getServiceNames()
        {
            ArrayList<String> answer = new ArrayList<String>();
            for ( String k : servicesByName.keySet() ) {
                answer.add(k);
            }
            return answer;
        }

        synchronized ServiceSet getServiceByName(String n)
        {
            return servicesByName.get(n);
        }

        synchronized ServiceSet getServiceByFriendly(long id)
        {
            return servicesByFriendly.get( id );
        }

        // API passes in a friendly (maybe) and an endpiont (maybe) but only one of these
        // Here we look up the service by whatever was passed in.
        synchronized ServiceSet  getServiceForApi(long id, String n)
        {
            if ( n == null ) return getServiceByFriendly(id);
            return getServiceByName(n);
        }

        synchronized List<ServiceSet> getRegisteredServices()
        {
            ArrayList<ServiceSet> answer = new ArrayList<ServiceSet>();
            for ( ServiceSet sset : servicesByName.values() ) {
                if ( sset.isRegistered() ) {
                    answer.add(sset);
                }
            }
            return answer;
        }

        synchronized void putServiceByName(String n, ServiceSet s)
        {
            servicesByName.put(n, s);
            DuccId id = s.getId();
            if ( id != null ) {
                servicesByFriendly.put(id.getFriendly(), s);
            }
        }

        synchronized ServiceSet removeService(String n)
        {
            ServiceSet s = servicesByName.remove(n);
            if ( s != null ) {
                DuccId id = s.getId();
                if ( id != null ) {
                    servicesByFriendly.remove(id.getFriendly());
                }
            }
            return s;
        }

        synchronized void removeService(long id)
        {
            ServiceSet sset = servicesByFriendly.remove(id);
            if ( sset != null ) {
                String key = sset.getKey();
                servicesByName.remove(key);
            }
        }

        synchronized void removeService(String n, long id)
        {
            if ( n == null ) removeService(id);
            else             removeService(n);
        }

        synchronized Map<String, ServiceSet> getServicesForJob(DuccId id)
        {

        	return servicesByJob.get(id);
        }

        synchronized void putServiceForJob(DuccId id, ServiceSet s)
        {
            Map<String, ServiceSet> services = servicesByJob.get(id);
            if ( services == null ) {
                services = new HashMap<String, ServiceSet>();
                servicesByJob.put(id, services);
            }
            services.put(s.getKey(), s);
        }

        synchronized void removeServicesForJob(DuccId id)
        {
            servicesByJob.remove(id);
        }

        synchronized void recordNewServices(Map<String, ServiceSet> services) 
        {
            servicesByName.putAll(services);
        }

    }

    /**
     * Tester for topo sorter.
     * Input is props file, e.g. for the graph:
     *   A -> B, A -> C, B -> C:
     *
     *    services = A B C
     *    svc.A = B C
     *    svc.B = C
     *    svc.C = 
     *
     */
    private void runSortTester(String propsfile)
    {
        int friendly = 1;
        DuccProperties props = new DuccProperties();
        try {
			props.load(propsfile);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
            System.exit(1);
		}

        
        String svcnames = props.getStringProperty("services");
        String[] svcs = svcnames.split("\\s");
        ServiceSet[] allServices = new ServiceSet[svcs.length];
        int ndx = 0;
        for ( String svc : svcs ) {
            svc = svc.trim();
            String key = "UIMA-AS:" + svc + ":tcp://foo:123";
            ServiceSet dep = serviceStateHandler.getServiceByName(key);
            if ( dep == null ) {
                dep = new ServiceSet(new DuccId(friendly++), new DuccId(0), key, null);
                serviceStateHandler.putServiceByName(key, dep);
                allServices[ndx++] = dep;
            }

            String depnames = props.getStringProperty("svc." + svc);
            String[] deps = depnames.split("\\s");
            List<String> subdeps = new ArrayList<String>();
            for ( String subsvc : deps ) {
                subsvc = subsvc.trim();
                if ( subsvc.equals("")) continue;

                String subkey = "UIMA-AS:" + subsvc + ":tcp://foo:123";
                ServiceSet subdep = serviceStateHandler.getServiceByName(subkey);
                if ( subdep == null ) {
                    subdep = new ServiceSet(new DuccId(friendly++), new DuccId(0), subkey, null);
                    serviceStateHandler.putServiceByName(subkey, subdep);
                    allServices[ndx++] = subdep;
                }
                subdeps.add(subkey);
            }
            if ( subdeps.size() > 0 ) {
            	dep.setIndependentServices(subdeps.toArray(new String[subdeps.size()]));
            }
        }

        CycleChecker cc = new CycleChecker(allServices[0]);
        if ( cc.hasCycle() ) {
            System.out.println("Service dependencies contain a cycle with " + cc.getCycles());
        } else {
            System.out.println("No cycles detected");
        }

    }

    // tester for the topo sorter
    public static void main(String[] args)
    {
        ServiceHandler sh = new ServiceHandler(null);
        sh.runSortTester(args[0]);
    }
}