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/01/06 16:37:27 UTC
svn commit: r1649836 - in
/uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container:
common/ common/fault/ common/fault/injector/ common/fsm/ jd/ jd/fsm/wi/
jd/log/ jd/mh/ jd/timeout/ jd/wi/
Author: degenaro
Date: Tue Jan 6 15:37:26 2015
New Revision: 1649836
URL: http://svn.apache.org/r1649836
Log: (empty)
Added:
uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/common/fault/
uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/common/fault/injector/
uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/common/fault/injector/FaultInjector.java (with props)
uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionAckTimeout.java (with props)
uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionEndTimeout.java (with props)
uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/wi/WiTracker.java (with props)
Modified:
uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/common/Standardize.java
uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/common/fsm/Fsm.java
uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/JobDriver.java
uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionAck.java
uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionEnd.java
uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionGet.java
uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionInProgress.java
uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/WiFsm.java
uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/log/LoggerHelper.java
uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/mh/MessageHandler.java
uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/timeout/ITimeoutManager.java
uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/timeout/TimeoutManager.java
uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/timeout/TimeoutTask.java
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/common/Standardize.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/common/Standardize.java?rev=1649836&r1=1649835&r2=1649836&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/common/Standardize.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/common/Standardize.java Tue Jan 6 15:37:26 2015
@@ -21,6 +21,10 @@ package org.apache.uima.ducc.container.c
public class Standardize {
public enum Label {
+ id,
+ skip,
+ dispatched,
+ deadline,
sum,
count,
total,
Added: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/common/fault/injector/FaultInjector.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/common/fault/injector/FaultInjector.java?rev=1649836&view=auto
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/common/fault/injector/FaultInjector.java (added)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/common/fault/injector/FaultInjector.java Tue Jan 6 15:37:26 2015
@@ -0,0 +1,63 @@
+/*
+ * 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.container.common.fault.injector;
+
+import java.util.Random;
+
+import org.apache.uima.ducc.container.net.iface.IMetaCasTransaction;
+import org.apache.uima.ducc.container.net.iface.IMetaCasTransaction.Type;
+
+public class FaultInjector {
+
+ public static boolean quitAfterGet = false;
+ public static boolean quitAfterAck = false;
+
+ private static int seed = 1;
+ private static int pctFail = 25;
+ private static Random random = new Random(seed);
+
+ public static long getTimeToFailure(IMetaCasTransaction trans) {
+ long time = 0;
+ Type type = trans.getType();
+ switch(type) {
+ case Get:
+ if(quitAfterGet) {
+ if(random.nextInt() < pctFail) {
+ long slack = 10;
+ long delay = 60;
+ time = (delay+slack)*1000;
+ }
+ }
+ break;
+ case Ack:
+ if(quitAfterAck) {
+ if(random.nextInt() < pctFail) {
+ long slack = 10;
+ long delay = 60;
+ time = (delay+slack)*1000;
+ }
+ }
+ break;
+ case End:
+ break;
+ }
+ return time;
+ }
+
+}
Propchange: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/common/fault/injector/FaultInjector.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/common/fault/injector/FaultInjector.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/common/fsm/Fsm.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/common/fsm/Fsm.java?rev=1649836&r1=1649835&r2=1649836&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/common/fsm/Fsm.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/common/fsm/Fsm.java Tue Jan 6 15:37:26 2015
@@ -126,6 +126,7 @@ public class Fsm implements IFsmBuilder
MessageBuffer mb = new MessageBuffer();
mb.append(Standardize.Label.curr.get()+getStateCurrent().getName());
mb.append(Standardize.Label.prev.get()+getStatePrevious().getName());
+ mb.append(Standardize.Label.event.get()+event.getName());
logger.trace(location, ILogger.null_id, mb.toString());
}
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/JobDriver.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/JobDriver.java?rev=1649836&r1=1649835&r2=1649836&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/JobDriver.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/JobDriver.java Tue Jan 6 15:37:26 2015
@@ -69,6 +69,7 @@ public class JobDriver {
private String jobId = null;
private String logDir = null;
+ private long workItemTimeoutMillis = 24*60*60*1000;
private ConcurrentHashMap<IRemoteWorkerThread, IWorkItem> remoteThreadMap = null;
private ConcurrentHashMap<IRemotePid, IProcessStatistics> remoteProcessMap = null;
private IWorkItemStatistics wis = null;
@@ -96,6 +97,7 @@ public class JobDriver {
FlagsExtendedHelper feh = FlagsExtendedHelper.getInstance();
jobId = feh.getJobId();
logDir = feh.getLogDirectory();
+ setWorkItemTimeout();
remoteThreadMap = new ConcurrentHashMap<IRemoteWorkerThread, IWorkItem>();
remoteProcessMap = new ConcurrentHashMap<IRemotePid, IProcessStatistics>();
wis = new WorkItemStatistics();
@@ -112,6 +114,21 @@ public class JobDriver {
}
}
+ private void setWorkItemTimeout() {
+ String location = "setWorkItemTimeout";
+ try {
+ FlagsExtendedHelper feh = FlagsExtendedHelper.getInstance();
+ String workItemTimeout = feh.getWorkItemTimeout();
+ workItemTimeoutMillis = Long.parseLong(workItemTimeout)*(60*1000);
+ MessageBuffer mb = new MessageBuffer();
+ mb.append(Standardize.Label.value.get()+workItemTimeoutMillis);
+ logger.trace(location, ILogger.null_id, mb.toString());
+ }
+ catch(Exception e) {
+ logger.error(location, ILogger.null_id, e);
+ }
+ }
+
public String getJobId() {
return jobId;
}
@@ -144,6 +161,10 @@ public class JobDriver {
return ddManager;
}
+ public long getWorkItemTimeoutMillis() {
+ return workItemTimeoutMillis;
+ }
+
public IWorkItemStateKeeper getWorkItemStateKeeper() {
return wisk;
}
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionAck.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionAck.java?rev=1649836&r1=1649835&r2=1649836&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionAck.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionAck.java Tue Jan 6 15:37:26 2015
@@ -26,7 +26,6 @@ import org.apache.uima.ducc.container.co
import org.apache.uima.ducc.container.common.logger.ILogger;
import org.apache.uima.ducc.container.common.logger.Logger;
import org.apache.uima.ducc.container.jd.JobDriver;
-import org.apache.uima.ducc.container.jd.fault.injector.FaultInjector;
import org.apache.uima.ducc.container.jd.log.LoggerHelper;
import org.apache.uima.ducc.container.jd.timeout.TimeoutManager;
import org.apache.uima.ducc.container.jd.wi.IWorkItem;
@@ -53,9 +52,6 @@ public class ActionAck implements IActio
IWorkItemStateKeeper wisk = jd.getWorkItemStateKeeper();
MetaCasHelper metaCasHelper = new MetaCasHelper(metaCas);
if(metaCas != null) {
- if(FaultInjector.missingAck(actionData)) {
- return;
- }
//
TimeoutManager toMgr = TimeoutManager.getInstance();
toMgr.receivedAck(actionData);
Added: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionAckTimeout.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionAckTimeout.java?rev=1649836&view=auto
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionAckTimeout.java (added)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionAckTimeout.java Tue Jan 6 15:37:26 2015
@@ -0,0 +1,97 @@
+/*
+ * 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.container.jd.fsm.wi;
+
+import org.apache.uima.ducc.common.jd.files.workitem.IWorkItemStateKeeper;
+import org.apache.uima.ducc.container.common.MessageBuffer;
+import org.apache.uima.ducc.container.common.MetaCasHelper;
+import org.apache.uima.ducc.container.common.fsm.iface.IAction;
+import org.apache.uima.ducc.container.common.logger.IComponent;
+import org.apache.uima.ducc.container.common.logger.ILogger;
+import org.apache.uima.ducc.container.common.logger.Logger;
+import org.apache.uima.ducc.container.jd.JobDriver;
+import org.apache.uima.ducc.container.jd.JobDriverHelper;
+import org.apache.uima.ducc.container.jd.cas.CasManager;
+import org.apache.uima.ducc.container.jd.cas.CasManagerStats.RetryReason;
+import org.apache.uima.ducc.container.jd.log.LoggerHelper;
+import org.apache.uima.ducc.container.jd.mh.iface.remote.IRemoteWorkerProcess;
+import org.apache.uima.ducc.container.jd.wi.IProcessStatistics;
+import org.apache.uima.ducc.container.jd.wi.IWorkItem;
+import org.apache.uima.ducc.container.jd.wi.WiTracker;
+import org.apache.uima.ducc.container.net.iface.IMetaCas;
+
+public class ActionAckTimeout extends Action implements IAction {
+
+ private static Logger logger = Logger.getLogger(ActionAckTimeout.class, IComponent.Id.JD.name());
+
+ @Override
+ public String getName() {
+ return ActionAckTimeout.class.getName();
+ }
+
+ private void preemptWorkItem(IActionData actionData, CasManager cm, IMetaCas metaCas) {
+ String location = "preemptWorkItem";
+ cm.putMetaCas(metaCas, RetryReason.TimeoutRetry);
+ cm.getCasManagerStats().incEndRetry();
+ MessageBuffer mb = LoggerHelper.getMessageBuffer(actionData);
+ logger.info(location, ILogger.null_id, mb.toString());
+ }
+
+ @Override
+ public void engage(Object objectData) {
+ String location = "engage";
+ logger.trace(location, ILogger.null_id, "");
+ IActionData actionData = (IActionData) objectData;
+ try {
+ IWorkItem wi = actionData.getWorkItem();
+ IMetaCas metaCas = wi.getMetaCas();
+ JobDriver jd = JobDriver.getInstance();
+ CasManager cm = jd.getCasManager();
+ JobDriverHelper jdh = JobDriverHelper.getInstance();
+ IRemoteWorkerProcess rwp = jdh.getRemoteWorkerProcess(wi);
+ if(rwp != null) {
+ if(metaCas != null) {
+ WiTracker.getInstance().unassign(wi);
+ preemptWorkItem(actionData, cm, metaCas);
+ IWorkItemStateKeeper wisk = jd.getWorkItemStateKeeper();
+ MetaCasHelper metaCasHelper = new MetaCasHelper(metaCas);
+ IProcessStatistics pStats = jdh.getProcessStatistics(rwp);
+ int seqNo = metaCasHelper.getSystemKey();
+ wisk.preempt(seqNo);
+ pStats.preempt(wi);
+ displayProcessStatistics(logger, actionData, wi, pStats);
+ wi.reset();
+ }
+ else {
+ MessageBuffer mb = LoggerHelper.getMessageBuffer(actionData);
+ mb.append("No CAS found for processing");
+ logger.info(location, ILogger.null_id, mb.toString());
+ }
+ }
+ else {
+ MessageBuffer mb = LoggerHelper.getMessageBuffer(actionData);
+ mb.append("No remote worker process entry found for processing");
+ logger.info(location, ILogger.null_id, mb.toString());
+ }
+ }
+ catch(Exception e) {
+ logger.error(location, ILogger.null_id, e);
+ }
+ }
+}
Propchange: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionAckTimeout.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionAckTimeout.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionEnd.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionEnd.java?rev=1649836&r1=1649835&r2=1649836&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionEnd.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionEnd.java Tue Jan 6 15:37:26 2015
@@ -37,15 +37,17 @@ import org.apache.uima.ducc.container.jd
import org.apache.uima.ducc.container.jd.cas.CasManagerStats.RetryReason;
import org.apache.uima.ducc.container.jd.classload.ProxyJobDriverDirective;
import org.apache.uima.ducc.container.jd.classload.ProxyJobDriverErrorHandler;
-import org.apache.uima.ducc.container.jd.fault.injector.FaultInjector;
import org.apache.uima.ducc.container.jd.log.ErrorLogger;
import org.apache.uima.ducc.container.jd.log.LoggerHelper;
import org.apache.uima.ducc.container.jd.mh.RemoteWorkerProcess;
+import org.apache.uima.ducc.container.jd.mh.RemoteWorkerThread;
import org.apache.uima.ducc.container.jd.mh.iface.remote.IRemoteWorkerProcess;
+import org.apache.uima.ducc.container.jd.mh.iface.remote.IRemoteWorkerThread;
import org.apache.uima.ducc.container.jd.timeout.TimeoutManager;
import org.apache.uima.ducc.container.jd.wi.IProcessStatistics;
import org.apache.uima.ducc.container.jd.wi.IWorkItem;
import org.apache.uima.ducc.container.jd.wi.IWorkItemStatistics;
+import org.apache.uima.ducc.container.jd.wi.WiTracker;
import org.apache.uima.ducc.container.jd.wi.perf.IWorkItemPerformanceKeeper;
import org.apache.uima.ducc.container.net.iface.IMetaCas;
import org.apache.uima.ducc.container.net.iface.IMetaCasTransaction;
@@ -229,6 +231,7 @@ public class ActionEnd extends Action im
try {
IWorkItem wi = actionData.getWorkItem();
IMetaCasTransaction trans = actionData.getMetaCasTransaction();
+ IRemoteWorkerThread rwt = new RemoteWorkerThread(trans);
IRemoteWorkerProcess rwp = new RemoteWorkerProcess(trans);
IMetaCas metaCas = wi.getMetaCas();
JobDriver jd = JobDriver.getInstance();
@@ -240,13 +243,11 @@ public class ActionEnd extends Action im
IProcessStatistics pStats = jdh.getProcessStatistics(rwp);
//
if(metaCas != null) {
- if(FaultInjector.missingEnd(actionData)) {
- return;
- }
+ WiTracker.getInstance().unassign(rwt);
//
TimeoutManager toMgr = TimeoutManager.getInstance();
toMgr.receivedAck(actionData);
- toMgr.pendingEnd(actionData);
+ toMgr.receivedEnd(actionData);
//
int seqNo = metaCasHelper.getSystemKey();
Object exception = metaCas.getUserSpaceException();
Added: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionEndTimeout.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionEndTimeout.java?rev=1649836&view=auto
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionEndTimeout.java (added)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionEndTimeout.java Tue Jan 6 15:37:26 2015
@@ -0,0 +1,97 @@
+/*
+ * 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.container.jd.fsm.wi;
+
+import org.apache.uima.ducc.common.jd.files.workitem.IWorkItemStateKeeper;
+import org.apache.uima.ducc.container.common.MessageBuffer;
+import org.apache.uima.ducc.container.common.MetaCasHelper;
+import org.apache.uima.ducc.container.common.fsm.iface.IAction;
+import org.apache.uima.ducc.container.common.logger.IComponent;
+import org.apache.uima.ducc.container.common.logger.ILogger;
+import org.apache.uima.ducc.container.common.logger.Logger;
+import org.apache.uima.ducc.container.jd.JobDriver;
+import org.apache.uima.ducc.container.jd.JobDriverHelper;
+import org.apache.uima.ducc.container.jd.cas.CasManager;
+import org.apache.uima.ducc.container.jd.cas.CasManagerStats.RetryReason;
+import org.apache.uima.ducc.container.jd.log.LoggerHelper;
+import org.apache.uima.ducc.container.jd.mh.iface.remote.IRemoteWorkerProcess;
+import org.apache.uima.ducc.container.jd.wi.IProcessStatistics;
+import org.apache.uima.ducc.container.jd.wi.IWorkItem;
+import org.apache.uima.ducc.container.jd.wi.WiTracker;
+import org.apache.uima.ducc.container.net.iface.IMetaCas;
+
+public class ActionEndTimeout extends Action implements IAction {
+
+ private static Logger logger = Logger.getLogger(ActionEndTimeout.class, IComponent.Id.JD.name());
+
+ @Override
+ public String getName() {
+ return ActionEndTimeout.class.getName();
+ }
+
+ private void preemptWorkItem(IActionData actionData, CasManager cm, IMetaCas metaCas) {
+ String location = "preemptWorkItem";
+ cm.putMetaCas(metaCas, RetryReason.TimeoutRetry);
+ cm.getCasManagerStats().incEndRetry();
+ MessageBuffer mb = LoggerHelper.getMessageBuffer(actionData);
+ logger.info(location, ILogger.null_id, mb.toString());
+ }
+
+ @Override
+ public void engage(Object objectData) {
+ String location = "engage";
+ logger.trace(location, ILogger.null_id, "");
+ IActionData actionData = (IActionData) objectData;
+ try {
+ IWorkItem wi = actionData.getWorkItem();
+ IMetaCas metaCas = wi.getMetaCas();
+ JobDriver jd = JobDriver.getInstance();
+ CasManager cm = jd.getCasManager();
+ JobDriverHelper jdh = JobDriverHelper.getInstance();
+ IRemoteWorkerProcess rwp = jdh.getRemoteWorkerProcess(wi);
+ if(rwp != null) {
+ if(metaCas != null) {
+ WiTracker.getInstance().unassign(wi);
+ preemptWorkItem(actionData, cm, metaCas);
+ IWorkItemStateKeeper wisk = jd.getWorkItemStateKeeper();
+ MetaCasHelper metaCasHelper = new MetaCasHelper(metaCas);
+ IProcessStatistics pStats = jdh.getProcessStatistics(rwp);
+ int seqNo = metaCasHelper.getSystemKey();
+ wisk.preempt(seqNo);
+ pStats.preempt(wi);
+ displayProcessStatistics(logger, actionData, wi, pStats);
+ wi.reset();
+ }
+ else {
+ MessageBuffer mb = LoggerHelper.getMessageBuffer(actionData);
+ mb.append("No CAS found for processing");
+ logger.info(location, ILogger.null_id, mb.toString());
+ }
+ }
+ else {
+ MessageBuffer mb = LoggerHelper.getMessageBuffer(actionData);
+ mb.append("No remote worker process entry found for processing");
+ logger.info(location, ILogger.null_id, mb.toString());
+ }
+ }
+ catch(Exception e) {
+ logger.error(location, ILogger.null_id, e);
+ }
+ }
+}
Propchange: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionEndTimeout.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionEndTimeout.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionGet.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionGet.java?rev=1649836&r1=1649835&r2=1649836&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionGet.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionGet.java Tue Jan 6 15:37:26 2015
@@ -39,6 +39,7 @@ import org.apache.uima.ducc.container.jd
import org.apache.uima.ducc.container.jd.mh.iface.remote.IRemoteWorkerThread;
import org.apache.uima.ducc.container.jd.wi.IProcessStatistics;
import org.apache.uima.ducc.container.jd.wi.IWorkItem;
+import org.apache.uima.ducc.container.jd.wi.WiTracker;
import org.apache.uima.ducc.container.net.iface.IMetaCas;
import org.apache.uima.ducc.container.net.iface.IMetaCasTransaction;
import org.apache.uima.ducc.container.net.iface.IMetaCasTransaction.JdState;
@@ -90,6 +91,7 @@ public class ActionGet implements IActio
IEvent event = null;
//
if(metaCas != null) {
+ WiTracker.getInstance().assign(rwt, wi);
int seqNo = metaCasHelper.getSystemKey();
String wiId = metaCas.getUserKey();
String node = rwt.getNodeAddress();
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionInProgress.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionInProgress.java?rev=1649836&r1=1649835&r2=1649836&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionInProgress.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/ActionInProgress.java Tue Jan 6 15:37:26 2015
@@ -18,19 +18,18 @@
*/
package org.apache.uima.ducc.container.jd.fsm.wi;
+import java.util.concurrent.ConcurrentHashMap;
+
import org.apache.uima.ducc.container.common.MessageBuffer;
import org.apache.uima.ducc.container.common.fsm.iface.IAction;
-import org.apache.uima.ducc.container.common.fsm.iface.IEvent;
-import org.apache.uima.ducc.container.common.fsm.iface.IFsm;
import org.apache.uima.ducc.container.common.logger.IComponent;
import org.apache.uima.ducc.container.common.logger.ILogger;
import org.apache.uima.ducc.container.common.logger.Logger;
import org.apache.uima.ducc.container.jd.JobDriver;
-import org.apache.uima.ducc.container.jd.cas.CasManager;
import org.apache.uima.ducc.container.jd.log.LoggerHelper;
+import org.apache.uima.ducc.container.jd.mh.iface.remote.IRemoteWorkerThread;
import org.apache.uima.ducc.container.jd.wi.IWorkItem;
import org.apache.uima.ducc.container.net.iface.IMetaCas;
-import org.apache.uima.ducc.container.net.iface.IMetaCasTransaction;
public class ActionInProgress implements IAction {
@@ -47,30 +46,22 @@ public class ActionInProgress implements
logger.trace(location, ILogger.null_id, "");
IActionData actionData = (IActionData) objectData;
try {
- IWorkItem wi = actionData.getWorkItem();
- IFsm fsm = wi.getFsm();
- IMetaCasTransaction trans = actionData.getMetaCasTransaction();
- //
- CasManager cm = JobDriver.getInstance().getCasManager();
- IMetaCas metaCas = cm.getMetaCas();
- trans.setMetaCas(metaCas);
- //
- IEvent event = null;
- //
- if(metaCas != null) {
- wi.setTodGet();
- event = WiFsm.CAS_Available;
- MessageBuffer mb = LoggerHelper.getMessageBuffer(actionData);
- logger.info(location, ILogger.null_id, mb.toString());
- }
- else {
- event = WiFsm.CAS_Unavailable;
- MessageBuffer mb = LoggerHelper.getMessageBuffer(actionData);
- mb.append("No CAS found for processing");
- logger.info(location, ILogger.null_id, mb.toString());
+ IRemoteWorkerThread rwt = actionData.getRemoteWorkerThread();
+ ConcurrentHashMap<IRemoteWorkerThread, IWorkItem> map = JobDriver.getInstance().getRemoteThreadMap();
+ IWorkItem wi = map.get(rwt);
+ if(wi != null) {
+ IMetaCas metaCas = wi.getMetaCas();
+ if(metaCas != null) {
+ MessageBuffer mb = LoggerHelper.getMessageBuffer(actionData);
+ logger.debug(location, ILogger.null_id, mb.toString());
+ actionData.getWorkItem().setMetaCas(metaCas);
+ }
+ else {
+ MessageBuffer mb = LoggerHelper.getMessageBuffer(actionData);
+ mb.append("No CAS found for processing");
+ logger.info(location, ILogger.null_id, mb.toString());
+ }
}
- //
- fsm.transition(event, actionData);
}
catch(Exception e) {
logger.error(location, ILogger.null_id, e);
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/WiFsm.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/WiFsm.java?rev=1649836&r1=1649835&r2=1649836&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/WiFsm.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/fsm/wi/WiFsm.java Tue Jan 6 15:37:26 2015
@@ -65,6 +65,9 @@ public class WiFsm extends Fsm {
public IAction ActionProcessFailure = new ActionProcessFailure();
public IAction ActionProcessPreempt = new ActionProcessPreempt();
+ public IAction ActionAckTimeout = new ActionAckTimeout();
+ public IAction ActionEndTimeout = new ActionEndTimeout();
+
public IAction ActionIgnore = new ActionIgnore();
public IAction ActionError = new ActionError();
@@ -93,22 +96,31 @@ public class WiFsm extends Fsm {
add(Start, CAS_Available, ActionIgnore, Start);
add(Start, CAS_Unavailable, ActionIgnore, Start);
add(Start, Ack_Request, ActionError, Start);
+ add(Start, End_Request, ActionError, Start);
add(Start, Process_Preempt, ActionIgnore, Start);
add(Start, Process_Failure, ActionIgnore, Start);
+ add(Start, Ack_Timer_Pop, ActionIgnore, Start);
+ add(Start, End_Timer_Pop, ActionIgnore, Start);
add(Get_Pending, Get_Request, ActionInProgress, Get_Pending);
add(Get_Pending, CAS_Available, ActionSend, CAS_Send);
add(Get_Pending, CAS_Unavailable, ActionIgnore, Start);
add(Get_Pending, Ack_Request, ActionError, Get_Pending);
+ add(Get_Pending, End_Request, ActionError, Get_Pending);
add(Get_Pending, Process_Preempt, ActionProcessPreempt, Start);
add(Get_Pending, Process_Failure, ActionProcessFailure, Start);
+ add(Get_Pending, Ack_Timer_Pop, ActionIgnore, Get_Pending);
+ add(Get_Pending, End_Timer_Pop, ActionIgnore, Get_Pending);
add(CAS_Send, Get_Request, ActionInProgress, CAS_Send);
add(CAS_Send, CAS_Available, ActionIgnore, CAS_Send);
add(CAS_Send, CAS_Unavailable, ActionIgnore, CAS_Send);
add(CAS_Send, Ack_Request, ActionAck, CAS_Active);
+ add(CAS_Send, End_Request, ActionError, CAS_Send);
add(CAS_Send, Process_Preempt, ActionProcessPreempt, Start);
add(CAS_Send, Process_Failure, ActionProcessFailure, Start);
+ add(CAS_Send, Ack_Timer_Pop, ActionAckTimeout, Start);
+ add(CAS_Send, End_Timer_Pop, ActionIgnore, CAS_Send);
add(CAS_Active, Get_Request, ActionError, CAS_Active);
add(CAS_Active, CAS_Available, ActionIgnore, CAS_Active);
@@ -117,6 +129,8 @@ public class WiFsm extends Fsm {
add(CAS_Active, End_Request, ActionEnd, Start);
add(CAS_Active, Process_Preempt, ActionProcessPreempt, Start);
add(CAS_Active, Process_Failure, ActionProcessFailure, Start);
+ add(CAS_Active, Ack_Timer_Pop, ActionIgnore, CAS_Active);
+ add(CAS_Active, End_Timer_Pop, ActionEndTimeout, Start);
MessageBuffer mb2 = new MessageBuffer();
mb2.append(Standardize.Label.exit.name());
@@ -141,6 +155,7 @@ public class WiFsm extends Fsm {
MessageBuffer mb = LoggerHelper.getMessageBuffer(actionData);
mb.append(Standardize.Label.curr.get()+getStateCurrent().getName());
mb.append(Standardize.Label.prev.get()+getStatePrevious().getName());
+ mb.append(Standardize.Label.event.get()+event.getName());
logger.info(location, ILogger.null_id, mb.toString());
}
}
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/log/LoggerHelper.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/log/LoggerHelper.java?rev=1649836&r1=1649835&r2=1649836&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/log/LoggerHelper.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/log/LoggerHelper.java Tue Jan 6 15:37:26 2015
@@ -50,6 +50,17 @@ public class LoggerHelper {
String retVal = "?";
try {
IWorkItem wi = actionData.getWorkItem();
+ retVal = getSeqNo(wi);
+ }
+ catch(Exception e) {
+ // oh well
+ }
+ return retVal;
+ }
+
+ public static String getSeqNo(IWorkItem wi) {
+ String retVal = "?";
+ try {
IMetaCas metaCas = wi.getMetaCas();
retVal = metaCas.getSystemKey();
}
@@ -59,6 +70,18 @@ public class LoggerHelper {
return retVal;
}
+ public static String getId(IWorkItem wi) {
+ String retVal = "?";
+ try {
+ IMetaCas metaCas = wi.getMetaCas();
+ retVal = metaCas.getUserKey();
+ }
+ catch(Exception e) {
+ // oh well
+ }
+ return retVal;
+ }
+
public static String getRemote(IActionData actionData) {
String retVal = "?";
try {
@@ -99,6 +122,33 @@ public class LoggerHelper {
}
catch(Exception e) {
logger.error(location, ILogger.null_id, e);
+ }
+ return mb;
+ }
+
+ public static MessageBuffer getMessageBuffer(IMetaCasTransaction trans, IRemoteWorkerThread rwt) {
+ String location = "getMessageBuffer";
+ MessageBuffer mb = new MessageBuffer();
+ try {
+ //mb.append(Standardize.Label.transNo.get()+transNo);
+ mb.append(Standardize.Label.remote.get()+rwt.toString());
+ }
+ catch(Exception e) {
+ logger.error(location, ILogger.null_id, e);
+ }
+ return mb;
+ }
+
+ public static MessageBuffer getMessageBuffer(IRemoteWorkerThread rwt, IWorkItem wi) {
+ String location = "getMessageBuffer";
+ MessageBuffer mb = new MessageBuffer();
+ try {
+ mb.append(Standardize.Label.seqNo.get()+getSeqNo(wi));
+ mb.append(Standardize.Label.id.get()+getSeqNo(wi));
+ mb.append(Standardize.Label.remote.get()+rwt.toString());
+ }
+ catch(Exception e) {
+ logger.error(location, ILogger.null_id, e);
}
return mb;
}
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/mh/MessageHandler.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/mh/MessageHandler.java?rev=1649836&r1=1649835&r2=1649836&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/mh/MessageHandler.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/mh/MessageHandler.java Tue Jan 6 15:37:26 2015
@@ -154,6 +154,7 @@ public class MessageHandler implements I
mb.append(Standardize.Label.endSuccess.get()+oi.getWorkItemEndSuccesses());
mb.append(Standardize.Label.endFailure.get()+oi.getWorkItemEndFailures());
mb.append(Standardize.Label.killJob.get()+oi.isKillJob());
+ mb.append(Standardize.Label.dispatched.get()+oi.getWorkItemDispatcheds());
mb.append(Standardize.Label.retrys.get()+oi.getWorkItemRetrys());
mb.append(Standardize.Label.preemptions.get()+oi.getWorkItemPreemptions());
mb.append(Standardize.Label.finishedMillisMin.get()+oi.getWorkItemFinishedMillisMin());
@@ -321,28 +322,28 @@ public class MessageHandler implements I
return wi;
}
- private IWorkItem find(IRemoteWorkerThread rwi) {
+ private IWorkItem find(IRemoteWorkerThread rwt) {
String location = "find";
ConcurrentHashMap<IRemoteWorkerThread, IWorkItem> map = JobDriver.getInstance().getRemoteThreadMap();
- IWorkItem wi = map.get(rwi);
+ IWorkItem wi = map.get(rwt);
if(wi != null) {
IMetaCas metaCas = wi.getMetaCas();
if(metaCas != null) {
MessageBuffer mb = new MessageBuffer();
- mb.append(Standardize.Label.remote.get()+rwi.toString());
+ mb.append(Standardize.Label.remote.get()+rwt.toString());
mb.append(Standardize.Label.seqNo.get()+metaCas.getSystemKey());
logger.debug(location, ILogger.null_id, mb.toString());
}
else {
MessageBuffer mb = new MessageBuffer();
- mb.append(Standardize.Label.remote.get()+rwi.toString());
+ mb.append(Standardize.Label.remote.get()+rwt.toString());
mb.append("has no work assigned presently");
logger.debug(location, ILogger.null_id, mb.toString());
}
}
else {
MessageBuffer mb = new MessageBuffer();
- mb.append(Standardize.Label.remote.get()+rwi.toString());
+ mb.append(Standardize.Label.remote.get()+rwt.toString());
mb.append("has no work assigned presently");
logger.debug(location, ILogger.null_id, mb.toString());
}
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/timeout/ITimeoutManager.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/timeout/ITimeoutManager.java?rev=1649836&r1=1649835&r2=1649836&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/timeout/ITimeoutManager.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/timeout/ITimeoutManager.java Tue Jan 6 15:37:26 2015
@@ -28,4 +28,5 @@ public interface ITimeoutManager {
public void pendingEnd(IActionData actionData);
public void receivedEnd(IActionData actionData);
+ public void timeout(IActionData actionData);
}
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/timeout/TimeoutManager.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/timeout/TimeoutManager.java?rev=1649836&r1=1649835&r2=1649836&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/timeout/TimeoutManager.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/timeout/TimeoutManager.java Tue Jan 6 15:37:26 2015
@@ -26,11 +26,13 @@ import java.util.concurrent.ScheduledFut
import java.util.concurrent.TimeUnit;
import org.apache.uima.ducc.container.common.MessageBuffer;
+import org.apache.uima.ducc.container.common.Standardize;
import org.apache.uima.ducc.container.common.fsm.iface.IEvent;
import org.apache.uima.ducc.container.common.fsm.iface.IFsm;
import org.apache.uima.ducc.container.common.logger.IComponent;
import org.apache.uima.ducc.container.common.logger.ILogger;
import org.apache.uima.ducc.container.common.logger.Logger;
+import org.apache.uima.ducc.container.jd.JobDriver;
import org.apache.uima.ducc.container.jd.fsm.wi.IActionData;
import org.apache.uima.ducc.container.jd.fsm.wi.WiFsm;
import org.apache.uima.ducc.container.jd.log.LoggerHelper;
@@ -57,14 +59,15 @@ public class TimeoutManager implements I
public void pendingAck(IActionData actionData) {
String location = "pendingAck";
try {
- MessageBuffer mb = LoggerHelper.getMessageBuffer(actionData);
- logger.debug(location, ILogger.null_id, mb.toString());
IWorkItem wi = actionData.getWorkItem();
IFsm fsm = wi.getFsm();
IEvent event = WiFsm.Ack_Timer_Pop;
long deadline = System.currentTimeMillis()+ackTimeout;
ITimeoutTask timeoutTask = new TimeoutTask(fsm, event, actionData, deadline);
register(fsm, timeoutTask);
+ MessageBuffer mb = LoggerHelper.getMessageBuffer(actionData);
+ mb.append(Standardize.Label.deadline+"+"+ackTimeout/1000);
+ logger.debug(location, ILogger.null_id, mb.toString());
}
catch(Exception e) {
logger.error(location, ILogger.null_id, e);
@@ -77,6 +80,9 @@ public class TimeoutManager implements I
try {
MessageBuffer mb = LoggerHelper.getMessageBuffer(actionData);
logger.debug(location, ILogger.null_id, mb.toString());
+ IWorkItem wi = actionData.getWorkItem();
+ IFsm fsm = wi.getFsm();
+ unregister(fsm);
}
catch(Exception e) {
logger.error(location, ILogger.null_id, e);
@@ -87,7 +93,16 @@ public class TimeoutManager implements I
public void pendingEnd(IActionData actionData) {
String location = "pendingEnd";
try {
+ IWorkItem wi = actionData.getWorkItem();
+ IFsm fsm = wi.getFsm();
+ IEvent event = WiFsm.End_Timer_Pop;
+ JobDriver jd = JobDriver.getInstance();
+ long endTimeout = jd.getWorkItemTimeoutMillis();
+ long deadline = System.currentTimeMillis()+endTimeout;
+ ITimeoutTask timeoutTask = new TimeoutTask(fsm, event, actionData, deadline);
+ register(fsm, timeoutTask);
MessageBuffer mb = LoggerHelper.getMessageBuffer(actionData);
+ mb.append(Standardize.Label.deadline+"+"+endTimeout/1000);
logger.debug(location, ILogger.null_id, mb.toString());
}
catch(Exception e) {
@@ -101,19 +116,17 @@ public class TimeoutManager implements I
try {
MessageBuffer mb = LoggerHelper.getMessageBuffer(actionData);
logger.debug(location, ILogger.null_id, mb.toString());
+ IWorkItem wi = actionData.getWorkItem();
+ IFsm fsm = wi.getFsm();
+ unregister(fsm);
}
catch(Exception e) {
logger.error(location, ILogger.null_id, e);
}
}
- private boolean wip = true;
-
- public void register(IFsm fsm, ITimeoutTask timeoutTask) {
+ private void register(IFsm fsm, ITimeoutTask timeoutTask) {
String location = "register";
- if(wip) {
- return;
- }
try {
mapTask.put(fsm, timeoutTask);
Callable<?> callable = timeoutTask;
@@ -124,18 +137,15 @@ public class TimeoutManager implements I
//
IActionData actionData = timeoutTask.getActionData();
MessageBuffer mb = LoggerHelper.getMessageBuffer(actionData);
- logger.info(location, ILogger.null_id, mb.toString());
+ logger.debug(location, ILogger.null_id, mb.toString());
}
catch(Exception e) {
logger.error(location, ILogger.null_id, e);
}
}
- public void unregister(IFsm fsm) {
+ private void unregister(IFsm fsm) {
String location = "unregister";
- if(wip) {
- return;
- }
try {
ScheduledFuture<?> scheduledFuture = mapFuture.remove(fsm);
if(scheduledFuture != null) {
@@ -145,11 +155,27 @@ public class TimeoutManager implements I
if(timeoutTask != null) {
IActionData actionData = timeoutTask.getActionData();
MessageBuffer mb = LoggerHelper.getMessageBuffer(actionData);
- logger.info(location, ILogger.null_id, mb.toString());
+ logger.debug(location, ILogger.null_id, mb.toString());
}
}
catch(Exception e) {
logger.error(location, ILogger.null_id, e);
}
}
+
+ @Override
+ public void timeout(IActionData actionData) {
+ String location = "timeout";
+ try {
+ MessageBuffer mb = LoggerHelper.getMessageBuffer(actionData);
+ logger.debug(location, ILogger.null_id, mb.toString());
+ IWorkItem wi = actionData.getWorkItem();
+ IFsm fsm = wi.getFsm();
+ unregister(fsm);
+ }
+ catch(Exception e) {
+ logger.error(location, ILogger.null_id, e);
+ }
+ }
+
}
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/timeout/TimeoutTask.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/timeout/TimeoutTask.java?rev=1649836&r1=1649835&r2=1649836&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/timeout/TimeoutTask.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/timeout/TimeoutTask.java Tue Jan 6 15:37:26 2015
@@ -86,7 +86,7 @@ public class TimeoutTask implements ITim
MessageBuffer mb = LoggerHelper.getMessageBuffer(actionData);
mb.append(Standardize.Label.event.get()+event.getName());
logger.warn(location, ILogger.null_id, mb.toString());
- TimeoutManager.getInstance().unregister(fsm);
+ TimeoutManager.getInstance().timeout(actionData);
fsm.transition(event, actionData);
return null;
}
Added: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/wi/WiTracker.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/wi/WiTracker.java?rev=1649836&view=auto
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/wi/WiTracker.java (added)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/wi/WiTracker.java Tue Jan 6 15:37:26 2015
@@ -0,0 +1,97 @@
+/*
+ * 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.container.jd.wi;
+
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.uima.ducc.container.common.MessageBuffer;
+import org.apache.uima.ducc.container.common.Standardize;
+import org.apache.uima.ducc.container.common.logger.IComponent;
+import org.apache.uima.ducc.container.common.logger.ILogger;
+import org.apache.uima.ducc.container.common.logger.Logger;
+import org.apache.uima.ducc.container.jd.log.LoggerHelper;
+import org.apache.uima.ducc.container.jd.mh.iface.remote.IRemoteWorkerThread;
+
+public class WiTracker {
+
+ private ConcurrentHashMap<IRemoteWorkerThread, IWorkItem> map = new ConcurrentHashMap<IRemoteWorkerThread, IWorkItem>();
+
+ private static Logger logger = Logger.getLogger(WiTracker.class, IComponent.Id.JD.name());
+
+ private static WiTracker instance = new WiTracker();
+
+ public static WiTracker getInstance() {
+ return instance;
+ }
+
+ public void assign(IRemoteWorkerThread rwt, IWorkItem wi) {
+ String location = "assign";
+ try {
+ map.put(rwt, wi);
+ report();
+ }
+ catch(Exception e) {
+ logger.error(location, ILogger.null_id, e);
+ }
+ }
+
+ public void unassign(IRemoteWorkerThread rwt) {
+ String location = "unassign";
+ try {
+ map.remove(rwt);
+ report();
+ }
+ catch(Exception e) {
+ logger.error(location, ILogger.null_id, e);
+ }
+ }
+
+ private IRemoteWorkerThread find(IWorkItem wi) {
+ IRemoteWorkerThread rwt = null;
+ for(Entry<IRemoteWorkerThread, IWorkItem> entry : map.entrySet()) {
+ if(wi.getSeqNo() == entry.getValue().getSeqNo()) {
+ rwt = entry.getKey();
+ break;
+ }
+ }
+ return rwt;
+ }
+
+ public void unassign(IWorkItem value) {
+ IRemoteWorkerThread rwt = find(value);
+ if(rwt != null) {
+ map.remove(rwt);
+ report();
+ }
+ }
+
+ private void report() {
+ String location = "report";
+ MessageBuffer mb = new MessageBuffer();
+ mb.append(Standardize.Label.size.get()+map.size());
+ logger.debug(location, ILogger.null_id, mb.toString());
+ for(Entry<IRemoteWorkerThread, IWorkItem> entry : map.entrySet()) {
+ IRemoteWorkerThread rwt = entry.getKey();
+ IWorkItem wi = entry.getValue();
+ MessageBuffer mb1 = LoggerHelper.getMessageBuffer(rwt, wi);
+ logger.debug(location, ILogger.null_id, mb1.toString());
+ }
+ }
+}
Propchange: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/wi/WiTracker.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: uima/sandbox/uima-ducc/trunk/uima-ducc-container/src/main/java/org/apache/uima/ducc/container/jd/wi/WiTracker.java
------------------------------------------------------------------------------
svn:mime-type = text/plain