<commit_before>/**
 * Reducer.java
 *
 * Performs a reduce2all(max, double).
 *
 * Uses PoolInfo.IPCluster() to determine which cluster any host is in.
 * For each cluster, build a standard spanning tree.
 * Between clusters, do the reduce as in MagPIe.
 *
 * @author <NAME>
 */

import java.io.IOException;

import ibis.ipl.Ibis;
import ibis.ipl.PortType;
import ibis.ipl.SendPort;
import ibis.ipl.ReceivePort;
import ibis.ipl.SendPortIdentifier;
import ibis.ipl.ReceivePortIdentifier;
import ibis.ipl.ReadMessage;
import ibis.ipl.WriteMessage;
import ibis.ipl.Registry;
import ibis.ipl.StaticProperties;
import ibis.ipl.IbisException;

import ibis.util.PoolInfo;

public class ClusterReducer extends TreeReducer {

    private ReceivePort[]	reduceRinter;
    private SendPort		reduceSinter;

    private int 		myCluster;
    private int			clusterSize;


    public ClusterReducer(Ibis ibis, PoolInfo info)
	    throws IOException, IbisException {

	int rank = info.rank();
	int size = info.size();

	int[] cluster = info.clusterIPRank();
	clusterSize = info.clusterIPSize();
	myCluster = cluster[rank];

	/*
	 * Number of hosts in myCluster
	 */
	int localSize = 0;
	for (int i = 0; i < size; i++) {
	    if (cluster[i] == myCluster) {
		localSize++;
	    }
	}

	/*
	 * Intra-cluster, we also have a ranking of hosts to set up the local
	 * spanning tree.
	 * These two arrays translate between global ranking and local ranking.
	 */
	int[] globalRank = new int[localSize];
	int[] localRank = new int[size];
	int r = 0;
	for (int i = 0; i < size; i++) {
	    if (cluster[i] == myCluster) {
		globalRank[r] = i;
		localRank[i]  = r;
		r++;
	    }
	}

	/*
	 * The global rank of the root of each cluster
	 */
	int[] clusterRoot = new int[clusterSize];
	for (int i = 0; i < clusterSize; i++) {
	    clusterRoot[i] = -1;
	}
	for (int i = 0; i < size; i++) {
	    if (clusterRoot[cluster[i]] == -1) {
		clusterRoot[cluster[i]] = i;
	    }
	}

	// Sanity check
	if (localRank[rank] == 0 && clusterRoot[myCluster] != rank) {
	    System.err.println("Oh ho oh no cluster root != cluster first");
	    System.exit(33);
	}
	if (localRank[rank] == 0) {
	    System.err.println(rank + ": I am root for cluster " + myCluster);
	}

	StaticProperties reqprops = new StaticProperties();
	reqprops.add("serialization", "data");
	// reqprops.add("communication", "OneToOne, Reliable, ExplicitReceipt");
	reqprops.add("communication",
		"OneToOne, Reliable, ExplicitReceipt");

	PortType portTypeReduce = ibis.createPortType("SOR Reduce", reqprops);


	reqprops = new StaticProperties();
	reqprops.add("serialization", "data");
	reqprops.add("communication",
		"OneToMany, OneToOne, Reliable, ExplicitReceipt");

	PortType portTypeBroadcast = ibis.createPortType("SOR Broadcast", reqprops);


	if (localRank[rank] == 0) {
	    parent = LEAF_NODE;
	} else {
	    parent = (localRank[rank] - 1) / 2;
	    parent = globalRank[parent];
	}

	int children = 0;
	for (int c = 0; c < 2; c++) {
	    child[c] = 2 * localRank[rank] + c + 1;
	    if (child[c] >= localSize) {
		child[c] = LEAF_NODE;
	    } else {
		children++;
		child[c] = globalRank[child[c]];
	    }
	}

	Registry registry = ibis.registry();

	/* Create and connect ports for the intra-cluster reduce phase */
	if (children > 0) {
	    reduceRreduce = new ReceivePort[2];
	    for (int c = 0; c < 2; c++) {
		if (child[c] != LEAF_NODE) {
		    reduceRreduce[c] = portTypeReduce.createReceivePort("SOR" + rank + "_" + c + "_reduceR");
		    reduceRreduce[c].enableConnections();
		}
	    }
	}

	if (parent != LEAF_NODE) {
	    int childrank = localRank[rank] - 2 * localRank[parent] - 1;
	    reduceSreduce = portTypeReduce.createSendPort("SOR" + rank + "reduceS");
	    ReceivePortIdentifier id;
	    id = registry.lookup("SOR" + parent + "_" + childrank + "_reduceR");
	    reduceSreduce.connect(id);
	}

	/* Create and connect ports for the intra-cluster bcast phase */
	if (parent != LEAF_NODE) {
	    reduceRbcast = portTypeBroadcast.createReceivePort("SOR" + rank + "reduceR");
	    reduceRbcast.enableConnections();
	}

	if (children > 0) {
	    reduceSbcast = portTypeBroadcast.createSendPort("SOR" + rank + "reduceS");
	    for (int c = 0; c < 2; c++) {
		if (child[c] != LEAF_NODE) {
		    ReceivePortIdentifier id;
		    id = registry.lookup("SOR" + child[c] + "reduceR");
		    reduceSbcast.connect(id);
		}
	    }
	}


	if (rank == clusterRoot[myCluster]) {
	    /* Create and connect ports for the inter-cluster reduce phase */

	    reqprops = new StaticProperties();
	    reqprops.add("serialization", "data");
	    reqprops.add("communication",
		    "OneToMany, OneToOne, Reliable, ExplicitReceipt");

	    PortType portTypeInter = ibis.createPortType("SOR InterCluster", reqprops);

	    reduceRinter = new ReceivePort[clusterSize];
	    for (int i = 0; i < clusterSize; i++) {
		if (i != myCluster) {
		    reduceRinter[i] = portTypeInter.createReceivePort("SOR" + rank + "_" + i + "_interR");
		    reduceRinter[i].enableConnections();
		}
	    }

	    reduceSinter = portTypeInter.createSendPort("SOR" + rank + "interS");
	    for (int i = 0; i < clusterSize; i++) {
		if (i != myCluster) {
		    ReceivePortIdentifier id;
		    id = registry.lookup("SOR" + clusterRoot[i] + "_" + i + "interR");
		    reduceSinter.connect(id);
		}
	    }
	}

    }


    public double reduce(double value) throws IOException {

	for (int c = 0; c < 2; c++) {
	    if (child[c] != LEAF_NODE) {
		ReadMessage rm = reduceRreduce[c].receive();
		value = Math.max(value, rm.readDouble());
		rm.finish();
	    }
	}

	if (parent != LEAF_NODE) {
	    WriteMessage wm = reduceSreduce.newMessage();
	    wm.writeDouble(value);<commit_msg>Patch ClusterReducer<commit_after>/**
 * Reducer.java
 *
 * Performs a reduce2all(max, double).
 *
 * Uses PoolInfo.IPCluster() to determine which cluster any host is in.
 * For each cluster, build a standard spanning tree.
 * Between clusters, do the reduce as in MagPIe.
 *
 * @author <NAME>
 */

import java.io.IOException;

import ibis.ipl.Ibis;
import ibis.ipl.PortType;
import ibis.ipl.SendPort;
import ibis.ipl.ReceivePort;
import ibis.ipl.SendPortIdentifier;
import ibis.ipl.ReceivePortIdentifier;
import ibis.ipl.ReadMessage;
import ibis.ipl.WriteMessage;
import ibis.ipl.Registry;
import ibis.ipl.StaticProperties;
import ibis.ipl.IbisException;

import ibis.util.PoolInfo;

public class ClusterReducer extends TreeReducer {

    private ReceivePort[]	reduceRinter;
    private SendPort		reduceSinter;

    private int 		myCluster;
    private int			clusterSize;


    public ClusterReducer(Ibis ibis, PoolInfo info)
	    throws IOException, IbisException {

	int rank = info.rank();
	int size = info.size();

	int[] cluster = info.clusterIPRank();
	clusterSize = info.clusterIPSize();
	myCluster = cluster[rank];

	/*
	 * Number of hosts in myCluster
	 */
	int localSize = 0;
	for (int i = 0; i < size; i++) {
	    if (cluster[i] == myCluster) {
		localSize++;
	    }
	}

	/*
	 * Intra-cluster, we also have a ranking of hosts to set up the local
	 * spanning tree.
	 * These two arrays translate between global ranking and local ranking.
	 */
	int[] globalRank = new int[localSize];
	int[] localRank = new int[size];
	int r = 0;
	for (int i = 0; i < size; i++) {
	    if (cluster[i] == myCluster) {
		globalRank[r] = i;
		localRank[i]  = r;
		r++;
	    }
	}

	/*
	 * The global rank of the root of each cluster
	 */
	int[] clusterRoot = new int[clusterSize];
	for (int i = 0; i < clusterSize; i++) {
	    clusterRoot[i] = -1;
	}
	for (int i = 0; i < size; i++) {
	    if (clusterRoot[cluster[i]] == -1) {
		clusterRoot[cluster[i]] = i;
	    }
	}

	// Sanity check
	if (localRank[rank] == 0 && clusterRoot[myCluster] != rank) {
	    System.err.println("Oh ho oh no cluster root != cluster first");
	    System.exit(33);
	}
// if (localRank[rank] == 0)
// System.err.println(rank + ": I am root for cluster " + myCluster);
// System.err.print(rank + ": my cluster = {");
// for (int i = 0; i < clusterSize; i++) System.err.print(globalRank[i] + " ");
// System.err.println("}");

	StaticProperties reqprops = new StaticProperties();
	reqprops.add("serialization", "data");
	// reqprops.add("communication", "OneToOne, Reliable, ExplicitReceipt");
	reqprops.add("communication",
		"OneToOne, Reliable, ExplicitReceipt");

	PortType portTypeReduce = ibis.createPortType("SOR Reduce", reqprops);


	reqprops = new StaticProperties();
	reqprops.add("serialization", "data");
	reqprops.add("communication",
		"OneToMany, OneToOne, Reliable, ExplicitReceipt");

	PortType portTypeBroadcast = ibis.createPortType("SOR Broadcast", reqprops);


	if (localRank[rank] == 0) {
	    parent = LEAF_NODE;
	} else {
	    parent = (localRank[rank] - 1) / 2;
	    parent = globalRank[parent];
	}

	int children = 0;
	for (int c = 0; c < 2; c++) {
	    child[c] = 2 * localRank[rank] + c + 1;
	    if (child[c] >= localSize) {
		child[c] = LEAF_NODE;
	    } else {
		children++;
		child[c] = globalRank[child[c]];
	    }
	}

	Registry registry = ibis.registry();

	/* Create and connect ports for the intra-cluster reduce phase */
	if (children > 0) {
	    reduceRreduce = new ReceivePort[2];
	    for (int c = 0; c < 2; c++) {
		if (child[c] != LEAF_NODE) {
		    reduceRreduce[c] = portTypeReduce.createReceivePort("SOR" + rank + "_" + c + "_reduceR");
		    reduceRreduce[c].enableConnections();
		}
	    }
	}

	if (parent != LEAF_NODE) {
	    int childrank = localRank[rank] - 2 * localRank[parent] - 1;
	    reduceSreduce = portTypeReduce.createSendPort("SOR" + rank + "reduceS");
	    ReceivePortIdentifier id;
	    id = registry.lookup("SOR" + parent + "_" + childrank + "_reduceR");
	    reduceSreduce.connect(id);
	}

	/* Create and connect ports for the intra-cluster bcast phase */
	if (parent != LEAF_NODE) {
	    reduceRbcast = portTypeBroadcast.createReceivePort("SOR" + rank + "reduceR");
	    reduceRbcast.enableConnections();
	}

	if (children > 0) {
	    reduceSbcast = portTypeBroadcast.createSendPort("SOR" + rank + "reduceS");
	    for (int c = 0; c < 2; c++) {
		if (child[c] != LEAF_NODE) {
		    ReceivePortIdentifier id;
		    id = registry.lookup("SOR" + child[c] + "reduceR");
		    reduceSbcast.connect(id);
		}
	    }
	}
// System.err.println(rank + ": local connection OK");


	if (rank == clusterRoot[myCluster]) {
	    /* Create and connect ports for the inter-cluster reduce phase */

	    reqprops = new StaticProperties();
	    reqprops.add("serialization", "data");
	    reqprops.add("communication",
		    "OneToMany, OneToOne, Reliable, ExplicitReceipt");

	    PortType portTypeInter = ibis.createPortType("SOR InterCluster", reqprops);

	    reduceRinter = new ReceivePort[clusterSize];
	    for (int i = 0; i < clusterSize; i++) {
		if (i != myCluster) {
		    reduceRinter[i] = portTypeInter.createReceivePort("SOR" + rank + "_" + i + "_interR");
		    reduceRinter[i].enableConnections();
		}
	    }

	    reduceSinter = portTypeInter.createSendPort("SOR" + rank + "interS");
	    for (int i = 0; i < clusterSize; i++) {
		if (i != myCluster) {
		    ReceivePortIdentifier id;
		    id = registry.lookup("SOR" + clusterRoot[i] + "_" + myCluster + "_interR");
		    reduceSinter.connect(id);
		}
	    }
// System.err.println(rank + ": interlocal connection OK");
	}

    }


    public double reduce(double value) throws IOException {

	for (int c = 0; c < 2; c++) {
	    if (child[c] != LEAF_NODE) {
		ReadMessage rm = reduceRreduce[c].receive();
		value = Math.max(value, rm.readDouble());
		rm.finish();
	    }
	}

	if (parent != LEAF_NODE) {
	    WriteMessage wm = reduceSreduce.newMessage();
	    wm.writeDouble(value);
{'fraction_non_alphanumeric': 0.14640646492434664, 'fraction_numerical': 0.00593191196698762, 'mean_word_length': 3.36182977127859, 'pattern_counts': {'":': 5, '<': 26, '<?xml version=': 0, '>': 11, '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': '3426255', 'n_tokens_mistral': 4306, 'n_tokens_neox': 3883, 'n_words': 1004}