StackOverflow2013

Note that there are some explanatory texts on larger screens.

plurals

POExecutorCompletionService not taking an item returned by Callable?
text
Body
copied!<p>I'm having strange behavior from the ExecutorCompletionService. The item gets added to the ExecutorCompletionService.submit() fine. Then it gets worked on and is returned by the Callable worker thread which was previously submitted. After that return the ExecutorCompletionService.take() never sees it so never sees the blocking to return anymore items? I'm really not sure what is going on. I have created print lines and can see it completing the Callable worker thread. As soon as that happens the ExecutorCompletionService.take should be ready to take but in some cases the thing locks up and sometimes its fine?</p> <p>I have created a test case if you run it a few times you will see it will in some cases lock up and never take any of the finished threads </p> <p>ThreadDeadlockDemo</p> <pre><code>import java.util.Observable; import java.util.concurrent.BlockingQueue; import java.util.concurrent.Callable; import java.util.concurrent.CompletionService; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorCompletionService; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.LinkedBlockingQueue; public class ThreadDeadlockDemo extends Observable implements Runnable { private CompletionService<String> pool; private ExecutorService executor ; private Thread responseWorkerThread; private HttpSchedulerWorker schedulerWorker; private boolean shouldRun = true; private int numThreadsInPool; private BlockingQueue<String> queue; public ThreadDeadlockDemo(int numThreads) { numThreadsInPool = numThreads; executor = Executors.newFixedThreadPool(numThreads); pool = new ExecutorCompletionService<String>(executor); schedulerWorker = new HttpSchedulerWorker(); responseWorkerThread = new Thread(schedulerWorker); responseWorkerThread.start(); queue = new LinkedBlockingQueue<String>(); new Thread(this).start(); } public ThreadDeadlockDemo() { numThreadsInPool = 1; executor = Executors.newFixedThreadPool(1); pool = new ExecutorCompletionService<String>(executor); schedulerWorker = new HttpSchedulerWorker(); responseWorkerThread = new Thread(schedulerWorker); responseWorkerThread.start(); queue = new LinkedBlockingQueue<String>(); new Thread(this).start(); } public void setThreadCount(int numThreads) { executor = Executors.newFixedThreadPool(numThreads); pool = new ExecutorCompletionService<String>(executor); numThreadsInPool = numThreads; } public void add(String info) { queue.add(info); } @Override public void run() { // TODO Auto-generated method stub while(shouldRun) { try { String info = queue.take(); Callable<String> worker = new WorkerThread(info); System.out.println("submitting to pooler: " + info); pool.submit(worker); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } } /** * Inner class of proxy is a worker thread blocks until the pool has transactions complete as soon as they * are complete it will send them to server for completion. * @author Steve * */ class HttpSchedulerWorker implements Runnable{ public void run() { // TODO Auto-generated method stub while(true) { String vulnInfo = null; try { Future<String> tmp = pool.take(); // Future<VulnInfo> tmp = pool.poll(); if(tmp != null) vulnInfo = tmp.get(); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (ExecutionException e) { // TODO Auto-generated catch block e.printStackTrace(); } if(vulnInfo != null) { System.out.println("info was taken from pool completed: " + vulnInfo); } } } } </code></pre> <p>}</p> <p>WorkerClass: this is the thread worker added to the executor pool and returns but in some cases never gets notified in the ThreadlockDemos ExecutorCompletionService pool?</p> <pre><code>import java.util.concurrent.Callable; public class WorkerThread implements Callable<String>{ String info; WorkerThread(String info) { this.info = info; } //@Override public String call() throws Exception { System.out.println("sending vuln info: " + info); return info; } } </code></pre> <p>Here is my test class simply adds items to queue. Here is a print out from my console of one that looks to have failed. Its adds to the queue works on it and returns the value. But the take() is never called any ideas why? It works sometimes and sometimes fails making it very hard for me to see what is wrong.I'd love to say its bug in java but I looked around did not see any issues with theses classes?</p> <pre><code> public class HttpSchedulerThreadedUnitTest { ThreadDeadlockDemo scheduler; public HttpSchedulerThreadedUnitTest(){ setupScheduler(); for(int i=0; i < 5;i++) { scheduler.add(i+""); } } private void setupScheduler() { scheduler = new ThreadDeadlockDemo(); scheduler.setThreadCount(1); } public static void main(String[] args) { new HttpSchedulerThreadedUnitTest(); } </code></pre> <p>}</p> <p>Console Print: this is it run never taking from the pool when the WorkerThread completes submitting to pooler: 0 submitting to pooler: 1 submitting to pooler: 2 sending vuln info: 0 submitting to pooler: 3 sending vuln info: 1 submitting to pooler: 4 sending vuln info: 2 sending vuln info: 3 sending vuln info: 4</p> <p>Console Print : it actually doing taking itenms from pool returning! submitting to pooler: 0 submitting to pooler: 1 submitting to pooler: 2 submitting to pooler: 3 submitting to pooler: 4 sending vuln info: 0 info was taken from pool completed: 0 sending vuln info: 1 info was taken from pool completed: 1 sending vuln info: 2 info was taken from pool completed: 2 sending vuln info: 3 info was taken from pool completed: 3 sending vuln info: 4 info was taken from pool completed: 4</p>

Querying!

Guidance

An individual column

Larger individual text columns get their own page to allow for proper reading.

SQuiL has stopped working due to an internal error.

If you are curious you may find further information in the browser console, which is accessible through the devtools (F12).

Reload