<commit_before>package roart.thread;

import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import roart.util.Constants;
import roart.util.MyQueue;
import roart.util.MyQueues;
import roart.dir.Traverse;
import roart.dir.TraverseFile;
import roart.filesystem.FileSystemDao;
import roart.model.FileObject;
import roart.queue.Queues;
import roart.queue.TraverseQueueElement;

public class TraverseQueueRunner implements Runnable {
	
    static Logger log = LoggerFactory.getLogger(TraverseQueueRunner.class);

    private static final java.util.Queue<Object[]> execQueue = new ConcurrentLinkedQueue<Object[]>();
   
    public void run() {

        int nThreads = 50;
        ThreadPoolExecutor /*ExecutorService*/ executorService = (ThreadPoolExecutor) Executors.newFixedThreadPool(nThreads);

        String queueid = Constants.TRAVERSEQUEUE;
        MyQueue<TraverseQueueElement> queue = MyQueues.get(queueid);

    	while (true) {
    	    TraverseQueueElement trav = queue.poll();
    		if (trav == null) {
    		    try {
                    TimeUnit.SECONDS.sleep(1);
                    continue;
                } catch (InterruptedException e) {
                    log.error(Constants.EXCEPTION, e); 
                }    		    
    		}
    	    
    		String filename = trav.getFilename();
    		log.info("trav cli " + trav.getClientQueueElement());
            FileObject fo = FileSystemDao.get(filename);
    		
    		if (Traverse.isLocal(fo)) {
    		    try {
                    if (Queues.traverseQueueHeavyLoaded()) {
                                                    try {
                                                            TimeUnit.SECONDS.sleep(1);
                                                    } catch (InterruptedException e) {
                                                            // TODO Auto-generated catch block              
                                                            log.error(Constants.EXCEPTION, e);
                                                    }
                                                    continue;
                                            }
                   Callable<Object> callable = new Callable<Object>() {
                        public Object call() /* throws Exception*/ {
                                try {
                                    Queues.traverseQueue.add(trav);
                                        doHandleFo();
                                } catch (Exception e) {
                                    log.error(Constants.EXCEPTION, e);
                                } catch (Error e) {
                                    System.gc();
                                log.error("Error " + Thread.currentThread().getId());
                                    log.error(Constants.ERROR, e);
                                }
                                finally {
                                        //log.info("myend");            
                                }
                                return null; //myMethod();              
                        }
                };
                    Future<Object> task = executorService.submit(callable);

                    //TraverseFile.handleFo3(trav);
                } catch (Exception e) {
                    log.error(Constants.EXCEPTION, e);
                }
    		} else {
                    queue.offer(trav);
    		}
    		
      	}
    }
    
    public static String doHandleFo() {<commit_msg>testing simplifying use for thread pool<commit_after>package roart.thread;

import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import roart.util.Constants;
import roart.util.MyQueue;
import roart.util.MyQueues;
import roart.dir.Traverse;
import roart.dir.TraverseFile;
import roart.filesystem.FileSystemDao;
import roart.model.FileObject;
import roart.queue.Queues;
import roart.queue.TraverseQueueElement;

public class TraverseQueueRunner implements Runnable {
	
    static Logger log = LoggerFactory.getLogger(TraverseQueueRunner.class);

    private static final java.util.Queue<Object[]> execQueue = new ConcurrentLinkedQueue<Object[]>();
   
    public void run() {

        int nThreads = 10;
        ThreadPoolExecutor /*ExecutorService*/ executorService = (ThreadPoolExecutor) Executors.newFixedThreadPool(nThreads);

        String queueid = Constants.TRAVERSEQUEUE;
        MyQueue<TraverseQueueElement> queue = MyQueues.get(queueid);

    	while (true) {
    	    TraverseQueueElement trav = queue.poll();
    		if (trav == null) {
    		    try {
                    TimeUnit.SECONDS.sleep(1);
                    continue;
                } catch (InterruptedException e) {
                    log.error(Constants.EXCEPTION, e); 
                }    		    
    		}
    	    
    		String filename = trav.getFilename();
    		log.info("trav cli " + trav.getClientQueueElement());
            FileObject fo = FileSystemDao.get(filename);
    		
    		if (Traverse.isLocal(fo)) {
    		    try {
                    if (Queues.traverseQueueHeavyLoaded()) {
                                                    try {
                                                            TimeUnit.SECONDS.sleep(1);
                                                    } catch (InterruptedException e) {
                                                            // TODO Auto-generated catch block              
                                                            log.error(Constants.EXCEPTION, e);
                                                    }
                                                    continue;
                                            }
                   Callable<Object> callable = new Callable<Object>() {
                        public Object call() /* throws Exception*/ {
                                try {
                                    Queues.traverseQueue.add(trav);
                                        doHandleFo();
                                } catch (Exception e) {
                                    log.error(Constants.EXCEPTION, e);
                                } catch (Error e) {
                                    System.gc();
                                log.error("Error " + Thread.currentThread().getId());
                                    log.error(Constants.ERROR, e);
                                }
                                finally {
                                        //log.info("myend");            
                                }
                                return null; //myMethod();              
                        }
                };
                   Callable<Object> callablesimple = new Callable<Object>() {
                        public Object call() /* throws Exception*/ {
                                try {
				    TraverseFile.handleFo3(trav);
                                } catch (Exception e) {
                                    log.error(Constants.EXCEPTION, e);
                                } catch (Error e) {
                                    System.gc();
                                log.error("Error " + Thread.currentThread().getId());
                                    log.error(Constants.ERROR, e);
                                }
                                finally {
                                        //log.info("myend");            
                                }
                                return null; //myMethod();              
                        }
		   };
		   log.info("submitting " + executorService.getCompletedTaskCount() + " " + executorService.getPoolSize() + " " + executorService.getActiveCount());
                    Future<Object> task = executorService.submit(callablesimple);

                    //TraverseFile.handleFo3(trav);
                } catch (Exception e) {
                    log.error(Constants.EXCEPTION, e);
                }
    		} else {
                    queue.offer(trav);
    		}
    		
      	}
    }
    
    public static String doHandleFo() {
{'fraction_non_alphanumeric': 0.08472206036592472, 'fraction_numerical': 0.0017480480130520918, 'mean_word_length': 1.0789728682170543, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 54, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4327386', 'n_tokens_mistral': 2024, 'n_tokens_neox': 1908, 'n_words': 350}