You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@uima.apache.org by de...@apache.org on 2015/05/20 18:05:25 UTC
svn commit: r1680619 - in
/uima/sandbox/uima-ducc/trunk/uima-ducc-orchestrator/src/main/java/org/apache/uima/ducc/orchestrator:
Orchestrator.java OrchestratorComponent.java
config/OrchestratorConfiguration.java
Author: degenaro
Date: Wed May 20 16:05:24 2015
New Revision: 1680619
URL: http://svn.apache.org/r1680619
Log:
UIMA-4394 DUCC Orchestrator (OR) reduce publication size comprising command line & especially classpath
Modified:
uima/sandbox/uima-ducc/trunk/uima-ducc-orchestrator/src/main/java/org/apache/uima/ducc/orchestrator/Orchestrator.java
uima/sandbox/uima-ducc/trunk/uima-ducc-orchestrator/src/main/java/org/apache/uima/ducc/orchestrator/OrchestratorComponent.java
uima/sandbox/uima-ducc/trunk/uima-ducc-orchestrator/src/main/java/org/apache/uima/ducc/orchestrator/config/OrchestratorConfiguration.java
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-orchestrator/src/main/java/org/apache/uima/ducc/orchestrator/Orchestrator.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-orchestrator/src/main/java/org/apache/uima/ducc/orchestrator/Orchestrator.java?rev=1680619&r1=1680618&r2=1680619&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-orchestrator/src/main/java/org/apache/uima/ducc/orchestrator/Orchestrator.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-orchestrator/src/main/java/org/apache/uima/ducc/orchestrator/Orchestrator.java Wed May 20 16:05:24 2015
@@ -24,7 +24,6 @@ import org.apache.uima.ducc.transport.ev
import org.apache.uima.ducc.transport.event.DuccWorkRequestEvent;
import org.apache.uima.ducc.transport.event.JdRequestEvent;
import org.apache.uima.ducc.transport.event.NodeInventoryUpdateDuccEvent;
-import org.apache.uima.ducc.transport.event.OrchestratorAbbreviatedStateDuccEvent;
import org.apache.uima.ducc.transport.event.OrchestratorStateDuccEvent;
import org.apache.uima.ducc.transport.event.RmStateDuccEvent;
import org.apache.uima.ducc.transport.event.SmStateDuccEvent;
@@ -39,7 +38,6 @@ public interface Orchestrator {
public void reconcileJdState(JdRequestEvent duccEvent);
public void reconcileNodeInventory(NodeInventoryUpdateDuccEvent duccEvent);
public OrchestratorStateDuccEvent getState();
- public OrchestratorAbbreviatedStateDuccEvent getAbbreviatedState();
public void startJob(SubmitJobDuccEvent duccEvent);
public void stopJob(CancelJobDuccEvent duccEvent);
public void stopJobProcess(CancelJobDuccEvent duccEvent);
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-orchestrator/src/main/java/org/apache/uima/ducc/orchestrator/OrchestratorComponent.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-orchestrator/src/main/java/org/apache/uima/ducc/orchestrator/OrchestratorComponent.java?rev=1680619&r1=1680618&r2=1680619&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-orchestrator/src/main/java/org/apache/uima/ducc/orchestrator/OrchestratorComponent.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-orchestrator/src/main/java/org/apache/uima/ducc/orchestrator/OrchestratorComponent.java Wed May 20 16:05:24 2015
@@ -57,7 +57,6 @@ import org.apache.uima.ducc.transport.ev
import org.apache.uima.ducc.transport.event.IDuccContext.DuccContext;
import org.apache.uima.ducc.transport.event.JdRequestEvent;
import org.apache.uima.ducc.transport.event.NodeInventoryUpdateDuccEvent;
-import org.apache.uima.ducc.transport.event.OrchestratorAbbreviatedStateDuccEvent;
import org.apache.uima.ducc.transport.event.OrchestratorStateDuccEvent;
import org.apache.uima.ducc.transport.event.RmStateDuccEvent;
import org.apache.uima.ducc.transport.event.SmStateDuccEvent;
@@ -446,7 +445,7 @@ implements Orchestrator {
public OrchestratorStateDuccEvent getState() {
String methodName = "getState";
logger.trace(methodName, null, messages.fetch("enter"));
- OrchestratorStateDuccEvent orchestratorStateDuccEvent = new OrchestratorStateDuccEvent();
+ OrchestratorStateDuccEvent orchestratorStateDuccEvent = new OrchestratorStateDuccEvent(logger);
try {
DuccWorkMap workMapCopy = WorkMapHelper.deepCopy(workMap, this, methodName);
int activeJobs = workMapCopy.getJobCount();
@@ -472,39 +471,7 @@ implements Orchestrator {
logger.trace(methodName, null, messages.fetch("exit"));
return orchestratorStateDuccEvent;
}
- /**
- * Publish Orchestrator Abbreviated State
- */
-
- public OrchestratorAbbreviatedStateDuccEvent getAbbreviatedState() {
- String methodName = "getAbbreviatedState";
- logger.trace(methodName, null, messages.fetch("enter"));
- OrchestratorAbbreviatedStateDuccEvent orchestratorAbbreviatedStateDuccEvent = new OrchestratorAbbreviatedStateDuccEvent();
- try {
- DuccWorkMap workMapCopy = WorkMapHelper.deepCopy(workMap, this, methodName);
- int activeJobs = workMapCopy.getJobCount();
- int activeReservations = workMapCopy.getReservationCount();
- int activeServices = workMapCopy.getServiceCount();
- logger.debug(methodName, null, messages.fetch("publishing state")+" "+
- messages.fetchLabel("active job count")+activeJobs
- +" "+
- messages.fetchLabel("active reservation count")+activeReservations
- +" "+
- messages.fetchLabel("active service count")+activeServices
- );
- int jobDriverNodeCount = hostManager.nodes();
- workMapCopy.setJobDriverNodeCount(jobDriverNodeCount);
- orchestratorAbbreviatedStateDuccEvent.setWorkMap(workMapCopy);
- //stateManager.prune(workMapCopy);
- //healthMonitor.cancelNonViableJobs();
- //mqReaper.removeUnusedJdQueues(workMapCopy);
- }
- catch(Throwable t) {
- logger.error(methodName, null, t);
- }
- logger.trace(methodName, null, messages.fetch("exit"));
- return orchestratorAbbreviatedStateDuccEvent;
- }
+
@SuppressWarnings("unchecked")
private void submitError(Properties properties, String error_message) {
String key = SpecificationProperties.key_submit_errors;
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-orchestrator/src/main/java/org/apache/uima/ducc/orchestrator/config/OrchestratorConfiguration.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-orchestrator/src/main/java/org/apache/uima/ducc/orchestrator/config/OrchestratorConfiguration.java?rev=1680619&r1=1680618&r2=1680619&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-orchestrator/src/main/java/org/apache/uima/ducc/orchestrator/config/OrchestratorConfiguration.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-orchestrator/src/main/java/org/apache/uima/ducc/orchestrator/config/OrchestratorConfiguration.java Wed May 20 16:05:24 2015
@@ -48,7 +48,6 @@ import org.apache.uima.ducc.transport.ev
import org.apache.uima.ducc.transport.event.DuccWorkRequestEvent;
import org.apache.uima.ducc.transport.event.JdReplyEvent;
import org.apache.uima.ducc.transport.event.JdRequestEvent;
-import org.apache.uima.ducc.transport.event.OrchestratorAbbreviatedStateDuccEvent;
import org.apache.uima.ducc.transport.event.OrchestratorStateDuccEvent;
import org.apache.uima.ducc.transport.event.SubmitJobDuccEvent;
import org.apache.uima.ducc.transport.event.SubmitJobReplyDuccEvent;
@@ -271,63 +270,6 @@ public class OrchestratorConfiguration {
}
}
-
- /**
- * Creates Camel router that will publish Orchestrator abbreviated state at regular intervals.
- *
- * @param targetEndpointToReceiveOrchestratorAbbreviatedStateUpdate - endpoint where to publish state
- * @param statePublishRate - how often to publish state
- * @return
- * @throws Exception
- */
- private RouteBuilder routeBuilderForOrchestratorAbbreviatedStatePost(final Orchestrator orchestrator, final String targetEndpointToReceiveOrchestratorAbbreviatedStateUpdate, final int statePublishRate) throws Exception {
- final OrchestratorAbbreviatedStateProcessor orchestratorp = // an object responsible for generating the state
- new OrchestratorAbbreviatedStateProcessor(orchestrator);
-
- return new RouteBuilder() {
- public void configure() {
-
- final Predicate blastFilter = new DuccBlastGuardPredicate(duccLogger);
-
- from("timer:orchestratorAbbreviatedStateDumpTimer?fixedRate=true&period=" + statePublishRate)
- // This route uses a filter to prevent sudden bursts of messages which
- // may flood DUCC daemons causing chaos. The filter disposes any event
- // that appears in a window of 1 sec or less.
- .filter(blastFilter)
- //.process(xmStart)
- .process(orchestratorp)
- //.process(xmEnded)
- .to(targetEndpointToReceiveOrchestratorAbbreviatedStateUpdate)
- ;
- }
- };
- }
-
- /**
- * Camel Processor responsible for generating Orchestrator's state.
- *
- */
- private class OrchestratorAbbreviatedStateProcessor implements Processor {
- private Orchestrator orchestrator;
-
- private OrchestratorAbbreviatedStateProcessor(Orchestrator orchestrator) {
- this.orchestrator = orchestrator;
- }
- public void process(Exchange exchange) throws Exception {
- String location = "OrchestratorAbbreviatedStateProcessor.process";
- // Fetch new state from Orchestrator
- OrchestratorAbbreviatedStateDuccEvent jse = orchestrator.getAbbreviatedState();
- // add sequence number to the outgoing message. This should be used to manage
- // processing order in the consumer
- OrchestratorState orchestratorState = OrchestratorState.getInstance();
- long seqNo = orchestratorState.getNextSequenceNumberStateAbbreviated();
- duccLogger.debug(location, jobid, ""+seqNo);
- jse.setSequence(seqNo);
- // Add the state object to the Message
- exchange.getIn().setBody(jse);
- }
- }
-
/**
* Instantiate a listener to which Camel will route a body of the incoming message.
* The listener should provide a method for each object class it expects to receive.
@@ -358,7 +300,6 @@ public class OrchestratorConfiguration {
orchestrator.getContext().addRoutes(this.routeBuilderForEndpoint(common.jdStateUpdateEndpoint,delegateListener));
orchestrator.getContext().addRoutes(this.routeBuilderForEndpoint(common.nodeInventoryEndpoint,delegateListener));
orchestrator.getContext().addRoutes(this.routeBuilderForOrchestratorStatePost(orchestrator, common.orchestratorStateUpdateEndpoint, Integer.parseInt(common.orchestratorStatePublishRate)));
- orchestrator.getContext().addRoutes(this.routeBuilderForOrchestratorAbbreviatedStatePost(orchestrator, common.orchestratorAbbreviatedStateUpdateEndpoint, Integer.parseInt(common.orchestratorAbbreviatedStatePublishRate)));
return orchestrator;
}
public class ErrorProcessor implements Processor {