Author: cwiklik Date: Mon Aug 3 17:29:10 2015 New Revision: 1693942 URL: http://svn.apache.org/r1693942 Log: UIMA-4540 Moved inventory publication beyond logging
Modified:
uima/sandbox/uima-ducc/trunk/uima-ducc-agent/src/main/java/org/apache/uima/ducc/agent/processors/DefaultNodeInventoryProcessor.java
Modified:
uima/sandbox/uima-ducc/trunk/uima-ducc-agent/src/main/java/org/apache/uima/ducc/agent/processors/DefaultNodeInventoryProcessor.java
URL:
http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-agent/src/main/java/org/apache/uima/ducc/agent/processors/DefaultNodeInventoryProcessor.java?rev=1693942&r1=1693941&r2=1693942&view=diff
==============================================================================
---
uima/sandbox/uima-ducc/trunk/uima-ducc-agent/src/main/java/org/apache/uima/ducc/agent/processors/DefaultNodeInventoryProcessor.java
(original)
+++
uima/sandbox/uima-ducc/trunk/uima-ducc-agent/src/main/java/org/apache/uima/ducc/agent/processors/DefaultNodeInventoryProcessor.java
Mon Aug 3 17:29:10 2015
@@ -15,7 +15,7 @@
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
-*/
+ */
package org.apache.uima.ducc.agent.processors;
import java.util.ArrayList;
@@ -38,178 +38,217 @@ import org.apache.uima.ducc.transport.ev
*/
public class DefaultNodeInventoryProcessor implements NodeInventoryProcessor {
DuccLogger logger = new DuccLogger(this.getClass(), "AGENT");
- boolean inventoryChanged=true;
+ boolean inventoryChanged = true;
private NodeAgent agent;
- private HashMap<DuccId, IDuccProcess> previousInventory;
- private int forceInventoryUpdateMaxThreshold=0;
- private long counter=0;
-
- public DefaultNodeInventoryProcessor(NodeAgent agent,String
inventoryPublishRateSkipCount) {
+ private HashMap<DuccId, IDuccProcess> previousInventory;
+ private int forceInventoryUpdateMaxThreshold = 0;
+ private long counter = 0;
+
+ public DefaultNodeInventoryProcessor(NodeAgent agent,
+ String inventoryPublishRateSkipCount) {
this.agent = agent;
try {
- forceInventoryUpdateMaxThreshold =
Integer.parseInt(inventoryPublishRateSkipCount);
- } catch( Exception e) {
+ forceInventoryUpdateMaxThreshold = Integer
+
.parseInt(inventoryPublishRateSkipCount);
+ } catch (Exception e) {
}
- // Dont allow 0
- if ( forceInventoryUpdateMaxThreshold == 0 ) {
- forceInventoryUpdateMaxThreshold = 1;
+ // Dont allow 0
+ if (forceInventoryUpdateMaxThreshold == 0) {
+ forceInventoryUpdateMaxThreshold = 1;
}
}
+
/**
* Get a copy of agent {@code Process} inventory
*/
public HashMap<DuccId, IDuccProcess> getInventory() {
return agent.getInventoryCopy();
}
+
/**
*
*/
public void process(Exchange outgoingMessage) throws Exception {
- String methodName="process";
- // Get a deep copy of agent's inventory
+ String methodName = "process";
+ // Get a deep copy of agent's inventory
HashMap<DuccId, IDuccProcess> inventory = getInventory();
// Determine if the inventory changed since the last publishing
was done
- // First check if the inventory expanded or shrunk. If the same
in size,
+ // First check if the inventory expanded or shrunk. If the same
in size,
// compare process states and PID. If either of the two changed
for any
- // of the processes trigger immediate publish. If no changes
found, publish
- // according to skip counter
(ducc.agent.node.inventory.publish.rate.skip)
+ // of the processes trigger immediate publish. If no changes
found,
+ // publish
+ // according to skip counter
+ // (ducc.agent.node.inventory.publish.rate.skip)
// configured in ducc.properties.
- if ( previousInventory != null ) {
+ if (previousInventory != null) {
if (inventory.size() != previousInventory.size()) {
inventoryChanged = true;
} else {
- // Inventory maps are equal in size, check if
all processes in the current
- // inventory exist in the previous inventory
snapshot. If not, it means that
- // that perhaps a new process was added and one
was removed. In this case,
+ // Inventory maps are equal in size, check if
all processes in
+ // the current
+ // inventory exist in the previous inventory
snapshot. If not,
+ // it means that
+ // that perhaps a new process was added and one
was removed. In
+ // this case,
// force the publish, since there was a change.
- for( Map.Entry<DuccId, IDuccProcess>
currentProcess: inventory.entrySet()) {
- // Check if a process in the
current inventory exists in a previous
- // inventory snapshot
- if (
previousInventory.containsKey(currentProcess.getKey())) {
- IDuccProcess previousProcess =
-
previousInventory.get(currentProcess.getKey());
- // check if either PID or
process state has changed
- if (
currentProcess.getValue().getPID() != null &&
-
previousProcess.getPID() == null) {
+ for (Map.Entry<DuccId, IDuccProcess>
currentProcess : inventory
+ .entrySet()) {
+ // Check if a process in the current
inventory exists in a
+ // previous
+ // inventory snapshot
+ if
(previousInventory.containsKey(currentProcess.getKey())) {
+ IDuccProcess previousProcess =
previousInventory
+
.get(currentProcess.getKey());
+ // check if either PID or
process state has changed
+ if
(currentProcess.getValue().getPID() != null
+ &&
previousProcess.getPID() == null) {
inventoryChanged = true;
break;
- } else if (
!currentProcess.getValue().getProcessState().equals(previousProcess.getProcessState()))
{
+ } else if
(!currentProcess.getValue().getProcessState()
+
.equals(previousProcess.getProcessState())) {
inventoryChanged = true;
break;
} else {
-
List<IUimaPipelineAEComponent> breakdown =
-
currentProcess.getValue().getUimaPipelineComponents();
- if ( breakdown != null
&& breakdown.size() > 0 ) {
-
List<IUimaPipelineAEComponent> previousBreakdown =
-
previousProcess.getUimaPipelineComponents();
- if (
previousBreakdown == null || previousBreakdown.size() == 0 ||
-
breakdown.size() != previousBreakdown.size()) {
+
List<IUimaPipelineAEComponent> breakdown = currentProcess
+
.getValue().getUimaPipelineComponents();
+ if (breakdown != null
&& breakdown.size() > 0) {
+
List<IUimaPipelineAEComponent> previousBreakdown = previousProcess
+
.getUimaPipelineComponents();
+ if
(previousBreakdown == null
+
|| previousBreakdown.size() == 0
+
|| breakdown.size() != previousBreakdown
+
.size()) {
inventoryChanged = true;
} else {
- for
(IUimaPipelineAEComponent uimaAeState : breakdown ) {
- boolean found = false;
- for
(IUimaPipelineAEComponent previousUimaAeState : previousBreakdown ) {
- if
( uimaAeState.getAeName().equals(previousUimaAeState.getAeName()) ) {
-
found = true;
-
if ( !uimaAeState.getAeState().equals(previousUimaAeState.getAeState()) ||
-
uimaAeState.getInitializationTime() !=
previousUimaAeState.getInitializationTime() ) {
-
inventoryChanged = true;
+ for
(IUimaPipelineAEComponent uimaAeState : breakdown) {
+
boolean found = false;
+
for (IUimaPipelineAEComponent previousUimaAeState : previousBreakdown) {
+
if (uimaAeState.getAeName().equals(
+
previousUimaAeState
+
.getAeName())) {
+
found = true;
+
if (!uimaAeState
+
.getAeState()
+
.equals(previousUimaAeState
+
.getAeState())
+
|| uimaAeState
+
.getInitializationTime() !=
previousUimaAeState
+
.getInitializationTime()) {
+
inventoryChanged = true;
+
break;
+
}
+
}
+
}
+
if (!found) {
+
inventoryChanged = true;
+
}
+
+
if (inventoryChanged) {
break;
-
}
- }
- }
- if (
!found ) {
-
inventoryChanged = true;
- }
-
- if (
inventoryChanged ) {
-
break;
- }
+
}
- }
+ }
}
-
+
}
}
} else {
- // New inventory contains a
process not in the previous snapshot
+ // New inventory contains a
process not in the previous
+ // snapshot
inventoryChanged = true;
break;
}
}
}
}
-
- // Get this inventory snapshot
+
+ // Get this inventory snapshot
previousInventory = inventory;
- // Broadcast inventory if there is a change or configured
number of epochs
- // passed since the last broadcast. This is configured in
ducc.properties with
- // property ducc.agent.node.inventory.publish.rate.skip
+ // Broadcast inventory if there is a change or configured
number of
+ // epochs
+ // passed since the last broadcast. This is configured in
+ // ducc.properties with
+ // property ducc.agent.node.inventory.publish.rate.skip
try {
- if ( inventory.size() > 0 &&
- ( inventoryChanged || // if there is
inventory change, publish
- forceInventoryUpdateMaxThreshold == 0
|| // skip rate in ducc.properties is zero, publish
- ( counter > 0 && (counter %
forceInventoryUpdateMaxThreshold ) == 0)) ) { // if reached skip rate, publish
-
- outgoingMessage.getIn().setBody(new
NodeInventoryUpdateDuccEvent(inventory));
- StringBuffer sb = new StringBuffer("Node
Inventory ("+inventory.size()+")");
- for( Map.Entry<DuccId, IDuccProcess> p :
inventory.entrySet()) {
+ if (inventory.size() > 0 && (inventoryChanged || // if
there is
+
// inventory
+
// change,
+
// publish
+ forceInventoryUpdateMaxThreshold == 0
|| // skip rate in
+
// ducc.properties
+
// is zero,
+
// publish
+ (counter > 0 && (counter %
forceInventoryUpdateMaxThreshold) == 0))) { // if
+
// reached
+
// skip
+
// rate,
+
// publish
+
+ StringBuffer sb = new StringBuffer("Node
Inventory ("
+ + inventory.size() + ")");
+ for (Map.Entry<DuccId, IDuccProcess> p :
inventory.entrySet()) {
/*
- long endInitLong = 0;
- String endInit = "";
- ITimeWindow wInit =
p.getValue().getTimeWindowInit();
- if(wInit != null) {
- endInit = wInit.getEnd();
- endInitLong =
wInit.getEndLong();
- }
- long startRunLong = 0;
- String startRun = "";
- ITimeWindow wRun =
p.getValue().getTimeWindowRun();
- if(wRun != null) {
- startRun = wRun.getStart();
- startRunLong =
wRun.getStartLong();
+ * long endInitLong = 0; String endInit
= ""; ITimeWindow
+ * wInit =
p.getValue().getTimeWindowInit(); if(wInit !=
+ * null) { endInit = wInit.getEnd();
endInitLong =
+ * wInit.getEndLong(); } long
startRunLong = 0; String
+ * startRun = ""; ITimeWindow wRun =
+ * p.getValue().getTimeWindowRun();
if(wRun != null) {
+ * startRun = wRun.getStart();
startRunLong =
+ * wRun.getStartLong(); }
if(endInitLong > startRunLong) {
+ * logger.warn(methodName, null,
+ * "endInit:"+endInitLong+"
"+"startRun:"+startRunLong); }
+ */
+ int pipelineInitStats = (p.getValue()
+
.getUimaPipelineComponents() == null) ? 0 : p
+
.getValue().getUimaPipelineComponents().size();
+
+ if
(p.getValue().getUimaPipelineComponents() == null) {
+
p.getValue().setUimaPipelineComponents(
+ new
ArrayList<IUimaPipelineAEComponent>());
}
- if(endInitLong > startRunLong) {
- logger.warn(methodName, null,
"endInit:"+endInitLong+" "+"startRun:"+startRunLong);
+ sb.append("\n\t[Process Type=")
+
.append(p.getValue().getProcessType())
+ .append(" DUCC ID=")
+
.append(p.getValue().getDuccId())
+ .append(" PID=")
+
.append(p.getValue().getPID())
+ .append(" State=")
+
.append(p.getValue().getProcessState())
+ .append(" Resident
Memory=")
+
.append(p.getValue().getResidentMemory())
+ .append(" Init Stats
List Size:"
+ +
pipelineInitStats).
+ // append(" end
init:"+endInit).
+ // append(" start
run:"+startRun).
+ append("] ");
+ if (p.getValue().getProcessState()
+
.equals(ProcessState.Stopped)
+ ||
p.getValue().getProcessState()
+
.equals(ProcessState.Failed)
+ ||
p.getValue().getProcessState()
+
.equals(ProcessState.Killed)) {
+ sb.append(" Reason:"
+ +
p.getValue().getReasonForStoppingProcess());
}
- */
- int pipelineInitStats =
(p.getValue().getUimaPipelineComponents() == null ) ? 0 :
- p.getValue().getUimaPipelineComponents().size();
-
- if ( p.getValue().getUimaPipelineComponents() == null) {
- p.getValue().setUimaPipelineComponents(new
ArrayList<IUimaPipelineAEComponent>());
- }
- sb.append("\n\t[Process Type=").
-
append(p.getValue().getProcessType()).
- append(" DUCC ID=").
- append(p.getValue().getDuccId()).
- append(" PID=").
- append(p.getValue().getPID()).
- append("
State=").append(p.getValue().getProcessState()).
- append(" Resident
Memory=").append(p.getValue().getResidentMemory()).
- append(" Init Stats List
Size:"+pipelineInitStats).
- //append(" end init:"+endInit).
- //append(" start run:"+startRun).
- append("] ");
- if ( p.getValue().getProcessState().equals(ProcessState.Stopped) ||
- p.getValue().getProcessState().equals(ProcessState.Failed) ||
- p.getValue().getProcessState().equals(ProcessState.Killed)) {
- sb.append(" Reason:"+p.getValue().getReasonForStoppingProcess());
- }
- sb.append(" Exit Code="+p.getValue().getProcessExitCode());
+ sb.append(" Exit Code=" +
p.getValue().getProcessExitCode());
}
- logger.info(methodName, null, "Agent
"+agent.getIdentity().getName()+" Posting Inventory:"+sb.toString());
+ logger.info(methodName, null, "Agent "
+ + agent.getIdentity().getName()
+ " Posting Inventory:"
+ + sb.toString());
+ outgoingMessage.getIn().setBody(new
NodeInventoryUpdateDuccEvent(inventory));
+
} else {
- // Add null to the body of the message. A filter
- // defined in the Camel route
(AgentConfiguration.java)
- // has a predicate to check for null body and throws
- // away such a message.
- outgoingMessage.getIn().setBody(null);
+ // Add null to the body of the message. A filter
+ // defined in the Camel route
(AgentConfiguration.java)
+ // has a predicate to check for null body and
throws
+ // away such a message.
+ outgoingMessage.getIn().setBody(null);
}
- } catch( Exception e) {
+ } catch (Exception e) {
logger.error(methodName, null, e);
} finally {
- if ( inventoryChanged ) {
+ if (inventoryChanged) {
counter = 0;
} else {
counter++;
