Merge pull request #2 from tangzhankun/zhankun-add-scheduler

Modify fair scheduler and DS

Author: tangzhankun

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/FPGASlot.java

@@ -1,27 +1,36 @@
 package org.apache.hadoop.yarn.api.records;
 
-public class FPGASlot {
+public class FPGASlot implements Comparable<FPGASlot>{
+
+  private FPGAType fpgaType;
+  private String slotId;
+  private String afuId;
+
+  public static FPGASlot newInstance(FPGAType type, String slotId, String afuId) {
+    FPGASlot fpgaSlot = new FPGASlot();
+    fpgaSlot.setFpgaType(type);
+    fpgaSlot.setSlotId(slotId);
+    fpgaSlot.setAfuId(afuId);
+    return fpgaSlot;
+  }
 
-  private final FPGAType fpgaType;
-  private final String socketId;
-  private final String slotId;
-  private final String afuId;
+  public static FPGASlot newInstance(FPGASlot a) {
+    FPGASlot fpgaSlot = new FPGASlot();
+    fpgaSlot.setSlotId(a.getSlotId());
+    fpgaSlot.setFpgaType(a.getFpgaType());
+    fpgaSlot.setAfuId(a.getAfuId());
+    return fpgaSlot;
+  }
 
-  private FPGASlot(Builder builder) {
-    this.fpgaType = builder.fpgaType;
-    this.socketId = builder.socketId;
-    this.slotId = builder.slotId;
-    this.afuId = builder.afuId;
+  public static FPGASlot newInstance() {
+    FPGASlot fpgaSlot = new FPGASlot();
+    return fpgaSlot;
   }
 
   public FPGAType getFpgaType() {
     return fpgaType;
   }
 
-  public String getSocketId() {
-    return socketId;
-  }
-
   public String getSlotId() {
     return slotId;
   }
@@ -31,41 +40,54 @@ public String getAfuId() {
   }
 
   @Override
-  public String toString() {
-    return "fpga type: " + fpgaType + " socket id: " + socketId + " slot id: " + slotId + " afu id: " + afuId + ".";
+  public boolean equals(Object obj) {
+    if (this == obj)
+      return true;
+    if (obj == null)
+      return false;
+    if (!(obj instanceof FPGASlot))
+      return false;
+    FPGASlot other = (FPGASlot) obj;
+    return !(getFpgaType() != other.getFpgaType()
+        || !getAfuId().equals(other.getAfuId())
+        || !getSlotId().equals(other.getSlotId()));
   }
 
-  public static class Builder {
-
-    private FPGAType fpgaType;
-    private String socketId;
-    private String slotId;
-    private String afuId;
-
-    public Builder fpgaType(FPGAType fpgaType) {
-      this.fpgaType = fpgaType;
-      return this;
-    }
-
-    public Builder socketId(String socketId) {
-      this.socketId = socketId;
-      return this;
-    }
+  @Override
+  public int hashCode() {
+    final int prime = 263167;
+    int result = 939769357 + fpgaType.hashCode();
+    result = prime * result + slotId.hashCode();
+    result = prime * result + afuId.hashCode();
+    return result;
+  }
 
-    public Builder slotId(String slotId) {
-      this.slotId = slotId;
-      return this;
-    }
+  @Override
+  public String toString() {
+    return "fpga type: " + fpgaType + " slot id: " + slotId + " afu id: " + afuId + ".";
+  }
 
-    public Builder afuId(String afuId) {
-      this.afuId = afuId;
-      return this;
+  @Override
+  public int compareTo(FPGASlot o) {
+    int diff = this.fpgaType.compareTo(o.fpgaType);
+    if (diff == 0) {
+      diff = this.slotId.compareTo(o.slotId);
+      if (diff == 0) {
+        diff = this.afuId.compareTo(o.afuId);
+      }
     }
+    return diff;
+  }
 
-    synchronized public FPGASlot build() {
-      return new FPGASlot(this);
-    }
+  public void setFpgaType(FPGAType fpgaType) {
+    this.fpgaType = fpgaType;
+  }
 
+  public void setSlotId(String slotId) {
+    this.slotId = slotId;
   }
 
+  public void setAfuId(String afuId) {
+    this.afuId = afuId;
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/FPGAType.java

@@ -8,5 +8,7 @@ public enum FPGAType {
 
   RSD,
 
-  PHI
+  PHI,
+
+  UNKNOWN,
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/Resource.java

@@ -6,9 +6,9 @@
  * 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
- *
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
  * 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.
@@ -24,28 +24,30 @@
 import org.apache.hadoop.classification.InterfaceStability.Stable;
 import org.apache.hadoop.yarn.api.ApplicationMasterProtocol;
 import org.apache.hadoop.yarn.util.Records;
-import java.util.List;
+
+import java.util.Iterator;
+import java.util.Set;
 
 /**
  * <p><code>Resource</code> models a set of computer resources in the 
  * cluster.</p>
- * 
+ *
  * <p>Currently it models both <em>memory</em> and <em>CPU</em>.</p>
- * 
+ *
  * <p>The unit for memory is megabytes. CPU is modeled with virtual cores
  * (vcores), a unit for expressing parallelism. A node's capacity should
  * be configured with virtual cores equal to its number of physical cores. A
  * container should be requested with the number of cores it can saturate, i.e.
  * the average number of threads it expects to have runnable at a time.</p>
- * 
+ *
  * <p>Virtual cores take integer values and thus currently CPU-scheduling is
  * very coarse.  A complementary axis for CPU requests that represents processing
  * power will likely be added in the future to enable finer-grained resource
  * configuration.</p>
- * 
+ *
  * <p>Typically, applications request <code>Resource</code> of suitable
  * capability to run their component tasks.</p>
- * 
+ *
  * @see ResourceRequest
  * @see ApplicationMasterProtocol#allocate(org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest)
  */
@@ -73,7 +75,7 @@ public static Resource newInstance(long memory, int vCores) {
 
   @Public
   @Stable
-  public static Resource newInstance(int memory, int vCores, List<FPGASlot> fpgaSlots) {
+  public static Resource newInstance(int memory, int vCores, Set<FPGASlot> fpgaSlots) {
     Resource resource = Records.newRecord(Resource.class);
     resource.setMemorySize(memory);
     resource.setVirtualCores(vCores);
@@ -83,7 +85,7 @@ public static Resource newInstance(int memory, int vCores, List<FPGASlot> fpgaSl
 
   @Public
   @Stable
-  public static Resource newInstance(long memory, int vCores, List<FPGASlot> fpgaSlots) {
+  public static Resource newInstance(long memory, int vCores, Set<FPGASlot> fpgaSlots) {
     Resource resource = Records.newRecord(Resource.class);
     resource.setMemorySize(memory);
     resource.setVirtualCores(vCores);
@@ -110,7 +112,7 @@ public static Resource newInstance(long memory, int vCores, List<FPGASlot> fpgaS
   @Stable
   public long getMemorySize() {
     throw new NotImplementedException(
-        "This method is implemented by ResourcePBImpl");
+            "This method is implemented by ResourcePBImpl");
   }
 
   /**
@@ -129,32 +131,32 @@ public long getMemorySize() {
   @Stable
   public void setMemorySize(long memory) {
     throw new NotImplementedException(
-        "This method is implemented by ResourcePBImpl");
+            "This method is implemented by ResourcePBImpl");
   }
 
 
   /**
    * Get <em>number of virtual cpu cores</em> of the resource.
-   * 
+   *
    * Virtual cores are a unit for expressing CPU parallelism. A node's capacity
    * should be configured with virtual cores equal to its number of physical cores.
    * A container should be requested with the number of cores it can saturate, i.e.
    * the average number of threads it expects to have runnable at a time.
-   *   
+   *
    * @return <em>num of virtual cpu cores</em> of the resource
    */
   @Public
   @Evolving
   public abstract int getVirtualCores();
-  
+
   /**
    * Set <em>number of virtual cpu cores</em> of the resource.
-   * 
+   *
    * Virtual cores are a unit for expressing CPU parallelism. A node's capacity
    * should be configured with virtual cores equal to its number of physical cores.
    * A container should be requested with the number of cores it can saturate, i.e.
    * the average number of threads it expects to have runnable at a time.
-   *    
+   *
    * @param vCores <em>number of virtual cpu cores</em> of the resource
    */
 
@@ -164,19 +166,22 @@ public void setMemorySize(long memory) {
 
   @Public
   @Evolving
-  public abstract List<FPGASlot> getFPGASlots();
+  public abstract Set<FPGASlot> getFPGASlots();
 
   @Public
   @Evolving
-  public abstract void setFPGASlots(List<FPGASlot> fpgaSlots);
+  public abstract void setFPGASlots(Set<FPGASlot> fpgaSlots);
 
   @Override
   public int hashCode() {
     final int prime = 263167;
 
     int result = (int) (939769357
-        + getMemorySize()); // prime * result = 939769357 initially
+            + getMemorySize()); // prime * result = 939769357 initially
     result = prime * result + getVirtualCores();
+    for (FPGASlot fpgaSlot : getFPGASlots()) {
+      result = prime * result + fpgaSlot.hashCode();
+    }
     return result;
   }
 
@@ -190,23 +195,48 @@ public boolean equals(Object obj) {
       return false;
     Resource other = (Resource) obj;
     if (getMemorySize() != other.getMemorySize() ||
-        getVirtualCores() != other.getVirtualCores()) {
+            getVirtualCores() != other.getVirtualCores()) {
       return false;
     }
-    return true;
+    return this.getFPGASlots().equals(other.getFPGASlots());
   }
 
   @Override
   public String toString() {
     StringBuilder fpgaInfo = new StringBuilder();
 
-    List<FPGASlot> fpgaSlots = getFPGASlots();
+    Set<FPGASlot> fpgaSlots = getFPGASlots();
+    if (fpgaSlots == null) {
+      return "<memory:" + getMemorySize() + ", vCores:" + getVirtualCores() + ">";
+    }
     fpgaInfo.append("\t\t\t\tFPGA accelerator number:" + fpgaSlots.size() + "\n");
+    if (fpgaSlots.size() > 0) {
+      fpgaInfo.append("\t\t\t\tFPGA accelerator details: \n");
+      for (FPGASlot fpgaSlot : fpgaSlots) {
+        fpgaInfo.append("\t\t\t\t     fpga type:" + fpgaSlot.getFpgaType() + ", slot id:" + fpgaSlot.getSlotId() + ", afu id:" + fpgaSlot.getAfuId() + "\n");
+      }
+    }
+    return "<memory:" + getMemorySize() + ", vCores:" + getVirtualCores() + ">\n" + fpgaInfo;
+  }
 
-    fpgaInfo.append("\t\t\t\tFPGA accelerator details: \n");
-    for(FPGASlot fpgaSlot : fpgaSlots) {
-      fpgaInfo.append("\t\t\t\t     fpga type:" + fpgaSlot.getFpgaType() + ", socket id:" + fpgaSlot.getSocketId() + ", slot id:" + fpgaSlot.getSlotId() + ", afu id:" + fpgaSlot.getAfuId() + "\n");
+  @Override
+  public int compareTo(Resource o) {
+    Set<FPGASlot> thisFPGA = this.getFPGASlots();
+    Set<FPGASlot> otherFPGA = o.getFPGASlots();
+    int diff = thisFPGA.size() - otherFPGA.size();
+    if (diff == 0) {
+      diff = this.getVirtualCores() - o.getVirtualCores();
+      if (diff == 0) {
+        diff = this.getMemory() - o.getMemory();
+        if (diff == 0) {
+          Iterator<FPGASlot> it1 = thisFPGA.iterator();
+          Iterator<FPGASlot> it2 = otherFPGA.iterator();
+          for (; it1.hasNext();) {
+            diff = it1.next().compareTo(it2.next());
+          }
+        }
+      }
     }
-    return "<memory:" + getMemorySize() + ", vCores:" + getVirtualCores() + ">\n" + fpgaInfo ;
+    return diff;
   }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java

@@ -1123,7 +1123,7 @@ public static boolean isAclEnabled(Configuration conf) {
   public static final int DEFAULT_NM_VCORES = 8;
 
   public static final String NM_FPGA = NM_PREFIX + "resource.fpga";
-  public static final String DEFAULT_NM_FPGA = "MCP:0:1:88f2b0e6-b3d0-4b18-a7e1-0b53eaaf74e1,DSC:0:1:50d64a0f-da0d-4057-ba94-67290d25eeee";
+  public static final String DEFAULT_NM_FPGA = "MCP:0:88f2b0e6-b3d0-4b18-a7e1-0b53eaaf74e1,DSC:0:50d64a0f-da0d-4057-ba94-67290d25eeee,MCP:1:00000000-0000-0000-0000-000000000000";
 
   /** Count logical processors(like hyperthreads) as cores. */
   public static final String NM_COUNT_LOGICAL_PROCESSORS_AS_CORES = NM_PREFIX

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto

@@ -64,11 +64,11 @@ enum FPGATypeProto {
   MCP = 2;
   RSD = 3;
   PHI = 4;
+  UNKNOWN = 5;
 }
 
 message FPGAOptionProto {
   optional FPGATypeProto fpga_type = 1;
-  optional string socket_id = 2;
   optional string slot_id = 3;
   optional string afu_id = 4;
 }
@@ -79,10 +79,6 @@ message ResourceUtilizationProto {
   optional float cpu = 3;
 }
 
-message FPGAResourceProto {
-  optional string type = 1;
-  optional string accelerator = 2;
-}
 
 message ResourceOptionProto {
   optional ResourceProto resource = 1;