realtime output of finished threads in multithreading program (CompletionService)

Question

I try to make multithreading program on Java print results that returned finished threads. The thing is, when i run this code it simply gets stuck on second value that was in queue:

        System.out.println("[!] Creaing pool");
        int max_threads     = 50;
        ExecutorService threadPool  = Executors.newFixedThreadPool(max_threads);
        CompletionService<String> taskCompletionService =
        new ExecutorCompletionService<String>(threadPool);
        String url;

        while(our_file.hasNext()){

            url = our_file.next();
            if (url.length()>0){

                futures.add(
                    taskCompletionService.submit(
                    new GoGo(url)
                    )
                    );
                    }



            int total_tasks = futures.size();

            while(total_tasks>0){
            for (int i=0; i<futures.size(); i++){

                try{    

                    Future result = taskCompletionService.poll();
                    if(result!=null && result.isDone()){
                        System.out.println(result.get());
                        total_tasks--;

                    }
                }
                catch (InterruptedException e) {
                    // Something went wrong with a task submitted
                    System.out.println("Error Interrupted exception");
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    // Something went wrong with the result
                    e.printStackTrace();
                    System.out.println("Error get() threw exception");
                }

                }


            }

                 }


        threadPool.shutdown();
        try {
        threadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
        }
        catch (InterruptedException e ) {
        }



...



     class GoGo implements Callable{

        private String url;

        public GoGo(String received_url){
            this.url = received_url;
        }

        public String call(){

            String URL = this.url;
            return url;

        }
    }

output is like this:

[!] Creaing pool
http://www.www1.com/
http://www.www2.ch/

and at this point program just stucks. I tried to move loop that iterates futures array out of main loop that submits threads, and it worked fine, but in case if i will go through very large file i need real-time output. Please help me figure out where is the bottleneck, I was unable to find any suitable piece of code that using non-blocking poll() method from CompletionService. Thanks for any answer or reference.

Answer 1

Thanks to Robin Green for advice, putting future harvester class to separate thread solved the problem! So, i just start the endless loop thread that pops argument with poll() check if the popped future object indicates that thread isDone() and write output. And after shutting down the fixedThreadPool, output writer class is stopped. Here's the code (except the GoGo class):

    public class headScanner {


        public static List<Future<String>> gloabal_futures   = new ArrayList<Future<String>>();


        public static void main(String args[]){

            Scanner our_file                 = null;
            ArrayList<String> our_urls       = new ArrayList<String>();
            List<Future<String>> futures     = new ArrayList<Future<String>>();
            ArrayList<String> urls_buffer    = new ArrayList<String>();


            try {
            our_file = new Scanner (new File ("list.txt"));
            }
            catch(IOException e){
                System.out.println("[-] Cant open the file!");
                System.exit(0);
            }

            System.out.println("[!] Creaing pool");
            int max_threads     = 50;
            ExecutorService threadPool  = Executors.newFixedThreadPool(max_threads);
            CompletionService<String> taskCompletionService =
            new ExecutorCompletionService<String>(threadPool);
            String url;



            Thread result_thread = new Thread(new ResultHarvester(futures.size(), taskCompletionService));
            result_thread.start();  

            while(our_file.hasNext()){

                url = our_file.next();
                if (url.length()>0){

                    futures.add(
                        taskCompletionService.submit(
                        new GoGo(url)
                        )
                        );
                        }

                }



            threadPool.shutdown();
            try {
            threadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
            }
            catch (InterruptedException e ) {
            }

            result_thread.stop();

        }
    }


 class ResultHarvester implements Runnable {

     private int size;
     private CompletionService<String> all_service;

     public ResultHarvester (int size, CompletionService<String> service){
        this.size = size;
        this.all_service = service;
     }

     public void run(){
            int future_size = 1;
            CompletionService<String> this_service = this.all_service; 
                while(true){
                    Future result = this_service.poll();
                    try {
                            if(result!=null && result.isDone()){
                                String output = result.get().toString();
                                if(output.length()>1){
                                    System.out.println(output);
                                }

                            }
                        }

                    catch (InterruptedException e) {
                        // Something went wrong with a task submitted
                        System.out.println("Error Interrupted exception");
                        e.printStackTrace();
                    } catch (ExecutionException e) {
                        // Something went wrong with the result
                        e.printStackTrace();
                        System.out.println("Error get() threw exception");
                    }


                    }
                }

        }

Answer 2

The problem is you are trying to do two things simultaneously (submit work, and read work results) in one thread.

That doesn't make sense - for simultaneous tasks, you need multiple threads.

So create another thread to read the results. Or another thread to submit the tasks. It doesn't matter which way you do it; either way, you end up with 2 threads instead of one.