<commit_before>package hadoop;

import java.io.IOException;
import java.net.InetAddress;
import java.net.URI;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Scanner;

import lightLogger.Logger;

import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import stopwatch.StopWatch;
import utils.Points;
import cl_util.CLInstance;

public class KMMapperReducerCL {

	public static class KMMapper extends
			Mapper<NullWritable, PointWritable, PointWritable, PointWritable> {

		private static final Class<KMMapper> CLAZZ = KMMapper.class;

		private List<PointWritable> centroids;
		private CLPointFloat clPoint;
	//	private List<PointWritable> pointBuffer;
	//	private int pointCounter;

		private StopWatch swPhase = new StopWatch(
				KMeansHadoop.Timer.MAPPHASE.prefix,
				KMeansHadoop.Timer.MAPPHASE.suffix);
		private StopWatch swMethod = new StopWatch(
				KMeansHadoop.Timer.MAPMETHOD.prefix,
				KMeansHadoop.Timer.MAPMETHOD.suffix);

		@Override
		protected void setup(KMMapper.Context context) {
			swPhase.start();
			swMethod.start();
			swMethod.pause();

			Logger.logDebug(CLAZZ,
					"TaskAttemptID: " + context.getTaskAttemptID());
			try {
				Logger.logDebug(CLAZZ, "Hostname: "
						+ InetAddress.getLocalHost().getHostName());
			} catch (Exception e) {
				Logger.logDebug(CLAZZ, "Hostname: unknown");
			}

			// TODO read max k from conf to use ArrayList
			this.centroids = new LinkedList<PointWritable>();

			Scanner sc = null;
			try {
				URI[] uris = DistributedCache.getCacheFiles(context
						.getConfiguration());
				FileSystem fs = FileSystem.get(context.getConfiguration());
				for (FileStatus fst : fs
						.listStatus(new Path(uris[0].toString()))) {
					if (!fst.isDir()) {
						Logger.logDebug(CLAZZ, "centroids: " + fst.getPath());
						sc = new Scanner(fs.open(fst.getPath()));
						while (sc.hasNext())
							this.centroids.add(new PointWritable(Points
									.createPoint(sc.next())));
						sc.close();
					}
				}
			} catch (IOException e) {
				Logger.logError(CLAZZ, "Could not get local cache files");
				e.printStackTrace();
			} finally {
				if (sc != null)
					sc.close();
				// don't close FileSystem!
			}

			this.clPoint = new CLPointFloat(new CLInstance(
					CLInstance.TYPES.CL_GPU), this.centroids.get(0).getDim(), context);
			List<PointWritable> tmpCentroids = new ArrayList<PointWritable>(
					this.centroids.size());
			tmpCentroids.addAll(this.centroids);
			Logger.logDebug(CLAZZ, "Centroid size: " + tmpCentroids.size());
			this.clPoint.prepareNearestPoints(tmpCentroids);
			//this.pointBuffer = new LinkedList<PointWritable>();
			this.clPoint.reset();
		//	this.pointCounter = 0;
		}

		@Override
		protected void map(NullWritable key, PointWritable value,
				KMMapper.Context context) throws IOException,
				InterruptedException {
			swMethod.resume();

			//ICPoint<Float> cp = new CPoint(value);

//			if (pointCounter < pointBuffer.size()) {
//				this.pointBuffer.add(value);
//				this.clPoint.put(value);
//				this.pointCounter++;
//			} else {
//				this.clPoint.setNearestPoints();
//				write(context);
//
//				this.pointBuffer.add(value);
//				this.clPoint.put(value);
//				this.pointCounter++;
//			}
			
			//this.pointBuffer.add(value);
			this.clPoint.put(value);
			// this.clPoint.setNearestPoints();
			//
			// PointWritable centroid = new PointWritable(cp.getCentroid());
			//
			// context.write(centroid, value);

			swMethod.pause();
		}

//		private void write(KMMapper.Context context) throws IOException,
//				InterruptedException {
//			for (PointWritable cp : pointBuffer) {
//				context.write((PointWritable)cp.getCentroid(), cp);
//			}
//			//this.pointBuffer.clear();
//			//this.pointCounter = 0;
//		}

		@Override
		protected void cleanup(KMMapper.Context context) {
			try {
				this.clPoint.setNearestPoints();
				//write(context);
			} catch (Exception e) {
				Logger.logError(CLAZZ, "cleanup: " + e.getMessage());
			}

			swMethod.stop();
			Logger.log(KMeansHadoop.TIME_LEVEL, CLAZZ, swMethod.getTimeString());

			swPhase.stop();
			Logger.log(KMeansHadoop.TIME_LEVEL, CLAZZ, swPhase.getTimeString());
		}

	}

	// public static class KMReducer extends
	// Reducer<PointWritable, PointWritable, PointWritable, PointWritable> {
	//
	// private static final Class<KMReducer> CLAZZ = KMReducer.class;
	//
	// private StopWatch swPhase = new StopWatch(
	// KMeansHadoop.Timer.REDUCEPHASE.prefix,
	// KMeansHadoop.Timer.REDUCEPHASE.suffix);
	// private StopWatch swMethod = new StopWatch(
	// KMeansHadoop.Timer.REDUCEMETHOD.prefix,
	// KMeansHadoop.Timer.REDUCEMETHOD.suffix);
	//
	// private ICLSummarizer<Float>[] clFloat;
	// private CLInstance clInstance;
	//
	// @Override
	// protected void setup(KMReducer.Context context) {
	// swPhase.start();<commit_msg>New KMeans-Hadoop-OCL branching #2 ...<commit_after>package hadoop;

import java.io.IOException;
import java.net.InetAddress;
import java.net.URI;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Scanner;

import lightLogger.Logger;

import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import stopwatch.StopWatch;
import utils.Points;
import cl_util.CLInstance;
import cl_util.CLPointFloat;
import clustering.CPoint;
import clustering.ICPoint;
import clustering.IPoint;

public class KMMapperReducerCL {

	public static class KMMapper extends
			Mapper<NullWritable, PointWritable, PointWritable, PointWritable> {

		private static final Class<KMMapper> CLAZZ = KMMapper.class;

		private List<PointWritable> centroids;
		private CLPointFloat clPoint;

		private StopWatch swPhase = new StopWatch(
				KMeansHadoop.Timer.MAPPHASE.prefix,
				KMeansHadoop.Timer.MAPPHASE.suffix);
		private StopWatch swMethod = new StopWatch(
				KMeansHadoop.Timer.MAPMETHOD.prefix,
				KMeansHadoop.Timer.MAPMETHOD.suffix);

		@Override
		protected void setup(KMMapper.Context context) {
			swPhase.start();
			swMethod.start();
			swMethod.pause();

			Logger.logDebug(CLAZZ,
					"TaskAttemptID: " + context.getTaskAttemptID());
			try {
				Logger.logDebug(CLAZZ, "Hostname: "
						+ InetAddress.getLocalHost().getHostName());
			} catch (Exception e) {
				Logger.logDebug(CLAZZ, "Hostname: unknown");
			}

			// TODO read max k from conf to use ArrayList
			this.centroids = new LinkedList<PointWritable>();

			Scanner sc = null;
			try {
				URI[] uris = DistributedCache.getCacheFiles(context
						.getConfiguration());
				FileSystem fs = FileSystem.get(context.getConfiguration());
				for (FileStatus fst : fs
						.listStatus(new Path(uris[0].toString()))) {
					if (!fst.isDir()) {
						Logger.logDebug(CLAZZ, "centroids: " + fst.getPath());
						sc = new Scanner(fs.open(fst.getPath()));
						while (sc.hasNext())
							this.centroids.add(new PointWritable(Points
									.createPoint(sc.next())));
						sc.close();
					}
				}
			} catch (IOException e) {
				Logger.logError(CLAZZ, "Could not get local cache files");
				e.printStackTrace();
			} finally {
				if (sc != null)
					sc.close();
				// don't close FileSystem!
			}

			this.clPoint = new CLPointFloat(new CLInstance(
					CLInstance.TYPES.CL_GPU), this.centroids.get(0).getDim());
			List<IPoint<Float>> tmpCentroids = new ArrayList<IPoint<Float>>(
					this.centroids.size());
			tmpCentroids.addAll(this.centroids);
			Logger.logDebug(CLAZZ, "Centroid size: " + tmpCentroids.size());
			this.clPoint.prepareNearestPoints(tmpCentroids);
			this.clPoint.reset(1);
		}

		@Override
		protected void map(NullWritable key, PointWritable value,
				KMMapper.Context context) throws IOException,
				InterruptedException {
			swMethod.resume();

			ICPoint<Float> cp = new CPoint(value);

			this.clPoint.put(cp);
			this.clPoint.setNearestPoints();

			PointWritable centroid = new PointWritable(cp.getCentroid());

			context.write(centroid, value);

			swMethod.pause();
		}

		@Override
		protected void cleanup(KMMapper.Context context) {
			swMethod.stop();
			Logger.log(KMeansHadoop.TIME_LEVEL, CLAZZ, swMethod.getTimeString());

			swPhase.stop();
			Logger.log(KMeansHadoop.TIME_LEVEL, CLAZZ, swPhase.getTimeString());
		}

	}

	// public static class KMReducer extends
	// Reducer<PointWritable, PointWritable, PointWritable, PointWritable> {
	//
	// private static final Class<KMReducer> CLAZZ = KMReducer.class;
	//
	// private StopWatch swPhase = new StopWatch(
	// KMeansHadoop.Timer.REDUCEPHASE.prefix,
	// KMeansHadoop.Timer.REDUCEPHASE.suffix);
	// private StopWatch swMethod = new StopWatch(
	// KMeansHadoop.Timer.REDUCEMETHOD.prefix,
	// KMeansHadoop.Timer.REDUCEMETHOD.suffix);
	//
	// private ICLSummarizer<Float>[] clFloat;
	// private CLInstance clInstance;
	//
	// @Override
	// protected void setup(KMReducer.Context context) {
	// swPhase.start();
{'fraction_non_alphanumeric': 0.13191170167166738, 'fraction_numerical': 0.0008572653236176596, 'mean_word_length': 3.7837006663249615, 'pattern_counts': {'":': 0, '<': 28, '<?xml version=': 0, '>': 27, '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': '7545772', 'n_tokens_mistral': 3646, 'n_tokens_neox': 3088, 'n_words': 575}