Author: cwiklik
Date: Wed Jul 18 20:42:44 2018
New Revision: 1836233
URL: http://svn.apache.org/viewvc?rev=1836233&view=rev
Log:
UIMA-5815 support quiesce on SIGTERM
Modified:
uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/Lifecycle.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/main/ServiceWrapper.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/http/HttpServiceTransport.java
uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/service/JunitPullServiceTestCase.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
Modified:
uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/Lifecycle.java
URL:
http://svn.apache.org/viewvc/uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/Lifecycle.java?rev=1836233&r1=1836232&r2=1836233&view=diff
==============================================================================
---
uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/Lifecycle.java
(original)
+++
uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/Lifecycle.java
Wed Jul 18 20:42:44 2018
@@ -25,4 +25,5 @@ import org.apache.uima.ducc.ps.service.e
public interface Lifecycle {
public void start() throws ExecutionException, ServiceException;
public void stop();
+ public void quiesceAndStop();
}
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=1836233&r1=1836232&r2=1836233&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 Jul 18 20:42:44 2018
@@ -250,12 +250,21 @@ public class PullService implements ISer
stopProcessThreads();
// close connection to remote client and cleanup
stopTransport();
- stopProtocolHandler();
+ stopProtocolHandler(false);
stopServiceProcessor();
- // monitor should be stopped last to keep posting updates to observer
+ // monitor should be stopped last to keep posting updates to
observer
stopMonitor();
}
-
+ public void quiesceAndStop() {
+ // when quiescing, let the process threads finish processing
+ stopProtocolHandler(true);
+
+ // close connection to remote client and cleanup
+ stopTransport();
+ stopServiceProcessor();
+ // monitor should be stopped last to keep posting updates to observer
+ stopMonitor();
+ }
private void waitForProcessThreads() throws InterruptedException,
ExecutionException {
for (Future<String> future : threadHandleList) {
// print the return value of Future, notice the output
delay in console
@@ -306,8 +315,12 @@ public class PullService implements ISer
serviceProcessor.stop();
}
}
- private void stopProtocolHandler() {
-
+ private void stopProtocolHandler(boolean quiesce) {
+ if ( quiesce ) {
+ protocolHandler.quiesceAndStop();
+ } else {
+ protocolHandler.stop();
+ }
}
private void stopTransport() {
transport.stop();
Modified:
uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/main/ServiceWrapper.java
URL:
http://svn.apache.org/viewvc/uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/main/ServiceWrapper.java?rev=1836233&r1=1836232&r2=1836233&view=diff
==============================================================================
---
uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/main/ServiceWrapper.java
(original)
+++
uima/uima-ducc/trunk/uima-ducc-pullservice/src/main/java/org/apache/uima/ducc/ps/service/main/ServiceWrapper.java
Wed Jul 18 20:42:44 2018
@@ -19,7 +19,6 @@
package org.apache.uima.ducc.ps.service.main;
-import java.io.IOException;
import java.lang.reflect.Constructor;
import java.util.Objects;
import java.util.concurrent.ExecutionException;
@@ -186,15 +185,25 @@ public class ServiceWrapper {
}
public void stop() {
- service.stop();
try {
+ service.stop();
jmxAgent.stop();
- } catch( IOException e ) {
-
+ } catch( Exception e ) {
+ logger.log(Level.WARNING,"",e);
+
}
}
+ public void quiesceAndStop() {
+ try {
+ service.quiesceAndStop();
+ jmxAgent.stop();
+ } catch( Exception e ) {
+ logger.log(Level.WARNING,"",e);
+ }
+
+ }
public static void main(String[] args) {
ServiceWrapper wrapper = null;
try {
@@ -219,9 +228,9 @@ public class ServiceWrapper {
@Override
public void run() {
try {
- logger.log(Level.INFO, "Pull Service Caught SIGTERM
Signal - Stopping ...");
+ logger.log(Level.INFO, "Pull Service Caught SIGTERM
Signal - Stopping (Quiescing) ...");
- serviceWrapper.stop();
+ serviceWrapper.quiesceAndStop();
} catch (Exception e) {
logger.log(Level.WARNING,"", e);
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=1836233&r1=1836232&r2=1836233&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 Jul 18 20:42:44 2018
@@ -52,6 +52,7 @@ public class DefaultServiceProtocolHandl
Logger logger =
UIMAFramework.getLogger(DefaultServiceProtocolHandler.class);
private volatile boolean initError = false;
private volatile boolean running = false;
+ private volatile boolean quiescing = false;
private IServiceTransport transport;
private IServiceProcessor processor;
private INoTaskAvailableStrategy noTaskStrategy;
@@ -214,8 +215,10 @@ public class DefaultServiceProtocolHandl
// send GET Request
transaction = callGet(new
MetaTaskTransaction());
// the code may have blocked in callGet for
awhile, so check
- // if service is still running
- if ( !running ) {
+ // if service is still running. If this service
is in quiescing
+ // mode, finish processing current task. The
while-loop will
+ // terminate when the task is finished.
+ if ( !running && !quiescing ) {
break;
}
if (transaction.getMetaTask() == null ||
transaction.getMetaTask().getUserSpaceTask() == null ) {
@@ -227,7 +230,7 @@ public class DefaultServiceProtocolHandl
// send ACK
transaction = callAck(transaction);
- if (!running) {
+ if (!running && !quiescing ) {
break;
}
IProcessResult processResult =
processor.process((String) task);
@@ -241,7 +244,6 @@ public class DefaultServiceProtocolHandl
mc.setUserSpaceException(errorAsString);
} else {
// success
- // System.out.println("Performance
Metrics:"+processResult.getResult());
transaction.getMetaTask().setPerformanceMetrics(processResult.getResult());
}
// send END Request
@@ -282,16 +284,31 @@ public class DefaultServiceProtocolHandl
private void delegateStop() {
- service.stop();
+ service.stop(); // dont quiesce
}
@Override
public void stop() {
+ quiescing = false;
running = false;
if ( logger.isLoggable(Level.INFO)) {
logger.log(Level.INFO, this.getClass().getName()+"
stop() called");
}
}
@Override
+ public void quiesceAndStop() {
+ quiescing = true;
+ running = false;
+ if ( logger.isLoggable(Level.INFO)) {
+ logger.log(Level.INFO, this.getClass().getName()+"
quiesceAndStop() called");
+ }
+ try {
+ // wait for process threads to terminate
+ stopLatch.await();
+ logger.log(Level.INFO, this.getClass().getName()+" All
process threads completed quiesce");
+ } catch( Exception e ) {
+ }
+ }
+ @Override
public void start() {
running = true;
// process threads are initialized and are awaiting latch
countdown
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=1836233&r1=1836232&r2=1836233&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 Jul 18 20:42:44 2018
@@ -66,7 +66,6 @@ public class HttpServiceTransport implem
private int clientMaxConnectionsPerRoute = 60;
private int clientMaxConnectionsPerHostPort = 0;
private ReentrantLock lock = new ReentrantLock();
-// private CountDownLatch initLatch = new CountDownLatch(1);
private ReentrantLock registryLookupLock = new ReentrantLock();
private long threadSleepTime=10000; // millis
private final String nodeIP;
@@ -82,30 +81,7 @@ 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) {
- // create instance of HttpServiceTransport with RegistryClient.
The assumption
- // is that the implementation of the client has been fully
configured with
- // registry URI and target id. We just pass in a NoOpTarget
instead of null.
- // The initialize() will use registry to lookup the client
TargetURI and will
- // create correct instance of ITargetURI based on what registry
returns
-// this(new NoOpTargetURI(), registryClient, scaleout);
- this(new NoOpTargetURI(), registryClient, scaleout);
-
- }
- */
-// private HttpServiceTransport(ITargetURI targetUrl, IRegistryClient
registryClient, int scaleout) {
- public HttpServiceTransport(IRegistryClient registryClient, int
scaleout) throws ServiceException {
- //TargetURIFactory.newTarget(registryClient.lookUp(new
NoOpTargetURI().asString()));
- /*
- if ( registryClient == null ) {
- // the default client just returns the same targetUrl
- // No lookups
- this.registryClient = new
DefaultRegistryClient(targetUrl);
- } else {
- this.registryClient = registryClient;
- }
- */
+ public HttpServiceTransport(IRegistryClient registryClient, int
scaleout) throws ServiceException {
this.registryClient = registryClient;
clientMaxConnections = scaleout;
@@ -152,7 +128,6 @@ public class HttpServiceTransport implem
while( !stopping ) {
try {
String newTarget =
registryClient.lookUp(currentTargetUrl.asString());
- // logger.log(Level.INFO, "Registry lookup
succesfull - current target URL:"+newTarget);
currentTargetUrl =
TargetURIFactory.newTarget(newTarget);
break;
} catch( Exception e) {
@@ -184,11 +159,6 @@ public class HttpServiceTransport implem
logger.log(Level.FINE,
"threadId:"+transaction.getRequesterThreadId());
}
-//
-// transaction.setRequesterNodeName(nodeName);
-// transaction.setRequesterProcessId(pid);
-// transaction.setRequesterProcessName(value);
-//
transaction.setRequesterThreadId((int)Thread.currentThread().getId());
}
public void initialize() throws ServiceInitializationException {
@@ -211,19 +181,6 @@ public class HttpServiceTransport implem
cMgr.setMaxPerRoute(new HttpRoute(httpHost),
clientMaxConnectionsPerHostPort);
}
- int timeout = 30;
-// SocketConfig socketConfig =
SocketConfig.custom().setSoTimeout(timeout*1000).build();
- // RequestConfig requestConfig = RequestConfig.custom()
- // .setConnectTimeout(timeout * 1000)
- // .setConnectionRequestTimeout(timeout * 1000)
- // .setSocketTimeout(0).build();
- // cMgr.setDefaultSocketConfig(socketConfig);
-
-// System.out.println("HttpTransport Max
Connections:"+cMgr.getMaxTotal());
-// httpClient = HttpClients.custom().
- // setConnectionManager(cMgr).
- // setDefaultRequestConfig(requestConfig).build();
-
httpClient =
HttpClients.custom().setConnectionManager(cMgr).build();
running = true;
@@ -258,7 +215,6 @@ public class HttpServiceTransport implem
// retry until service is stopped
while (isRunning()) {
try {
- //response = dispatch(request);
response = doPost(postMethod);
// success, so release the lock so that other
waiting threads
// can retry command
@@ -307,7 +263,6 @@ public class HttpServiceTransport implem
}
@Override
public String dispatch(String serializedRequest) throws
TransportException {
-// System.out.println(".... in dispatch()...stopping="+stopping);
if ( stopping ) {
throw new IllegalStateException("Service transport has
been stopped, unable to dispatch request");
}
@@ -326,14 +281,11 @@ public class HttpServiceTransport implem
} catch ( NoHttpResponseException ex ) {
if ( stopping ) {
System.out.println("Process
Thread:"+Thread.currentThread().getId()+" NoHttpResponseException ");
- //ex.printStackTrace();
throw new TransportException(ex);
} else {
serializedResponse =
retryUntilSuccessfull(serializedRequest, postMethod);
}
- // timeout so try again
- //ex.printStackTrace();
} catch (HttpHostConnectException | UnknownHostException ex ) {
if ( stopping ) {
System.out.println("Process
Thread:"+Thread.currentThread().getId()+" HttpHostConnectException ");
@@ -368,7 +320,6 @@ public class HttpServiceTransport implem
} catch (SocketException ex) {
if ( stopping ) {
- //System.out.println("Process
Thread:"+Thread.currentThread().getId()+" SocketException ");
throw new TransportException(ex);
}
@@ -404,7 +355,6 @@ public class HttpServiceTransport implem
stopping = true;
running = false;
- //initLatch.countDown();
logger.log(Level.INFO,this.getClass().getName()+" stop()
called");
if ( cMgr != null ) {
cMgr.shutdown();
Modified:
uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/service/JunitPullServiceTestCase.java
URL:
http://svn.apache.org/viewvc/uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/service/JunitPullServiceTestCase.java?rev=1836233&r1=1836232&r2=1836233&view=diff
==============================================================================
---
uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/service/JunitPullServiceTestCase.java
(original)
+++
uima/uima-ducc/trunk/uima-ducc-pullservice/src/test/java/org/apache/uima/ducc/ps/service/JunitPullServiceTestCase.java
Wed Jul 18 20:42:44 2018
@@ -19,6 +19,7 @@ public class JunitPullServiceTestCase ex
@Test
public void testPullService() throws Exception {
+ System.out.println("----------------- testPullService
-------------------");
int scaleout = 2;
super.startJetty(false); // don't block
String analysisEngineDescriptor = "TestAAE";
@@ -36,7 +37,7 @@ public class JunitPullServiceTestCase ex
service.initialize();
Timer fTimer = new Timer("testPullService Timer");
// after 5secs stop the pull service
- fTimer.schedule(new MyTimerTask(service, fTimer),
35000);
+ fTimer.schedule(new MyTimerTask(service, fTimer,
false), 35000);
service.start();
@@ -47,7 +48,39 @@ public class JunitPullServiceTestCase ex
}
}
@Test
+ public void testPullServiceQuiesce() throws Exception {
+ System.out.println("----------------- testPullServiceQuiesce
-------------------");
+ int scaleout = 2;
+ super.startJetty(false); // don't block
+ String analysisEngineDescriptor = "TestAAE";
+ System.setProperty("ducc.deploy.JpType", "uima");
+ IServiceProcessor processor = new
+ UimaServiceProcessor(analysisEngineDescriptor);
+
+ String tasURL = "http://localhost:8080/test";;
+
+ IService service =
PullServiceStepBuilder.newBuilder().withProcessor(processor)
+ .withClientURL(tasURL).withType("Note
Service").withScaleout(scaleout)
+ .withOptionalsDone().build();
+
+ try {
+ service.initialize();
+ Timer fTimer = new Timer("testPullService Timer");
+ // after 5secs stop the pull service
+ fTimer.schedule(new MyTimerTask(service, fTimer, true),
35000);
+
+ service.start();
+
+ } catch (ServiceInitializationException e) {
+ throw e;
+ } catch (Exception e) {
+ throw e;
+ }
+ }
+
+ @Test
public void testPullServiceTimeout() throws Exception {
+ System.out.println("----------------- testPullServiceTimeout
-------------------");
super.startJetty(true); // true=client blocks all POST requests
int scaleout = 12;
String analysisEngineDescriptor = "TestAAE";
@@ -65,7 +98,7 @@ public class JunitPullServiceTestCase ex
System.out.println("----------- Starting Service
.....");
Timer fTimer = new Timer();
//after 10sec stop the service
- fTimer.schedule(new MyTimerTask(service, fTimer),
40000);
+ fTimer.schedule(new MyTimerTask(service, fTimer,
false), 40000);
service.start();
@@ -79,6 +112,7 @@ public class JunitPullServiceTestCase ex
@Test
public void testPullServiceWithProcessFailure() throws Exception {
+ System.out.println("-----------------
testPullServiceWithProcessFailure -------------------");
int scaleout = 2;
super.startJetty(false); // don't block
String analysisEngineDescriptor = "NoOpAE";
@@ -96,7 +130,7 @@ public class JunitPullServiceTestCase ex
service.initialize();
Timer fTimer = new Timer("testPullService Timer");
// after 5secs stop the pull service
- fTimer.schedule(new MyTimerTask(service, fTimer),
35000);
+ fTimer.schedule(new MyTimerTask(service, fTimer,
false), 35000);
service.start();
@@ -139,24 +173,28 @@ public class JunitPullServiceTestCase ex
class MyTimerTask extends TimerTask {
final IService service;
final Timer fTimer;
- MyTimerTask(IService service, Timer fTimer) {
+ final boolean quiesce;
+
+ MyTimerTask(IService service, Timer fTimer, boolean quiesce) {
this.service = service;
this.fTimer = fTimer;
+ this.quiesce = quiesce;
}
-
- @Override
-
- public void run() {
- this.cancel();
- fTimer.purge();
- fTimer.cancel();
- System.out.println("Timmer popped - stopping
service");
- service.stop();
-
- }
-
-
-
- }
+
+ @Override
+
+ public void run() {
+ this.cancel();
+ fTimer.purge();
+ fTimer.cancel();
+ System.out.println("Timmer popped - stopping service");
+ if (quiesce ) {
+ service.quiesceAndStop();
+ } else {
+ service.stop();
+ }
+ }
+
+ }
}
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=1836233&r1=1836232&r2=1836233&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 Jul 18 20:42:44 2018
@@ -31,6 +31,8 @@ public class JUnitServiceWrapperTestCase
@Test
public void testPullServiceWrapper() throws Exception {
+ System.out.println("--------------------------
testPullServiceWrapper ----------------------");;
+
//int scaleout = 2;
StateMonitor monitor = new StateMonitor();
monitor.start();
@@ -44,7 +46,7 @@ public class JUnitServiceWrapperTestCase
System.setProperty("ducc.deploy.JdURL", tasURL);
System.setProperty("ducc.deploy.JpThreadCount","4");
System.setProperty("ducc.deploy.service.type",
"NotesService");
- System.getProperty("ducc.deploy.JpType", "uima");
+ System.setProperty("ducc.deploy.JpType", "uima");
ServiceWrapper service = new ServiceWrapper();
@@ -69,6 +71,7 @@ public class JUnitServiceWrapperTestCase
@Test
public void testPullServiceWrapperWithProcessFailure() throws Exception
{
+ System.out.println("--------------------------
testPullServiceWrapperWithProcessFailure ----------------------");;
//int scaleout = 2;
StateMonitor monitor = new StateMonitor();
monitor.start();
@@ -84,7 +87,7 @@ public class JUnitServiceWrapperTestCase
System.setProperty("ducc.deploy.JdURL", tasURL);
System.setProperty("ducc.deploy.JpThreadCount","4");
System.setProperty("ducc.deploy.service.type",
"NotesService");
- System.getProperty("ducc.deploy.JpType", "uima");
+ System.setProperty("ducc.deploy.JpType", "uima");
ServiceWrapper service = new ServiceWrapper();
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=1836233&r1=1836232&r2=1836233&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 Jul 18 20:42:44 2018
@@ -125,30 +125,35 @@ public class JunitProtocolHandlerTestCas
transport.stop();
processor.stop();
}
-
+ @Override
+ public void quiesceAndStop() {
+ protocolHandler.quiesceAndStop();
+ threadPool.shutdown();
+ transport.stop();
+ processor.stop();
+ }
@Override
public void initialize() throws ServiceInitializationException {
-
- List<Future<String>> threadHandleList =
- new ArrayList<Future<String>>();
- threadPool =
- new
ScheduledThreadPoolExecutor(scaleout, new ServiceThreadFactory());
-
-
- // Create and start worker threads that pull Work Items
from a client
- for (int j = 0; j < scaleout; j++) {
- threadHandleList.add(
threadPool.submit(protocolHandler));
- }
- try {
- // wait until all process threads
initialize
- threadsReady.await();
-
- } catch( InterruptedException e) {
- Thread.currentThread().interrupt();
- threadPool.shutdownNow();
- throw new
ServiceInitializationException("Service interrupted during initialization -
shutting down process threads");
- }
+
+ List<Future<String>> threadHandleList = new
ArrayList<Future<String>>();
+ threadPool = new ScheduledThreadPoolExecutor(scaleout,
new ServiceThreadFactory());
+
+ // Create and start worker threads that pull Work Items
from a client
+ for (int j = 0; j < scaleout; j++) {
+
threadHandleList.add(threadPool.submit(protocolHandler));
+ }
+ try {
+ // wait until all process threads initialize
+ threadsReady.await();
+
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ threadPool.shutdownNow();
+ throw new ServiceInitializationException(
+ "Service interrupted during
initialization - shutting down process threads");
+ }
}
+
@Override
public String getType() {
// TODO Auto-generated method stub
