<commit_before>package org.deeplearning4j.models.glove;

import akka.actor.ActorSystem;
import com.google.common.collect.Lists;
import org.apache.commons.io.IOUtils;
import org.apache.commons.io.LineIterator;
import org.apache.commons.math3.random.MersenneTwister;
import org.apache.commons.math3.random.RandomGenerator;
import org.deeplearning4j.bagofwords.vectorizer.TextVectorizer;
import org.deeplearning4j.bagofwords.vectorizer.TfidfVectorizer;
import org.deeplearning4j.berkeley.Counter;
import org.deeplearning4j.berkeley.Pair;
import org.deeplearning4j.models.embeddings.inmemory.InMemoryLookupTable;
import org.deeplearning4j.models.word2vec.VocabWord;
import org.deeplearning4j.models.word2vec.Word2Vec;
import org.deeplearning4j.models.word2vec.wordstore.VocabCache;
import org.deeplearning4j.models.word2vec.wordstore.inmemory.InMemoryLookupCache;
import org.deeplearning4j.parallel.Parallelization;
import org.deeplearning4j.text.movingwindow.Util;
import org.deeplearning4j.text.sentenceiterator.SentenceIterator;
import org.deeplearning4j.text.stopwords.StopWords;
import org.deeplearning4j.text.tokenization.tokenizerfactory.DefaultTokenizerFactory;
import org.deeplearning4j.text.tokenization.tokenizerfactory.TokenizerFactory;
import org.deeplearning4j.util.MathUtils;
import org.deeplearning4j.util.SetUtils;
import org.nd4j.linalg.api.ndarray.INDArray;
import org.nd4j.linalg.factory.Nd4j;
import org.nd4j.linalg.ops.transforms.Transforms;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.io.Serializable;
import java.util.*;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Glove by socher et. al
 *
 * @author <NAME>
 */
public class Glove implements Serializable {

    private VocabCache cache;
    private transient SentenceIterator sentenceIterator;
    private transient TextVectorizer textVectorizer;
    private transient TokenizerFactory tokenizerFactory;
    private GloveWeightLookupTable lookupTable;
    private int layerSize = 100;
    private double learningRate = 0.05;
    private double xMax = 0.75;
    private int windowSize = 15;
    private CoOccurrences coOccurrences;
    private List<String> stopWords = StopWords.getStopWords();
    private boolean stem = false;
    protected Queue<Pair<Integer,List<Pair<VocabWord,VocabWord>>>> jobQueue = new LinkedBlockingDeque<>();
    private int batchSize = 1000;
    private int minWordFrequency = 5;
    private double maxCount = 100;
    public final static String UNK = Word2Vec.UNK;
    private int iterations = 5;
    private static Logger log = LoggerFactory.getLogger(Glove.class);
    private boolean symmetric = true;
    private transient RandomGenerator gen;
    private boolean shuffle = true;
    private transient Random shuffleRandom;
    private int numWorkers = Runtime.getRuntime().availableProcessors();

    private Glove(){}

    public Glove(VocabCache cache, SentenceIterator sentenceIterator, TextVectorizer textVectorizer, TokenizerFactory tokenizerFactory, GloveWeightLookupTable lookupTable, int layerSize, double learningRate, double xMax, int windowSize, CoOccurrences coOccurrences, List<String> stopWords, boolean stem,int batchSize,int minWordFrequency,double maxCount,int iterations,boolean symmetric,RandomGenerator gen,boolean shuffle,long seed,int numWorkers) {
        this.numWorkers = numWorkers;
        this.gen = gen;
        this.cache = cache;
        this.shuffle = shuffle;
        this.sentenceIterator = sentenceIterator;
        this.textVectorizer = textVectorizer;
        this.tokenizerFactory = tokenizerFactory;
        this.lookupTable = lookupTable;
        this.layerSize = layerSize;
        this.learningRate = learningRate;
        this.xMax = xMax;
        this.windowSize = windowSize;
        this.coOccurrences = coOccurrences;
        this.stopWords = stopWords;
        this.stem = stem;
        this.batchSize = batchSize;
        this.minWordFrequency = minWordFrequency;
        this.maxCount = maxCount;
        this.iterations = iterations;
        this.symmetric = symmetric;
        shuffleRandom = new Random(seed);
    }

    public void fit() {
        boolean cacheFresh = false;

        if(cache == null) {
            cacheFresh  = true;
            cache = new InMemoryLookupCache();
        }

        if(textVectorizer == null && cacheFresh) {
            textVectorizer = new TfidfVectorizer.Builder().tokenize(tokenizerFactory)
                    .cache(cache).iterate(sentenceIterator).minWords(minWordFrequency)
                    .stopWords(stopWords).stem(stem).build();

            textVectorizer.fit();
        }

        if(sentenceIterator != null)
            sentenceIterator.reset();

        if(coOccurrences == null) {
            coOccurrences = new CoOccurrences.Builder()
                    .cache(cache).iterate(sentenceIterator).symmetric(symmetric)
                    .tokenizer(tokenizerFactory).windowSize(windowSize)
                    .build();

            coOccurrences.fit();

        }

        if(lookupTable == null)
            lookupTable = new GloveWeightLookupTable.Builder().xMax(xMax).maxCount(maxCount)
                    .cache(cache).lr(learningRate).vectorLength(layerSize).gen(gen)
                    .build();


        if(lookupTable.getSyn0() == null)
            lookupTable.resetWeights();
        final List<Pair<String,String>> pairList = coOccurrences.coOccurrenceList();
        if(shuffle)
            Collections.shuffle(pairList,shuffleRandom);



        final AtomicInteger countUp = new AtomicInteger(0);
        final Counter<Integer> erroriPerIteration = Util.parallelCounter();
        log.info("Processing # of co occurrences " + coOccurrences.numCoOccurrences());
        for(int i = 0; i < iterations; i++) {
            final AtomicInteger processed = new AtomicInteger(coOccurrences.numCoOccurrences());
            doIteration(i, pairList, erroriPerIteration, processed, countUp);
            log.info("Processed " + countUp.doubleValue() + " out of " + (pairList.size() * iterations) + " error was " + erroriPerIteration.getCount(i));

        }


    }


    public void doIteration(final int i,List<Pair<String,String>> pairList, final Counter<Integer> errorPerIteration,final AtomicInteger processed,final AtomicInteger countUp) {
        log.info("Iteration " + i);
        if(shuffle)
            Collections.shuffle(pairList,shuffleRandom);
        List<List<Pair<String,String>>> miniBatches = Lists.partition(pairList,batchSize);
        ActorSystem actor = ActorSystem.create();
        Parallelization.iterateInParallel(miniBatches,new Parallelization.RunnableWithParams<List<Pair<String, String>>>() {
            @Override
            public void run(List<Pair<String, String>> currentItem, Object[] args) {
                List<Pair<VocabWord,VocabWord>> send = new ArrayList<>();
                for (Pair<String, String> next : currentItem) {
                    String w1 = next.getFirst();
                    String w2 = next.getSecond();
                    VocabWord vocabWord = cache.wordFor(w1);
                    VocabWord vocabWord1 = cache.wordFor(w2);
                    send.add(new Pair<>(vocabWord, vocabWord1));

                }

                jobQueue.add(new Pair<>(i, send));
            }
        },actor);



        Parallelization.runInParallel(numWorkers,new Runnable() {
            @Override
            public void run() {
                while(processed.get() > 0 || !jobQueue.isEmpty()) {
                    Pair<Integer,List<Pair<VocabWord,VocabWord>>> work = jobQueue.poll();
                    if(work == null)
                        continue;
                    List<Pair<VocabWord,VocabWord>> batch = work.getSecond();

                    for(Pair<VocabWord,VocabWord> pair : batch) {
                        VocabWord w1 = pair.getFirst();
                        VocabWord w2 = pair.getSecond();
                        double weight = getCount(w1.getWord(),w2.getWord());
                        if(weight <= 0) {<commit_msg>iterate in parallel for glove<commit_after>package org.deeplearning4j.models.glove;

import akka.actor.ActorSystem;
import com.google.common.collect.Lists;
import org.apache.commons.io.IOUtils;
import org.apache.commons.io.LineIterator;
import org.apache.commons.math3.random.MersenneTwister;
import org.apache.commons.math3.random.RandomGenerator;
import org.deeplearning4j.bagofwords.vectorizer.TextVectorizer;
import org.deeplearning4j.bagofwords.vectorizer.TfidfVectorizer;
import org.deeplearning4j.berkeley.Counter;
import org.deeplearning4j.berkeley.Pair;
import org.deeplearning4j.models.embeddings.inmemory.InMemoryLookupTable;
import org.deeplearning4j.models.word2vec.VocabWord;
import org.deeplearning4j.models.word2vec.Word2Vec;
import org.deeplearning4j.models.word2vec.wordstore.VocabCache;
import org.deeplearning4j.models.word2vec.wordstore.inmemory.InMemoryLookupCache;
import org.deeplearning4j.parallel.Parallelization;
import org.deeplearning4j.text.movingwindow.Util;
import org.deeplearning4j.text.sentenceiterator.SentenceIterator;
import org.deeplearning4j.text.stopwords.StopWords;
import org.deeplearning4j.text.tokenization.tokenizerfactory.DefaultTokenizerFactory;
import org.deeplearning4j.text.tokenization.tokenizerfactory.TokenizerFactory;
import org.deeplearning4j.util.MathUtils;
import org.deeplearning4j.util.SetUtils;
import org.nd4j.linalg.api.ndarray.INDArray;
import org.nd4j.linalg.factory.Nd4j;
import org.nd4j.linalg.ops.transforms.Transforms;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.io.Serializable;
import java.util.*;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Glove by socher et. al
 *
 * @author <NAME>
 */
public class Glove implements Serializable {

    private VocabCache cache;
    private transient SentenceIterator sentenceIterator;
    private transient TextVectorizer textVectorizer;
    private transient TokenizerFactory tokenizerFactory;
    private GloveWeightLookupTable lookupTable;
    private int layerSize = 100;
    private double learningRate = 0.05;
    private double xMax = 0.75;
    private int windowSize = 15;
    private CoOccurrences coOccurrences;
    private List<String> stopWords = StopWords.getStopWords();
    private boolean stem = false;
    protected Queue<Pair<Integer,List<Pair<VocabWord,VocabWord>>>> jobQueue = new LinkedBlockingDeque<>();
    private int batchSize = 1000;
    private int minWordFrequency = 5;
    private double maxCount = 100;
    public final static String UNK = Word2Vec.UNK;
    private int iterations = 5;
    private static Logger log = LoggerFactory.getLogger(Glove.class);
    private boolean symmetric = true;
    private transient RandomGenerator gen;
    private boolean shuffle = true;
    private transient Random shuffleRandom;
    private int numWorkers = Runtime.getRuntime().availableProcessors();

    private Glove(){}

    public Glove(VocabCache cache, SentenceIterator sentenceIterator, TextVectorizer textVectorizer, TokenizerFactory tokenizerFactory, GloveWeightLookupTable lookupTable, int layerSize, double learningRate, double xMax, int windowSize, CoOccurrences coOccurrences, List<String> stopWords, boolean stem,int batchSize,int minWordFrequency,double maxCount,int iterations,boolean symmetric,RandomGenerator gen,boolean shuffle,long seed,int numWorkers) {
        this.numWorkers = numWorkers;
        this.gen = gen;
        this.cache = cache;
        this.shuffle = shuffle;
        this.sentenceIterator = sentenceIterator;
        this.textVectorizer = textVectorizer;
        this.tokenizerFactory = tokenizerFactory;
        this.lookupTable = lookupTable;
        this.layerSize = layerSize;
        this.learningRate = learningRate;
        this.xMax = xMax;
        this.windowSize = windowSize;
        this.coOccurrences = coOccurrences;
        this.stopWords = stopWords;
        this.stem = stem;
        this.batchSize = batchSize;
        this.minWordFrequency = minWordFrequency;
        this.maxCount = maxCount;
        this.iterations = iterations;
        this.symmetric = symmetric;
        shuffleRandom = new Random(seed);
    }

    public void fit() {
        boolean cacheFresh = false;

        if(cache == null) {
            cacheFresh  = true;
            cache = new InMemoryLookupCache();
        }

        if(textVectorizer == null && cacheFresh) {
            textVectorizer = new TfidfVectorizer.Builder().tokenize(tokenizerFactory)
                    .cache(cache).iterate(sentenceIterator).minWords(minWordFrequency)
                    .stopWords(stopWords).stem(stem).build();

            textVectorizer.fit();
        }

        if(sentenceIterator != null)
            sentenceIterator.reset();

        if(coOccurrences == null) {
            coOccurrences = new CoOccurrences.Builder()
                    .cache(cache).iterate(sentenceIterator).symmetric(symmetric)
                    .tokenizer(tokenizerFactory).windowSize(windowSize)
                    .build();

            coOccurrences.fit();

        }

        if(lookupTable == null)
            lookupTable = new GloveWeightLookupTable.Builder().xMax(xMax).maxCount(maxCount)
                    .cache(cache).lr(learningRate).vectorLength(layerSize).gen(gen)
                    .build();


        if(lookupTable.getSyn0() == null)
            lookupTable.resetWeights();
        final List<Pair<String,String>> pairList = coOccurrences.coOccurrenceList();
        if(shuffle)
            Collections.shuffle(pairList,shuffleRandom);



        final AtomicInteger countUp = new AtomicInteger(0);
        final Counter<Integer> erroriPerIteration = Util.parallelCounter();
        log.info("Processing # of co occurrences " + coOccurrences.numCoOccurrences());
        for(int i = 0; i < iterations; i++) {
            final AtomicInteger processed = new AtomicInteger(coOccurrences.numCoOccurrences());
            doIteration(i, pairList, erroriPerIteration, processed, countUp);
            log.info("Processed " + countUp.doubleValue() + " out of " + (pairList.size() * iterations) + " error was " + erroriPerIteration.getCount(i));

        }


    }


    public void doIteration(final int i,List<Pair<String,String>> pairList, final Counter<Integer> errorPerIteration,final AtomicInteger processed,final AtomicInteger countUp) {
        log.info("Iteration " + i);
        if(shuffle)
            Collections.shuffle(pairList,shuffleRandom);
        List<List<Pair<String,String>>> miniBatches = Lists.partition(pairList,batchSize);
        ActorSystem actor = ActorSystem.create();
        Parallelization.iterateInParallel(miniBatches,new Parallelization.RunnableWithParams<List<Pair<String, String>>>() {
            @Override
            public void run(List<Pair<String, String>> currentItem, Object[] args) {
                List<Pair<VocabWord,VocabWord>> send = new ArrayList<>();
                for (Pair<String, String> next : currentItem) {
                    String w1 = next.getFirst();
                    String w2 = next.getSecond();
                    VocabWord vocabWord = cache.wordFor(w1);
                    VocabWord vocabWord1 = cache.wordFor(w2);
                    send.add(new Pair<>(vocabWord, vocabWord1));

                }

                jobQueue.add(new Pair<>(i, send));
            }
        },actor);



        actor.shutdown();

        Parallelization.runInParallel(numWorkers,new Runnable() {
            @Override
            public void run() {
                while(processed.get() > 0 || !jobQueue.isEmpty()) {
                    Pair<Integer,List<Pair<VocabWord,VocabWord>>> work = jobQueue.poll();
                    if(work == null)
                        continue;
                    List<Pair<VocabWord,VocabWord>> batch = work.getSecond();

                    for(Pair<VocabWord,VocabWord> pair : batch) {
                        VocabWord w1 = pair.getFirst();
                        VocabWord w2 = pair.getSecond();
                        double weight = getCount(w1.getWord(),w2.getWord());
                        if(weight <= 0) {
{'fraction_non_alphanumeric': 0.09706367132653684, 'fraction_numerical': 0.008180208778462853, 'mean_word_length': 2.995609756097561, 'pattern_counts': {'":': 0, '<': 75, '<?xml version=': 0, '>': 73, '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': '1924446', 'n_tokens_mistral': 4817, 'n_tokens_neox': 4499, 'n_words': 887}