<commit_before>package model.supervised.cart;

import com.google.common.util.concurrent.AtomicDouble;
import data.DataSet;
import gnu.trove.list.array.TIntArrayList;
import gnu.trove.set.hash.TIntHashSet;
import model.Predictable;
import model.Trainable;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import utils.sort.SortIntDoubleUtils;
//import org.neu.util.sort.SortIntDoubleUtils;

import java.util.Arrays;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.IntPredicate;
import java.util.stream.IntStream;

/**
 * Created by hanxuan on 9/17/15.
 */
public abstract class Tree implements Trainable, Predictable{

    private static final Logger log = LogManager.getLogger(Tree.class);

    public static int TREE_ID = 0;

    public static int MAX_DEPTH = 10;

    public static int MIN_INSTANCE_COUNT = 5;

    public static int MAX_THREADS = 4;

    public static int THREAD_WORK_LOAD = 1;

    protected int td;

    protected Tree left = null;

    protected Tree right = null;

    protected double meanResponse = 0;

    protected int depth = 0;

    protected int[] existIds = null;

    protected DataSet dataSet = null;

    private int featureId = Integer.MIN_VALUE;

    private double featureThreshold = Integer.MAX_VALUE;

    private ExecutorService service = null;

    private CountDownLatch countDownLatch = null;

    public Tree() {}

    public Tree(int depth, DataSet dataSet, int[] existIds) {

        this.depth = depth;
        this.dataSet = dataSet;
        this.existIds = existIds;
        td = TREE_ID ++;
    }


    protected abstract double gainByCriteria(double[] labels, int position, int[] sortedIds);

    protected abstract boolean lessThanImpurityGainThreshold(double gain);

    protected abstract void setTreeLabel();

    protected abstract void split(int[] leftGroup, int[] rightGroup);

    @Override
    public double predict(double[] feature) {

        if (left == null) {
            return meanResponse;
        }else if(feature[featureId] < featureThreshold) {
            return this.left.predict(feature);
        }else {
            return this.right.predict(feature);
        }
    }

    @Override
    public void train() {

        if (pure()) {
            log.debug("[STOP GROW] pure node {} ...", td);
            setTreeLabel();
            return;
        }

        if (stopGrow()) {
            log.debug("[STOP GROW] depth({}) >= MAX_DEPTH({}) || existIds.length({}) <= MIN_INSTANCE_COUNT({})",
                    depth, MAX_DEPTH, existIds.length, MIN_INSTANCE_COUNT);
            setTreeLabel();
            return;
        }

        final int featureLength = dataSet.getFeatureLength();
        final AtomicInteger bestFeatureId = new AtomicInteger(Integer.MIN_VALUE);
        final AtomicDouble bestThreshold = new AtomicDouble(Integer.MIN_VALUE);
        final AtomicDouble bestGain = new AtomicDouble(Integer.MIN_VALUE);

        service = Executors.newFixedThreadPool(MAX_THREADS);

        log.debug("featureLength {}, existIds {}", featureLength, existIds.length);

        int realTaskCount = (int) Math.ceil(featureLength / (double) THREAD_WORK_LOAD);
        countDownLatch = new CountDownLatch(realTaskCount);
        log.debug("Task Count: {}", countDownLatch.getCount());
        TIntArrayList taskPackage = new TIntArrayList();
        IntStream.range(0, featureLength).forEach(i -> {
            taskPackage.add(i);
            if (taskPackage.size() == THREAD_WORK_LOAD || i == featureLength - 1) {
                TIntArrayList taskPackage2 = new TIntArrayList(taskPackage);
                service.submit(()->
                {
                    try {

                        int currentBestFeatureId = -1;
                        double currentBestGain = Integer.MIN_VALUE;
                        double currentBestThreshold = Integer.MIN_VALUE;
                        for(int taskId : taskPackage2.toArray()) {
                            double[] report = doTask(taskId);
                            if (report[0] > currentBestGain) {
                                currentBestFeatureId = taskId;
                                currentBestGain = report[0];
                                currentBestThreshold = report[1];
                            }
                        }
                        if (currentBestGain > bestGain.get()) {
                            bestGain.getAndSet(currentBestGain);
                            bestFeatureId.getAndSet(currentBestFeatureId);
                            bestThreshold.getAndSet(currentBestThreshold);
                        }
                        log.debug("one package processed/{}", realTaskCount);
                    }catch (Throwable t) {
                        log.error(t.getMessage(), t);
                    }
                    countDownLatch.countDown();
                });
                taskPackage.clear();
            }
        });

        try {
            TimeUnit.MILLISECONDS.sleep(10);
            countDownLatch.await();
        } catch (InterruptedException e) {
            log.error(e.getMessage(), e);
        }
        service.shutdown();

        log.debug("All task finished, service shutdown ...");

        log.debug("Best FeatureId: {}, threshold: {}, gain: {}", bestFeatureId.get(), bestThreshold.get(), bestGain.get());

        if (lessThanImpurityGainThreshold(bestGain.get())) {

            log.debug("[STOP GROW] lessThanImpurityGainThreshold, best gain: {}", bestGain.get());
            setTreeLabel();
            return;
        }

        featureId = bestFeatureId.get();
        featureThreshold = bestThreshold.get();

        grow();
    }

    @Override
    public Predictable offer() {
        return this;
    }

    private void grow() {

        TIntHashSet left = new TIntHashSet();
        TIntHashSet right = new TIntHashSet();
        for (int i : existIds) {
            if (dataSet.getEntry(i, featureId) < featureThreshold) {
                left.add(i);
            }else {
                right.add(i);
            }
        }

        if (left.size() == 0 || right.size() == 0) {
            setTreeLabel();
            return;
        }

        split(left.toArray(), right.toArray());

        this.left.train();
        this.right.train();<commit_msg>add stop grow logic featureId = Integer.MIN<commit_after>package model.supervised.cart;

import com.google.common.util.concurrent.AtomicDouble;
import data.DataSet;
import gnu.trove.list.array.TIntArrayList;
import gnu.trove.set.hash.TIntHashSet;
import model.Predictable;
import model.Trainable;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import utils.sort.SortIntDoubleUtils;
//import org.neu.util.sort.SortIntDoubleUtils;

import java.util.Arrays;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.IntPredicate;
import java.util.stream.IntStream;

/**
 * Created by hanxuan on 9/17/15.
 */
public abstract class Tree implements Trainable, Predictable{

    private static final Logger log = LogManager.getLogger(Tree.class);

    public static int TREE_ID = 0;

    public static int MAX_DEPTH = 10;

    public static int MIN_INSTANCE_COUNT = 5;

    public static int MAX_THREADS = 4;

    public static int THREAD_WORK_LOAD = 1;

    protected int td;

    protected Tree left = null;

    protected Tree right = null;

    protected double meanResponse = 0;

    protected int depth = 0;

    protected int[] existIds = null;

    protected DataSet dataSet = null;

    private int featureId = Integer.MIN_VALUE;

    private double featureThreshold = Integer.MAX_VALUE;

    private ExecutorService service = null;

    private CountDownLatch countDownLatch = null;

    public Tree() {}

    public Tree(int depth, DataSet dataSet, int[] existIds) {

        this.depth = depth;
        this.dataSet = dataSet;
        this.existIds = existIds;
        td = TREE_ID ++;
    }


    protected abstract double gainByCriteria(double[] labels, int position, int[] sortedIds);

    protected abstract boolean lessThanImpurityGainThreshold(double gain);

    protected abstract void setTreeLabel();

    protected abstract void split(int[] leftGroup, int[] rightGroup);

    @Override
    public double predict(double[] feature) {

        if (left == null) {
            return meanResponse;
        }else if(feature[featureId] < featureThreshold) {
            return this.left.predict(feature);
        }else {
            return this.right.predict(feature);
        }
    }

    @Override
    public void train() {

        if (pure()) {
            log.debug("[STOP GROW] pure node {} ...", td);
            setTreeLabel();
            return;
        }

        if (stopGrow()) {
            log.debug("[STOP GROW] depth({}) >= MAX_DEPTH({}) || existIds.length({}) <= MIN_INSTANCE_COUNT({})",
                    depth, MAX_DEPTH, existIds.length, MIN_INSTANCE_COUNT);
            setTreeLabel();
            return;
        }

        final int featureLength = dataSet.getFeatureLength();
        final AtomicInteger bestFeatureId = new AtomicInteger(Integer.MIN_VALUE);
        final AtomicDouble bestThreshold = new AtomicDouble(Integer.MIN_VALUE);
        final AtomicDouble bestGain = new AtomicDouble(Integer.MIN_VALUE);

        service = Executors.newFixedThreadPool(MAX_THREADS);

        log.debug("featureLength {}, existIds {}", featureLength, existIds.length);

        int realTaskCount = (int) Math.ceil(featureLength / (double) THREAD_WORK_LOAD);
        countDownLatch = new CountDownLatch(realTaskCount);
        log.debug("Task Count: {}", countDownLatch.getCount());
        TIntArrayList taskPackage = new TIntArrayList();
        IntStream.range(0, featureLength).forEach(i -> {
            taskPackage.add(i);
            if (taskPackage.size() == THREAD_WORK_LOAD || i == featureLength - 1) {
                TIntArrayList taskPackage2 = new TIntArrayList(taskPackage);
                service.submit(()->
                {
                    try {

                        int currentBestFeatureId = -1;
                        double currentBestGain = Integer.MIN_VALUE;
                        double currentBestThreshold = Integer.MIN_VALUE;
                        for(int taskId : taskPackage2.toArray()) {
                            double[] report = doTask(taskId);
                            if (report[0] > currentBestGain) {
                                currentBestFeatureId = taskId;
                                currentBestGain = report[0];
                                currentBestThreshold = report[1];
                            }
                        }
                        if (currentBestGain > bestGain.get()) {
                            bestGain.getAndSet(currentBestGain);
                            bestFeatureId.getAndSet(currentBestFeatureId);
                            bestThreshold.getAndSet(currentBestThreshold);
                        }
                        log.debug("one package processed/{}", realTaskCount);
                    }catch (Throwable t) {
                        log.error(t.getMessage(), t);
                    }
                    countDownLatch.countDown();
                });
                taskPackage.clear();
            }
        });

        try {
            TimeUnit.MILLISECONDS.sleep(10);
            countDownLatch.await();
        } catch (InterruptedException e) {
            log.error(e.getMessage(), e);
        }
        service.shutdown();

        log.debug("All task finished, service shutdown ...");

        log.debug("Best FeatureId: {}, threshold: {}, gain: {}", bestFeatureId.get(), bestThreshold.get(), bestGain.get());

        if (lessThanImpurityGainThreshold(bestGain.get())) {

            log.debug("[STOP GROW] lessThanImpurityGainThreshold, best gain: {}", bestGain.get());
            setTreeLabel();
            return;
        }

        featureId = bestFeatureId.get();
        featureThreshold = bestThreshold.get();

        if(featureId == Integer.MIN_VALUE) {
            log.warn("featureId == Integer.MIN_VALUE ~ now stop growing ...");
            setTreeLabel();
            return;
        }

        grow();
    }

    @Override
    public Predictable offer() {
        return this;
    }

    private void grow() {

        TIntHashSet left = new TIntHashSet();
        TIntHashSet right = new TIntHashSet();
        for (int i : existIds) {
            if (dataSet.getEntry(i, featureId) < featureThreshold) {
                left.add(i);
            }else {
                right.add(i);
            }
        }

        if (left.size() == 0 || right.size() == 0) {
            setTreeLabel();
            return;
        }

        split(left.toArray(), right.toArray());

        this.left.train();
        this.right.train();
{'fraction_non_alphanumeric': 0.10566910913999229, 'fraction_numerical': 0.004165059776320864, 'mean_word_length': 1.962979890310786, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 13, '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': '652789', 'n_tokens_mistral': 3559, 'n_tokens_neox': 3380, 'n_words': 767}