You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@uima.apache.org by cw...@apache.org on 2019/01/23 18:58:01 UTC
svn commit: r1851960 - in /uima/uima-ducc/trunk/uima-ducc-pullservice/src:
main/java/org/apache/uima/ducc/ps/sd/task/
main/java/org/apache/uima/ducc/ps/service/main/
main/java/org/apache/uima/ducc/ps/service/protocol/builtin/
main/java/org/apache/uima/...
Author: cwiklik
Date: Wed Jan 23 18:58:01 2019
New Revision: 1851960
URL: http://svn.apache.org/viewvc?rev=1851960&view=rev
Log:
UIMA-5960 refactored to synchronize service threads to allow the pull by a single thread at a time. Remove sleep from client when out of tasks. Sleeping is implemented at the service
Added:
uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/protocol/builtin/WaitWithLockWhenNoTaskAvailable.java
Modified:
uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/sd/task/DuccServiceTaskProtocolHandler.java
uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/main/PullService.java
uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/protocol/builtin/DefaultNoTaskAvailableStrategy.java
uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/protocol/builtin/DefaultServiceProtocolHandler.java
uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/transport/IServiceTransport.java
uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/transport/http/HttpServiceTransport.java
uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/Client.java
uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/service/wrapper/JUnitServiceWrapperTestCase.java
uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/transport/JunitProtocolHandlerTestCase.java
uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/transport/JunitTransportTestCase.java
Modified: uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/sd/task/DuccServiceTaskProtocolHandler.java
URL: http://svn.apache.org/viewvc/uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/sd/task/DuccServiceTaskProtocolHandler.java?rev=1851960&r1=1851959&r2=1851960&view=diff
==============================================================================
--- uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/sd/task/DuccServiceTaskProtocolHandler.java (original)
+++ uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/sd/task/DuccServiceTaskProtocolHandler.java Wed Jan 23 18:58:01 2019
@@ -20,7 +20,6 @@
package org.apache.uima.ducc.ps.sd.task;
import java.util.ArrayList;
-import java.util.Objects;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicInteger;
@@ -47,16 +46,12 @@ public class DuccServiceTaskProtocolHand
private volatile boolean running = true;;
private static AtomicInteger atomicCounter =
new AtomicInteger(0);
- Properties props = new Properties();
public DuccServiceTaskProtocolHandler(TaskAllocatorCallbackListener taskAllocator) {
}
@Override
public String initialize(Properties props) throws TaskProtocolException {
- if ( Objects.nonNull(props)) {
- this.props = props;
- }
return null;
}
@@ -134,54 +129,23 @@ public class DuccServiceTaskProtocolHand
}
return new MetaTask(atomicCounter.incrementAndGet(), "", serializedCas);
}
- private int convertToInt(String value, int defaultValue) {
- try {
- return Integer.valueOf(value);
- } catch(NumberFormatException e) {
- return defaultValue;
- }
- }
+
private synchronized IMetaMetaTask getMetaMetaTask(TaskConsumer taskConsumer) {
IMetaMetaTask mmc = new MetaMetaTask();
ServiceDriver sd = DuccServiceDriver.getInstance();
TaskAllocatorCallbackListener taskAllocator =
sd.getTaskAllocator();
ITask task;
-
- // By default, the max time we are willing to wait for a task is 30 secs
- // with 1 secs wait time between retries. When we reach the max
- // retry, we return empty task to the service.
- int retryCount = convertToInt(props.getProperty(ServiceDriver.DriverTaskRetryCount),30);
- if ( retryCount == 0 ) {
- retryCount = 1;
- }
- int waitTime = convertToInt(props.getProperty(ServiceDriver.DriverTaskWaitTime),1000);
- while( retryCount > 0 ) {
- task = taskAllocator.getTask(taskConsumer);
- // if allocation system does not return a task (or empty)
- // block this thread and retry until a task becomes
- // available or until max retry count is exhausted
- if ( task == null || task.isEmpty() ) {
- if ( waitTime > 0 ) {
- try {
- Thread.sleep(waitTime);
- } catch(InterruptedException ee) {
- Thread.currentThread().interrupt();
- }
- }
-
- } else {
- IMetaTask metaTask = getMetaTask(task.asString());
- mmc.setMetaCas(metaTask);
- if ( logger.isLoggable(Level.FINE)) {
- logger.log(Level.FINE,"Returning TASK with appdata:"+task.getMetadata()+" to the service");
- }
- mmc.getMetaCas().setAppData(task.getMetadata());
- break;
+
+ task = taskAllocator.getTask(taskConsumer);
+ if ( task != null && !task.isEmpty() ) {
+ IMetaTask metaTask = getMetaTask(task.asString());
+ mmc.setMetaCas(metaTask);
+ if ( logger.isLoggable(Level.FINE)) {
+ logger.log(Level.FINE,"Returning TASK with appdata:"+task.getMetadata()+" to the service");
}
- retryCount--;
+ mmc.getMetaCas().setAppData(task.getMetadata());
}
-
return mmc;
}
private void handleMetaTaskTransationAck(IMetaTaskTransaction trans, TaskConsumer taskConsumer) {
Modified: uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/main/PullService.java
URL: http://svn.apache.org/viewvc/uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/main/PullService.java?rev=1851960&r1=1851959&r2=1851960&view=diff
==============================================================================
--- uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/main/PullService.java (original)
+++ uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/main/PullService.java Wed Jan 23 18:58:01 2019
@@ -34,7 +34,6 @@ import java.util.concurrent.locks.Reentr
import org.apache.uima.UIMAFramework;
import org.apache.uima.ducc.ps.ServiceThreadFactory;
import org.apache.uima.ducc.ps.service.IService;
-//import org.apache.uima.ducc.ps.service.ServiceConfiguration;
import org.apache.uima.ducc.ps.service.errors.IServiceErrorHandler;
import org.apache.uima.ducc.ps.service.errors.ServiceException;
import org.apache.uima.ducc.ps.service.errors.ServiceInitializationException;
@@ -61,7 +60,7 @@ public class PullService implements ISer
// how many processing threads
private int scaleout=1;
// amount of time to wait when client has no tasks to give
- private int waitTimeInMillis=0;
+ private int waitTimeInMillis=1000;
// application assigned service label
private String type;
@@ -170,9 +169,12 @@ public class PullService implements ISer
// the following will throw exception if client URL not specified
initializeDefaultRegistryClient();
}
+ // Use default wait strategy when a driver is out of work.
+ INoTaskAvailableStrategy waitStrategy =
+ new DefaultNoTaskAvailableStrategy(waitTimeInMillis);
// add default transport
- transport = new HttpServiceTransport(registryClient, scaleout);
+ transport = new HttpServiceTransport(registryClient, scaleout, waitStrategy);
// contract is that the service will block in this method until
// all process threads initialize. Use a latch to block until this
@@ -185,13 +187,11 @@ public class PullService implements ISer
// this down just before thread dies.
CountDownLatch stopLatch = new CountDownLatch(scaleout);
serviceProcessor.setScaleout(scaleout);
- INoTaskAvailableStrategy waitStrategy =
- new DefaultNoTaskAvailableStrategy(waitTimeInMillis);
// add default protocol handler
protocolHandler =
new DefaultServiceProtocolHandler.Builder()
.withProcessor(serviceProcessor)
- .withNoTaskStrategy(waitStrategy)
+ //.withNoTaskStrategy(waitStrategy)
.withService(this)
.withTransport(transport)
.withDoneLatch(stopLatch)
Modified: uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/protocol/builtin/DefaultNoTaskAvailableStrategy.java
URL: http://svn.apache.org/viewvc/uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/protocol/builtin/DefaultNoTaskAvailableStrategy.java?rev=1851960&r1=1851959&r2=1851960&view=diff
==============================================================================
--- uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/protocol/builtin/DefaultNoTaskAvailableStrategy.java (original)
+++ uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/protocol/builtin/DefaultNoTaskAvailableStrategy.java Wed Jan 23 18:58:01 2019
@@ -38,7 +38,7 @@ public class DefaultNoTaskAvailableStrat
*/
@Override
public void interrupt() {
-
+ Thread.currentThread().interrupt();
}
@Override
Modified: uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/protocol/builtin/DefaultServiceProtocolHandler.java
URL: http://svn.apache.org/viewvc/uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/protocol/builtin/DefaultServiceProtocolHandler.java?rev=1851960&r1=1851959&r2=1851960&view=diff
==============================================================================
--- uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/protocol/builtin/DefaultServiceProtocolHandler.java (original)
+++ uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/protocol/builtin/DefaultServiceProtocolHandler.java Wed Jan 23 18:58:01 2019
@@ -6,9 +6,9 @@
* 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
@@ -19,7 +19,6 @@
package org.apache.uima.ducc.ps.service.protocol.builtin;
import java.io.ByteArrayOutputStream;
-import java.io.InvalidClassException;
import java.io.ObjectOutputStream;
import java.util.Objects;
import java.util.concurrent.CountDownLatch;
@@ -48,8 +47,8 @@ import org.apache.uima.util.Level;
import org.apache.uima.util.Logger;
/**
- *
- * This protocol handler is a Runnable
+ *
+ * This protocol handler is a Runnable
*
*/
public class DefaultServiceProtocolHandler implements IServiceProtocolHandler {
@@ -71,16 +70,16 @@ public class DefaultServiceProtocolHandl
private IService service;
// forces process threads to initialize serially
private static ReentrantLock initLock = new ReentrantLock();
-
+
private static AtomicInteger idGenerator = new AtomicInteger();
-
+
private static ReentrantLock retryLock = new ReentrantLock();
-
+
private Thread retryThread = null;
-
- private DefaultServiceProtocolHandler(Builder builder) {
- this.initLatch = builder.initLatch;
- this.stopLatch = builder.stopLatch;
+
+ private DefaultServiceProtocolHandler(Builder builder) {
+ this.initLatch = builder.initLatch;
+ this.stopLatch = builder.stopLatch;
this.service = builder.service;
this.transport = builder.transport;
this.processor = builder.processor;
@@ -126,7 +125,7 @@ public class DefaultServiceProtocolHandl
public boolean initialized() {
return ( initError==false );
}
- private IMetaTaskTransaction send(IMetaTaskTransaction transaction) throws Exception {
+ private IMetaTaskTransaction sendAndReceive(IMetaTaskTransaction transaction) throws Exception {
TransactionId tid;
if (Type.Get.equals(transaction.getType())) {
int major = idGenerator.addAndGet(1);
@@ -138,30 +137,26 @@ public class DefaultServiceProtocolHandl
// increment minor
tid.next();
}
-
+
transaction.setRequesterProcessName(service.getType());
transport.addRequestorInfo(transaction);
- Object o = null;
+ IMetaTaskTransaction reply = null;
try {
String body = XStreamUtils.marshall(transaction);
- String content = transport.dispatch(body);
- if ( content == null ) {
+ // dispatch implements waiting if no task is given by the driver
+ reply = transport.dispatch(body);
+
+ if ( Objects.isNull(reply) ) {
throw new TransportException("Received invalid content (null) in response from client - rejecting request");
}
- o = XStreamUtils.unmarshall(content);
-
+
} catch ( Exception e) {
if ( !running ) {
throw new TransportException("Service stopping - rejecting request");
}
throw e;
}
- if (o instanceof IMetaTaskTransaction) {
- return (MetaTaskTransaction) o;
- } else {
- throw new InvalidClassException(
- "Expected IMetaTaskTransaction - Instead Received " + o.getClass().getName());
- }
+ return reply;
}
private IMetaTaskTransaction callEnd(IMetaTaskTransaction transaction) throws Exception {
@@ -169,7 +164,7 @@ public class DefaultServiceProtocolHandl
if ( logger.isLoggable(Level.FINE)) {
logger.log(Level.FINE, "ProtocolHandler calling END");
}
- return send(transaction);
+ return sendAndReceive(transaction);
}
@@ -178,19 +173,19 @@ public class DefaultServiceProtocolHandl
if ( logger.isLoggable(Level.FINE)) {
logger.log(Level.FINE, "ProtocolHandler calling ACK");
}
- return send(transaction);
+ return sendAndReceive(transaction);
}
private IMetaTaskTransaction callGet(IMetaTaskTransaction transaction) throws Exception {
- transaction.setType(Type.Get);
+ transaction.setType(Type.Get);
if ( logger.isLoggable(Level.FINE)) {
logger.log(Level.FINE, "ProtocolHandler calling GET");
}
- return send(transaction);
+ return sendAndReceive(transaction);
}
/**
* Block until service start() is called
- *
+ *
* @throws ServiceInitializationException
*/
private void awaitStart() throws ServiceInitializationException {
@@ -209,7 +204,7 @@ public class DefaultServiceProtocolHandl
while (running) {
// send GET Request
- transaction =
+ transaction =
callGet(new MetaTaskTransaction());
// the code may have blocked in callGet for awhile, so check
// if service is still running. If this service is in quiescing
@@ -220,10 +215,10 @@ public class DefaultServiceProtocolHandl
}
if (transaction.getMetaTask() != null && transaction.getMetaTask().getUserSpaceTask() != null) {
break;
- }
+ }
retryThread = Thread.currentThread();
-
- //System.out.println("Thread:"+Thread.currentThread().getId()+" ------------- No Task -------------- Retrying GET until success");
+
+ System.out.println("Thread:"+Thread.currentThread().getId()+" ------------- No Task -------------- Retrying GET until success");
// the client has no tasks to give.
noTaskStrategy.handleNoTaskSupplied();
@@ -242,19 +237,19 @@ public class DefaultServiceProtocolHandl
// we may fail in initialize() in which case the ServiceInitializationException
// is thrown
initialize();
-
+
// now wait for application to call start
awaitStart();
-
+
// all threads intialized, enter running state
IMetaTaskTransaction transaction = null;
-
+
if ( logger.isLoggable(Level.INFO)) {
logger.log(Level.INFO, ".............. Thread "+Thread.currentThread().getId() + " ready to process");
}
-
+
while (running) {
try {
@@ -267,12 +262,12 @@ public class DefaultServiceProtocolHandl
if ( !running && !quiescing ) {
break;
}
- if (transaction.getMetaTask() == null || transaction.getMetaTask().getUserSpaceTask() == null ) {
- // synchronize retry. Allow single thread to test if client has
- // more tasks to give. If so, unblock other threads and proceed
- // to normal processing.
- transaction = retryUntilSuccessfull();
- }
+// if (transaction.getMetaTask() == null || transaction.getMetaTask().getUserSpaceTask() == null ) {
+// // synchronize retry. Allow single thread to test if client has
+// // more tasks to give. If so, unblock other threads and proceed
+// // to normal processing.
+// transaction = retryUntilSuccessfull();
+// }
// transaction may be null if retryUntilSuccessfull was interrupted
// due to stop
if (Objects.isNull(transaction) || (!running && !quiescing)) {
@@ -281,8 +276,8 @@ public class DefaultServiceProtocolHandl
logger.log(Level.INFO, ".............. Thread "+Thread.currentThread().getId() + " processing new task");
Object task = transaction.getMetaTask().getUserSpaceTask();
-
- // send ACK
+
+ // send ACK
transaction = callAck(transaction);
if (!running && !quiescing ) {
break;
@@ -291,22 +286,22 @@ public class DefaultServiceProtocolHandl
// assume success
Action action = Action.CONTINUE;
- // check if process error occurred.
+ // check if process error occurred.
String errorAsString = processResult.getError();
if (processResult.terminateProcess()) {
action = Action.TERMINATE;
} else if ( Objects.isNull(errorAsString)){
- // success
+ // success
transaction.getMetaTask().setPerformanceMetrics(processResult.getResult());
- }
+ }
if ( Objects.nonNull(errorAsString ) ) {
IMetaTask mc = transaction.getMetaTask();
// the ducc.deploy.JpType is only present for jobs. If not specified
// we return stringified exception to the client. The JD expects
// Java Exception object for its error handling
if ( Objects.isNull(System.getProperty("ducc.deploy.JpType")) ) {
-
+
mc.setUserSpaceException(errorAsString);
} else {
logger.log(Level.INFO, "Sending Exception to JD:\n" +
@@ -316,7 +311,7 @@ public class DefaultServiceProtocolHandl
}
}
-
+
// send END Request
callEnd(transaction);
if (running && Action.TERMINATE.equals(action)) {
@@ -333,8 +328,8 @@ public class DefaultServiceProtocolHandl
}).start();
running = false;
}
-
-
+
+
} catch( IllegalStateException e) {
break;
@@ -342,9 +337,9 @@ public class DefaultServiceProtocolHandl
break;
}
catch (Exception e) {
-
+
logger.log(Level.WARNING,"",e);
- }
+ }
}
stopLatch.countDown();
System.out.println(Utils.getTimestamp()+">>>>>>> "+Utils.getShortClassname(this.getClass())+".call() >>>>>>>>>> Thread ["+Thread.currentThread().getId()+"] "+ " ProtocolHandler stopped requesting new tasks - Stopping processor");
@@ -365,13 +360,13 @@ public class DefaultServiceProtocolHandl
} catch (Exception e) {
try {
logger.log(Level.WARNING, "Unable to Serialize "+t.getClass().getName()+" - Will Stringify It Instead");
-
+
} catch( Exception ee) {}
throw e;
} finally {
oos.close();
}
-
+
return baos.toByteArray();
}
private void delegateStop() {
@@ -404,7 +399,7 @@ public class DefaultServiceProtocolHandl
// change state of transport to not running but keep connection open
// so that other threads can quiesce (send results)
transport.stop(true);
-
+
quiescing = true;
running = false;
try {
@@ -442,8 +437,8 @@ public class DefaultServiceProtocolHandl
public void setTransport(IServiceTransport transport) {
this.transport = transport;
}
-
-
+
+
public static class Builder {
private IServiceTransport transport;
private IServiceProcessor processor;
@@ -460,15 +455,15 @@ public class DefaultServiceProtocolHandl
public Builder withProcessor(IServiceProcessor processor) {
this.processor = processor;
return this;
- }
+ }
public Builder withInitCompleteLatch(CountDownLatch initLatch) {
this.initLatch = initLatch;
return this;
- }
+ }
public Builder withDoneLatch(CountDownLatch stopLatch) {
this.stopLatch = stopLatch;
return this;
- }
+ }
public Builder withNoTaskStrategy(INoTaskAvailableStrategy strategy) {
this.strategy = strategy;
return this;
Added: uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/protocol/builtin/WaitWithLockWhenNoTaskAvailable.java
URL: http://svn.apache.org/viewvc/uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/protocol/builtin/WaitWithLockWhenNoTaskAvailable.java?rev=1851960&view=auto
==============================================================================
--- uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/protocol/builtin/WaitWithLockWhenNoTaskAvailable.java (added)
+++ uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/protocol/builtin/WaitWithLockWhenNoTaskAvailable.java Wed Jan 23 18:58:01 2019
@@ -0,0 +1,68 @@
+/*
+ * 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.ps.service.protocol.builtin;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.apache.uima.UIMAFramework;
+import org.apache.uima.ducc.ps.service.protocol.INoTaskAvailableStrategy;
+import org.apache.uima.util.Level;
+import org.apache.uima.util.Logger;
+
+public class WaitWithLockWhenNoTaskAvailable implements INoTaskAvailableStrategy {
+ private int waitTime;
+ private final ReentrantLock lock = new ReentrantLock();
+ Logger logger = UIMAFramework.getLogger(WaitWithLockWhenNoTaskAvailable.class);
+ public WaitWithLockWhenNoTaskAvailable(int waitTimeInMillis) {
+ this.waitTime = waitTimeInMillis;
+ logger.log(Level.INFO, ">>>>>>>> Service Wait Time For Task:"+waitTimeInMillis+" ms");
+ }
+ /**
+ * This methods is called when a service is stopping. There is no
+ * need to wait. We want to stop as soon as possible so we just
+ * interrupt the lock which might be blocking in await()
+ */
+ @Override
+ public void interrupt() {
+ lock.unlock();
+ }
+
+ @Override
+ public void handleNoTaskSupplied() {
+ Condition waitAwhileCondition = lock.newCondition();
+ try {
+ lock.lock();
+ // wait only if wait time > 0. Indefinite wait is not supported. If waitTime=0, it means no wait
+ if ( waitTime > 0 ) {
+ try {
+ waitAwhileCondition.await(waitTime, TimeUnit.MILLISECONDS);
+ } catch( InterruptedException e) {
+ System.out.println("DefaultNoTaskAvailableStrategy.handleNoTaskSupplied() - Waiting interrupted "+" Thread:"+Thread.currentThread().getId());
+ }
+ }
+ } finally {
+ lock.unlock();
+
+ }
+ }
+
+}
Modified: uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/transport/IServiceTransport.java
URL: http://svn.apache.org/viewvc/uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/transport/IServiceTransport.java?rev=1851960&r1=1851959&r2=1851960&view=diff
==============================================================================
--- uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/transport/IServiceTransport.java (original)
+++ uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/transport/IServiceTransport.java Wed Jan 23 18:58:01 2019
@@ -25,7 +25,7 @@ import org.apache.uima.ducc.ps.service.e
public interface IServiceTransport extends IServiceComponent {
// called by Protocal Handler. Any errors will be handled
// by instance of IServiceErrorHandler
- public String dispatch(String request) throws TransportException;
+ public IMetaTaskTransaction dispatch(String request) throws TransportException;
// initialize transport
public void initialize() throws ServiceInitializationException;
// stop transport
Modified: uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/transport/http/HttpServiceTransport.java
URL: http://svn.apache.org/viewvc/uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/transport/http/HttpServiceTransport.java?rev=1851960&r1=1851959&r2=1851960&view=diff
==============================================================================
--- uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/transport/http/HttpServiceTransport.java (original)
+++ uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/transport/http/HttpServiceTransport.java Wed Jan 23 18:58:01 2019
@@ -19,6 +19,7 @@
package org.apache.uima.ducc.ps.service.transport.http;
import java.io.IOException;
+import java.io.InvalidClassException;
import java.lang.management.ManagementFactory;
import java.net.InetAddress;
import java.net.NoRouteToHostException;
@@ -28,6 +29,7 @@ import java.net.URISyntaxException;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;
+import java.util.Objects;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.http.HttpEntity;
@@ -49,11 +51,13 @@ import org.apache.uima.ducc.ps.net.iface
import org.apache.uima.ducc.ps.service.errors.IServiceErrorHandler.Action;
import org.apache.uima.ducc.ps.service.errors.ServiceException;
import org.apache.uima.ducc.ps.service.errors.ServiceInitializationException;
+import org.apache.uima.ducc.ps.service.protocol.INoTaskAvailableStrategy;
import org.apache.uima.ducc.ps.service.registry.IRegistryClient;
import org.apache.uima.ducc.ps.service.transport.IServiceTransport;
import org.apache.uima.ducc.ps.service.transport.ITargetURI;
import org.apache.uima.ducc.ps.service.transport.TransportException;
import org.apache.uima.ducc.ps.service.transport.TransportStats;
+import org.apache.uima.ducc.ps.service.transport.XStreamUtils;
import org.apache.uima.ducc.ps.service.transport.target.NoOpTargetURI;
import org.apache.uima.ducc.ps.service.transport.target.TargetURIFactory;
import org.apache.uima.ducc.ps.service.utils.Utils;
@@ -65,11 +69,11 @@ public class HttpServiceTransport implem
private HttpClient httpClient = null;
private PoolingHttpClientConnectionManager cMgr = null;
private int clientMaxConnections = 1;
- private int clientMaxConnectionsPerRoute = 60;
+ private int clientMaxConnectionsPerRoute = 1;
private int clientMaxConnectionsPerHostPort = 0;
private ReentrantLock lock = new ReentrantLock();
private ReentrantLock registryLookupLock = new ReentrantLock();
- private long threadSleepTime=10000; // millis
+ private long threadSleepTime=1000; // millis
private final String nodeIP;
private final String nodeName;
private final String pid;
@@ -83,10 +87,12 @@ public class HttpServiceTransport implem
private volatile boolean stopping = false;
private volatile boolean running = false;
private volatile boolean log = true;
- public HttpServiceTransport(IRegistryClient registryClient, int scaleout) throws ServiceException {
+ private INoTaskAvailableStrategy waitStrategy=null;
+
+ public HttpServiceTransport(IRegistryClient registryClient, int scaleout, INoTaskAvailableStrategy waitStrategy) throws ServiceException {
this.registryClient = registryClient;
clientMaxConnections = scaleout;
-
+ this.waitStrategy = waitStrategy;
try {
nodeIP = InetAddress.getLocalHost().getHostAddress();
nodeName=InetAddress.getLocalHost().getCanonicalHostName();
@@ -208,8 +214,8 @@ public class HttpServiceTransport implem
return running;
}
- private String retryUntilSuccessfull(String request, HttpPost postMethod) {
- String response="";
+ private IMetaTaskTransaction retryUntilSuccessfull(String request, HttpPost postMethod) {
+ IMetaTaskTransaction response=null;
// Only one thread attempts recovery. Other threads will block here
// until connection to the remote is restored.
lock.lock();
@@ -242,30 +248,61 @@ public class HttpServiceTransport implem
return response;
}
- private String doPost(HttpPost postMethod) throws URISyntaxException, IOException, TransportException {
+ private synchronized IMetaTaskTransaction doPost(HttpPost postMethod) throws URISyntaxException, IOException, TransportException {
postMethod.setURI(new URI(currentTargetUrl.asString()));
- HttpResponse response = httpClient.execute(postMethod);
- if ( stopping ) {
- throw new TransportException("Service stopping - rejecting request");
- }
- HttpEntity entity = response.getEntity();
- String serializedResponse = EntityUtils.toString(entity);
- StatusLine statusLine = response.getStatusLine();
- if (statusLine.getStatusCode() != 200 ) {
- // all IOExceptions are retried
- throw new IOException(
- "Unexpected HttpClient response status:"+statusLine+ " Content causing error:"+serializedResponse);
+ IMetaTaskTransaction metaTransaction=null;
+ while(running) {
+ HttpResponse response = httpClient.execute(postMethod);
+ if ( stopping ) {
+ throw new TransportException("Service stopping - rejecting request");
+ }
+ HttpEntity entity = response.getEntity();
+ String serializedResponse = EntityUtils.toString(entity);
+ Object transaction=null;
+ try {
+ transaction = XStreamUtils.unmarshall(serializedResponse);
+ } catch(Exception e) {
+ logger.log(Level.WARNING,"Process Thread:"+Thread.currentThread().getId()+" Error while deserializing response with XStream",e);
+ throw new TransportException(e);
+ }
+ if (transaction instanceof IMetaTaskTransaction) {
+ metaTransaction = (IMetaTaskTransaction) transaction;
+ if ( metaTransaction.getMetaTask() == null || metaTransaction.getMetaTask().getUserSpaceTask() == null) {
+ logger.log(Level.INFO,"Process Thread:"+Thread.currentThread().getId()+" - Driver is out of tasks - waiting for awhile ("+threadSleepTime+" ms) and will try again ");
+ waitStrategy.handleNoTaskSupplied();
+ } else {
+ // Got a task
+ break;
+ }
+ } else {
+ if ( Objects.isNull(transaction)) {
+ throw new InvalidClassException(
+ "Expected IMetaTaskTransaction - Instead Received NULL");
+
+ } else {
+ throw new InvalidClassException(
+ "Expected IMetaTaskTransaction - Instead Received " + transaction.getClass().getName());
+ }
+ }
+ StatusLine statusLine = response.getStatusLine();
+ if (statusLine.getStatusCode() != 200 ) {
+ // all IOExceptions are retried
+ throw new IOException(
+ "Unexpected HttpClient response status:"+statusLine+ " Content causing error:"+serializedResponse);
+ }
+
+
}
-
stats.incrementSuccessCount();
- return serializedResponse;
+ return metaTransaction;
}
@Override
- public String dispatch(String serializedRequest) throws TransportException {
+ public IMetaTaskTransaction dispatch(String serializedRequest) throws TransportException {
if ( stopping ) {
throw new IllegalStateException("Service transport has been stopped, unable to dispatch request");
}
+ IMetaTaskTransaction transaction=null;
HttpEntity e = wrapRequest(serializedRequest);
// Each thread has its own HttpPost method. If current thread
// doesnt have one, it will be created and added to the local
@@ -294,7 +331,7 @@ public class HttpServiceTransport implem
new HttpClientExceptionGenerator(simulatedException);
mockExceptionGenerator.throwSimulatedException();
} else {
- serializedResponse = doPost(postMethod);
+ transaction = doPost(postMethod);
}
} catch( IOException | URISyntaxException ex) {
if ( stopping ) {
@@ -308,7 +345,7 @@ public class HttpServiceTransport implem
logger.log(Level.WARNING, this.getClass().getName()+".dispatch() >>>>>>>>>> Handling Exception \n"+ex);
logger.log(Level.INFO, ">>>>>>>>>> Unable to communicate with target:"+currentTargetUrl.asString()+" - retrying until successfull - with "+threadSleepTime/1000+" seconds wait between retries ");
}
- serializedResponse = retryUntilSuccessfull(serializedRequest, postMethod);
+ transaction = retryUntilSuccessfull(serializedRequest, postMethod);
log = true;
logger.log(Level.INFO, "Established connection to target:"+currentTargetUrl.asString());
}
@@ -317,7 +354,7 @@ public class HttpServiceTransport implem
} finally {
postMethod.releaseConnection();
}
- return serializedResponse;
+ return transaction;
}
Modified: uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/Client.java
URL: http://svn.apache.org/viewvc/uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/Client.java?rev=1851960&r1=1851959&r2=1851960&view=diff
==============================================================================
--- uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/Client.java (original)
+++ uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/Client.java Wed Jan 23 18:58:01 2019
@@ -166,7 +166,7 @@ public class Client {
String nodeName = request.getHeader("Hostname");
String threadID = request.getHeader("ThreadID");
String pid = request.getHeader("PID");
- // System.out.println( "Sender ID:::Node IP"+nodeIP+" Node Name:"+nodeName+" PID:"+pid+" ThreadID:"+threadID);
+ System.out.println( "Sender ID:::Node IP"+nodeIP+" Node Name:"+nodeName+" PID:"+pid+" ThreadID:"+threadID);
IMetaTaskTransaction imt = null;
Modified: uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/service/wrapper/JUnitServiceWrapperTestCase.java
URL: http://svn.apache.org/viewvc/uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/service/wrapper/JUnitServiceWrapperTestCase.java?rev=1851960&r1=1851959&r2=1851960&view=diff
==============================================================================
--- uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/service/wrapper/JUnitServiceWrapperTestCase.java (original)
+++ uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/service/wrapper/JUnitServiceWrapperTestCase.java Wed Jan 23 18:58:01 2019
@@ -41,7 +41,7 @@ public class JUnitServiceWrapperTestCase
public void testPullServiceWrapperNoTask() throws Exception {
// make client return null task in response to GET
System.setProperty("simulate.no.work", "3");
- System.setProperty("ducc.process.thread.sleep.time", "2000");
+ System.setProperty("ducc.process.thread.sleep.time", "1000");
try {
testPullServiceWrapper();
} finally {
@@ -71,7 +71,7 @@ public class JUnitServiceWrapperTestCase
Timer fTimer = new Timer("testPullService Timer");
// after 5secs stop the pull service
- fTimer.schedule(new MyTimerTask(service, fTimer), 20000);
+ fTimer.schedule(new MyTimerTask(service, fTimer), 40000);
service.initialize(new String[] {analysisEngineDescriptor});
Modified: uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/transport/JunitProtocolHandlerTestCase.java
URL: http://svn.apache.org/viewvc/uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/transport/JunitProtocolHandlerTestCase.java?rev=1851960&r1=1851959&r2=1851960&view=diff
==============================================================================
--- uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/transport/JunitProtocolHandlerTestCase.java (original)
+++ uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/transport/JunitProtocolHandlerTestCase.java Wed Jan 23 18:58:01 2019
@@ -31,7 +31,9 @@ import org.apache.uima.ducc.ps.service.e
import org.apache.uima.ducc.ps.service.errors.ServiceInitializationException;
import org.apache.uima.ducc.ps.service.processor.IServiceProcessor;
import org.apache.uima.ducc.ps.service.processor.uima.UimaServiceProcessor;
+import org.apache.uima.ducc.ps.service.protocol.INoTaskAvailableStrategy;
import org.apache.uima.ducc.ps.service.protocol.IServiceProtocolHandler;
+import org.apache.uima.ducc.ps.service.protocol.builtin.DefaultNoTaskAvailableStrategy;
import org.apache.uima.ducc.ps.service.protocol.builtin.DefaultServiceProtocolHandler;
import org.apache.uima.ducc.ps.service.registry.DefaultRegistryClient;
import org.apache.uima.ducc.ps.service.transport.IServiceTransport;
@@ -50,8 +52,10 @@ public class JunitProtocolHandlerTestCas
ITargetURI targetUrl = new HttpTargetURI("http://localhost:"+super.getPort()+"/"+super.getApp());
DefaultRegistryClient registryClient =
new DefaultRegistryClient(targetUrl);
+ INoTaskAvailableStrategy waitStrategy =
+ new DefaultNoTaskAvailableStrategy(1000);
HttpServiceTransport transport =
- new HttpServiceTransport(registryClient, scaleout);
+ new HttpServiceTransport(registryClient, scaleout,waitStrategy);
transport.initialize();
return transport;
}
Modified: uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/transport/JunitTransportTestCase.java
URL: http://svn.apache.org/viewvc/uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/transport/JunitTransportTestCase.java?rev=1851960&r1=1851959&r2=1851960&view=diff
==============================================================================
--- uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/transport/JunitTransportTestCase.java (original)
+++ uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/transport/JunitTransportTestCase.java Wed Jan 23 18:58:01 2019
@@ -27,8 +27,15 @@ import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
+import org.apache.uima.ducc.ps.net.iface.IMetaTaskTransaction;
+import org.apache.uima.ducc.ps.net.impl.MetaMetaTask;
+import org.apache.uima.ducc.ps.net.impl.MetaTask;
+import org.apache.uima.ducc.ps.net.impl.MetaTaskTransaction;
+import org.apache.uima.ducc.ps.service.protocol.INoTaskAvailableStrategy;
+import org.apache.uima.ducc.ps.service.protocol.builtin.DefaultNoTaskAvailableStrategy;
import org.apache.uima.ducc.ps.service.registry.DefaultRegistryClient;
import org.apache.uima.ducc.ps.service.transport.ITargetURI;
+import org.apache.uima.ducc.ps.service.transport.XStreamUtils;
import org.apache.uima.ducc.ps.service.transport.http.HttpServiceTransport;
import org.apache.uima.ducc.ps.service.transport.http.HttpServiceTransport.HttpClientExceptionGenerator.ERROR;
import org.apache.uima.ducc.ps.service.transport.target.HttpTargetURI;
@@ -131,7 +138,10 @@ public class JunitTransportTestCase {
ITargetURI targetUrl = new HttpTargetURI("http://localhost:"+getPort()+"/"+app);
DefaultRegistryClient registryClient =
new DefaultRegistryClient(targetUrl);
- HttpServiceTransport transport = new HttpServiceTransport(registryClient, scaleout);
+ INoTaskAvailableStrategy waitStrategy =
+ new DefaultNoTaskAvailableStrategy(1000);
+
+ HttpServiceTransport transport = new HttpServiceTransport(registryClient, scaleout, waitStrategy);
transport.initialize();
//String response = transport.getWork("Test");
//System.out.println("Test Received Response:"+response);
@@ -146,10 +156,13 @@ public class JunitTransportTestCase {
ITargetURI targetUrl = new HttpTargetURI("http://localhost:"+getPort()+"/"+app);
DefaultRegistryClient registryClient =
new DefaultRegistryClient(targetUrl);
- HttpServiceTransport transport = new HttpServiceTransport(registryClient, scaleout);
+ INoTaskAvailableStrategy waitStrategy =
+ new DefaultNoTaskAvailableStrategy(1000);
+ HttpServiceTransport transport = new HttpServiceTransport(registryClient, scaleout, waitStrategy);
transport.initialize();
System.setProperty("MockHttpPostError", ERROR.IOException.name());
- transport.dispatch("Dummy Message");
+ IMetaTaskTransaction transaction = new MetaTaskTransaction();
+ transport.dispatch(XStreamUtils.marshall(transaction));
wait(registryClient);
}
@@ -161,10 +174,13 @@ public class JunitTransportTestCase {
ITargetURI targetUrl = new HttpTargetURI("http://localhost:"+getPort()+"/"+app);
DefaultRegistryClient registryClient =
new DefaultRegistryClient(targetUrl);
- HttpServiceTransport transport = new HttpServiceTransport(registryClient, scaleout);
+ INoTaskAvailableStrategy waitStrategy =
+ new DefaultNoTaskAvailableStrategy(1000);
+ HttpServiceTransport transport = new HttpServiceTransport(registryClient, scaleout, waitStrategy);
transport.initialize();
System.setProperty("MockHttpPostError", ERROR.NoRouteToHostException.name());
- transport.dispatch("Dummy Message");
+ IMetaTaskTransaction transaction = new MetaTaskTransaction();
+ transport.dispatch(XStreamUtils.marshall(transaction));
wait(registryClient);
}
@@ -176,10 +192,13 @@ public class JunitTransportTestCase {
ITargetURI targetUrl = new HttpTargetURI("http://localhost:"+getPort()+"/"+app);
DefaultRegistryClient registryClient =
new DefaultRegistryClient(targetUrl);
- HttpServiceTransport transport = new HttpServiceTransport(registryClient, scaleout);
+ INoTaskAvailableStrategy waitStrategy =
+ new DefaultNoTaskAvailableStrategy(1000);
+ HttpServiceTransport transport = new HttpServiceTransport(registryClient, scaleout, waitStrategy);
transport.initialize();
System.setProperty("MockHttpPostError", ERROR.URISyntaxException.name());
- transport.dispatch("Dummy Message");
+ IMetaTaskTransaction transaction = new MetaTaskTransaction();
+ transport.dispatch(XStreamUtils.marshall(transaction));
wait(registryClient);
}
@@ -203,7 +222,7 @@ public class JunitTransportTestCase {
}
String content = sb.toString().trim();
- System.out.println( "Http Request Body:::"+String.valueOf(content));
+ //System.out.println( "Http Request Body:::"+content);
String nodeIP = request.getHeader("IP");
@@ -211,8 +230,12 @@ public class JunitTransportTestCase {
String threadID = request.getHeader("ThreadID");
String pid = request.getHeader("PID");
System.out.println( "Sender ID:::Node IP"+nodeIP+" Node Name:"+nodeName+" PID:"+pid+" ThreadID:"+threadID);
-
- response.getWriter().write(content);
+ IMetaTaskTransaction transaction = (IMetaTaskTransaction)XStreamUtils.unmarshall(content);
+ transaction.setMetaTask(new MetaTask(1,"",null));
+ transaction.getMetaTask().setUserSpaceTask("Test Message");
+ String serializedResponse = XStreamUtils.marshall(transaction);
+ //System.out.println(serializedResponse);
+ response.getWriter().write(serializedResponse);
} catch (Throwable e) {
e.printStackTrace();
throw new ServletException(e);