You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@systemds.apache.org by mb...@apache.org on 2025/01/25 09:51:49 UTC
(systemds) branch main updated: [SYSTEMDS-3790] Extraction of optimal FedPlans and conflict handling
This is an automated email from the ASF dual-hosted git repository.
mboehm7 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/systemds.git
The following commit(s) were added to refs/heads/main by this push:
new c5ab81c6cf [SYSTEMDS-3790] Extraction of optimal FedPlans and conflict handling
c5ab81c6cf is described below
commit c5ab81c6cf8d4288762b2355f060755433e7e720
Author: min-guk <ko...@gmail.com>
AuthorDate: Sat Jan 25 10:22:27 2025 +0100
[SYSTEMDS-3790] Extraction of optimal FedPlans and conflict handling
Closes #2175.
---
.../sysds/hops/fedplanner/FederatedMemoTable.java | 214 +++++++--------------
.../hops/fedplanner/FederatedMemoTablePrinter.java | 139 +++++++++++++
.../fedplanner/FederatedPlanCostEnumerator.java | 112 ++++++++++-
.../fedplanner/FederatedPlanCostEstimator.java | 130 ++++++++++++-
.../federated/FederatedPlanCostEnumeratorTest.java | 18 +-
.../FederatedPlanCostEnumeratorTest1.dml} | 0
.../FederatedPlanCostEnumeratorTest2.dml} | 5 +-
.../FederatedPlanCostEnumeratorTest3.dml} | 5 +-
8 files changed, 460 insertions(+), 163 deletions(-)
diff --git a/src/main/java/org/apache/sysds/hops/fedplanner/FederatedMemoTable.java b/src/main/java/org/apache/sysds/hops/fedplanner/FederatedMemoTable.java
index 16240f0281..b2b58871f6 100644
--- a/src/main/java/org/apache/sysds/hops/fedplanner/FederatedMemoTable.java
+++ b/src/main/java/org/apache/sysds/hops/fedplanner/FederatedMemoTable.java
@@ -20,7 +20,6 @@
package org.apache.sysds.hops.fedplanner;
import org.apache.sysds.hops.Hop;
-import org.apache.sysds.hops.OptimizerUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.sysds.runtime.instructions.fed.FEDInstruction.FederatedOutput;
@@ -29,8 +28,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.ArrayList;
import java.util.Map;
-import java.util.HashSet;
-import java.util.Set;
/**
* A Memoization Table for managing federated plans (FedPlan) based on combinations of Hops and fedOutTypes.
@@ -71,12 +68,11 @@ public class FederatedMemoTable {
* Retrieves the minimum cost child plan considering the parent's output type.
* The cost is calculated using getParentViewCost to account for potential type mismatches.
*
- * @param childHopID ?
- * @param childFedOutType ?
- * @return ?
+ * @param fedPlanPair ???
+ * @return min cost fed plan
*/
- public FedPlan getMinCostChildFedPlan(long childHopID, FederatedOutput childFedOutType) {
- FedPlanVariants fedPlanVariantList = hopMemoTable.get(new ImmutablePair<>(childHopID, childFedOutType));
+ public FedPlan getMinCostFedPlan(Pair<Long, FederatedOutput> fedPlanPair) {
+ FedPlanVariants fedPlanVariantList = hopMemoTable.get(fedPlanPair);
return fedPlanVariantList._fedPlanVariants.stream()
.min(Comparator.comparingDouble(FedPlan::getTotalCost))
.orElse(null);
@@ -86,6 +82,22 @@ public class FederatedMemoTable {
return hopMemoTable.get(new ImmutablePair<>(hopID, fedOutType));
}
+ public FedPlanVariants getFedPlanVariants(Pair<Long, FederatedOutput> fedPlanPair) {
+ return hopMemoTable.get(fedPlanPair);
+ }
+
+ public FedPlan getFedPlanAfterPrune(long hopID, FederatedOutput fedOutType) {
+ // Todo: Consider whether to verify if pruning has been performed
+ FedPlanVariants fedPlanVariantList = hopMemoTable.get(new ImmutablePair<>(hopID, fedOutType));
+ return fedPlanVariantList._fedPlanVariants.get(0);
+ }
+
+ public FedPlan getFedPlanAfterPrune(Pair<Long, FederatedOutput> fedPlanPair) {
+ // Todo: Consider whether to verify if pruning has been performed
+ FedPlanVariants fedPlanVariantList = hopMemoTable.get(fedPlanPair);
+ return fedPlanVariantList._fedPlanVariants.get(0);
+ }
+
/**
* Checks if the memo table contains an entry for a given Hop and fedOutType.
*
@@ -98,162 +110,77 @@ public class FederatedMemoTable {
}
/**
- * Prunes all entries in the memo table, retaining only the minimum-cost
- * FedPlan for each entry.
- */
- public void pruneMemoTable() {
- for (Map.Entry<Pair<Long, FederatedOutput>, FedPlanVariants> entry : hopMemoTable.entrySet()) {
- List<FedPlan> fedPlanList = entry.getValue().getFedPlanVariants();
- if (fedPlanList.size() > 1) {
- // Find the FedPlan with the minimum cost
- FedPlan minCostPlan = fedPlanList.stream()
- .min(Comparator.comparingDouble(FedPlan::getTotalCost))
- .orElse(null);
-
- // Retain only the minimum cost plan
- fedPlanList.clear();
- fedPlanList.add(minCostPlan);
- }
- }
- }
-
- /**
- * Recursively prints a tree representation of the DAG starting from the given root FedPlan.
- * Includes information about hopID, fedOutType, TotalCost, SelfCost, and NetCost for each node.
+ * Prunes the specified entry in the memo table, retaining only the minimum-cost
+ * FedPlan for the given Hop ID and federated output type.
*
- * @param rootFedPlan The starting point FedPlan to print
+ * @param hopID The ID of the Hop to prune
+ * @param federatedOutput The federated output type associated with the Hop
*/
- public void printFedPlanTree(FedPlan rootFedPlan) {
- Set<FedPlan> visited = new HashSet<>();
- printFedPlanTreeRecursive(rootFedPlan, visited, 0, true);
+ public void pruneFedPlan(long hopID, FederatedOutput federatedOutput) {
+ hopMemoTable.get(new ImmutablePair<>(hopID, federatedOutput)).prune();
}
/**
- * Helper method to recursively print the FedPlan tree.
- *
- * @param plan The current FedPlan to print
- * @param visited Set to keep track of visited FedPlans (prevents cycles)
- * @param depth The current depth level for indentation
- * @param isLast Whether this node is the last child of its parent
+ * Represents common properties and costs associated with a Hop.
+ * This class holds a reference to the Hop and tracks its execution and network transfer costs.
*/
- private void printFedPlanTreeRecursive(FedPlan plan, Set<FedPlan> visited, int depth, boolean isLast) {
- if (plan == null || visited.contains(plan)) {
- return;
- }
-
- visited.add(plan);
-
- Hop hop = plan.getHopRef();
- StringBuilder sb = new StringBuilder();
-
- // Add FedPlan information
- sb.append(String.format("(%d) ", plan.getHopRef().getHopID()))
- .append(plan.getHopRef().getOpString())
- .append(" [")
- .append(plan.getFedOutType())
- .append("]");
-
- StringBuilder childs = new StringBuilder();
- childs.append(" (");
- boolean childAdded = false;
- for( Hop input : hop.getInput()){
- childs.append(childAdded?",":"");
- childs.append(input.getHopID());
- childAdded = true;
- }
- childs.append(")");
- if( childAdded )
- sb.append(childs.toString());
-
-
- sb.append(String.format(" {Total: %.1f, Self: %.1f, Net: %.1f}",
- plan.getTotalCost(),
- plan.getSelfCost(),
- plan.getNetTransferCost()));
-
- // Add matrix characteristics
- sb.append(" [")
- .append(hop.getDim1()).append(", ")
- .append(hop.getDim2()).append(", ")
- .append(hop.getBlocksize()).append(", ")
- .append(hop.getNnz());
-
- if (hop.getUpdateType().isInPlace()) {
- sb.append(", ").append(hop.getUpdateType().toString().toLowerCase());
- }
- sb.append("]");
-
- // Add memory estimates
- sb.append(" [")
- .append(OptimizerUtils.toMB(hop.getInputMemEstimate())).append(", ")
- .append(OptimizerUtils.toMB(hop.getIntermediateMemEstimate())).append(", ")
- .append(OptimizerUtils.toMB(hop.getOutputMemEstimate())).append(" -> ")
- .append(OptimizerUtils.toMB(hop.getMemEstimate())).append("MB]");
-
- // Add reblock and checkpoint requirements
- if (hop.requiresReblock() && hop.requiresCheckpoint()) {
- sb.append(" [rblk, chkpt]");
- } else if (hop.requiresReblock()) {
- sb.append(" [rblk]");
- } else if (hop.requiresCheckpoint()) {
- sb.append(" [chkpt]");
- }
-
- // Add execution type
- if (hop.getExecType() != null) {
- sb.append(", ").append(hop.getExecType());
- }
-
- System.out.println(sb);
-
- // Process child nodes
- List<Pair<Long, FederatedOutput>> childRefs = plan.getChildFedPlans();
- for (int i = 0; i < childRefs.size(); i++) {
- Pair<Long, FederatedOutput> childRef = childRefs.get(i);
- FedPlanVariants childVariants = getFedPlanVariants(childRef.getLeft(), childRef.getRight());
- if (childVariants == null || childVariants.getFedPlanVariants().isEmpty())
- continue;
+ public static class HopCommon {
+ protected final Hop hopRef; // Reference to the associated Hop
+ protected double selfCost; // Current execution cost (compute + memory access)
+ protected double netTransferCost; // Network transfer cost
- boolean isLastChild = (i == childRefs.size() - 1);
- for (FedPlan childPlan : childVariants.getFedPlanVariants()) {
- printFedPlanTreeRecursive(childPlan, visited, depth + 1, isLastChild);
- }
+ protected HopCommon(Hop hopRef) {
+ this.hopRef = hopRef;
+ this.selfCost = 0;
+ this.netTransferCost = 0;
}
}
/**
- * Represents a collection of federated execution plan variants for a specific Hop.
- * Contains cost information and references to the associated plans.
+ * Represents a collection of federated execution plan variants for a specific Hop and FederatedOutput.
+ * This class contains cost information and references to the associated plans.
+ * It uses HopCommon to store common properties and costs related to the Hop.
*/
public static class FedPlanVariants {
- protected final Hop hopRef; // Reference to the associated Hop
- protected double selfCost; // Current execution cost (compute + memory access)
- protected double netTransferCost; // Network transfer cost
- private final FederatedOutput fedOutType; // Output type (FOUT/LOUT)
- protected List<FedPlan> _fedPlanVariants; // List of plan variants
+ protected HopCommon hopCommon; // Common properties and costs for the Hop
+ private final FederatedOutput fedOutType; // Output type (FOUT/LOUT)
+ protected List<FedPlan> _fedPlanVariants; // List of plan variants
public FedPlanVariants(Hop hopRef, FederatedOutput fedOutType) {
- this.hopRef = hopRef;
+ this.hopCommon = new HopCommon(hopRef);
this.fedOutType = fedOutType;
- this.selfCost = 0;
- this.netTransferCost = 0;
this._fedPlanVariants = new ArrayList<>();
}
- public int size() {return _fedPlanVariants.size();}
public void addFedPlan(FedPlan fedPlan) {_fedPlanVariants.add(fedPlan);}
public List<FedPlan> getFedPlanVariants() {return _fedPlanVariants;}
+ public boolean isEmpty() {return _fedPlanVariants.isEmpty();}
+
+ public void prune() {
+ if (_fedPlanVariants.size() > 1) {
+ // Find the FedPlan with the minimum cost
+ FedPlan minCostPlan = _fedPlanVariants.stream()
+ .min(Comparator.comparingDouble(FedPlan::getTotalCost))
+ .orElse(null);
+
+ // Retain only the minimum cost plan
+ _fedPlanVariants.clear();
+ _fedPlanVariants.add(minCostPlan);
+ }
+ }
}
/**
* Represents a single federated execution plan with its associated costs and dependencies.
- * Contains:
+ * This class contains:
* 1. selfCost: Cost of current hop (compute + input/output memory access)
* 2. totalCost: Cumulative cost including this plan and all child plans
* 3. netTransferCost: Network transfer cost for this plan to parent plan.
+ *
+ * FedPlan is linked to FedPlanVariants, which in turn uses HopCommon to manage common properties and costs.
*/
public static class FedPlan {
- private double totalCost; // Total cost including child plans
+ private double totalCost; // Total cost including child plans
private final FedPlanVariants fedPlanVariants; // Reference to variant list
private final List<Pair<Long, FederatedOutput>> childFedPlans; // Child plan references
@@ -264,25 +191,26 @@ public class FederatedMemoTable {
}
public void setTotalCost(double totalCost) {this.totalCost = totalCost;}
- public void setSelfCost(double selfCost) {fedPlanVariants.selfCost = selfCost;}
- public void setNetTransferCost(double netTransferCost) {fedPlanVariants.netTransferCost = netTransferCost;}
-
- public Hop getHopRef() {return fedPlanVariants.hopRef;}
+ public void setSelfCost(double selfCost) {fedPlanVariants.hopCommon.selfCost = selfCost;}
+ public void setNetTransferCost(double netTransferCost) {fedPlanVariants.hopCommon.netTransferCost = netTransferCost;}
+
+ public Hop getHopRef() {return fedPlanVariants.hopCommon.hopRef;}
+ public long getHopID() {return fedPlanVariants.hopCommon.hopRef.getHopID();}
public FederatedOutput getFedOutType() {return fedPlanVariants.fedOutType;}
public double getTotalCost() {return totalCost;}
- public double getSelfCost() {return fedPlanVariants.selfCost;}
- private double getNetTransferCost() {return fedPlanVariants.netTransferCost;}
+ public double getSelfCost() {return fedPlanVariants.hopCommon.selfCost;}
+ public double getNetTransferCost() {return fedPlanVariants.hopCommon.netTransferCost;}
public List<Pair<Long, FederatedOutput>> getChildFedPlans() {return childFedPlans;}
/**
* Calculates the conditional network transfer cost based on output type compatibility.
* Returns 0 if output types match, otherwise returns the network transfer cost.
- * @param parentFedOutType ?
- * @return ?
+ * @param parentFedOutType The federated output type of the parent plan.
+ * @return The conditional network transfer cost.
*/
public double getCondNetTransferCost(FederatedOutput parentFedOutType) {
if (parentFedOutType == getFedOutType()) return 0;
- return fedPlanVariants.netTransferCost;
+ return fedPlanVariants.hopCommon.netTransferCost;
}
}
}
diff --git a/src/main/java/org/apache/sysds/hops/fedplanner/FederatedMemoTablePrinter.java b/src/main/java/org/apache/sysds/hops/fedplanner/FederatedMemoTablePrinter.java
new file mode 100644
index 0000000000..f7b3343a98
--- /dev/null
+++ b/src/main/java/org/apache/sysds/hops/fedplanner/FederatedMemoTablePrinter.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.sysds.hops.fedplanner;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.sysds.hops.Hop;
+import org.apache.sysds.hops.OptimizerUtils;
+import org.apache.sysds.runtime.instructions.fed.FEDInstruction;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+public class FederatedMemoTablePrinter {
+ /**
+ * Recursively prints a tree representation of the DAG starting from the given root FedPlan.
+ * Includes information about hopID, fedOutType, TotalCost, SelfCost, and NetCost for each node.
+ * Additionally, prints the additional total cost once at the beginning.
+ *
+ * @param rootFedPlan The starting point FedPlan to print
+ * @param memoTable The memoization table containing FedPlan variants
+ * @param additionalTotalCost The additional cost to be printed once
+ */
+ public static void printFedPlanTree(FederatedMemoTable.FedPlan rootFedPlan, FederatedMemoTable memoTable,
+ double additionalTotalCost) {
+ System.out.println("Additional Cost: " + additionalTotalCost);
+ Set<FederatedMemoTable.FedPlan> visited = new HashSet<>();
+ printFedPlanTreeRecursive(rootFedPlan, memoTable, visited, 0);
+ }
+
+ /**
+ * Helper method to recursively print the FedPlan tree.
+ *
+ * @param plan The current FedPlan to print
+ * @param visited Set to keep track of visited FedPlans (prevents cycles)
+ * @param depth The current depth level for indentation
+ */
+ private static void printFedPlanTreeRecursive(FederatedMemoTable.FedPlan plan, FederatedMemoTable memoTable,
+ Set<FederatedMemoTable.FedPlan> visited, int depth) {
+ if (plan == null || visited.contains(plan)) {
+ return;
+ }
+
+ visited.add(plan);
+
+ Hop hop = plan.getHopRef();
+ StringBuilder sb = new StringBuilder();
+
+ // Add FedPlan information
+ sb.append(String.format("(%d) ", plan.getHopRef().getHopID()))
+ .append(plan.getHopRef().getOpString())
+ .append(" [")
+ .append(plan.getFedOutType())
+ .append("]");
+
+ StringBuilder childs = new StringBuilder();
+ childs.append(" (");
+ boolean childAdded = false;
+ for( Hop input : hop.getInput()){
+ childs.append(childAdded?",":"");
+ childs.append(input.getHopID());
+ childAdded = true;
+ }
+ childs.append(")");
+ if( childAdded )
+ sb.append(childs.toString());
+
+
+ sb.append(String.format(" {Total: %.1f, Self: %.1f, Net: %.1f}",
+ plan.getTotalCost(),
+ plan.getSelfCost(),
+ plan.getNetTransferCost()));
+
+ // Add matrix characteristics
+ sb.append(" [")
+ .append(hop.getDim1()).append(", ")
+ .append(hop.getDim2()).append(", ")
+ .append(hop.getBlocksize()).append(", ")
+ .append(hop.getNnz());
+
+ if (hop.getUpdateType().isInPlace()) {
+ sb.append(", ").append(hop.getUpdateType().toString().toLowerCase());
+ }
+ sb.append("]");
+
+ // Add memory estimates
+ sb.append(" [")
+ .append(OptimizerUtils.toMB(hop.getInputMemEstimate())).append(", ")
+ .append(OptimizerUtils.toMB(hop.getIntermediateMemEstimate())).append(", ")
+ .append(OptimizerUtils.toMB(hop.getOutputMemEstimate())).append(" -> ")
+ .append(OptimizerUtils.toMB(hop.getMemEstimate())).append("MB]");
+
+ // Add reblock and checkpoint requirements
+ if (hop.requiresReblock() && hop.requiresCheckpoint()) {
+ sb.append(" [rblk, chkpt]");
+ } else if (hop.requiresReblock()) {
+ sb.append(" [rblk]");
+ } else if (hop.requiresCheckpoint()) {
+ sb.append(" [chkpt]");
+ }
+
+ // Add execution type
+ if (hop.getExecType() != null) {
+ sb.append(", ").append(hop.getExecType());
+ }
+
+ System.out.println(sb);
+
+ // Process child nodes
+ List<Pair<Long, FEDInstruction.FederatedOutput>> childFedPlanPairs = plan.getChildFedPlans();
+ for (int i = 0; i < childFedPlanPairs.size(); i++) {
+ Pair<Long, FEDInstruction.FederatedOutput> childFedPlanPair = childFedPlanPairs.get(i);
+ FederatedMemoTable.FedPlanVariants childVariants = memoTable.getFedPlanVariants(childFedPlanPair);
+ if (childVariants == null || childVariants.isEmpty())
+ continue;
+
+ for (FederatedMemoTable.FedPlan childPlan : childVariants.getFedPlanVariants()) {
+ printFedPlanTreeRecursive(childPlan, memoTable, visited, depth + 1);
+ }
+ }
+ }
+}
diff --git a/src/main/java/org/apache/sysds/hops/fedplanner/FederatedPlanCostEnumerator.java b/src/main/java/org/apache/sysds/hops/fedplanner/FederatedPlanCostEnumerator.java
index 73e8d5d693..be1cfa7cdf 100644
--- a/src/main/java/org/apache/sysds/hops/fedplanner/FederatedPlanCostEnumerator.java
+++ b/src/main/java/org/apache/sysds/hops/fedplanner/FederatedPlanCostEnumerator.java
@@ -20,10 +20,14 @@
package org.apache.sysds.hops.fedplanner;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import java.util.Comparator;
+import java.util.HashMap;
import java.util.Objects;
+import java.util.LinkedHashMap;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.sysds.hops.Hop;
import org.apache.sysds.hops.fedplanner.FederatedMemoTable.FedPlan;
import org.apache.sysds.hops.fedplanner.FederatedMemoTable.FedPlanVariants;
@@ -36,12 +40,13 @@ import org.apache.sysds.runtime.instructions.fed.FEDInstruction.FederatedOutput;
*/
public class FederatedPlanCostEnumerator {
/**
- * Entry point for federated plan enumeration. Creates a memo table and returns
- * the minimum cost plan for the entire DAG.
+ * Entry point for federated plan enumeration. This method creates a memo table
+ * and returns the minimum cost plan for the entire Directed Acyclic Graph (DAG).
+ * It also resolves conflicts where FedPlans have different FederatedOutput types.
*
- * @param rootHop ?
- * @param printTree ?
- * @return ?
+ * @param rootHop The root Hop node from which to start the plan enumeration.
+ * @param printTree A boolean flag indicating whether to print the federated plan tree.
+ * @return The optimal FedPlan with the minimum cost for the entire DAG.
*/
public static FedPlan enumerateFederatedPlanCost(Hop rootHop, boolean printTree) {
// Create new memo table to store all plan variants
@@ -52,8 +57,12 @@ public class FederatedPlanCostEnumerator {
// Return the minimum cost plan for the root node
FedPlan optimalPlan = getMinCostRootFedPlan(rootHop.getHopID(), memoTable);
- memoTable.pruneMemoTable();
- if (printTree) memoTable.printFedPlanTree(optimalPlan);
+
+ // Detect conflicts in the federated plans where different FedPlans have different FederatedOutput types
+ double additionalTotalCost = detectAndResolveConflictFedPlan(optimalPlan, memoTable);
+
+ // Optionally print the federated plan tree if requested
+ if (printTree) FederatedMemoTablePrinter.printFedPlanTree(optimalPlan, memoTable, additionalTotalCost);
return optimalPlan;
}
@@ -106,6 +115,10 @@ public class FederatedPlanCostEnumerator {
FedPlan lOutPlan = memoTable.addFedPlan(hop, FederatedOutput.LOUT, planChilds);
FederatedPlanCostEstimator.computeFederatedPlanCost(lOutPlan, memoTable);
}
+
+ // Prune MemoTable for hop.
+ memoTable.pruneFedPlan(hop.getHopID(), FederatedOutput.LOUT);
+ memoTable.pruneFedPlan(hop.getHopID(), FederatedOutput.FOUT);
}
/**
@@ -133,4 +146,89 @@ public class FederatedPlanCostEnumerator {
}
return minlOutFedPlan;
}
+
+ /**
+ * Detects and resolves conflicts in federated plans starting from the root plan.
+ * This function performs a breadth-first search (BFS) to traverse the federated plan tree.
+ * It identifies conflicts where the same plan ID has different federated output types.
+ * For each conflict, it records the plan ID and its conflicting parent plans.
+ * The function ensures that each plan ID is associated with a consistent federated output type
+ * by resolving these conflicts iteratively.
+ *
+ * The process involves:
+ * - Using a map to track conflicts, associating each plan ID with its federated output type
+ * and a list of parent plans.
+ * - Storing detected conflicts in a linked map, each entry containing a plan ID and its
+ * conflicting parent plans.
+ * - Performing BFS traversal starting from the root plan, checking each child plan for conflicts.
+ * - If a conflict is detected (i.e., a plan ID has different output types), the conflicting plan
+ * is removed from the BFS queue and added to the conflict map to prevent duplicate calculations.
+ * - Resolving conflicts by ensuring a consistent federated output type across the plan.
+ * - Re-running BFS with resolved conflicts to ensure all inconsistencies are addressed.
+ *
+ * @param rootPlan The root federated plan from which to start the conflict detection.
+ * @param memoTable The memoization table used to retrieve pruned federated plans.
+ * @return The cumulative additional cost for resolving conflicts.
+ */
+ private static double detectAndResolveConflictFedPlan(FedPlan rootPlan, FederatedMemoTable memoTable) {
+ // Map to track conflicts: maps a plan ID to its federated output type and list of parent plans
+ Map<Long, Pair<FederatedOutput, List<FedPlan>>> conflictCheckMap = new HashMap<>();
+
+ // LinkedMap to store detected conflicts, each with a plan ID and its conflicting parent plans
+ LinkedHashMap<Long, List<FedPlan>> conflictLinkedMap = new LinkedHashMap<>();
+
+ // LinkedMap for BFS traversal starting from the root plan (Do not use value (boolean))
+ LinkedHashMap<FedPlan, Boolean> bfsLinkedMap = new LinkedHashMap<>();
+ bfsLinkedMap.put(rootPlan, true);
+
+ // Array to store cumulative additional cost for resolving conflicts
+ double[] cumulativeAdditionalCost = new double[]{0.0};
+
+ while (!bfsLinkedMap.isEmpty()) {
+ // Perform BFS to detect conflicts in federated plans
+ while (!bfsLinkedMap.isEmpty()) {
+ FedPlan currentPlan = bfsLinkedMap.keySet().iterator().next();
+ bfsLinkedMap.remove(currentPlan);
+
+ // Iterate over each child plan of the current plan
+ for (Pair<Long, FederatedOutput> childPlanPair : currentPlan.getChildFedPlans()) {
+ FedPlan childFedPlan = memoTable.getFedPlanAfterPrune(childPlanPair);
+
+ // Check if the child plan ID is already visited
+ if (conflictCheckMap.containsKey(childPlanPair.getLeft())) {
+ // Retrieve the existing conflict pair for the child plan
+ Pair<FederatedOutput, List<FedPlan>> conflictChildPlanPair = conflictCheckMap.get(childPlanPair.getLeft());
+ // Add the current plan to the list of parent plans
+ conflictChildPlanPair.getRight().add(currentPlan);
+
+ // If the federated output type differs, a conflict is detected
+ if (conflictChildPlanPair.getLeft() != childPlanPair.getRight()) {
+ // If this is the first detection, remove conflictChildFedPlan from the BFS queue and add it to the conflict linked map (queue)
+ // If the existing FedPlan is not removed from the bfsqueue or both actions are performed, duplicate calculations for the same FedPlan and its children occur
+ if (!conflictLinkedMap.containsKey(childPlanPair.getLeft())) {
+ conflictLinkedMap.put(childPlanPair.getLeft(), conflictChildPlanPair.getRight());
+ bfsLinkedMap.remove(childFedPlan);
+ }
+ }
+ } else {
+ // If no conflict exists, create a new entry in the conflict check map
+ List<FedPlan> parentFedPlanList = new ArrayList<>();
+ parentFedPlanList.add(currentPlan);
+
+ // Map the child plan ID to its output type and list of parent plans
+ conflictCheckMap.put(childPlanPair.getLeft(), new ImmutablePair<>(childPlanPair.getRight(), parentFedPlanList));
+ // Add the child plan to the BFS queue
+ bfsLinkedMap.put(childFedPlan, true);
+ }
+ }
+ }
+ // Resolve these conflicts to ensure a consistent federated output type across the plan
+ // Re-run BFS with resolved conflicts
+ bfsLinkedMap = FederatedPlanCostEstimator.resolveConflictFedPlan(memoTable, conflictLinkedMap, cumulativeAdditionalCost);
+ conflictLinkedMap.clear();
+ }
+
+ // Return the cumulative additional cost for resolving conflicts
+ return cumulativeAdditionalCost[0];
+ }
}
diff --git a/src/main/java/org/apache/sysds/hops/fedplanner/FederatedPlanCostEstimator.java b/src/main/java/org/apache/sysds/hops/fedplanner/FederatedPlanCostEstimator.java
index a716c3321d..7bc7339563 100644
--- a/src/main/java/org/apache/sysds/hops/fedplanner/FederatedPlanCostEstimator.java
+++ b/src/main/java/org/apache/sysds/hops/fedplanner/FederatedPlanCostEstimator.java
@@ -24,6 +24,11 @@ import org.apache.sysds.hops.cost.ComputeCost;
import org.apache.sysds.hops.fedplanner.FederatedMemoTable.FedPlan;
import org.apache.sysds.runtime.instructions.fed.FEDInstruction.FederatedOutput;
+import java.util.LinkedHashMap;
+import java.util.NoSuchElementException;
+import java.util.List;
+import java.util.Map;
+
/**
* Cost estimator for federated execution plans.
* Calculates computation, memory access, and network transfer costs for federated operations.
@@ -47,7 +52,7 @@ public class FederatedPlanCostEstimator {
* @param memoTable Table containing all plan variants
*/
public static void computeFederatedPlanCost(FedPlan currentPlan, FederatedMemoTable memoTable) {
- double totalCost = 0;
+ double totalCost;
Hop currentHop = currentPlan.getHopRef();
// Step 1: Calculate current node costs if not already computed
@@ -62,11 +67,11 @@ public class FederatedPlanCostEstimator {
}
// Step 2: Process each child plan and add their costs
- for (Pair<Long, FederatedOutput> planRefMeta : currentPlan.getChildFedPlans()) {
+ for (Pair<Long, FederatedOutput> childPlanPair : currentPlan.getChildFedPlans()) {
// Find minimum cost child plan considering federation type compatibility
// Note: This approach might lead to suboptimal or wrong solutions when a child has multiple parents
// because we're selecting child plans independently for each parent
- FedPlan planRef = memoTable.getMinCostChildFedPlan(planRefMeta.getLeft(), planRefMeta.getRight());
+ FedPlan planRef = memoTable.getMinCostFedPlan(childPlanPair);
// Add child plan cost (includes network transfer cost if federation types differ)
totalCost += planRef.getTotalCost() + planRef.getCondNetTransferCost(currentPlan.getFedOutType());
@@ -76,6 +81,125 @@ public class FederatedPlanCostEstimator {
currentPlan.setTotalCost(totalCost);
}
+ /**
+ * Resolves conflicts in federated plans where different plans have different FederatedOutput types.
+ * This function traverses the list of conflicting plans in reverse order to ensure that conflicts
+ * are resolved from the bottom-up, allowing for consistent federated output types across the plan.
+ * It calculates additional costs for each potential resolution and updates the cumulative additional cost.
+ *
+ * @param memoTable The FederatedMemoTable containing all federated plan variants.
+ * @param conflictFedPlanLinkedMap A map of plan IDs to lists of parent plans with conflicting federated outputs.
+ * @param cumulativeAdditionalCost An array to store the cumulative additional cost incurred by resolving conflicts.
+ * @return A LinkedHashMap of resolved federated plans, marked with a boolean indicating resolution status.
+ */
+ public static LinkedHashMap<FedPlan, Boolean> resolveConflictFedPlan(FederatedMemoTable memoTable, LinkedHashMap<Long, List<FedPlan>> conflictFedPlanLinkedMap, double[] cumulativeAdditionalCost) {
+ // LinkedHashMap to store resolved federated plans for BFS traversal.
+ LinkedHashMap<FedPlan, Boolean> resolvedFedPlanLinkedMap = new LinkedHashMap<>();
+
+ // Traverse the conflictFedPlanList in reverse order after BFS to resolve conflicts
+ for (Map.Entry<Long, List<FedPlan>> conflictFedPlanPair : conflictFedPlanLinkedMap.entrySet()) {
+ long conflictHopID = conflictFedPlanPair.getKey();
+ List<FedPlan> conflictParentFedPlans = conflictFedPlanPair.getValue();
+
+ // Retrieve the conflicting federated plans for LOUT and FOUT types
+ FedPlan confilctLOutFedPlan = memoTable.getFedPlanAfterPrune(conflictHopID, FederatedOutput.LOUT);
+ FedPlan confilctFOutFedPlan = memoTable.getFedPlanAfterPrune(conflictHopID, FederatedOutput.FOUT);
+
+ // Variables to store additional costs for LOUT and FOUT types
+ double lOutAdditionalCost = 0;
+ double fOutAdditionalCost = 0;
+
+ // Flags to check if the plan involves network transfer
+ // Network transfer cost is calculated only once, even if it occurs multiple times
+ boolean isLOutNetTransfer = false;
+ boolean isFOutNetTransfer = false;
+
+ // Determine the optimal federated output type based on the calculated costs
+ FederatedOutput optimalFedOutType;
+
+ // Iterate over each parent federated plan in the current conflict pair
+ for (FedPlan conflictParentFedPlan : conflictParentFedPlans) {
+ // Find the calculated FedOutType of the child plan
+ Pair<Long, FederatedOutput> cacluatedConflictPlanPair = conflictParentFedPlan.getChildFedPlans().stream()
+ .filter(pair -> pair.getLeft().equals(conflictHopID))
+ .findFirst()
+ .orElseThrow(() -> new NoSuchElementException("No matching pair found for ID: " + conflictHopID));
+
+ // CASE 1. Calculated LOUT / Parent LOUT / Current LOUT: Total cost remains unchanged.
+ // CASE 2. Calculated LOUT / Parent FOUT / Current LOUT: Total cost remains unchanged, subtract net cost, add net cost later.
+ // CASE 3. Calculated FOUT / Parent LOUT / Current LOUT: Change total cost, subtract net cost.
+ // CASE 4. Calculated FOUT / Parent FOUT / Current LOUT: Change total cost, add net cost later.
+ // CASE 5. Calculated LOUT / Parent LOUT / Current FOUT: Change total cost, add net cost later.
+ // CASE 6. Calculated LOUT / Parent FOUT / Current FOUT: Change total cost, subtract net cost.
+ // CASE 7. Calculated FOUT / Parent LOUT / Current FOUT: Total cost remains unchanged, subtract net cost, add net cost later.
+ // CASE 8. Calculated FOUT / Parent FOUT / Current FOUT: Total cost remains unchanged.
+
+ // Adjust LOUT, FOUT costs based on the calculated plan's output type
+ if (cacluatedConflictPlanPair.getRight() == FederatedOutput.LOUT) {
+ // When changing from calculated LOUT to current FOUT, subtract the existing LOUT total cost and add the FOUT total cost
+ // When maintaining calculated LOUT to current LOUT, the total cost remains unchanged.
+ fOutAdditionalCost += confilctFOutFedPlan.getTotalCost() - confilctLOutFedPlan.getTotalCost();
+
+ if (conflictParentFedPlan.getFedOutType() == FederatedOutput.LOUT) {
+ // (CASE 1) Previously, calculated was LOUT and parent was LOUT, so no network transfer cost occurred
+ // (CASE 5) If changing from calculated LOUT to current FOUT, network transfer cost occurs, but calculated later
+ isFOutNetTransfer = true;
+ } else {
+ // Previously, calculated was LOUT and parent was FOUT, so network transfer cost occurred
+ // (CASE 2) If maintaining calculated LOUT to current LOUT, subtract existing network transfer cost and calculate later
+ isLOutNetTransfer = true;
+ lOutAdditionalCost -= confilctLOutFedPlan.getNetTransferCost();
+
+ // (CASE 6) If changing from calculated LOUT to current FOUT, no network transfer cost occurs, so subtract it
+ fOutAdditionalCost -= confilctLOutFedPlan.getNetTransferCost();
+ }
+ } else {
+ lOutAdditionalCost += confilctLOutFedPlan.getTotalCost() - confilctFOutFedPlan.getTotalCost();
+
+ if (conflictParentFedPlan.getFedOutType() == FederatedOutput.FOUT) {
+ isLOutNetTransfer = true;
+ } else {
+ isFOutNetTransfer = true;
+ lOutAdditionalCost -= confilctLOutFedPlan.getNetTransferCost();
+ fOutAdditionalCost -= confilctLOutFedPlan.getNetTransferCost();
+ }
+ }
+ }
+
+ // Add network transfer costs if applicable
+ if (isLOutNetTransfer) {
+ lOutAdditionalCost += confilctLOutFedPlan.getNetTransferCost();
+ }
+ if (isFOutNetTransfer) {
+ fOutAdditionalCost += confilctFOutFedPlan.getNetTransferCost();
+ }
+
+ // Determine the optimal federated output type based on the calculated costs
+ if (lOutAdditionalCost <= fOutAdditionalCost) {
+ optimalFedOutType = FederatedOutput.LOUT;
+ cumulativeAdditionalCost[0] += lOutAdditionalCost;
+ resolvedFedPlanLinkedMap.put(confilctLOutFedPlan, true);
+ } else {
+ optimalFedOutType = FederatedOutput.FOUT;
+ cumulativeAdditionalCost[0] += fOutAdditionalCost;
+ resolvedFedPlanLinkedMap.put(confilctFOutFedPlan, true);
+ }
+
+ // Update only the optimal federated output type, not the cost itself or recursively
+ for (FedPlan conflictParentFedPlan : conflictParentFedPlans) {
+ for (Pair<Long, FederatedOutput> childPlanPair : conflictParentFedPlan.getChildFedPlans()) {
+ if (childPlanPair.getLeft() == conflictHopID && childPlanPair.getRight() != optimalFedOutType) {
+ int index = conflictParentFedPlan.getChildFedPlans().indexOf(childPlanPair);
+ conflictParentFedPlan.getChildFedPlans().set(index,
+ Pair.of(childPlanPair.getLeft(), optimalFedOutType));
+ break;
+ }
+ }
+ }
+ }
+ return resolvedFedPlanLinkedMap;
+ }
+
/**
* Computes the cost for the current Hop node.
*
diff --git a/src/test/java/org/apache/sysds/test/component/federated/FederatedPlanCostEnumeratorTest.java b/src/test/java/org/apache/sysds/test/component/federated/FederatedPlanCostEnumeratorTest.java
index 56de8cf3c4..20485588d3 100644
--- a/src/test/java/org/apache/sysds/test/component/federated/FederatedPlanCostEnumeratorTest.java
+++ b/src/test/java/org/apache/sysds/test/component/federated/FederatedPlanCostEnumeratorTest.java
@@ -39,7 +39,7 @@ import org.apache.sysds.hops.fedplanner.FederatedPlanCostEnumerator;
public class FederatedPlanCostEnumeratorTest extends AutomatedTestBase
{
- private static final String TEST_DIR = "functions/federated/";
+ private static final String TEST_DIR = "functions/federated/privacy/";
private static final String HOME = SCRIPT_DIR + TEST_DIR;
private static final String TEST_CLASS_DIR = TEST_DIR + FederatedPlanCostEnumeratorTest.class.getSimpleName() + "/";
@@ -47,8 +47,15 @@ public class FederatedPlanCostEnumeratorTest extends AutomatedTestBase
public void setUp() {}
@Test
- public void testDependencyAnalysis1() { runTest("cost.dml"); }
-
+ public void testFederatedPlanCostEnumerator1() { runTest("FederatedPlanCostEnumeratorTest1.dml"); }
+
+ @Test
+ public void testFederatedPlanCostEnumerator2() { runTest("FederatedPlanCostEnumeratorTest2.dml"); }
+
+ @Test
+ public void testFederatedPlanCostEnumerator3() { runTest("FederatedPlanCostEnumeratorTest3.dml"); }
+
+ // Todo: Need to write test scripts for the federated version
private void runTest( String scriptFilename ) {
int index = scriptFilename.lastIndexOf(".dml");
String testName = scriptFilename.substring(0, index > 0 ? index : scriptFilename.length());
@@ -72,10 +79,7 @@ public class FederatedPlanCostEnumeratorTest extends AutomatedTestBase
dmlt.constructHops(prog);
dmlt.rewriteHopsDAG(prog);
dmlt.constructLops(prog);
- /* TODO) In the current DAG, Hop's _outputMemEstimate is not initialized
- // This leads to incorrect fedplan generation, so test code needs to be modified
- // If needed, modify costEstimator to handle cases where _outputMemEstimate is not initialized
- */
+
Hop hops = prog.getStatementBlocks().get(0).getHops().get(0);
FederatedPlanCostEnumerator.enumerateFederatedPlanCost(hops, true);
}
diff --git a/src/test/scripts/functions/federated/cost.dml b/src/test/scripts/functions/federated/privacy/FederatedPlanCostEnumeratorTest1.dml
similarity index 100%
copy from src/test/scripts/functions/federated/cost.dml
copy to src/test/scripts/functions/federated/privacy/FederatedPlanCostEnumeratorTest1.dml
diff --git a/src/test/scripts/functions/federated/cost.dml b/src/test/scripts/functions/federated/privacy/FederatedPlanCostEnumeratorTest2.dml
similarity index 95%
copy from src/test/scripts/functions/federated/cost.dml
copy to src/test/scripts/functions/federated/privacy/FederatedPlanCostEnumeratorTest2.dml
index ec34d45bb6..3cc07eeb01 100644
--- a/src/test/scripts/functions/federated/cost.dml
+++ b/src/test/scripts/functions/federated/privacy/FederatedPlanCostEnumeratorTest2.dml
@@ -21,5 +21,6 @@
a = matrix(7,10,10);
b = a + a^2;
-c = sqrt(b);
-print(sum(c));
\ No newline at end of file
+c = a * b;
+d = b + sqrt(c);
+print(sum(d));
\ No newline at end of file
diff --git a/src/test/scripts/functions/federated/cost.dml b/src/test/scripts/functions/federated/privacy/FederatedPlanCostEnumeratorTest3.dml
similarity index 93%
rename from src/test/scripts/functions/federated/cost.dml
rename to src/test/scripts/functions/federated/privacy/FederatedPlanCostEnumeratorTest3.dml
index ec34d45bb6..7fe002df75 100644
--- a/src/test/scripts/functions/federated/cost.dml
+++ b/src/test/scripts/functions/federated/privacy/FederatedPlanCostEnumeratorTest3.dml
@@ -20,6 +20,9 @@
#-------------------------------------------------------------
a = matrix(7,10,10);
-b = a + a^2;
+if (sum(a) > 0.5)
+ b = a + a^2;
+else
+ b = a * a;
c = sqrt(b);
print(sum(c));
\ No newline at end of file