| /* |
| * 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.BufferedReader; |
| import java.io.FileNotFoundException; |
| import java.io.FileOutputStream; |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.io.InputStreamReader; |
| import java.net.MalformedURLException; |
| import java.net.URL; |
| import java.sql.Timestamp; |
| import java.util.ArrayList; |
| import java.util.Date; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Timer; |
| import java.util.TimerTask; |
| |
| import org.apache.uima.UIMAFramework; |
| import org.apache.uima.ducc.cli.IServiceApi.RegistrationOption; |
| import org.apache.uima.ducc.cli.IUiOptions.UiOption; |
| import org.apache.uima.ducc.cli.UimaAsPing; |
| import org.apache.uima.ducc.cli.UimaAsServiceMonitor; |
| import org.apache.uima.ducc.common.IServiceStatistics; |
| import org.apache.uima.ducc.common.TcpStreamHandler; |
| import org.apache.uima.ducc.common.utils.DuccLogger; |
| import org.apache.uima.ducc.common.utils.DuccProperties; |
| import org.apache.uima.ducc.common.utils.id.ADuccId; |
| import org.apache.uima.ducc.common.utils.id.DuccId; |
| import org.apache.uima.ducc.transport.event.common.IDuccState.JobState; |
| 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 static final long serialVersionUID = 1L; |
| |
| private DuccLogger logger = DuccLogger.getLogger(this.getClass().getName(), COMPONENT_NAME); |
| |
| // key is unique id of descriptor. The descriptor inherites key from a Job's DuccId, or from |
| // a unique-to SM key for implicit references. |
| HashMap<DuccId, JobState> implementors = new HashMap<DuccId, JobState>(); |
| |
| // key is job/service id, value is same. it's a map for fast existence check |
| HashMap<DuccId, DuccId> references = new HashMap<DuccId, DuccId>(); |
| |
| // For a registered service, here is my registered id |
| DuccId id; |
| HashMap<Long, DuccId> friendly_ids = new HashMap<Long, DuccId>(); |
| String history_key = "work-instances"; |
| |
| // 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; |
| |
| // Registered services, Automatically start at boot, and keep implementors alive |
| boolean autostart = false; |
| // Registered services, we've been stopped. Must remember in order to counteract autostart. |
| boolean stopped = false; |
| // Registered services, remember if was started by reference only so we can stop when refs die |
| boolean referenced_start = false; |
| |
| // Registered services, the number of instances to maintain |
| int instances = 1; |
| |
| // Service pinger |
| IServiceMeta serviceMeta = null; |
| |
| // After stopping a pinger we need to discard it,and we usually need to stop it well before |
| // it is ok to delete residual state such as UIMA-AS queues. Instead of discarding it, we |
| // stash it here to use once the last implementor seems to be dead. |
| IServiceMeta residualMeta = null; |
| |
| // registered services state files |
| DuccProperties job_props = null; |
| DuccProperties meta_props = null; |
| String props_filename = null; |
| String meta_filename = null; |
| boolean deregistered = false; |
| |
| ServiceType service_type = ServiceType.Undefined; |
| ServiceClass service_class = ServiceClass.Undefined; |
| ServiceState service_state = ServiceState.Undefined;; |
| |
| // structures to manage service linger after it exits |
| Timer timer = null; |
| LingerTask linger = null; |
| long linger_time = 5000; |
| |
| int failure_max = ServiceManagerComponent.failure_max; |
| int failure_run = 0; // max allowed consecutive failures, current failure count |
| |
| //JobState job_state = JobState.Undefined; |
| // |
| // This is the constructor for an implicit service |
| // |
| public ServiceSet(String key, DuccId id) |
| { |
| this.key = key; |
| this.id = id; |
| parseEndpoint(key); |
| |
| if ( this.service_type == ServiceType.Custom ) { |
| throw new IllegalStateException("Custom services may not be referenced as Implicit services."); |
| } |
| |
| this.service_type = ServiceType.UimaAs; |
| this.service_class = ServiceClass.Implicit; |
| |
| String state_dir = System.getProperty("DUCC_HOME") + "/state"; |
| |
| // Need job props and meta props for webserver. |
| // The pinger is always the default configured UIMA-AS pinger. |
| // |
| // job props: working_directory, log_directory |
| // meta props: endpoint, user |
| job_props = new DuccProperties(); |
| // job_props.put("working_directory", System.getProperty("user.dir")); // whatever my current dir is |
| // job_props.put("log_directory", System.getProperty("user.dir") + "/../logs"); |
| //job_props.put("service_ping_jvm_args", "-Xmx50M"); |
| props_filename = state_dir + "/services/" + id.toString() + ".svc"; |
| saveServiceProperties(); |
| |
| meta_props = new DuccProperties(); |
| meta_props.put("user", System.getProperty("user.name")); |
| meta_props.put("endpoint", key); |
| meta_props.put("service-class", ""+service_class.decode()); |
| meta_props.put("service-type", ""+service_type.decode()); |
| meta_props.put("stopped", ""+stopped); |
| meta_props.put("service-state", ""+getServiceState()); |
| meta_props.put("ping-active", "false"); |
| meta_props.put("service-alive", "false"); |
| meta_props.put("service-healthy", "false"); |
| meta_props.put("service-statistics", "N/A"); |
| meta_props.put("ping-only", "true"); |
| |
| meta_filename = state_dir + "/services/" + id.toString() + ".meta"; |
| saveMetaProperties(); |
| } |
| |
| // |
| // Constructor for a submitted service |
| // |
| public ServiceSet(DuccId id, DuccId first_implementor, String key, String[] independentServices) |
| { |
| this.key = key; |
| this.id = id; |
| this.friendly_ids.put(first_implementor.getFriendly(), null); |
| |
| this.independentServices = independentServices; |
| this.service_class = ServiceClass.Submitted; |
| |
| parseEndpoint(key); |
| |
| String state_dir = System.getProperty("DUCC_HOME") + "/state"; |
| // Need job props and meta props for webserver. |
| // The pinger is always the default configured UIMA-AS pinger. |
| // Submitted services must always be UIMA-AS services, for now, checked in caller. |
| // |
| // job props: working_directory, log_directory |
| // meta props: endpoint, user |
| job_props = new DuccProperties(); |
| //job_props.put("service_ping_jvm_args", "-Xmx50M"); |
| props_filename = state_dir + "/services/" + id.toString() + ".svc"; |
| saveServiceProperties(); |
| |
| meta_props = new DuccProperties(); |
| meta_props.put("user", System.getProperty("user.name")); |
| meta_props.put("endpoint", key); |
| meta_props.put("service-class", ""+service_class.decode()); |
| meta_props.put("service-type", ""+service_type.decode()); |
| meta_props.put("stopped", ""+stopped); |
| meta_props.put("service-state", ""+getServiceState()); |
| meta_props.put("ping-active", "false"); |
| meta_props.put("service-alive", "false"); |
| meta_props.put("service-healthy", "false"); |
| meta_props.put("service-statistics", "N/A"); |
| meta_props.put("implementors", ""+id.getFriendly()); |
| meta_props.put("ping-only", "false"); |
| |
| meta_filename = state_dir + "/services/" + id.toString() + ".meta"; |
| saveMetaProperties(); |
| } |
| |
| // |
| // Constructor for a registered service |
| // |
| public ServiceSet(DuccId id, String props_filename, String meta_filename, DuccProperties props, DuccProperties meta) |
| { |
| this.job_props = props; |
| this.meta_props = meta; |
| this.id = id; |
| this.props_filename = props_filename; |
| this.meta_filename = meta_filename; |
| this.service_state = ServiceState.NotAvailable; |
| this.linger_time = props.getLongProperty(RegistrationOption.ServiceLinger.decode(), 5000); |
| this.key = meta.getProperty("endpoint"); |
| this.failure_max = props.getIntProperty(UiOption.InstanceFailuresLimit.pname(), ServiceManagerComponent.failure_max); |
| |
| parseEndpoint(key); |
| |
| this.user = meta.getProperty("user"); |
| this.instances = meta.getIntProperty("instances", 1); |
| this.autostart = meta.getBooleanProperty("autostart", false); |
| |
| String idprop = meta.getProperty("implementors", null); |
| if ( idprop != null ) { |
| String[] ids = idprop.split("\\s"); |
| for ( String i : ids ) { |
| friendly_ids.put(Long.parseLong(i), null); |
| } |
| } |
| this.service_class = ServiceClass.Registered; |
| |
| parseIndependentServices(); |
| |
| if ( ! job_props.containsKey("service_ping_dolog")) { |
| job_props.put("service_ping_dolog", "false"); |
| } |
| if ( !job_props.containsKey("service_ping_timeout") ) { |
| job_props.put("service_ping_timeout", ""+ServiceManagerComponent.meta_ping_timeout); |
| } |
| |
| meta_props.put("service-class", ""+service_class.decode()); |
| meta_props.put("service-type", ""+service_type.decode()); |
| meta_props.put("stopped", ""+stopped); |
| meta_props.put("service-state", ""+getServiceState()); |
| meta_props.put("ping-active", "false"); |
| meta_props.put("service-alive", "false"); |
| meta_props.put("service-healthy", "false"); |
| meta_props.put("service-statistics", "N/A"); |
| |
| if ( isStartable() ) { |
| meta_props.put("ping-only", "false"); |
| } else { |
| meta_props.put("ping-only", "true"); |
| } |
| // 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; |
| } |
| |
| protected 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); |
| } |
| |
| } |
| |
| synchronized void deleteProperties() |
| { |
| |
| // be sure to move any services that seem not to have croaked yet to history |
| 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.remove("implementors"); |
| |
| ServiceManagerComponent.deleteProperties(id.toString(), meta_filename, meta_props, props_filename, job_props); |
| meta_filename = null; |
| props_filename = null; |
| } |
| |
| void setIncoming(ServiceSet sset) |
| { |
| predecessors.add(sset); |
| } |
| |
| void clearEdges() |
| { |
| predecessors.clear(); |
| successors.clear(); |
| } |
| |
| boolean isStopped() |
| { |
| return stopped; |
| } |
| |
| 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; |
| } |
| |
| private void parseIndependentServices() |
| { |
| String depstr = job_props.getProperty(RegistrationOption.ServiceDependency.decode()); |
| 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; |
| } |
| |
| /** |
| * A service is startable if it's UIMA-AS, or if it's CUSTOM and has a process_executable |
| * associated with it. A non-startable CUSTOM service may have only a pinger. |
| */ |
| boolean isStartable() |
| { |
| switch ( service_type ) { |
| case UimaAs: |
| return true; |
| case Custom: |
| return job_props.containsKey("process_executable"); |
| } |
| return false; // redundant but needed for compilation |
| } |
| |
| /** |
| * At boot only ... synchronize my state with published OR state. |
| */ |
| void synchronizeImplementors(Map<DuccId, JobState> work) |
| { |
| HashMap<Long, DuccId> newmap = new HashMap<Long, DuccId>(); |
| // first loop synchronized 'friendly_ids' with live jobs comining in |
| for ( DuccId id : work.keySet() ) { |
| |
| long fid = id.getFriendly(); |
| if ( friendly_ids.containsKey(fid) ) { |
| JobState js = work.get(id); |
| implementors.put(id, js); |
| newmap.put(fid, id); |
| } |
| } |
| |
| // second loop synchronizes history with jobs that used to be live and aren't any more (because of restart) |
| String history = meta_props.getStringProperty(history_key, ""); |
| for ( Long friendly : friendly_ids.keySet() ) { |
| DuccId id = newmap.get(friendly); |
| if ( id == null ) { |
| history = history + " " + friendly; |
| } |
| } |
| meta_props.put(history_key, history); |
| |
| friendly_ids = newmap; // replace persisted version with validated version from OR state |
| persistImplementors(); |
| } |
| |
| /** |
| * |
| */ |
| void enforceAutostart() |
| { |
| //String methodName = "enforceAutostart"; |
| if ( ! autostart ) return; // not doing auto, nothing to do |
| if ( stopped ) return; // doing auto, but we've been manually stopped |
| if ( failure_run >= failure_max ) return; // too many failures, no more enforcement |
| |
| if ( (!isStartable()) && (serviceMeta == null) ) { // ping-only and pinger not alive |
| start(); // ... then it needs to be started |
| return; |
| } |
| |
| // could have more implementors than instances if some were started dynamically but the count not persisted |
| int needed = Math.max(0, instances - countImplementors()); |
| |
| while ( (needed--) > 0 ) { |
| if ( ! start() ) break; |
| } |
| } |
| |
| /** |
| * Add implementor that we have an OR-assigned DuccId for |
| */ |
| public void addImplementor(DuccId id, JobState js) |
| { |
| if ( isSubmitted() ) { |
| friendly_ids.put(id.getFriendly(), id); |
| } |
| implementors.put(id, js); |
| persistImplementors(); |
| } |
| |
| void promote() |
| { |
| String methodName = "promote"; |
| logger.debug(methodName, null, "Promoting", key); |
| switch ( service_class ) { |
| case Implicit : this.service_class = ServiceClass.Submitted; break; |
| case Submitted: break; |
| default : throw new IllegalStateException("Trying to promote a Registered service!"); |
| } |
| } |
| |
| boolean isUimaAs() |
| { |
| return (service_type == ServiceType.UimaAs); |
| } |
| |
| boolean isCustom() |
| { |
| return (service_type == ServiceType.Custom); |
| } |
| |
| Map<DuccId, JobState> getImplementors() |
| { |
| return implementors; |
| } |
| |
| DuccProperties getJobProperties() |
| { |
| return job_props; |
| } |
| |
| DuccProperties getMetaProperties() |
| { |
| return meta_props; |
| } |
| |
| boolean isImplicit() |
| { |
| return (service_class == ServiceClass.Implicit); |
| } |
| |
| boolean isSubmitted() |
| { |
| return (service_class == ServiceClass.Submitted); |
| } |
| |
| boolean isPingOnly() |
| { |
| return meta_props.containsKey("ping-only"); |
| } |
| |
| boolean isRegistered() |
| { |
| return (service_class == ServiceClass.Registered) && (!deregistered); |
| } |
| |
| void setReferencedStart(boolean val) |
| { |
| this.referenced_start = val; |
| } |
| |
| boolean isReferencedStart() |
| { |
| return this.referenced_start; |
| } |
| |
| String getUser() |
| { |
| return user; |
| } |
| |
| /** |
| * True only if |
| * a) this is a registereed service, and |
| * b) it has been deregistered. |
| * because you can't deregister a non-registered service. |
| */ |
| boolean isDeregistered() |
| { |
| return (service_class == ServiceClass.Registered) && (deregistered); |
| } |
| |
| void deregister() |
| { |
| deregistered = true; |
| } |
| |
| String getMetaFilename() |
| { |
| return meta_filename; |
| } |
| |
| String getPropsFilename() |
| { |
| return props_filename; |
| } |
| |
| synchronized int getNInstances() |
| { |
| return instances; |
| } |
| |
| synchronized void saveMetaProperties() |
| { |
| String methodName = "saveMetaProperties"; |
| |
| if ( meta_filename == null ) { |
| // if this is null it was deleted and this is some kind of lingering thread updating, that |
| // we don't really want any more |
| logger.warn(methodName, id, "Meta properties is deleted, bypassing attempt to save."); |
| return; |
| } |
| |
| meta_props.put("stopped", ""+stopped); |
| meta_props.put("service-state", ""+ getServiceState()); |
| meta_props.put("ping-active", "" + (serviceMeta != null)); |
| meta_props.put("service-alive", "false"); |
| meta_props.put("service-healthy", "false"); |
| meta_props.put("service-statistics", "N/A"); |
| |
| if ( serviceMeta != null ) { |
| IServiceStatistics ss = serviceMeta.getServiceStatistics(); |
| if ( ss != null ) { |
| meta_props.put("service-alive", "" + ss.isAlive()); |
| meta_props.put("service-healthy", "" + ss.isHealthy()); |
| meta_props.put("service-statistics", "" + ss.getInfo()); |
| } |
| } |
| |
| FileOutputStream fos = null; |
| try { |
| fos = new FileOutputStream(meta_filename); |
| meta_props.store(fos, "Meta descriptor"); |
| } catch (FileNotFoundException e) { |
| logger.warn(methodName, id, "Cannot save meta properties, file does not exist."); |
| } catch (IOException e) { |
| logger.warn(methodName, id, "I/O Error saving meta properties:", e); |
| } finally { |
| try { |
| if ( fos != null ) fos.close(); |
| } catch (IOException e) { |
| } |
| } |
| return; |
| } |
| |
| void saveServiceProperties() |
| { |
| String methodName = "saveServiceProperties"; |
| FileOutputStream fos = null; |
| try { |
| fos = new FileOutputStream(props_filename); |
| job_props.store(fos, "Service descriptor"); |
| } catch (FileNotFoundException e) { |
| logger.warn(methodName, id, "Cannot save service properties, file does not exist."); |
| } catch (IOException e) { |
| logger.warn(methodName, id, "I/O Error saving service properties:", e); |
| } finally { |
| try { |
| if ( fos != null ) fos.close(); |
| } catch (IOException e) { |
| } |
| } |
| return; |
| } |
| |
| synchronized void setNInstances(int n) |
| { |
| if ( n != meta_props.getIntProperty("instances") ) { |
| meta_props.setProperty("instances", Integer.toString(n)); |
| this.instances = n; |
| saveMetaProperties(); |
| } |
| } |
| |
| synchronized void setAutostart(boolean auto) |
| { |
| meta_props.setProperty("autostart", auto ? "true" : "false"); |
| this.autostart = auto; |
| saveMetaProperties(); |
| if ( auto ) { |
| // turning this on gives benefit of the doubt on failure management |
| failure_run = 0; |
| } |
| } |
| |
| synchronized boolean isAutostart() |
| { |
| return autostart; |
| } |
| |
| synchronized void persistImplementors() |
| { |
| if ( isImplicit() ) return; |
| |
| if ( friendly_ids.size() == 0 ) { |
| meta_props.remove("implementors"); |
| } else { |
| StringBuffer sb = new StringBuffer(); |
| for ( Long l : friendly_ids.keySet() ) { |
| sb.append(Long.toString(l)); |
| sb.append(" "); |
| } |
| String s = sb.toString().trim(); |
| meta_props.setProperty("implementors", s); |
| } |
| saveMetaProperties(); |
| } |
| |
| synchronized void persistReferences() |
| { |
| if ( references.size() == 0 ) { |
| meta_props.remove("references"); |
| } else { |
| StringBuffer sb = new StringBuffer(); |
| for ( DuccId id : references.keySet() ) { |
| sb.append(id.toString()); |
| sb.append(" "); |
| } |
| String s = sb.toString().trim(); |
| meta_props.setProperty("references", s); |
| } |
| saveMetaProperties(); |
| } |
| |
| /** |
| * If we're are registered service, and one of the "stringids" from submit |
| * happens to match the friendly id passed in then this ServiceSet is |
| * the representative object for the service. |
| */ |
| boolean matches(DuccId did) |
| { |
| if ( ! isRegistered() ) return false; |
| |
| if ( friendly_ids.containsKey(did.getFriendly()) ) { |
| friendly_ids.put(did.getFriendly(), did); |
| return true; |
| } |
| |
| return false; |
| } |
| |
| boolean containsImplementor(DuccId id) |
| { |
| // must use friendly, in case the thing was just started and not into the implementors set yet |
| return friendly_ids.containsKey(id.getFriendly()); |
| } |
| |
| public void removeImplementor(DuccId id) |
| { |
| String methodName = "removeImplementors"; |
| if ( ! implementors.containsKey(id ) ) return; // quick short circuit if it's already gone |
| |
| logger.debug(methodName, this.id, "Removing implementor", id); |
| implementors.remove(id); |
| friendly_ids.remove(id.getFriendly()); |
| if ( implementors.size() == 0 ) { |
| stopPingThread(); |
| } |
| String history = meta_props.getStringProperty(history_key, ""); |
| history = history + " " + id.toString(); |
| meta_props.put(history_key, history); |
| persistImplementors(); |
| |
| // |
| // So much to check here |
| // - all implementors gone? |
| // - is it a registered, not-ping-only service, or a submitted service (startable) |
| // - is it a uima-as service, with an internal pinger? |
| // Only if all that is true, we'll clear out the queues. |
| //== |
| logger.debug(methodName, id, "implementors.size()", implementors.size(), |
| "service_class", service_class, |
| "isStartable()", isStartable(), |
| "isSubmitted()", isSubmitted(), |
| "service_type", service_type, |
| "ping_class", job_props.getStringProperty("service_ping_class", UimaAsPing.class.getName()) |
| ); |
| |
| // This block is to clear the service queue from ActiveMq if it's no longer being used. |
| if ( implementors.size() == 0 ) { // Went to 0 and there was a pinger? |
| if ( ( (service_class == ServiceClass.Registered) && isStartable()) || isSubmitted() ) { // Is one of our happy cases (not ping-only, we don't know much about it.) |
| if ( service_type == ServiceType.UimaAs ) { |
| // Either no pinger specified, in which case the default is used. Or, it is specified, and if it |
| // matches the default, we get to clear anyway. |
| String pingclass = job_props.getStringProperty("service_ping_class", UimaAsPing.class.getName()); |
| if ( pingclass.equals(UimaAsPing.class.getName()) ) { |
| UimaAsServiceMonitor monitor = new UimaAsServiceMonitor(endpoint, broker_host, broker_jmx_port); |
| logger.debug(methodName, id, "Clearing queues"); |
| try { |
| monitor.init(null); |
| monitor.clearQueues(); |
| } catch (Throwable e) { |
| // TODO Auto-generated catch block |
| e.printStackTrace(); |
| } |
| } |
| } |
| } |
| } |
| } |
| |
| public synchronized int countImplementors() |
| { |
| // The implementos and friendly_ids sets track each other carefully. The former |
| // tracks process state via the remote process's DuccId. The latter tracks |
| // what we've tried to start, but may not know the actual state for until we get |
| // Orchestrator publications. |
| // |
| // To avoid ugly races, we consider a process to be an implementor as soon as we |
| // get the "friendly" id from the Orchestrator. |
| return friendly_ids.size(); |
| } |
| |
| public synchronized int reference(DuccId id) |
| { |
| String methodName = "reference"; |
| logger.debug(methodName, this.id, " ---------------- Service ", this.id, "references", id); |
| |
| if ( linger != null ) { |
| logger.debug(methodName, this.id, " ---------------- Canceling linger task"); |
| linger.cancel(); |
| linger = null; |
| } |
| references.put(id, id); |
| |
| persistReferences(); |
| return references.size(); |
| } |
| |
| public synchronized int dereference(DuccId id) |
| { |
| if ( references.remove(id) == null ) { |
| throw new IllegalStateException("Id " + id + " not found in map for " + getKey()); |
| } |
| |
| // stop the pinger if no longer needed |
| if ( (references.size() == 0) && // nothing left |
| ( isImplicit() || ( isCustom() && !isStartable()) ) // implicit UIMA-AS || implicit CUSTOM |
| ) |
| { |
| stopPingThread(); // must stop pinger because there's no state machine |
| // to do it for us in this situation |
| } |
| |
| persistReferences(); |
| return references.size(); |
| } |
| |
| 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(); |
| } |
| |
| 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 |
| return ServiceState.Waiting; |
| case WaitingForResources: // Scheduler is assigning resources to Job |
| 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 |
| case Completed: // Job processing is completed |
| case Undefined: // None of the above |
| default: |
| return ServiceState.NotAvailable; |
| } |
| } |
| |
| /** |
| * The MAX of the states of the implementors. |
| */ |
| private ServiceState cumulativeJobState() |
| { |
| ServiceState response = ServiceState.NotAvailable; |
| for ( JobState s : implementors.values() ) { |
| ServiceState translated = translateJobState(s); |
| if ( translated.ordinality() > response.ordinality() ) response = translated; |
| } |
| |
| // if we're stopped, we cannot advance global state beyond whatever it is, i.e. an "active" |
| // implementor that hasn't quit yet is not allowed to progress state. |
| if ( isStopped() ) { |
| ServiceState current = getServiceState(); |
| if ( current.ordinality() < response.ordinality() ) { |
| response = current; |
| } |
| } |
| return response; |
| } |
| |
| // synchronized void establish() |
| // { |
| // if ( implementors.size() == 0 ) { |
| // startPingThread(); |
| // } |
| // // Otherwise we let the form that passes in implementor state start the pinger |
| // } |
| |
| public synchronized void establish() |
| { |
| String methodName = "establish.0"; |
| |
| logger.debug(methodName, id, "service_class", service_class, "nimplementors", implementors.size(), |
| "instances", instances); |
| switch ( service_class ) { |
| case Implicit: |
| startPingThread(); |
| break; |
| case Submitted: |
| for ( DuccId id : implementors.keySet() ) { // there's only one |
| establish(id, implementors.get(id)); |
| } |
| break; |
| case Registered: |
| // use friendly to find number of potential instances because it reflects both |
| // the number of actually running instances plus those just started but not yet |
| // checked with their ducc ids yet. |
| |
| if ( friendly_ids.size() > 0 ) { |
| for ( DuccId id : implementors.keySet() ) { // there's only one |
| establish(id, implementors.get(id)); |
| } |
| } else { |
| if ( service_type == ServiceType.Custom ) { |
| startPingThread(); |
| } else { |
| int needed = Math.max(0, instances - friendly_ids.size()); |
| while ( (needed--) > 0 ) { |
| start(); |
| } |
| } |
| } |
| break; |
| } |
| } |
| |
| /** |
| * Starts a ping thread because I've started up and I need to monitor myself. |
| */ |
| public synchronized void establish(DuccId id, JobState job_state) |
| { |
| String methodName = "establish.1"; |
| |
| if ( service_class == ServiceClass.Implicit ) { |
| startPingThread(); |
| return; |
| } |
| |
| if ( service_class == ServiceClass.Submitted ) { |
| // |
| // Annoying edge case - cancel a service, then resubmit before the first one has time |
| // to wind down - need to be sure that the winding-down service is not handled by the |
| // new instance's state machine. |
| // |
| if ( ! implementors.containsKey(id) ) { |
| logger.debug(methodName, id, "Submitted service: Skipping state machine because the service set has no implemetor for", id); |
| return; |
| } |
| } |
| |
| if ( true ) { |
| implementors.put(id, job_state); |
| 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.debug(methodName, id, "serviceState", getServiceState(), "cumulativeState", cumulative); |
| switch ( getServiceState() ) { |
| // 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 Undefined: |
| switch ( cumulative ) { |
| case Initializing: |
| setServiceState(ServiceState.Initializing); |
| break; |
| case Available: |
| startPingThread(); |
| break; |
| case Waiting: |
| setServiceState(ServiceState.Waiting); |
| break; |
| case NotAvailable: |
| break; |
| } |
| break; |
| |
| // If I'm initting and now something is running we can start a pinger |
| case Initializing: |
| switch ( cumulative ) { |
| case Available: |
| startPingThread(); |
| break; |
| case Initializing: |
| break; |
| case Waiting: |
| setServiceState(ServiceState.Initializing); |
| break; |
| case NotAvailable: |
| if ( failure_run >= failure_max ) { |
| setServiceState(ServiceState.NotAvailable); |
| stopPingThread(); |
| } else { |
| // don't regress if we're in retry |
| logger.info(methodName, id, "RETRY RETRY RETRY prevents state regression from Initializing"); |
| } |
| break; |
| } |
| break; |
| |
| case NotAvailable: |
| switch ( cumulative ) { |
| case Available: |
| startPingThread(); |
| setServiceState(ServiceState.Available); |
| break; |
| case Initializing: |
| setServiceState(ServiceState.Initializing); |
| break; |
| case Waiting: |
| setServiceState(ServiceState.Waiting); |
| break; |
| case NotAvailable: |
| break; |
| } |
| break; |
| |
| case Available: |
| switch ( cumulative ) { |
| case Available: |
| startPingThread(); |
| 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:", getServiceState(), "->", cumulative); // can't do anything about it but complain |
| setServiceState(ServiceState.Initializing); |
| break; |
| case NotAvailable: |
| if ( failure_run >= failure_max ) { |
| setServiceState(ServiceState.NotAvailable); |
| stopPingThread(); |
| } else { |
| // don't regress if we're in retry |
| logger.info(methodName, id, "RETRY RETRY RETRY prevents state regression from Available"); |
| } |
| break; |
| } |
| |
| break; |
| |
| case Waiting: |
| switch ( cumulative ) { |
| case Available: |
| break; |
| case Initializing: |
| // state regression can happen with a promoted implicit service, where the service is referenced |
| // and a pinger starts before the actual service is available. the ping keeps us in 'waiting' state |
| // up until expiration, but if the service is actually initializing, we regress. |
| logger.warn(methodName, id, "STATE REGRESSION:", getServiceState(), "->", cumulative); // can't do anything about it but complain |
| setServiceState(ServiceState.Initializing); |
| break; |
| case Waiting: |
| break; |
| case NotAvailable: |
| if ( failure_run >= failure_max ) { |
| setServiceState(ServiceState.NotAvailable); |
| } else { |
| // don't regress if we're in retry |
| logger.info(methodName, id, "RETRY RETRY RETRY prevents state regression from Waiting"); |
| } |
| stopPingThread(); |
| break; |
| } |
| break; |
| |
| case Stopping: |
| switch ( cumulative ) { |
| case Available: |
| case Initializing: |
| case Waiting: |
| break; |
| |
| case NotAvailable: |
| // all the implementors died finally |
| setServiceState(ServiceState.NotAvailable); |
| break; |
| } |
| break; |
| } |
| } |
| |
| } |
| |
| synchronized ServiceState getServiceState() |
| { |
| return service_state; |
| } |
| |
| synchronized void setServiceState(ServiceState ss) |
| { |
| this.service_state = ss; |
| saveMetaProperties(); |
| } |
| |
| synchronized String getKey() |
| { |
| return key; |
| } |
| |
| void resetRunFailures() |
| { |
| failure_run = 0; |
| } |
| |
| synchronized boolean excessiveRunFailures() |
| { |
| String methodName = "runFailures"; |
| if ( (++failure_run) >= failure_max ) { |
| logger.debug(methodName, id, "RUN FAILURES EXCEEDED"); |
| return true; |
| } |
| logger.debug(methodName, id, "RUN FAILURES NOT EXCEEDED YET", failure_run); |
| return false; |
| } |
| |
| private void startPingThread() |
| { |
| String methodName = "startPingThread"; |
| if ( serviceMeta != null ) return; // don't start multiple times. |
| |
| try { |
| logger.info(methodName, id, "Starting ping/monitor."); |
| serviceMeta = new PingDriver(this); |
| } catch ( Throwable t ) { |
| logger.error(methodName, id, "Cannot instantiate ping/monitor.", t); |
| return; |
| } |
| |
| setServiceState(ServiceState.Waiting); |
| Thread t = new Thread(serviceMeta); |
| t.start(); |
| } |
| |
| synchronized void pingExited() |
| { |
| String methodName = "pingExited"; |
| |
| if ( serviceMeta != null ) { |
| logger.warn(methodName, id, "Pinger exited voluntarily, setting state to Undefined. Endpoint", endpoint); |
| setServiceState(ServiceState.Undefined); // not really sure what state is. it will be |
| |
| // checked and updated next run through the |
| // main state machine, and maybe ping restarted. |
| residualMeta = serviceMeta; |
| serviceMeta = null; |
| } else { |
| if ( ! isStopped() ) { // state may still be Stopping, don't over regress state |
| setServiceState(ServiceState.NotAvailable); |
| } |
| } |
| |
| if ( isImplicit() ) { |
| deleteProperties(); |
| } else { |
| saveMetaProperties(); |
| } |
| } |
| |
| public synchronized void stopPingThread() |
| { |
| String methodName = "stopPingThread"; |
| |
| if ( serviceMeta != null ) { |
| logger.debug(methodName, id, "Stopping ping thread, endpoint", endpoint); |
| serviceMeta.stop(); |
| residualMeta = serviceMeta; |
| serviceMeta = null; |
| } |
| |
| if ( !isRegistered() ) { |
| deleteProperties(); |
| } else { |
| saveMetaProperties(); |
| } |
| } |
| |
| synchronized void setResponsive() |
| { |
| setServiceState(ServiceState.Available); |
| saveMetaProperties(); |
| } |
| |
| synchronized void setUnresponsive() |
| { |
| setServiceState(ServiceState.NotAvailable); |
| saveMetaProperties(); |
| } |
| |
| synchronized void setWaiting() |
| { |
| // |
| // Only switch state sometimes .... |
| // |
| switch ( getServiceState() ) { |
| case Available: |
| case NotAvailable: |
| case Undefined: |
| setServiceState(ServiceState.Waiting); |
| break; |
| case Initializing: |
| case Waiting: |
| case Stopping: |
| break; |
| } |
| } |
| |
| // boolean ping() |
| // { |
| // //String methodName = "ping"; |
| // boolean answer = true; |
| // |
| // // Instantiate Uima AS Client |
| // BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl(); |
| // Map<String, Object> appCtx = new HashMap<String, Object>(); |
| // appCtx.put(UimaAsynchronousEngine.ServerUri, broker); |
| // appCtx.put(UimaAsynchronousEngine.Endpoint, endpoint); |
| // appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, ServiceManagerComponent.meta_ping_timeout); // 500 ms should be enough to get GetMeta reply |
| // |
| // try { |
| // // this sends GetMeta request and blocks waiting for a reply |
| // uimaAsEngine.initialize(appCtx); |
| // // logger.info(methodName, null, "Dependent Service Available:", getKey()); |
| // } catch( ResourceInitializationException e) { |
| // // either broker is down or service not available |
| // // logger.error(methodName, null, "Remote service unavailable:", getKey()); |
| // answer = false; |
| // } finally { |
| // uimaAsEngine.stop(); |
| // } |
| // |
| // return answer; |
| // } |
| |
| |
| 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); |
| try { |
| Process p = pb.start(); |
| int rc = p.waitFor(); |
| logger.debug(methodName, null, "Log start errors returns with rc", rc); |
| } catch (IOException e) { |
| // TODO Auto-generated catch block |
| e.printStackTrace(); |
| } catch (InterruptedException e) { |
| // TODO Auto-generated catch block |
| e.printStackTrace(); |
| } |
| |
| } |
| |
| 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()); |
| } |
| |
| |
| /** |
| * This assumes the caller has already verified that I'm a registered service. |
| */ |
| boolean start() |
| { |
| String methodName = "start"; |
| |
| logger.debug(methodName, null, "START START START START START START START START"); |
| this.stopped = false; |
| |
| if ( ! isStartable() ) { |
| establish(); // this will just start the ping thread |
| return true; |
| } |
| |
| // Simple use of ducc_ling, just submit as the user. The specification will have the working directory |
| // and classpath needed for the service, handled by the Orchestrator and Job Driver. |
| String[] args = { |
| System.getProperty("ducc.agent.launcher.ducc_spawn_path"), |
| "-u", |
| user, |
| "--", |
| System.getProperty("ducc.jvm"), |
| "-cp", |
| System.getProperty("java.class.path"), |
| "org.apache.uima.ducc.cli.DuccServiceSubmit", |
| "--specification", |
| props_filename |
| }; |
| |
| for ( int i = 0; i < args.length; i++ ) { |
| if ( i > 0 && (args[i-1].equals("-cp") ) ) { |
| // The classpaths can be just awful filling the logs with junk. It will end up in the agent log |
| // anyway so let's inhibit it here. |
| logger.debug(methodName, null, "Args[", i, "]: <CLASSPATH>"); |
| } else { |
| logger.debug(methodName, null, "Args[", i, "]:", args[i]); |
| } |
| } |
| |
| ProcessBuilder pb = new ProcessBuilder(args); |
| Map<String, String> env = pb.environment(); |
| env.put("DUCC_HOME", System.getProperty("DUCC_HOME")); |
| |
| ArrayList<String> stdout_lines = new ArrayList<String>(); |
| ArrayList<String> stderr_lines = new ArrayList<String>(); |
| try { |
| Process p = pb.start(); |
| |
| int rc = p.waitFor(); |
| logger.debug(methodName, null, "DuccServiceSubmit returns with rc", rc); |
| |
| // TODO: we should attache these streams to readers in threads because too much output |
| // can cause blocking, deadlock, ugliness. |
| InputStream stdout = p.getInputStream(); |
| InputStream stderr = p.getErrorStream(); |
| BufferedReader stdout_reader = new BufferedReader(new InputStreamReader(stdout)); |
| BufferedReader stderr_reader = new BufferedReader(new InputStreamReader(stderr)); |
| String line = null; |
| while ( (line = stdout_reader.readLine()) != null ) { |
| stdout_lines.add(line); |
| } |
| |
| line = null; |
| while ( (line = stderr_reader.readLine()) != null ) { |
| stderr_lines.add(line); |
| } |
| |
| } catch (Throwable t) { |
| // TODO Auto-generated catch block |
| logger.error(methodName, null, t); |
| } |
| |
| for ( String s : stderr_lines ) { |
| logger.info(methodName, id, "Start stderr:", s); |
| } |
| |
| // That was annoying. Now search the lines for some hint of the id. |
| boolean inhibit_cp = false; |
| boolean started = false; |
| StringBuffer submit_buffer = new StringBuffer(); |
| boolean recording = false; |
| for ( String s : stdout_lines ) { |
| |
| // simple logic to inhibit printing the danged classpath |
| if ( inhibit_cp ) { |
| inhibit_cp = false; |
| logger.info(methodName, id, "<INHIBITED CP>"); |
| } else { |
| logger.info(methodName, id, "Start stdout:", s); |
| } |
| |
| if ( s.indexOf("-cp") >= 0 ) { |
| inhibit_cp = true; |
| } |
| |
| if ( recording ) { |
| submit_buffer.append(s.trim()); |
| submit_buffer.append(";"); |
| } |
| if ( s.startsWith("1001 Command launching...") ) { |
| recording = true; |
| continue; |
| } |
| |
| if ( s.startsWith("Service") && s.endsWith("submitted") ) { |
| String[] toks = s.split("\\s"); |
| long friendly = 0; |
| try { |
| friendly = Long.parseLong(toks[1]); |
| friendly_ids.put(friendly, null); |
| persistImplementors(); |
| started = true; |
| logger.info(methodName, null, "Request to start service " + id.toString() + " accepted as job ", friendly); |
| } catch ( NumberFormatException e ) { |
| logger.warn(methodName, null, "Request to start service " + id.toString() + " failed, can't interpret response.: " + s); |
| } |
| |
| } |
| } |
| |
| boolean rc = true; |
| if ( ! started ) { |
| logger.warn(methodName, null, "Request to start service " + id.toString() + " failed."); |
| meta_props.put("submit_error", submit_buffer.toString()); |
| setAutostart(false); |
| log_errors(stdout_lines, stderr_lines); |
| } else { |
| meta_props.remove("submit_error"); |
| setServiceState(ServiceState.Initializing); |
| } |
| saveMetaProperties(); |
| logger.debug(methodName, null, "ENDSTART ENDSTART ENDSTART ENDSTART ENDSTART ENDSTART"); |
| return rc; |
| } |
| |
| /** |
| * This assumes the caller has already verified that I'm a registered service. |
| */ |
| void stopOneProcess(DuccId id) |
| { |
| String methodName = "stop"; |
| |
| String[] args = { |
| System.getProperty("ducc.agent.launcher.ducc_spawn_path"), |
| "-u", |
| user, |
| "--", |
| System.getProperty("ducc.jvm"), |
| "-cp", |
| System.getProperty("java.class.path"), |
| "org.apache.uima.ducc.cli.DuccServiceCancel", |
| "--id", |
| id.toString() |
| }; |
| |
| for ( int i = 0; i < args.length; i++ ) { |
| if ( i > 0 && (args[i-1].equals("-cp") ) ) { |
| // The classpaths can be just awful filling the logs with junk. It will end up in the agent log |
| // anyway so let's inhibit it here. |
| logger.debug(methodName, null, "Args[", i, "]: <CLASSPATH>"); |
| } else { |
| logger.debug(methodName, null, "Args[", i, "]:", args[i]); |
| } |
| } |
| |
| ProcessBuilder pb = new ProcessBuilder(args); |
| Map<String, String> env = pb.environment(); |
| env.put("DUCC_HOME", System.getProperty("DUCC_HOME")); |
| |
| ArrayList<String> stdout_lines = new ArrayList<String>(); |
| ArrayList<String> stderr_lines = new ArrayList<String>(); |
| try { |
| Process p = pb.start(); |
| |
| int rc = p.waitFor(); |
| logger.debug(methodName, null, "DuccServiceCancel returns with rc", rc); |
| |
| InputStream stdout = p.getInputStream(); |
| InputStream stderr = p.getErrorStream(); |
| BufferedReader stdout_reader = new BufferedReader(new InputStreamReader(stdout)); |
| BufferedReader stderr_reader = new BufferedReader(new InputStreamReader(stderr)); |
| |
| String line = null; |
| while ( (line = stdout_reader.readLine()) != null ) { |
| stdout_lines.add(line); |
| } |
| |
| line = null; |
| while ( (line = stderr_reader.readLine()) != null ) { |
| stderr_lines.add(line); |
| } |
| |
| } catch (Throwable t) { |
| // TODO Auto-generated catch block |
| logger.error(methodName, null, t); |
| } |
| |
| boolean inhibit_cp = false; |
| for ( String s : stdout_lines ) { |
| // simple logic to inhibit printing the danged classpath |
| if ( inhibit_cp ) { |
| inhibit_cp = false; |
| logger.info(methodName, id, "<INHIBITED CP>"); |
| } else { |
| logger.info(methodName, id, "Stop stdout:", s); |
| } |
| |
| if ( s.indexOf("-cp") >= 0 ) { |
| inhibit_cp = true; |
| } |
| } |
| |
| for ( String s : stderr_lines ) { |
| logger.info(methodName, id, "Stop stderr:", s); |
| } |
| |
| // is this the last implementor? if so the service is no longer available. |
| // should not have to do this, the state should update correctly in the state machine |
| //if ( implementors.size() <= 1 ) { |
| // service_state = ServiceState.NotAvailable; |
| //} |
| //return new ServiceReplyEvent(ServiceCode.OK, "Start service " + id.toString() + " complete.", toks[1], null); |
| |
| } |
| |
| /** |
| * Stop everything |
| */ |
| void stop() |
| { |
| String methodName = "stop"; |
| logger.debug(methodName, id, "Stopping all implementors"); |
| this.stopped = true; |
| stopPingThread(); |
| setServiceState(ServiceState.Stopping); |
| |
| if ( ! isStartable() ) { // CUSTOM ping-only service |
| return; |
| } |
| |
| for ( DuccId id : implementors.keySet() ) { |
| stopOneProcess(id); |
| } |
| saveMetaProperties(); |
| } |
| |
| /** |
| * Stop 'count' services. |
| * TODO: Put in logic to stop intelligently, i.e. favor processes not yet running |
| */ |
| void stop(int count) |
| { |
| String methodName = "stop(count)"; |
| |
| if ( ! isStartable() ) { // CUSTOM ping-only, only one running, let common code do the honors |
| stop(); |
| return; |
| } |
| |
| logger.debug(methodName, id, "Stopping", count, "implementors"); |
| for ( DuccId id: implementors.keySet() ) { |
| if ( (count--) > 0 ) { |
| stopOneProcess(id); |
| } else { |
| break; |
| } |
| } |
| saveMetaProperties(); |
| } |
| |
| |
| private class LingerTask |
| extends TimerTask |
| { |
| ServiceSet sset; |
| LingerTask(ServiceSet sset) |
| { |
| String methodName = "LingerTask.init"; |
| logger.debug(methodName, id, "Linger starts", linger_time); |
| this.sset = sset; |
| } |
| |
| public void run() |
| { |
| String methodName = "LingerTask.run"; |
| logger.debug(methodName, id, "Lingering stop completes."); |
| // doesn't matter how its started i think, we have to set this flag off when we stop |
| sset.setReferencedStart(false); |
| linger = null; |
| sset.stop(); |
| } |
| } |
| |
| void lingeringStop() |
| { |
| if ( timer == null ) { |
| timer = new Timer(); |
| } |
| linger = new LingerTask(this); |
| timer.schedule(linger, linger_time); |
| } |
| |
| IServiceDescription query() |
| { |
| IServiceDescription sd = new ServiceDescription(); |
| |
| ArrayList<ADuccId> imp = new ArrayList<ADuccId>(); |
| for ( DuccId id : implementors.keySet() ) { |
| imp.add(id); |
| } |
| sd.setImplementors(imp); |
| |
| ArrayList<ADuccId> ref = new ArrayList<ADuccId>(); |
| ref.clear(); |
| for ( DuccId id : references.keySet() ) { |
| ref.add(id); |
| } |
| sd.setReferences(ref); |
| |
| sd.setInstances(getNInstances()); |
| |
| sd.setType(service_type); |
| sd.setSubclass(service_class); |
| sd.setEndpoint(endpoint); |
| sd.setBroker(broker); |
| sd.setServiceState(getServiceState()); |
| //sd.setJobState(job_state); |
| sd.setActive(serviceMeta != null); |
| sd.setStopped(stopped); |
| sd.setAutostart(autostart); |
| sd.setLinger(linger_time); |
| sd.setId(id); |
| sd.setUser(user); |
| sd.setDeregistered(isDeregistered()); |
| |
| 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; |
| } |
| |
| |
| } |
| |
| /** |
| 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 |
| 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 |
| }; |
| */ |
