<commit_before>package tv.floe.metronome.deeplearning.dbn.iterativereduce;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.List;

import org.apache.commons.lang3.time.StopWatch;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.math3.random.MersenneTwister;
import org.apache.commons.math3.random.RandomGenerator;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.ToolRunner;

import tv.floe.metronome.deeplearning.datasets.DataSet;
import tv.floe.metronome.deeplearning.datasets.iterator.impl.MnistHDFSDataSetIterator;
import tv.floe.metronome.deeplearning.dbn.DeepBeliefNetwork;
import tv.floe.metronome.deeplearning.dbn.model.evaluation.ModelTester;
import tv.floe.metronome.io.records.CachedVectorReader;

import com.cloudera.iterativereduce.ComputableWorker;
import com.cloudera.iterativereduce.io.RecordParser;
import com.cloudera.iterativereduce.io.TextRecordParser;
import com.cloudera.iterativereduce.yarn.appworker.ApplicationWorker;

/**
 * Base IterativeReduce worker node
 * 
 * @author josh
 *
 */
public class WorkerNode implements ComputableWorker<DBNParameterVectorUpdateable> {
	
	private static final Log LOG = LogFactory.getLog(WorkerNode.class);
	
	  private enum TrainingState {
		    PRE_TRAIN, FINE_TUNE, TRAINING_COMPLETE
		  };
	
	private TrainingState currentTrainingState = TrainingState.PRE_TRAIN; // always start in PRE_TRAIN mode
	
	protected Configuration conf = null;	  

	DeepBeliefNetwork dbn = null;
	
	private boolean preTrainPhaseComplete = false;
	private boolean fineTunePhaseComplete = false;
	
	TextRecordParser lineParser = new TextRecordParser();
	CachedVectorReader cachedVecReader = null; //new CachedVectorReader(lineParser, rec_factory); 

	private boolean epochComplete = false;
	
	private int completedDatasetEpochs = 0;
	private int currentIteration = 0;
	

	int[] hiddenLayerSizes = { 500, 250, 100 };
	double learningRate = 0.01;
	
	// epochs pertaining to a batch
	int preTrainEpochs = 100;
	int fineTuneEpochs = 100;
	
	// passes over dataset
	int preTrainDatasetPasses = 1;
	//int currentPreTrainDatasetPass = 0;
	
	int fineTuneDatasetPasses = 1;
	
	// not used --- or needs to be calc'd?
	int totalTrainingDatasetSize = 1;	
	
	int batchSize = 1;
	boolean showNetworkStats = true;
	
	int numIns = 784; 
	int numLabels = 10; 

	int n_layers = hiddenLayerSizes.length;
	
	
	RandomGenerator rng = new MersenneTwister(123);
	
	MnistHDFSDataSetIterator hdfs_fetcher = null; //new MnistHDFSDataSetIterator( batchSize, totalNumExamples, txt_reader );
	
	
	StopWatch watch = new StopWatch();
//	watch.start();

	
	/**
	 * generates the local DBN parameter vector
	 *  
	 * 
	 */
	public DBNParameterVector GenerateParameterVectorUpdate() {

		DBNParameterVector vector = new DBNParameterVector();
		//vector.parameter_vector = this.polr.getBeta().clone(); // this.polr.getGamma().getMatrix().clone();

		ByteArrayOutputStream out = new ByteArrayOutputStream();

		this.dbn.write(out);
		
		
		vector.preTrainPhaseComplete = this.preTrainPhaseComplete;
		vector.dbn_payload = out.toByteArray();
		
		System.out.println( "----- GenerateParameterVectorUpdate -----" );
		


/*
		vector.batchTimeMS = this.lastBatchTimeMS;

		vector.AvgError = (new Double(metrics.AvgError * 100))
				.floatValue();
		vector.TrainedRecords = (new Long(metrics.TotalRecordsProcessed))
				.intValue();
		
*/
		return vector;

	}	
	
	/**
	 * Need to think hard about how we define a "pass through the dataset"
	 * - old definition: "iteration"
	 * - new definition: "epoch"
	 * 
	 */
	@Override
	public boolean IncrementIteration() {
		
		this.currentIteration++;
		//this.currentEpoch++;

		return false;
	}
	
	/**
	 * Run a training pass of a single batch of input records on the DBN
	 * 
	 * TODO
	 * - dileneate between pre-train and finetune pass through data
	 * 		- how?
	 * 
	 * - app.iteration.count
	 * 		- indicates how many times we're going to call the workers
	 * 
	 * - tv.floe.metronome.dbn.conf.batchSize=10
	 * 		- indicates that we're going to only process 10 records in a call to a worker
	 * 
	 * - we could either
	 * 
	 * 		1. make a complete pass through the batches in a split between iterations
	 * 
	 * 			- tends to skew away from good solutions
	 * 
	 * 		2. parameter average between batches
	 *
	 *			-	better quality, but more network overhead
	 *
	 * - if we paramete avg between batches, then our passes over the dataset become
	 * 
	 * 		- total number of examples / batch size
	 * 
	 * - might be pragmatic to let a command line tool calculate iterations
	 * 
	 * 		- given we need to know how many fine tune passes to make as well
	 * 
	 * 
	 * 
	 * 
	 * 
	 * 
	 * 
	 * 
	 */
	@Override
	public DBNParameterVectorUpdateable compute() {
		
		// TODO: setup a cached vector system from hdfs for batches
						
	//	System.out.println("Worker > Compute()");

		
		int recordsProcessed = 0;
		
		StopWatch batchWatch = new StopWatch();
		
		DataSet hdfs_recordBatch = null; //this.hdfs_fetcher.next();
		
		System.out.println("Iteration: " + this.currentIteration );
		
//		if (hdfs_recordBatch.getFirst().numRows() > 0) {
//		do  {
		
		if ( TrainingState.PRE_TRAIN == this.currentTrainingState ) {
		
 			if ( this.hdfs_fetcher.hasNext() ) {
				
				hdfs_recordBatch = this.hdfs_fetcher.next();

				// check for the straggler batch condition
				if (0 == this.currentIteration && hdfs_recordBatch.getFirst().numRows() > 0 && hdfs_recordBatch.getFirst().numRows() < this.batchSize) {
					
					// ok, only in this situation do we lower the batch size
					this.batchSize = hdfs_recordBatch.getFirst().numRows();

					// re-setup the dataset iterator
					try {
						this.hdfs_fetcher = new MnistHDFSDataSetIterator( this.batchSize, this.totalTrainingDatasetSize, (TextRecordParser)lineParser );
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}

					System.out.println( "Worker > PreTrain: Setting up for a straggler split... (sub batch size)" );					
					System.out.println( "New batch size: " + this.batchSize );
				}
				
				if (hdfs_recordBatch.getFirst().numRows() > 0) {
					
					if (hdfs_recordBatch.getFirst().numRows() < this.batchSize) {
						
						System.out.println( "Worker > PreTrain: [Jagged End of Split: Skipped] Processed Total " + recordsProcessed + " Total Time " + watch.toString() );
						
						
					} else {
						
						// calc stats on number records processed
						recordsProcessed += hdfs_recordBatch.getFirst().numRows();
						
						//System.out.println( "PreTrain: Batch Size: " + hdfs_recordBatch.getFirst().numRows() );
						
						batchWatch.reset();
						
						batchWatch.start();
				
						this.dbn.preTrain( hdfs_recordBatch.getFirst(), 1, this.learningRate, this.preTrainEpochs);
						
						batchWatch.stop();
		
						System.out.println( "Worker > PreTrain: Batch Mode, Processed Total " + recordsProcessed + ", Batch Time " + batchWatch.toString() + " Total Time " + watch.toString() );
		
					} // if
					
					
				} else {
				
					// in case we get a blank line
					System.out.println( "Worker > PreTrain > Idle pass, no records left to process in phase" );
					
				}
				
			}
			
		//	System.out.println( "Worker > Check PreTrain completion > completedEpochs: " + this.completedDatasetEpochs + ", preTrainDatasetPasses: " + this.preTrainDatasetPasses );
			
			// check for completion of split, to signal master on state change
			if (false == this.hdfs_fetcher.hasNext() && this.completedDatasetEpochs + 1 >= this.preTrainDatasetPasses ) {
				
				this.preTrainPhaseComplete = true;
			//	System.out.println( "Worker > Completion of pre-train phase" );
				
			}
			
					
		<commit_msg>quick commit to send debug out when worker is sitting idle in edge case during iterations on IR DBN<commit_after>package tv.floe.metronome.deeplearning.dbn.iterativereduce;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.List;

import org.apache.commons.lang3.time.StopWatch;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.math3.random.MersenneTwister;
import org.apache.commons.math3.random.RandomGenerator;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.ToolRunner;

import tv.floe.metronome.deeplearning.datasets.DataSet;
import tv.floe.metronome.deeplearning.datasets.iterator.impl.MnistHDFSDataSetIterator;
import tv.floe.metronome.deeplearning.dbn.DeepBeliefNetwork;
import tv.floe.metronome.deeplearning.dbn.model.evaluation.ModelTester;
import tv.floe.metronome.io.records.CachedVectorReader;

import com.cloudera.iterativereduce.ComputableWorker;
import com.cloudera.iterativereduce.io.RecordParser;
import com.cloudera.iterativereduce.io.TextRecordParser;
import com.cloudera.iterativereduce.yarn.appworker.ApplicationWorker;

/**
 * Base IterativeReduce worker node
 * 
 * @author josh
 *
 */
public class WorkerNode implements ComputableWorker<DBNParameterVectorUpdateable> {
	
	private static final Log LOG = LogFactory.getLog(WorkerNode.class);
	
	  private enum TrainingState {
		    PRE_TRAIN, FINE_TUNE, TRAINING_COMPLETE
		  };
	
	private TrainingState currentTrainingState = TrainingState.PRE_TRAIN; // always start in PRE_TRAIN mode
	
	protected Configuration conf = null;	  

	DeepBeliefNetwork dbn = null;
	
	private boolean preTrainPhaseComplete = false;
	private boolean fineTunePhaseComplete = false;
	
	TextRecordParser lineParser = new TextRecordParser();
	CachedVectorReader cachedVecReader = null; //new CachedVectorReader(lineParser, rec_factory); 

	private boolean epochComplete = false;
	
	private int completedDatasetEpochs = 0;
	private int currentIteration = 0;
	

	int[] hiddenLayerSizes = { 500, 250, 100 };
	double learningRate = 0.01;
	
	// epochs pertaining to a batch
	int preTrainEpochs = 100;
	int fineTuneEpochs = 100;
	
	// passes over dataset
	int preTrainDatasetPasses = 1;
	//int currentPreTrainDatasetPass = 0;
	
	int fineTuneDatasetPasses = 1;
	
	// not used --- or needs to be calc'd?
	int totalTrainingDatasetSize = 1;	
	
	int batchSize = 1;
	boolean showNetworkStats = true;
	
	int numIns = 784; 
	int numLabels = 10; 

	int n_layers = hiddenLayerSizes.length;
	
	
	RandomGenerator rng = new MersenneTwister(123);
	
	MnistHDFSDataSetIterator hdfs_fetcher = null; //new MnistHDFSDataSetIterator( batchSize, totalNumExamples, txt_reader );
	
	
	StopWatch watch = new StopWatch();
//	watch.start();

	
	/**
	 * generates the local DBN parameter vector
	 *  
	 * 
	 */
	public DBNParameterVector GenerateParameterVectorUpdate() {

		DBNParameterVector vector = new DBNParameterVector();
		//vector.parameter_vector = this.polr.getBeta().clone(); // this.polr.getGamma().getMatrix().clone();

		ByteArrayOutputStream out = new ByteArrayOutputStream();

		this.dbn.write(out);
		
		
		vector.preTrainPhaseComplete = this.preTrainPhaseComplete;
		vector.dbn_payload = out.toByteArray();
		
		System.out.println( "----- GenerateParameterVectorUpdate -----" );
		


/*
		vector.batchTimeMS = this.lastBatchTimeMS;

		vector.AvgError = (new Double(metrics.AvgError * 100))
				.floatValue();
		vector.TrainedRecords = (new Long(metrics.TotalRecordsProcessed))
				.intValue();
		
*/
		return vector;

	}	
	
	/**
	 * Need to think hard about how we define a "pass through the dataset"
	 * - old definition: "iteration"
	 * - new definition: "epoch"
	 * 
	 */
	@Override
	public boolean IncrementIteration() {
		
		this.currentIteration++;
		//this.currentEpoch++;

		return false;
	}
	
	/**
	 * Run a training pass of a single batch of input records on the DBN
	 * 
	 * TODO
	 * - dileneate between pre-train and finetune pass through data
	 * 		- how?
	 * 
	 * - app.iteration.count
	 * 		- indicates how many times we're going to call the workers
	 * 
	 * - tv.floe.metronome.dbn.conf.batchSize=10
	 * 		- indicates that we're going to only process 10 records in a call to a worker
	 * 
	 * - we could either
	 * 
	 * 		1. make a complete pass through the batches in a split between iterations
	 * 
	 * 			- tends to skew away from good solutions
	 * 
	 * 		2. parameter average between batches
	 *
	 *			-	better quality, but more network overhead
	 *
	 * - if we paramete avg between batches, then our passes over the dataset become
	 * 
	 * 		- total number of examples / batch size
	 * 
	 * - might be pragmatic to let a command line tool calculate iterations
	 * 
	 * 		- given we need to know how many fine tune passes to make as well
	 * 
	 * 
	 * 
	 * 
	 * 
	 * 
	 * 
	 * 
	 */
	@Override
	public DBNParameterVectorUpdateable compute() {
		
		// TODO: setup a cached vector system from hdfs for batches
						
	//	System.out.println("Worker > Compute()");

		
		int recordsProcessed = 0;
		
		StopWatch batchWatch = new StopWatch();
		
		DataSet hdfs_recordBatch = null; //this.hdfs_fetcher.next();
		
		System.out.println("Iteration: " + this.currentIteration );
		
//		if (hdfs_recordBatch.getFirst().numRows() > 0) {
//		do  {
		
		if ( TrainingState.PRE_TRAIN == this.currentTrainingState ) {
		
 			if ( this.hdfs_fetcher.hasNext() ) {
				
				hdfs_recordBatch = this.hdfs_fetcher.next();

				// check for the straggler batch condition
				if (0 == this.currentIteration && hdfs_recordBatch.getFirst().numRows() > 0 && hdfs_recordBatch.getFirst().numRows() < this.batchSize) {
					
					// ok, only in this situation do we lower the batch size
					this.batchSize = hdfs_recordBatch.getFirst().numRows();

					// re-setup the dataset iterator
					try {
						this.hdfs_fetcher = new MnistHDFSDataSetIterator( this.batchSize, this.totalTrainingDatasetSize, (TextRecordParser)lineParser );
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}

					System.out.println( "Worker > PreTrain: Setting up for a straggler split... (sub batch size)" );					
					System.out.println( "New batch size: " + this.batchSize );
				}
				
				if (hdfs_recordBatch.getFirst().numRows() > 0) {
					
					if (hdfs_recordBatch.getFirst().numRows() < this.batchSize) {
						
						System.out.println( "Worker > PreTrain: [Jagged End of Split: Skipped] Processed Total " + recordsProcessed + " Total Time " + watch.toString() );
						
						
					} else {
						
						// calc stats on number records processed
						recordsProcessed += hdfs_recordBatch.getFirst().numRows();
						
						//System.out.println( "PreTrain: Batch Size: " + hdfs_recordBatch.getFirst().numRows() );
						
						batchWatch.reset();
						
						batchWatch.start();
				
						this.dbn.preTrain( hdfs_recordBatch.getFirst(), 1, this.learningRate, this.preTrainEpochs);
						
						batchWatch.stop();
		
						System.out.println( "Worker > PreTrain: Batch Mode, Processed Total " + recordsProcessed + ", Batch Time " + batchWatch.toString() + " Total Time " + watch.toString() );
		
					} // if
					
					
				} else {
				
					// in case we get a blank line
					System.out.println( "Worker > PreTrain > Idle pass, no records left to process in phase" );
					
				}
				
			} else {
				
				System.out.println( "Worker > PreTrain > Idle pass, no records left to process in phase" );
				
			}
			
		//	System.out.println( "Worker > Check PreTrain completion > completedEpochs: " + this.completedDatasetEpochs + ", preTrainDatasetPasses: " + this.preTrainDatasetPasses );
			
			// check for completion of split, to signal master on state change
			if (false == this.hdfs_fetcher.hasNext() && this.completedDatasetEpochs + 1 >= this.preTrainDatasetPasses ) {
				
				this.preTrainPhaseComplete = true;
			//	System.out.println( "Worker > Completion of pre-train phase" );
				
			}
			
					
		
{'fraction_non_alphanumeric': 0.10415617128463477, 'fraction_numerical': 0.00654911838790932, 'mean_word_length': 3.2394554191137215, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 33, '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': '5047180', 'n_tokens_mistral': 5901, 'n_tokens_neox': 4752, 'n_words': 1217}