<commit_before>package central;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Multimap;
import com.google.common.collect.Multiset;

import reducemap.MappingCallable;
import threadbased.DistanceCalculationCallable;
import util.RunStrategy;

public class KMeans {

	private static List<Cluster> clusters = new LinkedList<Cluster>();

	// central method that runs the algorithm in a specific pattern according to
	// the chosen strategy
	public static void run(List<Point> points, int k, int iterations, RunStrategy strategy)
			throws InterruptedException, ExecutionException {

		switch (strategy) {

		case FORKJOIN:
			// runWithForkJoin(points, k, iterations);
			break;

		case PARALLEL:
			runInParallel(points, k, iterations);
			break;

		case SEQUENTIAL:
			runSequentially(points, k, iterations);
			break;

		case STREAM:
			// add andrea ferretti github version?
			break;

		case REDUCEMAP:
			runReduceMap(points, k, iterations);
			break;

		}

	}

	private static void runReduceMap(List<Point> points, int k, int iterations)
			throws InterruptedException, ExecutionException {

		// try cached threadpool, as well
		ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
		List<MappingCallable> callables = new ArrayList<MappingCallable>();

		for (int i = 1; i <= iterations; i++) {

			System.out.println("Iteration " + i + "/" + iterations);

			// threads need fast read-only access, updated after every iteration
			List<Point> centroids = points.subList(0, k);

			// Eight Simple Rules:
			// Split point list into sublists (watch out for views and real
			// sublists)

			int n = Runtime.getRuntime().availableProcessors();
			// use guava to split list
			List<List<Point>> sublists = Lists.partition(points, points.size() / n);

			// TODO: add loop that adds key(sublist) and value(cluster
			// number)

			// list of futures to catch result
			List<Future<Multiset>> results;

			// global mean?

			// process sublists in multiple threads, join
			System.out.println("Processing " + n + " multimaps");
			for (List<Point> sublist : sublists) {
				// calculate distances to have initial cluster assignments

				// create hashmultiset
				Multimap<Integer, Point> multimap = ArrayListMultimap.create();

				for (Point p : sublist) {
					double savedDistance = 100;

					// ugly
					int clusterNumber = 0;

					for (Point centroid : centroids) {
						int index = 1;
						double distance = calculateDistance(p, centroid);
						// System.out.println("distance is: " + distance);
						if (distance <= savedDistance) {
							clusterNumber = index;
							savedDistance = distance;
						}
						index++;
					}
					// assign point to cluster
					multimap.put(clusterNumber, p);

					// test
					System.out.println(multimap.size());
				}
				callables.add(new MappingCallable(multimap));
			}
		}

		// get future result (process callables); makes sure all futures are
		// done
		List<Future<Multimap>> futures = executor.invokeAll(callables);

		for (Future<Multimap> future : futures) {
			System.out.println(future.get());
		}

		// recalculate centroids here as all elements across sublists must
		// be considered (no local means)
		// or: collect local means and join them (to global mean)

		// revisit this<commit_msg>fix cluster assignment in reducemap strategy<commit_after>package central;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Multimap;

import reducemap.MappingCallable;
import threadbased.DistanceCalculationCallable;
import util.RunStrategy;

public class KMeans {

	private static List<Cluster> clusters = new LinkedList<Cluster>();

	// central method that runs the algorithm in a specific pattern according to
	// the chosen strategy
	public static void run(List<Point> points, int k, int iterations, RunStrategy strategy)
			throws InterruptedException, ExecutionException {

		switch (strategy) {

		case FORKJOIN:
			// runWithForkJoin(points, k, iterations);
			break;

		case PARALLEL:
			runInParallel(points, k, iterations);
			break;

		case SEQUENTIAL:
			runSequentially(points, k, iterations);
			break;

		case STREAM:
			// add <NAME> github version?
			break;

		case REDUCEMAP:
			runReduceMap(points, k, iterations);
			break;

		}

	}

	private static void runReduceMap(List<Point> points, int k, int iterations)
			throws InterruptedException, ExecutionException {

		// try cached threadpool, as well
		ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
		List<MappingCallable> callables = new ArrayList<MappingCallable>();

		for (int i = 1; i <= iterations; i++) {

			System.out.println("Iteration " + i + "/" + iterations);

			// threads need fast read-only access, updated after every iteration
			List<Point> centroids = points.subList(0, k);

			// Eight Simple Rules:
			// Split point list into sublists (watch out for views and real
			// sublists)

			int n = Runtime.getRuntime().availableProcessors();
			// use guava to split list
			List<List<Point>> sublists = Lists.partition(points, points.size() / n);

			// TODO: add loop that adds key(sublist) and value(cluster
			// number)

			// global mean?

			// process sublists in multiple threads, join
			System.out.println("Processing " + n + " multimaps");
			for (List<Point> sublist : sublists) {
				// calculate distances to have initial cluster assignments

				// create hashmultiset
				Multimap<Integer, Point> multimap = ArrayListMultimap.create();

				for (Point p : sublist) {
					double savedDistance = 100;

					// ugly
					int clusterNumber = 0;
					int index = 1;

					for (Point centroid : centroids) {
						double distance = calculateDistance(p, centroid);
						System.out.println("distance is: " + distance);
						if (distance <= savedDistance) {
							clusterNumber = index;
							savedDistance = distance;
						}
						index++;
						// assign point to cluster
						multimap.put(clusterNumber, p);
					}

					// test
					System.out.println(multimap.size());
				}
				callables.add(new MappingCallable(multimap));
			}
		}

		// get future result (process callables); makes sure all futures are
		// done
		List<Future<Multimap>> futures = executor.invokeAll(callables);
		List<Multimap> multimaps = new ArrayList<Multimap>();

		for (Future<Multimap> future : futures) {
			multimaps.add(future.get());
		}

		// recalculate centroids here as all elements across sublists must
		// be considered (no local means)

		for (Multimap m : multimaps) {
			// get cluster elements by integer key and sum them up (->local
			// means)

			// empty clusters?
			System.out.println(m.keySet().size());
		}
		// calculate global means

		// revisit this
{'fraction_non_alphanumeric': 0.10586988207234663, 'fraction_numerical': 0.001855041738439115, 'mean_word_length': 3.5745454545454547, 'pattern_counts': {'":': 0, '<': 40, '<?xml version=': 0, '>': 37, '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': '5811213', 'n_tokens_mistral': 2633, 'n_tokens_neox': 2154, 'n_words': 637}