<commit_before>package com.splicemachine.derby.impl.job.scheduler;

import com.google.common.base.Function;
import com.splicemachine.derby.stats.TaskStats;
import com.splicemachine.job.*;
import com.splicemachine.si.api.HTransactorFactory;
import com.splicemachine.si.api.TransactionStatus;
import com.splicemachine.si.api.TransactorControl;
import com.splicemachine.si.impl.TransactionId;
import com.splicemachine.tools.BalancedBlockingQueue;
import com.splicemachine.utils.SpliceLogUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.NotServingRegionException;
import org.apache.log4j.Logger;
import javax.annotation.Nullable;
import java.io.IOException;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author <NAME>
 * Created on: 4/9/13
 */
public class SimpleThreadedTaskScheduler<T extends Task> implements TaskScheduler<T>,TaskSchedulerManagement {
    private static final Logger WORKER_LOG = Logger.getLogger(TaskCallable.class);
    private static final int DEFAULT_MAX_WORKERS = 10;
    private static final int DEFAULT_PRIORITY_LEVELS = 5;
    private static final int DEFAULT_INTERLEAVE_COUNT = 10;

    private static final Function<Runnable,Integer> priorityMapper = new Function<Runnable, Integer>() {
        @Override
        public Integer apply(@Nullable Runnable input) {
            if(input instanceof PriorityRunnable){
                return ((PriorityRunnable)input).getPriority();
            }
            return 1;
        }
    };

    private final ThreadPoolExecutor executor;

    private StatsListener statsListener = new StatsListener();

    private SimpleThreadedTaskScheduler(ThreadPoolExecutor executor) {
        this.executor = executor;
    }

    public static <T extends Task> SimpleThreadedTaskScheduler<T> create(Configuration configuration){
        int maxWorkers = configuration.getInt("splice.task.maxWorkers",DEFAULT_MAX_WORKERS);
        int numPriorityLevels = configuration.getInt("splice.task.priorityLevels", DEFAULT_PRIORITY_LEVELS);
        int tasksPerLevel = configuration.getInt("splice.task.priorityInterleave", DEFAULT_INTERLEAVE_COUNT);

        ThreadPoolExecutor executor = new TaskThreadPool(maxWorkers,maxWorkers,60,TimeUnit.SECONDS,
                new BalancedBlockingQueue<Runnable>(numPriorityLevels,tasksPerLevel,priorityMapper),
                new NamedThreadFactory());
        executor.allowCoreThreadTimeOut(true);

        return new SimpleThreadedTaskScheduler<T>(executor);
    }

    @Override
    public TaskFuture submit(T task) throws ExecutionException {
        ListeningFuture future = new ListeningFuture(task,statsListener.numPending.get());
        task.getTaskStatus().attachListener(statsListener);
        future.future = executor.submit(new TaskCallable<T>(task, statsListener));
        return future;
    }

/*********************************************************************************************************************/
    /*Statistics gathering*/
    @Override public int getNumPendingTasks() { return statsListener.numPending.get(); }
    @Override public int getCurrentWorkers() { return executor.getPoolSize(); }
    @Override public int getMaxWorkers() { return executor.getCorePoolSize(); }
    @Override public void setMaxWorkers(int newMaxWorkers) { executor.setCorePoolSize(newMaxWorkers); }
    @Override public long getTotalSubmittedTasks() { return executor.getTaskCount(); }
    @Override public long getTotalCompletedTasks() { return statsListener.completedCount.get(); }
    @Override public long getTotalFailedTasks() { return statsListener.failedCount.get(); }
    @Override public long getTotalCancelledTasks() { return statsListener.cancelledCount.get(); }
    @Override public long getTotalInvalidatedTasks() { return statsListener.invalidatedCount.get(); }
    @Override public int getNumRunningTasks() { return statsListener.numExecuting.get(); }

    @Override
    public int getHighestWorkerLoad() {
        if(executor.getPoolSize()>0)
            return executor.getQueue().size()/executor.getPoolSize();
        return 0; //no threads means the executor is idle, so we can't possibly have any tasks waiting
    }

    @Override
    public int getLowestWorkerLoad() {
        return getHighestWorkerLoad(); //in this model, all workers have the same load
    }

    private static class TaskCallable<T extends Task> implements Callable<Void> {
        private final T task;
        private final StatsListener listener;

        public TaskCallable(T task, StatsListener listener) {
            this.task = task;
            this.listener = listener;
        }

        @Override
        public Void call() throws Exception {
            try{
                switch (task.getTaskStatus().getStatus()) {
                    case INVALID:
                        SpliceLogUtils.trace(WORKER_LOG, "Task %s has been invalidated, cleaning up and skipping", task.getTaskId());
                        cleanUpTask(task);
                        return null;
                    case FAILED:
                        SpliceLogUtils.trace(WORKER_LOG,"Task %s has failed, but was not removed from the queue, removing now and skipping",task.getTaskId());
                        cleanUpTask(task);
                        return null;
                    case COMPLETED:
                        SpliceLogUtils.trace(WORKER_LOG, "Task %s has completed, but was not removed from the queue, removing now and skipping", task.getTaskId());
                        cleanUpTask(task);
                        return null;
                    case CANCELLED:
                        SpliceLogUtils.trace(WORKER_LOG,"task %s has been cancelled, not executing",task.getTaskId());
                        cleanUpTask(task);
                        return null;
                }
            }catch(ExecutionException ee){
                SpliceLogUtils.error(WORKER_LOG,
                        "task "+ task.getTaskId()+" had an unexpected error while checking status, unable to execute",ee.getCause());
                return null;
            }
<commit_msg>added last chance logging in task<commit_after>package com.splicemachine.derby.impl.job.scheduler;

import com.google.common.base.Function;
import com.splicemachine.derby.stats.TaskStats;
import com.splicemachine.job.*;
import com.splicemachine.si.api.HTransactorFactory;
import com.splicemachine.si.api.TransactionStatus;
import com.splicemachine.si.api.TransactorControl;
import com.splicemachine.si.impl.TransactionId;
import com.splicemachine.tools.BalancedBlockingQueue;
import com.splicemachine.utils.SpliceLogUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.NotServingRegionException;
import org.apache.log4j.Logger;
import javax.annotation.Nullable;
import java.io.IOException;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author <NAME>
 * Created on: 4/9/13
 */
public class SimpleThreadedTaskScheduler<T extends Task> implements TaskScheduler<T>,TaskSchedulerManagement {
    private static final Logger WORKER_LOG = Logger.getLogger(TaskCallable.class);
    private static final int DEFAULT_MAX_WORKERS = 10;
    private static final int DEFAULT_PRIORITY_LEVELS = 5;
    private static final int DEFAULT_INTERLEAVE_COUNT = 10;

    private static final Function<Runnable,Integer> priorityMapper = new Function<Runnable, Integer>() {
        @Override
        public Integer apply(@Nullable Runnable input) {
            if(input instanceof PriorityRunnable){
                return ((PriorityRunnable)input).getPriority();
            }
            return 1;
        }
    };

    private final ThreadPoolExecutor executor;

    private StatsListener statsListener = new StatsListener();

    private SimpleThreadedTaskScheduler(ThreadPoolExecutor executor) {
        this.executor = executor;
    }

    public static <T extends Task> SimpleThreadedTaskScheduler<T> create(Configuration configuration){
        int maxWorkers = configuration.getInt("splice.task.maxWorkers",DEFAULT_MAX_WORKERS);
        int numPriorityLevels = configuration.getInt("splice.task.priorityLevels", DEFAULT_PRIORITY_LEVELS);
        int tasksPerLevel = configuration.getInt("splice.task.priorityInterleave", DEFAULT_INTERLEAVE_COUNT);

        ThreadPoolExecutor executor = new TaskThreadPool(maxWorkers,maxWorkers,60,TimeUnit.SECONDS,
                new BalancedBlockingQueue<Runnable>(numPriorityLevels,tasksPerLevel,priorityMapper),
                new NamedThreadFactory());
        executor.allowCoreThreadTimeOut(true);

        return new SimpleThreadedTaskScheduler<T>(executor);
    }

    @Override
    public TaskFuture submit(T task) throws ExecutionException {
        ListeningFuture future = new ListeningFuture(task,statsListener.numPending.get());
        task.getTaskStatus().attachListener(statsListener);
        future.future = executor.submit(new TaskCallable<T>(task, statsListener));
        return future;
    }

/*********************************************************************************************************************/
    /*Statistics gathering*/
    @Override public int getNumPendingTasks() { return statsListener.numPending.get(); }
    @Override public int getCurrentWorkers() { return executor.getPoolSize(); }
    @Override public int getMaxWorkers() { return executor.getCorePoolSize(); }
    @Override public void setMaxWorkers(int newMaxWorkers) { executor.setCorePoolSize(newMaxWorkers); }
    @Override public long getTotalSubmittedTasks() { return executor.getTaskCount(); }
    @Override public long getTotalCompletedTasks() { return statsListener.completedCount.get(); }
    @Override public long getTotalFailedTasks() { return statsListener.failedCount.get(); }
    @Override public long getTotalCancelledTasks() { return statsListener.cancelledCount.get(); }
    @Override public long getTotalInvalidatedTasks() { return statsListener.invalidatedCount.get(); }
    @Override public int getNumRunningTasks() { return statsListener.numExecuting.get(); }

    @Override
    public int getHighestWorkerLoad() {
        if(executor.getPoolSize()>0)
            return executor.getQueue().size()/executor.getPoolSize();
        return 0; //no threads means the executor is idle, so we can't possibly have any tasks waiting
    }

    @Override
    public int getLowestWorkerLoad() {
        return getHighestWorkerLoad(); //in this model, all workers have the same load
    }

    private static class TaskCallable<T extends Task> implements Callable<Void> {
        private final T task;
        private final StatsListener listener;

        public TaskCallable(T task, StatsListener listener) {
            this.task = task;
            this.listener = listener;
        }

        @Override
        public Void call() throws Exception {
            try {
                return callDirect();
            } catch (Exception e) {
                WORKER_LOG.error("uncaught exception", e);
                throw e;
            } catch (Throwable t) {
                WORKER_LOG.error("uncaught exception", t);
                throw new RuntimeException(t);
            }
        }

        private Void callDirect() throws ExecutionException {
            try{
                switch (task.getTaskStatus().getStatus()) {
                    case INVALID:
                        SpliceLogUtils.trace(WORKER_LOG, "Task %s has been invalidated, cleaning up and skipping", task.getTaskId());
                        cleanUpTask(task);
                        return null;
                    case FAILED:
                        SpliceLogUtils.trace(WORKER_LOG,"Task %s has failed, but was not removed from the queue, removing now and skipping",task.getTaskId());
                        cleanUpTask(task);
                        return null;
                    case COMPLETED:
                        SpliceLogUtils.trace(WORKER_LOG, "Task %s has completed, but was not removed from the queue, removing now and skipping", task.getTaskId());
                        cleanUpTask(task);
                        return null;
                    case CANCELLED:
                        SpliceLogUtils.trace(WORKER_LOG,"task %s has been cancelled, not executing",task.getTaskId());
                        cleanUpTask(task);
                        return null;
                }
            }catch(ExecutionException ee){
                SpliceLogUtils.error(WORKER_LOG,
                        "task "+ task.getTaskId()+" had an unexpected error while checking status, unable to execute",ee.getCause());
                return null;
            }

{'fraction_non_alphanumeric': 0.10396812997969067, 'fraction_numerical': 0.0023433838462740197, 'mean_word_length': 2.9662329615861216, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6726251', 'n_tokens_mistral': 3362, 'n_tokens_neox': 3200, 'n_words': 786}