Validate resource constraint (RAM and CPU) in RoundRobinPacking (apache#3142)

* init

* general resource constraint validation

* pass existing unit tests

* add more tests

* rename

* rename

* generic ResourceMeasure

* fixed wc example

* even more general generics

* address comments

* address comments by putting more tests

* set safe amount of cpu

* meaningful constants in ExampleResource

Author: xiaoyao1991

examples/src/java/org/apache/heron/examples/api/ExampleResources.java

@@ -23,7 +23,9 @@
 
 public final class ExampleResources {
 
+  static final long DEFAULT_RAM_PADDING_PER_CONTAINER = 2 * 1024;
   static final long COMPONENT_RAM_MB = 1024;
+  static final double DEFAULT_CPU_PADDING_PER_CONTAINER = 1.0;
 
   static ByteAmount getComponentRam() {
     return ByteAmount.fromMegabytes(COMPONENT_RAM_MB);
@@ -35,7 +37,13 @@ static ByteAmount getContainerDisk(int components, int containers) {
 
   static ByteAmount getContainerRam(int components, int containers) {
     final int componentsPerContainer = Math.max(components / containers, 1);
-    return ByteAmount.fromMegabytes(COMPONENT_RAM_MB * componentsPerContainer);
+    return ByteAmount.fromMegabytes(COMPONENT_RAM_MB * componentsPerContainer
+        + DEFAULT_RAM_PADDING_PER_CONTAINER);
+  }
+
+  static double getContainerCpu(int components, int containers) {
+    final int componentsPerContainer = Math.max(components / containers, 1);
+    return componentsPerContainer + DEFAULT_CPU_PADDING_PER_CONTAINER;
   }
 
   private ExampleResources() {

examples/src/java/org/apache/heron/examples/api/WordCountTopology.java

@@ -188,12 +188,16 @@ public static void main(String[] args) throws AlreadyAliveException, InvalidTopo
     conf.setComponentRam("consumer",
         ByteAmount.fromMegabytes(ExampleResources.COMPONENT_RAM_MB));
 
+    conf.setComponentCpu("word", 1.0);
+    conf.setComponentCpu("consumer", 1.0);
+
     // configure container resources
     conf.setContainerDiskRequested(
         ExampleResources.getContainerDisk(2 * parallelism, parallelism));
     conf.setContainerRamRequested(
         ExampleResources.getContainerRam(2 * parallelism, parallelism));
-    conf.setContainerCpuRequested(2);
+    conf.setContainerCpuRequested(
+        ExampleResources.getContainerCpu(2 * parallelism, parallelism));
 
     HeronSubmitter.submitTopology(args[0], conf, builder.createTopology());
   }

heron/common/src/java/org/apache/heron/common/basics/ByteAmount.java

@@ -22,7 +22,7 @@
 /**
  * Class that encapsulates number of bytes, with helpers to handle units properly.
  */
-public final class ByteAmount implements Comparable<ByteAmount> {
+public final class ByteAmount extends ResourceMeasure<Long> {
   private static final long KB = 1024L;
   private static final long MB = KB * 1024;
   private static final long GB = MB * 1024;
@@ -34,10 +34,9 @@ public final class ByteAmount implements Comparable<ByteAmount> {
   private static final long MAX_KB = Math.round(Long.MAX_VALUE / KB);
 
   public static final ByteAmount ZERO = ByteAmount.fromBytes(0);
-  private final long bytes;
 
-  private ByteAmount(long bytes) {
-    this.bytes = bytes;
+  private ByteAmount(Long value) {
+    super(value);
   }
 
   /**
@@ -67,7 +66,8 @@ public static ByteAmount fromMegabytes(long megabytes) {
 
   /**
    * Creates a ByteAmount value in gigabytes. If the gigabytes value
-   * is &gt;= Long.MAX_VALUE / 1024 / 1024 / 1024, the byte representation is capped at Long.MAX_VALUE.
+   * is &gt;= Long.MAX_VALUE / 1024 / 1024 / 1024,
+   * the byte representation is capped at Long.MAX_VALUE.
    *
    * @param gigabytes value in gigabytes to represent
    * @return a ByteAmount object repressing the number of GBs passed
@@ -85,7 +85,7 @@ public static ByteAmount fromGigabytes(long gigabytes) {
    * @return number of bytes
    */
   public long asBytes() {
-    return bytes;
+    return super.getValue();
   }
 
   /**
@@ -97,7 +97,7 @@ public long asBytes() {
    * @return returns the ByteValue in MBs or 0 if the value is &lt; (1024 * 1024) / 2
    */
   public long asMegabytes() {
-    return Math.round((double) bytes / MB);
+    return Math.round(value.doubleValue() / MB);
   }
 
   /**
@@ -109,7 +109,7 @@ public long asMegabytes() {
    * @return returns the ByteValue in KBs or 0 if the value is &lt; (1024) / 2
    */
   public long asKilobytes() {
-    return Math.round((double) bytes / KB);
+    return Math.round(value.doubleValue() / KB);
   }
 
   /**
@@ -121,15 +121,7 @@ public long asKilobytes() {
    * @return returns the ByteValue in GBs or 0 if the value is &lt; (1024 * 1024 * 1024) / 2
    */
   public long asGigabytes() {
-    return Math.round((double) bytes / GB);
-  }
-
-  /**
-   * Convenience methdod to determine if byte value is zero
-   * @return true if the byte value is 0
-   */
-  public boolean isZero() {
-    return ZERO.equals(this);
+    return Math.round(value.doubleValue() / GB);
   }
 
   /**
@@ -138,10 +130,11 @@ public boolean isZero() {
    * @return a new ByteValue of this minus other ByteValue
    * @throws IllegalArgumentException if subtraction would overshoot Long.MIN_VALUE
    */
-  public ByteAmount minus(ByteAmount other) {
-    checkArgument(Long.MIN_VALUE + other.asBytes() <= asBytes(), String.format(
-        "Subtracting %s from %s would overshoot Long.MIN_LONG", other, this));
-    return ByteAmount.fromBytes(asBytes() - other.asBytes());
+  @Override
+  public ByteAmount minus(ResourceMeasure<Long> other) {
+    checkArgument(Long.MIN_VALUE + other.value <= value,
+        String.format("Subtracting %s from %s would overshoot Long.MIN_LONG", other, this));
+    return ByteAmount.fromBytes(value - other.value);
   }
 
   /**
@@ -150,10 +143,11 @@ public ByteAmount minus(ByteAmount other) {
    * @return a new ByteValue of this plus other ByteValue
    * @throws IllegalArgumentException if addition would exceed Long.MAX_VALUE
    */
-  public ByteAmount plus(ByteAmount other) {
-    checkArgument(Long.MAX_VALUE - asBytes() >= other.asBytes(), String.format(
-        "Adding %s to %s would exceed Long.MAX_LONG", other, this));
-    return ByteAmount.fromBytes(asBytes() + other.asBytes());
+  @Override
+  public ByteAmount plus(ResourceMeasure<Long> other) {
+    checkArgument(Long.MAX_VALUE - value >= other.value,
+        String.format("Adding %s to %s would exceed Long.MAX_LONG", other, this));
+    return ByteAmount.fromBytes(value + other.value);
   }
 
   /**
@@ -162,10 +156,11 @@ public ByteAmount plus(ByteAmount other) {
    * @return a new ByteValue of this ByteValue multiplied by factor
    * @throws IllegalArgumentException if multiplication would exceed Long.MAX_VALUE
    */
+  @Override
   public ByteAmount multiply(int factor) {
-    checkArgument(asBytes() <= Long.MAX_VALUE / factor, String.format(
-        "Multiplying %s by %d would exceed Long.MAX_LONG", this, factor));
-    return ByteAmount.fromBytes(asBytes() * factor);
+    checkArgument(value <= Long.MAX_VALUE / factor,
+        String.format("Multiplying %s by %d would exceed Long.MAX_LONG", this, factor));
+    return ByteAmount.fromBytes(value * factor);
   }
 
   /**
@@ -175,9 +170,10 @@ public ByteAmount multiply(int factor) {
    * @param factor value to divide by
    * @return a new ByteValue of this ByteValue divided by factor
    */
+  @Override
   public ByteAmount divide(int factor) {
     checkArgument(factor != 0, String.format("Can not divide %s by 0", this));
-    return ByteAmount.fromBytes(Math.round((double) this.asBytes() / (double) factor));
+    return ByteAmount.fromBytes(Math.round(value.doubleValue() / factor));
   }
 
   /**
@@ -187,30 +183,15 @@ public ByteAmount divide(int factor) {
    * @return a new ByteValue of this ByteValue increased by percentage
    * @throws IllegalArgumentException if increase would exceed Long.MAX_VALUE
    */
+  @Override
   public ByteAmount increaseBy(int percentage) {
-    checkArgument(percentage >= 0, String.format(
-        "Increasing by negative percent (%d) not supported", percentage));
+    checkArgument(percentage >= 0,
+        String.format("Increasing by negative percent (%d) not supported", percentage));
     double factor = 1.0 + ((double) percentage / 100);
     long max = Math.round(Long.MAX_VALUE / factor);
-    checkArgument(asBytes() <= max, String.format(
-        "Increasing %s by %d percent would exceed Long.MAX_LONG", this, percentage));
-    return ByteAmount.fromBytes(Math.round((double) asBytes() * factor));
-  }
-
-  public boolean greaterThan(ByteAmount other) {
-    return this.asBytes() > other.asBytes();
-  }
-
-  public boolean greaterOrEqual(ByteAmount other) {
-    return this.asBytes() >= other.asBytes();
-  }
-
-  public boolean lessThan(ByteAmount other) {
-    return this.asBytes() < other.asBytes();
-  }
-
-  public boolean lessOrEqual(ByteAmount other) {
-    return this.asBytes() <= other.asBytes();
+    checkArgument(value <= max,
+        String.format("Increasing %s by %d percent would exceed Long.MAX_LONG", this, percentage));
+    return ByteAmount.fromBytes(Math.round(value.doubleValue() * factor));
   }
 
   public ByteAmount max(ByteAmount other) {
@@ -221,42 +202,19 @@ public ByteAmount max(ByteAmount other) {
     }
   }
 
-  @Override
-  public int compareTo(ByteAmount other) {
-    return Long.compare(asBytes(), other.asBytes());
-  }
-
-  @Override
-  public boolean equals(Object other) {
-    if (this == other) {
-      return true;
-    }
-    if (other == null || getClass() != other.getClass()) {
-      return false;
-    }
-
-    ByteAmount that = (ByteAmount) other;
-    return bytes == that.bytes;
-  }
-
-  @Override
-  public int hashCode() {
-    return (int) (bytes ^ (bytes >>> 32));
-  }
-
   @Override
   public String toString() {
-    String value;
+    String str;
     if (asGigabytes() > 0) {
-      value = String.format("%.1f GB (%d bytes)", (double) asBytes() / GB, asBytes());
+      str = String.format("%.1f GB (%d bytes)", value.doubleValue() / GB, value);
     } else if (asMegabytes() > 0) {
-      value = String.format("%.1f MB (%d bytes)", (double) asBytes() / MB, asBytes());
+      str = String.format("%.1f MB (%d bytes)", value.doubleValue() / MB, value);
     } else if (asKilobytes() > 0) {
-      value = String.format("%.1f KB (%d bytes)", (double) asBytes() / KB, asBytes());
+      str = String.format("%.1f KB (%d bytes)", value.doubleValue() / KB, value);
     } else {
-      value = bytes + " bytes";
+      str = value + " bytes";
     }
-    return String.format("ByteAmount{%s}", value);
+    return String.format("ByteAmount{%s}", str);
   }
 
   private void checkArgument(boolean condition, String errorMessage) {

heron/common/src/java/org/apache/heron/common/basics/CPUShare.java

@@ -0,0 +1,72 @@
+/**
+ * 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.heron.common.basics;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public final class CPUShare extends ResourceMeasure<Double> {
+
+  private CPUShare(Double value) {
+    super(value);
+  }
+
+  public static CPUShare fromDouble(double value) {
+    return new CPUShare(value);
+  }
+
+  @Override
+  public CPUShare minus(ResourceMeasure<Double> other) {
+    return new CPUShare(value - other.value);
+  }
+
+  @Override
+  public CPUShare plus(ResourceMeasure<Double> other) {
+    return new CPUShare(value + other.value);
+  }
+
+  @Override
+  public CPUShare multiply(int factor) {
+    return new CPUShare(value * factor);
+  }
+
+  @Override
+  public CPUShare divide(int factor) {
+    return new CPUShare(value / factor);
+  }
+
+  @Override
+  public CPUShare increaseBy(int percentage) {
+    return new CPUShare(value * (1.0 + percentage / 100.0));
+  }
+
+  public static Map<String, CPUShare> convertDoubleMapToCpuShareMap(Map<String, Double> doubleMap) {
+    Map<String, CPUShare> retval = new HashMap<>();
+    for (Map.Entry<String, Double> entry : doubleMap.entrySet()) {
+      retval.put(entry.getKey(), new CPUShare(entry.getValue()));
+    }
+    return retval;
+  }
+
+  @Override
+  public String toString() {
+    return String.format("CPUShare{%.3f}", value);
+  }
+}