[streaming] Async changes for resourcemanager part (#7955)

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/config/master/ResourceConfig.java

@@ -70,7 +70,7 @@ public interface ResourceConfig extends Config {
    */
   @DefaultValue(value = "500")
   @Key(MAX_ACTOR_NUM_PER_CONTAINER)
-  int maxActorNumPerContainer();
+  int actorNumPerContainer();
 
   /**
    * The interval between detecting ray cluster nodes.

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/config/types/ResourceAssignStrategyType.java

@@ -1,21 +1,21 @@
 package io.ray.streaming.runtime.config.types;
 
-public enum SlotAssignStrategyType {
+public enum ResourceAssignStrategyType {
 
   /**
    * Resource scheduling strategy based on FF(First Fit) algorithm and pipeline.
    */
   PIPELINE_FIRST_STRATEGY("pipeline_first_strategy", 0);
 
-  private String value;
+  private String name;
   private int index;
 
-  SlotAssignStrategyType(String value, int index) {
-    this.value = value;
+  ResourceAssignStrategyType(String name, int index) {
+    this.name = name;
     this.index = index;
   }
 
-  public String getValue() {
-    return value;
+  public String getName() {
+    return name;
   }
 }

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/common/AbstractID.java

@@ -1,123 +1,34 @@
 package io.ray.streaming.runtime.core.common;
 
+import com.google.common.base.MoreObjects;
+import io.ray.streaming.runtime.core.resource.ContainerID;
 import java.io.Serializable;
-import java.util.Random;
+import java.util.UUID;
 
 /**
  * Streaming system unique identity base class.
- * For example, ${@link io.ray.streaming.runtime.core.resource.ContainerID }
+ * For example, ${@link ContainerID }
  */
-public class AbstractID implements Comparable<AbstractID>, Serializable {
-
-  private static final long serialVersionUID = 1L;
-  private static final Random RANDOM = new Random();
-  private static final int SIZE_OF_LONG = 8;
-  private static final int SIZE_OF_UPPER_PART = 8;
-  private static final int SIZE_OF_LOWER_PART = 8;
-
-  //lowerPart(long type) + upperPart(long type)
-  public static final int SIZE = SIZE_OF_UPPER_PART + SIZE_OF_LOWER_PART;
-
-  protected final long upperPart;
-  protected final long lowerPart;
-
-  private String toString;
-
-  public AbstractID(byte[] bytes) {
-    if (bytes == null || bytes.length != SIZE) {
-      throw new IllegalArgumentException("Argument bytes must by an array of " + SIZE + " bytes");
-    }
-
-    this.lowerPart = byteArrayToLong(bytes, 0);
-    this.upperPart = byteArrayToLong(bytes, SIZE_OF_LONG);
-  }
-
-  public AbstractID(long lowerPart, long upperPart) {
-    this.lowerPart = lowerPart;
-    this.upperPart = upperPart;
-  }
-
-  public AbstractID(AbstractID id) {
-    if (id == null) {
-      throw new IllegalArgumentException("Id must not be null.");
-    }
-    this.lowerPart = id.lowerPart;
-    this.upperPart = id.upperPart;
-  }
+public class AbstractID implements Serializable {
+  private UUID id;
 
   public AbstractID() {
-    this.lowerPart = RANDOM.nextLong();
-    this.upperPart = RANDOM.nextLong();
-  }
-
-  public long getLowerPart() {
-    return lowerPart;
-  }
-
-  public long getUpperPart() {
-    return upperPart;
-  }
-
-  public byte[] getBytes() {
-    byte[] bytes = new byte[SIZE];
-    longToByteArray(lowerPart, bytes, 0);
-    longToByteArray(upperPart, bytes, SIZE_OF_LONG);
-    return bytes;
+    this.id = UUID.randomUUID();
   }
 
   @Override
   public boolean equals(Object obj) {
-    if (obj == this) {
-      return true;
-    } else if (obj != null && obj.getClass() == getClass()) {
-      AbstractID that = (AbstractID) obj;
-      return that.lowerPart == this.lowerPart && that.upperPart == this.upperPart;
-    } else {
-      return false;
-    }
+    return id.equals(((AbstractID)obj).getId());
   }
 
-  @Override
-  public int hashCode() {
-    return ((int)  this.lowerPart) ^
-        ((int) (this.lowerPart >>> 32)) ^
-        ((int)  this.upperPart) ^
-        ((int) (this.upperPart >>> 32));
+  public UUID getId() {
+    return id;
   }
 
   @Override
   public String toString() {
-    if (this.toString == null) {
-      final byte[] ba = new byte[SIZE];
-      longToByteArray(this.lowerPart, ba, 0);
-      longToByteArray(this.upperPart, ba, SIZE_OF_LONG);
-      this.toString = new String(ba);
-    }
-
-    return this.toString;
-  }
-
-  @Override
-  public int compareTo(AbstractID abstractID) {
-    int diff1 = Long.compare(this.upperPart, abstractID.upperPart);
-    int diff2 = Long.compare(this.lowerPart, abstractID.lowerPart);
-    return diff1 == 0 ? diff2 : diff1;
-  }
-
-  private static long byteArrayToLong(byte[] begin, int offset) {
-    long longNum = 0;
-
-    for (int i = 0; i < SIZE_OF_LONG; ++i) {
-      longNum |= (begin[offset + SIZE_OF_LONG - 1 - i] & 0xffL) << (i << 3);
-    }
-
-    return longNum;
-  }
-
-  private static void longToByteArray(long longNum, byte[] byteArray, int offset) {
-    for (int i = 0; i < SIZE_OF_LONG; ++i) {
-      final int shift = i << 3; // i * 8
-      byteArray[offset + SIZE_OF_LONG - 1 - i] = (byte) ((longNum & (0xffL << shift)) >>> shift);
-    }
+    return MoreObjects.toStringHelper(this)
+        .add("id", id)
+        .toString();
   }
 }

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/ExecutionGraph.java

@@ -12,8 +12,7 @@ import java.util.stream.Collectors;
  * Physical execution graph.
  *
  * <p>Notice: Temporary implementation for now to keep functional. This will be changed to
- * {@link io.ray.streaming.runtime.core.graph.executiongraph.ExecutionGraph} later when
- * new stream task implementation is ready.
+ * {@link ExecutionGraph} later when new stream task implementation is ready.
  */
 public class ExecutionGraph implements Serializable {
   private long buildTime;

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionGraph.java

@@ -54,7 +54,7 @@ public class ExecutionGraph implements Serializable {
     return jobName;
   }
 
-  public List<ExecutionJobVertex> getExecutionJobVertexLices() {
+  public List<ExecutionJobVertex> getExecutionJobVertexList() {
     return new ArrayList<ExecutionJobVertex>(executionJobVertexMap.values());
   }
 

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionVertex.java

@@ -7,8 +7,8 @@ import io.ray.streaming.api.Language;
 import io.ray.streaming.jobgraph.VertexType;
 import io.ray.streaming.operator.StreamOperator;
 import io.ray.streaming.runtime.config.master.ResourceConfig;
+import io.ray.streaming.runtime.core.resource.ContainerID;
 import io.ray.streaming.runtime.core.resource.ResourceType;
-import io.ray.streaming.runtime.core.resource.Slot;
 import io.ray.streaming.runtime.worker.JobWorker;
 import java.io.Serializable;
 import java.util.ArrayList;
@@ -47,7 +47,7 @@ public class ExecutionVertex implements Serializable {
   private int vertexIndex;
 
   private ExecutionVertexState state = ExecutionVertexState.TO_ADD;
-  private Slot slot;
+  private ContainerID containerId;
   private RayActor<JobWorker> workerActor;
   private List<ExecutionEdge> inputEdges = new ArrayList<>();
   private List<ExecutionEdge> outputEdges = new ArrayList<>();
@@ -157,17 +157,17 @@ public class ExecutionVertex implements Serializable {
     return resources;
   }
 
-  public Slot getSlot() {
-    return slot;
+  public ContainerID getContainerId() {
+    return containerId;
   }
 
-  public void setSlot(Slot slot) {
-    this.slot = slot;
+  public void setContainerId(ContainerID containerId) {
+    this.containerId = containerId;
   }
 
-  public void setSlotIfNotExist(Slot slot) {
-    if (null == this.slot) {
-      this.slot = slot;
+  public void setContainerIfNotExist(ContainerID containerId) {
+    if (null == this.containerId) {
+      this.containerId = containerId;
     }
   }
 
@@ -182,6 +182,14 @@ public class ExecutionVertex implements Serializable {
     return resourceMap;
   }
 
+  @Override
+  public boolean equals(Object obj) {
+    if (obj instanceof ExecutionVertex) {
+      return this.id == ((ExecutionVertex)obj).getId();
+    }
+    return false;
+  }
+
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
@@ -189,7 +197,7 @@ public class ExecutionVertex implements Serializable {
         .add("name", getVertexName())
         .add("resources", resources)
         .add("state", state)
-        .add("slot", slot)
+        .add("containerId", containerId)
         .add("workerActor", workerActor)
         .toString();
   }

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/resource/Container.java

@@ -1,47 +1,96 @@
 package io.ray.streaming.runtime.core.resource;
 
 import com.google.common.base.MoreObjects;
+import com.google.common.base.Preconditions;
 import io.ray.api.id.UniqueId;
+import io.ray.api.runtimecontext.NodeInfo;
+import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionVertex;
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
- * Resource manager unit abstraction.
- * Container identifies the available resource(cpu,mem) and allocated slots.
+ * Container is physical resource abstraction. It identifies the available resources(cpu,mem,etc.)
+ * and allocated actors.
  */
 public class Container implements Serializable {
 
-  private ContainerID containerId;
-  private UniqueId nodeId;
+  private static final Logger LOG = LoggerFactory.getLogger(Container.class);
+
+  /**
+   * container id
+   */
+  private ContainerID id;
+
+  /**
+   * Container address
+   */
   private String address;
+
+  /**
+   * Container hostname
+   */
   private String hostname;
 
-  private Map<String, Double> availableResource = new HashMap<>();
-  private List<Slot> slots = new ArrayList<>();
+  /**
+   * Container unique id fetched from raylet
+   */
+  private UniqueId nodeId;
+
+  /**
+   * Container available resources
+   */
+  private Map<String, Double> availableResources = new HashMap<>();
+
+  /**
+   * List of {@link ExecutionVertex} ids
+   * belong to the container.
+   */
+  private List<Integer> executionVertexIds = new ArrayList<>();
+
+  /**
+   * Capacity is max actor number could be allocated in the container
+   */
+  private int capacity = 0;
 
   public Container() {
   }
 
-  public Container(UniqueId nodeId, String address, String hostname) {
-    this.containerId = new ContainerID();
-    this.nodeId = nodeId;
+  public Container(
+      String address,
+      UniqueId nodeId, String hostname,
+      Map<String, Double> availableResources) {
+
+    this.id = new ContainerID();
     this.address = address;
     this.hostname = hostname;
+    this.nodeId = nodeId;
+    this.availableResources = availableResources;
   }
 
-  public void setContainerId(ContainerID containerId) {
-    this.containerId = containerId;
+  public static Container from(NodeInfo nodeInfo) {
+    return new Container(
+        nodeInfo.nodeAddress,
+        nodeInfo.nodeId,
+        nodeInfo.nodeHostname,
+        nodeInfo.resources
+    );
   }
 
-  public ContainerID getContainerId() {
-    return containerId;
+  public ContainerID getId() {
+    return id;
+  }
+
+  public void setId(ContainerID id) {
+    this.id = id;
   }
 
   public String getName() {
-    return containerId.toString();
+    return id.toString();
   }
 
   public String getAddress() {
@@ -56,31 +105,92 @@ public class Container implements Serializable {
     return hostname;
   }
 
-  public Map<String, Double> getAvailableResource() {
-    return availableResource;
+  public Map<String, Double> getAvailableResources() {
+    return availableResources;
   }
 
-  public void setAvailableResource(Map<String, Double> availableResource) {
-    this.availableResource = availableResource;
+  public int getCapacity() {
+    return capacity;
   }
 
-  public List<Slot> getSlots() {
-    return slots;
+
+  public void updateCapacity(int capacity) {
+    LOG.info("Update container capacity, old value: {}, new value: {}.", this.capacity, capacity);
+    this.capacity = capacity;
   }
 
-  public void setSlots(List<Slot> slots) {
-    this.slots = slots;
+  public int getRemainingCapacity() {
+    return capacity - getAllocatedActorNum();
+  }
+
+  public int getAllocatedActorNum() {
+    return executionVertexIds.size();
+  }
+
+  public boolean isFull() {
+    return getAllocatedActorNum() >= capacity;
+  }
+
+  public boolean isEmpty() {
+    return getAllocatedActorNum() == 0;
+  }
+
+  public void allocateActor(ExecutionVertex vertex) {
+    LOG.info("Allocating vertex [{}] in container [{}].", vertex, this);
+
+    executionVertexIds.add(vertex.getId());
+    vertex.setContainerIfNotExist(this.getId());
+    // Binding dynamic resource
+    vertex.getResources().put(getName(), 1.0);
+    decreaseResource(vertex.getResources());
+  }
+
+  public void releaseActor(ExecutionVertex vertex) {
+    LOG.info("Release actor, vertex: {}, container: {}.", vertex, vertex.getContainerId());
+    if (executionVertexIds.contains(vertex.getId())) {
+      executionVertexIds.removeIf(id -> id == vertex.getId());
+      reclaimResource(vertex.getResources());
+    } else {
+      throw new RuntimeException(String.format("Current container [%s] not found vertex [%s].",
+          this, vertex.getJobVertexName()));
+    }
+  }
+
+  public List<Integer> getExecutionVertexIds() {
+    return executionVertexIds;
+  }
+
+  private void decreaseResource(Map<String, Double> allocatedResource) {
+    allocatedResource.forEach((k, v) -> {
+      Preconditions.checkArgument(this.availableResources.get(k) >= v,
+          String.format("Available resource %s not >= decreased resource %s",
+              this.availableResources.get(k), v));
+      Double newValue = this.availableResources.get(k) - v;
+      LOG.info("Decrease container {} resource [{}], from {} to {}.",
+          this.address, k, this.availableResources.get(k), newValue);
+      this.availableResources.put(k, newValue);
+    });
+  }
+
+  private void reclaimResource(Map<String, Double> allocatedResource) {
+    allocatedResource.forEach((k, v) -> {
+      Double newValue = this.availableResources.get(k) + v;
+      LOG.info("Reclaim container {} resource [{}], from {} to {}.",
+          this.address, k, this.availableResources.get(k), newValue);
+      this.availableResources.put(k, newValue);
+    });
   }
 
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
-        .add("containerId", containerId)
-        .add("nodeId", nodeId)
+        .add("id", id)
         .add("address", address)
         .add("hostname", hostname)
-        .add("availableResource", availableResource)
-        .add("slots", slots)
+        .add("nodeId", nodeId)
+        .add("availableResources", availableResources)
+        .add("executionVertexIds", executionVertexIds)
+        .add("capacity", capacity)
         .toString();
   }
 }

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/resource/Resources.java

@@ -1,134 +1,64 @@
 package io.ray.streaming.runtime.core.resource;
 
 import com.google.common.base.MoreObjects;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import io.ray.api.id.UniqueId;
-import io.ray.streaming.runtime.config.master.ResourceConfig;
 import java.io.Serializable;
 import java.util.ArrayList;
-import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
-import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Resource description of ResourceManager.
  */
 public class Resources implements Serializable {
 
+  private static final Logger LOG = LoggerFactory.getLogger(Resources.class);
+
   /**
    * Available containers registered to ResourceManager.
    */
   private List<Container> registerContainers = new ArrayList<>();
 
-  /**
-   * Mapping of allocated container to slots.
-   */
-  private Map<ContainerID, List<Slot>> allocatingMap = new HashMap<>(16);
+  public Resources() {
+  }
 
   /**
-   * Number of slots per container.
+   * Get registered containers, the container list is read-only.
+   * @return container list.
    */
-  private int slotNumPerContainer = 0;
-
-  /**
-   * Number of actors per container.
-   */
-  private int actorPerContainer = 0;
-
-  /**
-   * Number of actors that the current container has allocated.
-   */
-  private int currentContainerAllocatedActorNum = 0;
-
-  /**
-   * The container index currently being allocated.
-   */
-  private int currentContainerIndex = 0;
-
-  private int maxActorNumPerContainer;
-
-
-  public Resources(ResourceConfig resourceConfig) {
-    maxActorNumPerContainer = resourceConfig.maxActorNumPerContainer();
+  public ImmutableList<Container> getRegisteredContainers() {
+    return ImmutableList.copyOf(registerContainers);
   }
 
-  public List<Container> getRegisterContainers() {
-    return registerContainers;
+  public void registerContainer(Container container) {
+    LOG.info("Add container {} to registry list.", container);
+    this.registerContainers.add(container);
   }
 
-  public void setSlotNumPerContainer(int slotNumPerContainer) {
-    this.slotNumPerContainer = slotNumPerContainer;
-  }
-
-  public int getSlotNumPerContainer() {
-    return slotNumPerContainer;
-  }
-
-  public void setRegisterContainers(
-      List<Container> registerContainers) {
-    this.registerContainers = registerContainers;
-  }
-
-  public void setCurrentContainerIndex(int currentContainerIndex) {
-    this.currentContainerIndex = currentContainerIndex;
-  }
-
-  public int getCurrentContainerIndex() {
-    return currentContainerIndex;
-  }
-
-  public void setCurrentContainerAllocatedActorNum(int currentContainerAllocatedActorNum) {
-    this.currentContainerAllocatedActorNum = currentContainerAllocatedActorNum;
-  }
-
-  public int getCurrentContainerAllocatedActorNum() {
-    return currentContainerAllocatedActorNum;
-  }
-
-  public int getActorPerContainer() {
-    return actorPerContainer;
-  }
-
-  public void setActorPerContainer(int actorPerContainer) {
-    this.actorPerContainer = actorPerContainer;
-  }
-
-  public Container getRegisterContainerByContainerId(ContainerID containerID) {
-    return registerContainers.stream()
-        .filter(container -> container.getContainerId().equals(containerID))
-        .findFirst().get();
-  }
-
-  public int getMaxActorNumPerContainer() {
-    return maxActorNumPerContainer;
-  }
-
-  public Map<ContainerID, List<Slot>> getAllocatingMap() {
-    return allocatingMap;
-  }
-
-  public void setAllocatingMap(
-      Map<ContainerID, List<Slot>> allocatingMap) {
-    this.allocatingMap = allocatingMap;
-  }
-
-  public Map<UniqueId, Map<String, Double>> getAllAvailableResource() {
-    Map<UniqueId, Map<String, Double>> availableResource = new HashMap<>();
-    for (Container container : registerContainers) {
-      availableResource.put(container.getNodeId(), container.getAvailableResource());
+  public void unRegisterContainer(List<UniqueId> deletedUniqueIds) {
+    Iterator<Container> iter = registerContainers.iterator();
+    while (iter.hasNext()) {
+      Container deletedContainer = iter.next();
+      if (deletedUniqueIds.contains(deletedContainer.getNodeId())) {
+        LOG.info("Remove container {} from registry list.", deletedContainer);
+        iter.remove();
+      }
     }
-    return availableResource;
+  }
+
+  public ImmutableMap<UniqueId, Container> getRegisteredContainerMap() {
+    return ImmutableMap.copyOf(registerContainers.stream()
+        .collect(java.util.stream.Collectors.toMap(Container::getNodeId, c -> c)));
   }
 
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
         .add("registerContainers", registerContainers)
-        .add("allocatingMap", allocatingMap)
-        .add("slotNumPerContainer", slotNumPerContainer)
-        .add("actorPerContainer", actorPerContainer)
-        .add("currentContainerAllocatedActorNum", currentContainerAllocatedActorNum)
-        .add("currentContainerIndex", currentContainerIndex)
-        .add("maxActorNumPerContainer", maxActorNumPerContainer)
         .toString();
   }
 }

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/resource/Slot.java

@@ -1,54 +0,0 @@
-package io.ray.streaming.runtime.core.resource;
-
-import com.google.common.base.MoreObjects;
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
-
-public class Slot implements Serializable {
-  private int id;
-  /**
-   * The slot belongs to a container.
-   */
-  private ContainerID containerID;
-  private AtomicInteger actorCount = new AtomicInteger(0);
-  /**
-   * List of ExecutionVertex ids belong to the slot.
-   */
-  private List<Integer> executionVertexIds = new ArrayList<>();
-
-  public Slot(int id, ContainerID containerID) {
-    this.id = id;
-    this.containerID = containerID;
-  }
-
-  public int getId() {
-    return id;
-  }
-
-  public ContainerID getContainerID() {
-    return containerID;
-  }
-
-  public AtomicInteger getActorCount() {
-    return actorCount;
-  }
-
-  public List<Integer> getExecutionVertexIds() {
-    return executionVertexIds;
-  }
-
-  public void setExecutionVertexIds(List<Integer> executionVertexIds) {
-    this.executionVertexIds = executionVertexIds;
-  }
-
-  @Override
-  public String toString() {
-    return MoreObjects.toStringHelper(this)
-        .add("id", id)
-        .add("containerID", containerID)
-        .add("actorCount", actorCount)
-        .toString();
-  }
-}

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/ResourceAssignmentView.java

@@ -0,0 +1,18 @@
+package io.ray.streaming.runtime.master.resourcemanager;
+
+import io.ray.streaming.runtime.core.resource.ContainerID;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Cluster resource allocation view, used to statically view cluster resource information.
+ */
+public class ResourceAssignmentView extends HashMap<ContainerID, List<Integer>> {
+
+  public static ResourceAssignmentView of(Map<ContainerID, List<Integer>> assignmentView) {
+    ResourceAssignmentView view = new ResourceAssignmentView();
+    view.putAll(assignmentView);
+    return view;
+  }
+}

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/ResourceManager.java

@@ -1,53 +1,19 @@
 package io.ray.streaming.runtime.master.resourcemanager;
 
+import com.google.common.collect.ImmutableList;
 import io.ray.streaming.runtime.core.resource.Container;
-import io.ray.streaming.runtime.core.resource.Resources;
-import io.ray.streaming.runtime.master.scheduler.strategy.SlotAssignStrategy;
-import java.util.List;
-import java.util.Map;
+import io.ray.streaming.runtime.master.resourcemanager.strategy.ResourceAssignStrategy;
 
 /**
- * The resource manager is responsible for resource de-/allocation and monitoring ray cluster.
+ * ResourceManager(RM) is responsible for resource de-/allocation and monitoring ray cluster.
  */
-public interface ResourceManager {
+public interface ResourceManager extends ResourceAssignStrategy {
 
   /**
-   * Get all registered container as a list.
+   * Get registered containers, the container list is read-only.
    *
-   * @return A list of containers.
+   * @return the registered container list
    */
-  List<Container> getRegisteredContainers();
+  ImmutableList<Container> getRegisteredContainers();
 
-  /**
-   * Allocate resource to actor.
-   *
-   * @param container Specify the container to be allocated.
-   * @param requireResource Resource size to be requested.
-   * @return Allocated resource.
-   */
-  Map<String, Double> allocateResource(final Container container,
-      final Map<String, Double> requireResource);
-
-  /**
-   * Deallocate resource to actor.
-   *
-   * @param container  Specify the container to be deallocate.
-   * @param releaseResource Resource to be released.
-   */
-  void deallocateResource(final Container container,
-      final Map<String, Double> releaseResource);
-
-  /**
-   * Get the current slot-assign strategy from manager.
-   *
-   * @return Current slot-assign strategy.
-   */
-  SlotAssignStrategy getSlotAssignStrategy();
-
-  /**
-   * Get resources from manager.
-   *
-   * @return Current resources in manager.
-   */
-  Resources getResources();
 }

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/ResourceManagerImpl.java

@@ -1,21 +1,24 @@
 package io.ray.streaming.runtime.master.resourcemanager;
 
+import com.google.common.collect.ImmutableList;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import io.ray.api.Ray;
+import io.ray.api.id.UniqueId;
 import io.ray.api.runtimecontext.NodeInfo;
 import io.ray.streaming.runtime.config.StreamingMasterConfig;
 import io.ray.streaming.runtime.config.master.ResourceConfig;
-import io.ray.streaming.runtime.config.types.SlotAssignStrategyType;
+import io.ray.streaming.runtime.config.types.ResourceAssignStrategyType;
+import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionGraph;
 import io.ray.streaming.runtime.core.resource.Container;
 import io.ray.streaming.runtime.core.resource.Resources;
 import io.ray.streaming.runtime.master.JobRuntimeContext;
-import io.ray.streaming.runtime.master.scheduler.strategy.SlotAssignStrategy;
-import io.ray.streaming.runtime.master.scheduler.strategy.SlotAssignStrategyFactory;
-import java.util.ArrayList;
-import java.util.HashMap;
+import io.ray.streaming.runtime.master.resourcemanager.strategy.ResourceAssignStrategy;
+import io.ray.streaming.runtime.master.resourcemanager.strategy.ResourceAssignStrategyFactory;
+import io.ray.streaming.runtime.util.RayUtils;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import org.slf4j.Logger;
@@ -41,160 +44,153 @@ public class ResourceManagerImpl implements ResourceManager {
   /**
    * Slot assign strategy.
    */
-  private SlotAssignStrategy slotAssignStrategy;
+  private ResourceAssignStrategy resourceAssignStrategy;
 
   /**
    * Resource description information.
    */
   private final Resources resources;
 
-  private final ScheduledExecutorService scheduledExecutorService;
+  /**
+   * Customized actor number for each container
+   */
+  private int actorNumPerContainer;
+
+  /**
+   * Timing resource updating thread
+   */
+  private final ScheduledExecutorService resourceUpdater = new ScheduledThreadPoolExecutor(1,
+      new ThreadFactoryBuilder().setNameFormat("resource-update-thread").build());
 
   public ResourceManagerImpl(JobRuntimeContext runtimeContext) {
     this.runtimeContext = runtimeContext;
     StreamingMasterConfig masterConfig = runtimeContext.getConf().masterConfig;
 
     this.resourceConfig = masterConfig.resourceConfig;
-    this.resources = new Resources(resourceConfig);
+    this.resources = new Resources();
     LOG.info("ResourceManagerImpl begin init, conf is {}, resources are {}.",
         resourceConfig, resources);
 
-    SlotAssignStrategyType slotAssignStrategyType = SlotAssignStrategyType.PIPELINE_FIRST_STRATEGY;
+    // Init custom resource configurations
+    this.actorNumPerContainer = resourceConfig.actorNumPerContainer();
 
-    this.slotAssignStrategy = SlotAssignStrategyFactory.getStrategy(slotAssignStrategyType);
-    this.slotAssignStrategy.setResources(resources);
-    LOG.info("Slot assign strategy: {}.", slotAssignStrategy.getName());
+    ResourceAssignStrategyType resourceAssignStrategyType =
+        ResourceAssignStrategyType.PIPELINE_FIRST_STRATEGY;
+    this.resourceAssignStrategy = ResourceAssignStrategyFactory.getStrategy(
+      resourceAssignStrategyType);
+    LOG.info("Slot assign strategy: {}.", resourceAssignStrategy.getName());
 
-    this.scheduledExecutorService = Executors.newScheduledThreadPool(1);
-    long intervalSecond = resourceConfig.resourceCheckIntervalSecond();
-    this.scheduledExecutorService.scheduleAtFixedRate(
-        Ray.wrapRunnable(this::checkAndUpdateResources), 0, intervalSecond, TimeUnit.SECONDS);
+    //Init resource
+    initResource();
+
+    checkAndUpdateResourcePeriodically();
 
     LOG.info("ResourceManagerImpl init success.");
   }
 
   @Override
-  public Map<String, Double> allocateResource(final Container container,
-      final Map<String, Double> requireResource) {
-    LOG.info("Start to allocate resource for actor with container: {}.", container);
-
-    // allocate resource to actor
-    Map<String, Double> resources = new HashMap<>();
-    Map<String, Double> containResource = container.getAvailableResource();
-    for (Map.Entry<String, Double> entry : containResource.entrySet()) {
-      if (requireResource.containsKey(entry.getKey())) {
-        double availableResource = entry.getValue() - requireResource.get(entry.getKey());
-        entry.setValue(availableResource);
-        resources.put(entry.getKey(), requireResource.get(entry.getKey()));
-      }
-    }
-
-    LOG.info("Allocate resource: {} to container {}.", requireResource, container);
-    return resources;
+  public ResourceAssignmentView assignResource(List<Container> containers,
+      ExecutionGraph executionGraph) {
+    return resourceAssignStrategy.assignResource(containers, executionGraph);
   }
 
   @Override
-  public void deallocateResource(final Container container,
-      final Map<String, Double> releaseResource) {
-    LOG.info("Deallocating resource for container {}.", container);
-
-    Map<String, Double> containResource = container.getAvailableResource();
-    for (Map.Entry<String, Double> entry : containResource.entrySet()) {
-      if (releaseResource.containsKey(entry.getKey())) {
-        double availableResource = entry.getValue() + releaseResource.get(entry.getKey());
-        LOG.info("Release source {}:{}", entry.getKey(), releaseResource.get(entry.getKey()));
-        entry.setValue(availableResource);
-      }
-    }
-
-    LOG.info("Deallocated resource for container {} success.", container);
+  public String getName() {
+    return resourceAssignStrategy.getName();
   }
 
   @Override
-  public List<Container> getRegisteredContainers() {
-    return new ArrayList<>(resources.getRegisterContainers());
-  }
-
-  @Override
-  public SlotAssignStrategy getSlotAssignStrategy() {
-    return slotAssignStrategy;
-  }
-
-  @Override
-  public Resources getResources() {
-    return this.resources;
+  public ImmutableList<Container> getRegisteredContainers() {
+    LOG.info("Current resource detail: {}.", resources.toString());
+    return resources.getRegisteredContainers();
   }
 
   /**
    * Check the status of ray cluster node and update the internal resource information of
    * streaming system.
    */
-  private void checkAndUpdateResources() {
-    // get all started nodes
-    List<NodeInfo> latestNodeInfos = Ray.getRuntimeContext().getAllNodeInfo();
+  private void checkAndUpdateResource() {
+    //Get add&del nodes(node -> container)
+    Map<UniqueId, NodeInfo> latestNodeInfos = RayUtils.getAliveNodeInfoMap();
 
-    List<NodeInfo> addNodes = latestNodeInfos.stream().filter(nodeInfo -> {
-      for (Container container : resources.getRegisterContainers()) {
-        if (container.getNodeId().equals(nodeInfo.nodeId)) {
-          return false;
-        }
-      }
-      return true;
-    }).collect(Collectors.toList());
+    List<UniqueId> addNodes = latestNodeInfos.keySet().stream()
+        .filter(this::isAddedNode).collect(Collectors.toList());
 
-    List<Container> deleteContainers = resources.getRegisterContainers().stream()
-        .filter(container -> {
-          for (NodeInfo nodeInfo : latestNodeInfos) {
-            if (nodeInfo.nodeId.equals(container.getNodeId())) {
-              return false;
-            }
-          }
-          return true;
-        }).collect(Collectors.toList());
+    List<UniqueId> deleteNodes = resources.getRegisteredContainerMap().keySet().stream()
+        .filter(nodeId -> !latestNodeInfos.containsKey(nodeId)).collect(Collectors.toList());
     LOG.info("Latest node infos: {}, current containers: {}, add nodes: {}, delete nodes: {}.",
-        latestNodeInfos, resources.getRegisterContainers(), addNodes, deleteContainers);
+        latestNodeInfos, resources.getRegisteredContainers(), addNodes, deleteNodes);
 
-    //Register new nodes.
-    if (!addNodes.isEmpty()) {
-      for (NodeInfo node : addNodes) {
-        registerContainer(node);
-      }
+    if (!addNodes.isEmpty() || !deleteNodes.isEmpty()) {
+      LOG.info("Latest node infos from GCS: {}", latestNodeInfos);
+      LOG.info("Resource details: {}.", resources.toString());
+      LOG.info("Get add nodes info: {}, del nodes info: {}.", addNodes, deleteNodes);
+
+      // unregister containers
+      unregisterDeletedContainer(deleteNodes);
+
+      // register containers
+      registerNewContainers(addNodes.stream().map(latestNodeInfos::get)
+          .collect(Collectors.toList()));
     }
-    //Clear deleted nodes
-    if (!deleteContainers.isEmpty()) {
-      for (Container container : deleteContainers) {
-        unregisterContainer(container);
-      }
+  }
+
+  private void registerNewContainers(List<NodeInfo> nodeInfos) {
+    LOG.info("Start to register containers. new add node infos are: {}.", nodeInfos);
+
+    if (nodeInfos == null || nodeInfos.isEmpty()) {
+      LOG.info("NodeInfos is null or empty, skip registry.");
+      return;
+    }
+
+    for (NodeInfo nodeInfo : nodeInfos) {
+      registerContainer(nodeInfo);
     }
   }
 
   private void registerContainer(final NodeInfo nodeInfo) {
     LOG.info("Register container {}.", nodeInfo);
 
-    Container container =
-        new Container(nodeInfo.nodeId, nodeInfo.nodeAddress, nodeInfo.nodeHostname);
-    container.setAvailableResource(nodeInfo.resources);
+    Container container = Container.from(nodeInfo);
+
+    // failover case: container has already allocated actors
+    double availableCapacity = actorNumPerContainer - container.getAllocatedActorNum();
+
 
     //Create ray resource.
-    Ray.setResource(container.getNodeId(),
-        container.getName(),
-        resources.getMaxActorNumPerContainer());
+    Ray.setResource(container.getNodeId(), container.getName(), availableCapacity);
     //Mark container is already registered.
-    Ray.setResource(container.getNodeId(),
-        CONTAINER_ENGAGED_KEY, 1);
+    Ray.setResource(container.getNodeId(), CONTAINER_ENGAGED_KEY, 1);
+
+    // update container's available dynamic resources
+    container.getAvailableResources()
+      .put(container.getName(), availableCapacity);
 
     // update register container list
-    resources.getRegisterContainers().add(container);
+    resources.registerContainer(container);
   }
 
-  private void unregisterContainer(final Container container) {
-    LOG.info("Unregister container {}.", container);
-
-    // delete resource with capacity to 0
-    Ray.setResource(container.getNodeId(), container.getName(), 0);
-    Ray.setResource(container.getNodeId(), CONTAINER_ENGAGED_KEY, 0);
-
-    // remove from container map
-    resources.getRegisterContainers().remove(container);
+  private void unregisterDeletedContainer(List<UniqueId> deletedIds) {
+    LOG.info("Unregister container, deleted node ids are: {}.", deletedIds);
+    if (null == deletedIds || deletedIds.isEmpty()) {
+      return;
+    }
+    resources.unRegisterContainer(deletedIds);
   }
+
+  private void initResource() {
+    LOG.info("Init resource.");
+    checkAndUpdateResource();
+  }
+
+  private void checkAndUpdateResourcePeriodically() {
+    long intervalSecond = resourceConfig.resourceCheckIntervalSecond();
+    this.resourceUpdater.scheduleAtFixedRate(
+        Ray.wrapRunnable(this::checkAndUpdateResource), 0, intervalSecond, TimeUnit.SECONDS);
+  }
+
+  private boolean isAddedNode(UniqueId uniqueId) {
+    return !resources.getRegisteredContainerMap().containsKey(uniqueId);
+  }
+
 }

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/ViewBuilder.java

@@ -0,0 +1,24 @@
+package io.ray.streaming.runtime.master.resourcemanager;
+
+import io.ray.streaming.runtime.core.resource.Container;
+import io.ray.streaming.runtime.core.resource.ContainerID;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * ViewBuilder describes current cluster's resource allocation detail information
+ */
+public class ViewBuilder {
+
+  // Default constructor for serialization.
+  public ViewBuilder() {
+  }
+
+  public static ResourceAssignmentView buildResourceAssignmentView(List<Container> containers) {
+    Map<ContainerID, List<Integer>> assignmentView = containers.stream()
+        .collect(java.util.stream.Collectors.toMap(Container::getId,
+            Container::getExecutionVertexIds));
+
+    return ResourceAssignmentView.of(assignmentView);
+  }
+}

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/strategy/ResourceAssignStrategy.java

@@ -0,0 +1,29 @@
+package io.ray.streaming.runtime.master.resourcemanager.strategy;
+
+import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionGraph;
+import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionVertex;
+import io.ray.streaming.runtime.core.resource.Container;
+import io.ray.streaming.runtime.master.resourcemanager.ResourceAssignmentView;
+import java.util.List;
+
+/**
+ * The ResourceAssignStrategy responsible assign {@link Container} to {@link ExecutionVertex}.
+ */
+public interface ResourceAssignStrategy {
+
+  /**
+   * Assign {@link Container} for
+   * {@link ExecutionVertex}
+   *
+   * @param containers registered container
+   * @param executionGraph execution graph
+   * @return allocating view
+   */
+  ResourceAssignmentView assignResource(List<Container> containers, ExecutionGraph executionGraph);
+
+  /**
+   * Get container assign strategy name
+   */
+  String getName();
+
+}

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/strategy/ResourceAssignStrategyFactory.java

@@ -0,0 +1,24 @@
+package io.ray.streaming.runtime.master.resourcemanager.strategy;
+
+import io.ray.streaming.runtime.config.types.ResourceAssignStrategyType;
+import io.ray.streaming.runtime.master.resourcemanager.strategy.impl.PipelineFirstStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ResourceAssignStrategyFactory {
+
+  private static final Logger LOG = LoggerFactory.getLogger(ResourceAssignStrategyFactory.class);
+
+  public static ResourceAssignStrategy getStrategy(final ResourceAssignStrategyType type) {
+    ResourceAssignStrategy strategy = null;
+    LOG.info("Slot assign strategy is: {}.", type);
+    switch (type) {
+      case PIPELINE_FIRST_STRATEGY:
+        strategy = new PipelineFirstStrategy();
+        break;
+      default:
+        throw new RuntimeException("strategy config error, no impl found for " + strategy);
+    }
+    return strategy;
+  }
+}

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/strategy/impl/PipelineFirstStrategy.java

@@ -0,0 +1,207 @@
+package io.ray.streaming.runtime.master.resourcemanager.strategy.impl;
+
+import io.ray.streaming.runtime.config.types.ResourceAssignStrategyType;
+import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionGraph;
+import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionJobVertex;
+import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionVertex;
+import io.ray.streaming.runtime.core.resource.Container;
+import io.ray.streaming.runtime.core.resource.ResourceType;
+import io.ray.streaming.runtime.master.resourcemanager.ResourceAssignmentView;
+import io.ray.streaming.runtime.master.resourcemanager.ViewBuilder;
+import io.ray.streaming.runtime.master.resourcemanager.strategy.ResourceAssignStrategy;
+import io.ray.streaming.runtime.master.scheduler.ScheduleException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Based on Ray dynamic resource function, resource details(by ray gcs get) and
+ * execution logic diagram, PipelineFirstStrategy provides a actor scheduling
+ * strategies to make the cluster load balanced and controllable scheduling.
+ * Assume that we have 2 containers and have a DAG graph composed of a source node with parallelism
+ * of 2 and a sink node with parallelism of 2, the structure will be like:
+ * <pre>
+ *   container_0
+ *             |- source_1
+ *             |- sink_1
+ *   container_1
+ *             |- source_2
+ *             |- sink_2
+ * </pre>
+ */
+public class PipelineFirstStrategy implements ResourceAssignStrategy {
+
+  public static final Logger LOG = LoggerFactory.getLogger(PipelineFirstStrategy.class);
+
+  private int currentContainerIndex = 0;
+
+  /**
+   *  Assign resource to each execution vertex in the given execution graph.
+   *
+   * @param containers registered containers
+   * @param executionGraph execution graph
+   * @return allocating map, key is container ID, value is list of vertextId, and contains vertices
+   */
+  @Override
+  public ResourceAssignmentView assignResource(
+      List<Container> containers,
+      ExecutionGraph executionGraph) {
+
+    Map<Integer, ExecutionJobVertex> vertices = executionGraph.getExecutionJobVertexMap();
+    Map<Integer, Integer> vertexRemainingNum = new HashMap<>();
+
+    vertices.forEach((k, v) -> {
+      int size = v.getExecutionVertices().size();
+      vertexRemainingNum.put(k, size);
+    });
+    int totalExecutionVerticesNum = vertexRemainingNum.values().stream()
+        .mapToInt(Integer::intValue)
+        .sum();
+    int containerNum = containers.size();
+    int capacityPerContainer = Math.max(totalExecutionVerticesNum / containerNum, 1);
+
+    updateContainerCapacity(containers, capacityPerContainer);
+
+    int enlargeCapacityThreshold = 0;
+    boolean enlarged = false;
+    if (capacityPerContainer * containerNum < totalExecutionVerticesNum) {
+      enlargeCapacityThreshold = capacityPerContainer * containerNum;
+      LOG.info("Need to enlarge capacity per container, threshold: {}.", enlargeCapacityThreshold);
+    }
+    LOG.info("Total execution vertices num: {}, container num: {}, capacity per container: {}.",
+        totalExecutionVerticesNum, containerNum, capacityPerContainer);
+
+    int maxParallelism = executionGraph.getMaxParallelism();
+
+    int allocatedVertexCount = 0;
+    for (int i = 0; i < maxParallelism; i++) {
+      for (ExecutionJobVertex jobVertex : vertices.values()) {
+        List<ExecutionVertex> exeVertices = jobVertex.getExecutionVertices();
+        // current job vertex assign finished
+        if (exeVertices.size() <= i) {
+          continue;
+        }
+        ExecutionVertex executionVertex = exeVertices.get(i);
+        Map<String, Double> requiredResource = executionVertex.getResources();
+        if (requiredResource.containsKey(ResourceType.CPU.getValue())) {
+          LOG.info("Required resource contain {} value : {}, no limitation by default.",
+              ResourceType.CPU, requiredResource.get(ResourceType.CPU.getValue()));
+          requiredResource.remove(ResourceType.CPU.getValue());
+        }
+
+        Container targetContainer = findMatchedContainer(requiredResource, containers);
+
+        targetContainer.allocateActor(executionVertex);
+        allocatedVertexCount++;
+        // Once allocatedVertexCount reaches threshold, we should enlarge capacity
+        if (!enlarged && enlargeCapacityThreshold > 0
+            && allocatedVertexCount >= enlargeCapacityThreshold) {
+          updateContainerCapacity(containers, capacityPerContainer + 1);
+          enlarged = true;
+          LOG.info("Enlarge capacity per container to: {}.", containers.get(0).getCapacity());
+        }
+      }
+    }
+
+    ResourceAssignmentView allocatingView = ViewBuilder.buildResourceAssignmentView(containers);
+    LOG.info("Assigning resource finished, allocating map: {}.", allocatingView);
+    return allocatingView;
+  }
+
+  @Override
+  public String getName() {
+    return ResourceAssignStrategyType.PIPELINE_FIRST_STRATEGY.getName();
+  }
+
+  /**
+   * Update container capacity. eg: we have 89 actors and 8 containers, capacity will be 11 when
+   * initialing, and will be increased to 12 when allocating actor#89, just for load balancing.
+   */
+  private void updateContainerCapacity(List<Container> containers, int capacity) {
+    containers.forEach(c -> c.updateCapacity(capacity));
+  }
+
+  /**
+   * Find a container which matches required resource
+   * @param requiredResource required resource
+   * @param containers registered containers
+   * @return container that matches the required resource
+   */
+  private Container findMatchedContainer(
+      Map<String, Double> requiredResource,
+      List<Container> containers) {
+
+    LOG.info("Check resource, required: {}.", requiredResource);
+
+    int checkedNum = 0;
+    // if current container does not have enough resource, go to the next one (loop)
+    while (!hasEnoughResource(requiredResource, getCurrentContainer(containers))) {
+      checkedNum++;
+      forwardToNextContainer(containers);
+      if (checkedNum >= containers.size()) {
+        throw new ScheduleException(
+            String.format("No enough resource left, required resource: %s, available resource: %s.",
+                requiredResource, containers));
+      }
+    }
+    return getCurrentContainer(containers);
+  }
+
+  /**
+   * Check if current container has enough resource
+   * @param requiredResource required resource
+   * @param container container
+   * @return true if matches, false else
+   */
+  private boolean hasEnoughResource(Map<String, Double> requiredResource, Container container) {
+    LOG.info("Check resource for index: {}, container: {}", currentContainerIndex, container);
+
+    if (null == requiredResource) {
+      return true;
+    }
+
+    if (container.isFull()) {
+      LOG.info("Container {} is full.", container);
+      return false;
+    }
+
+    Map<String, Double> availableResource = container.getAvailableResources();
+    for (Map.Entry<String, Double> entry : requiredResource.entrySet()) {
+      if (availableResource.containsKey(entry.getKey())) {
+        if (availableResource.get(entry.getKey()) < entry.getValue()) {
+          LOG.warn("No enough resource for container {}. required: {}, available: {}.",
+              container.getAddress(), requiredResource, availableResource);
+          return false;
+        }
+      } else {
+        LOG.warn("No enough resource for container {}. required: {}, available: {}.",
+            container.getAddress(), requiredResource, availableResource);
+        return false;
+      }
+    }
+
+    return true;
+  }
+
+  /**
+   * Forward to next container
+   *
+   * @param containers registered container list
+   * @return next container in the list
+   */
+  private Container forwardToNextContainer(List<Container> containers) {
+    this.currentContainerIndex = (this.currentContainerIndex + 1) % containers.size();
+    return getCurrentContainer(containers);
+  }
+
+  /**
+   * Get current container
+   * @param containers registered container
+   * @return current container to allocate actor
+   */
+  private Container getCurrentContainer(List<Container> containers) {
+    return containers.get(currentContainerIndex);
+  }
+}

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/ScheduleException.java

@@ -0,0 +1,25 @@
+package io.ray.streaming.runtime.master.scheduler;
+
+public class ScheduleException extends RuntimeException {
+
+  public ScheduleException() {
+    super();
+  }
+
+  public ScheduleException(String message) {
+    super(message);
+  }
+
+  public ScheduleException(String message, Throwable cause) {
+    super(message, cause);
+  }
+
+  public ScheduleException(Throwable cause) {
+    super(cause);
+  }
+
+  protected ScheduleException(String message, Throwable cause, boolean enableSuppression,
+      boolean writableStackTrace) {
+    super(message, cause, enableSuppression, writableStackTrace);
+  }
+}

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/strategy/SlotAssignStrategy.java

@@ -1,41 +0,0 @@
-package io.ray.streaming.runtime.master.scheduler.strategy;
-
-import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionGraph;
-import io.ray.streaming.runtime.core.resource.Container;
-import io.ray.streaming.runtime.core.resource.ContainerID;
-import io.ray.streaming.runtime.core.resource.Resources;
-import io.ray.streaming.runtime.core.resource.Slot;
-import java.util.List;
-import java.util.Map;
-
-/**
- * The SlotAssignStrategy managers a set of slots. When a container is
- * registered to ResourceManager, slots are assigned to it.
- */
-public interface SlotAssignStrategy {
-
-  /**
-   * Calculate slot number per container and set to resources.
-   */
-  int getSlotNumPerContainer(List<Container> containers, int maxParallelism);
-
-  /**
-   * Allocate slot to container
-   */
-  void allocateSlot(final List<Container> containers, final int slotNumPerContainer);
-
-  /**
-   * Assign slot to execution vertex
-   */
-  Map<ContainerID, List<Slot>> assignSlot(ExecutionGraph executionGraph);
-
-  /**
-   * Get slot assign strategy name
-   */
-  String getName();
-
-  /**
-   * Set resources.
-   */
-  void setResources(Resources resources);
-}

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/strategy/SlotAssignStrategyFactory.java

@@ -1,24 +0,0 @@
-package io.ray.streaming.runtime.master.scheduler.strategy;
-
-import io.ray.streaming.runtime.config.types.SlotAssignStrategyType;
-import io.ray.streaming.runtime.master.scheduler.strategy.impl.PipelineFirstStrategy;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class SlotAssignStrategyFactory {
-
-  private static final Logger LOG = LoggerFactory.getLogger(SlotAssignStrategyFactory.class);
-
-  public static SlotAssignStrategy getStrategy(final SlotAssignStrategyType type) {
-    SlotAssignStrategy strategy = null;
-    LOG.info("Slot assign strategy is: {}.", type);
-    switch (type) {
-      case PIPELINE_FIRST_STRATEGY:
-        strategy = new PipelineFirstStrategy();
-        break;
-      default:
-        throw new RuntimeException("strategy config error, no impl found for " + strategy);
-    }
-    return strategy;
-  }
-}

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/strategy/impl/PipelineFirstStrategy.java

@@ -1,205 +0,0 @@
-package io.ray.streaming.runtime.master.scheduler.strategy.impl;
-
-import com.google.common.base.Preconditions;
-import io.ray.streaming.runtime.config.types.SlotAssignStrategyType;
-import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionGraph;
-import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionJobVertex;
-import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionVertex;
-import io.ray.streaming.runtime.core.resource.Container;
-import io.ray.streaming.runtime.core.resource.ContainerID;
-import io.ray.streaming.runtime.core.resource.Resources;
-import io.ray.streaming.runtime.core.resource.Slot;
-import io.ray.streaming.runtime.master.scheduler.strategy.SlotAssignStrategy;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Based on Ray dynamic resource function, resource details(by ray gcs get) and
- * execution logic diagram, PipelineFirstStrategy provides a actor scheduling
- * strategies to make the cluster load balanced and controllable scheduling.
- * Assume that we have 2 containers and have a DAG graph composed of a source node with parallelism
- * of 2 and a sink node with parallelism of 2, the structure will be like:
- * <pre>
- *   container_0
- *             |- source_1
- *             |- sink_1
- *   container_1
- *             |- source_2
- *             |- sink_2
- * </pre>
- */
-public class PipelineFirstStrategy implements SlotAssignStrategy {
-
-  public static final Logger LOG = LoggerFactory.getLogger(PipelineFirstStrategy.class);
-
-  protected Resources resources;
-
-  @Override
-  public int getSlotNumPerContainer(List<Container> containers, int maxParallelism) {
-    LOG.info("max parallelism: {}, container size: {}.", maxParallelism, containers.size());
-    int slotNumPerContainer =
-        (int) Math.ceil(Math.max(maxParallelism, containers.size()) * 1.0 / containers.size());
-    LOG.info("slot num per container: {}.", slotNumPerContainer);
-    return slotNumPerContainer;
-  }
-
-  /**
-   * Allocate slot to target container, assume that we have 2 containers and max parallelism is 5,
-   * the structure will be like:
-   * <pre>
-   * container_0
-   *           |- slot_0
-   *           |- slot_2
-   *           |- slot_4
-   * container_1
-   *           |- slot_1
-   *           |- slot_3
-   *           |- slot_5
-   * </pre>
-   */
-  @Override
-  public void allocateSlot(List<Container> containers,
-      int slotNumPerContainer) {
-    int maxSlotSize = containers.size() * slotNumPerContainer;
-    LOG.info("Allocate slot, maxSlotSize: {}.", maxSlotSize);
-
-    for (int slotId = 0; slotId < maxSlotSize; ++slotId) {
-      Container targetContainer = containers.get(slotId % containers.size());
-      Slot slot = new Slot(slotId, targetContainer.getContainerId());
-      targetContainer.getSlots().add(slot);
-    }
-
-    // update new added containers' allocating map
-    containers.forEach(c -> {
-      List<Slot> slots = c.getSlots();
-      resources.getAllocatingMap().put(c.getContainerId(), slots);
-    });
-
-    LOG.info("Allocate slot result: {}.", resources.getAllocatingMap());
-  }
-
-  @Override
-  public Map<ContainerID, List<Slot>> assignSlot(ExecutionGraph executionGraph) {
-    LOG.info("Container available resources: {}.", resources.getAllAvailableResource());
-    Map<Integer, ExecutionJobVertex> vertices = executionGraph.getExecutionJobVertexMap();
-    Map<Integer, Integer> vertexRemainingNum = new HashMap<>();
-    vertices.forEach((k, v) -> {
-      int size = v.getExecutionVertices().size();
-      vertexRemainingNum.put(k, size);
-    });
-    int totalExecutionVerticesNum = vertexRemainingNum.values().stream()
-        .mapToInt(Integer::intValue)
-        .sum();
-    int containerNum = resources.getRegisterContainers().size();
-    resources.setActorPerContainer((int) Math
-        .ceil(totalExecutionVerticesNum * 1.0 / containerNum));
-    LOG.info("Total execution vertices num: {}, container num: {}, capacity per container: {}.",
-        totalExecutionVerticesNum, containerNum, resources.getActorPerContainer());
-
-    int maxParallelism = executionGraph.getMaxParallelism();
-
-    for (int i = 0; i < maxParallelism; i++) {
-      for (ExecutionJobVertex executionJobVertex : vertices.values()) {
-        List<ExecutionVertex> exeVertices = executionJobVertex.getExecutionVertices();
-        // current job vertex assign finished
-        if (exeVertices.size() <= i) {
-          continue;
-        }
-
-        ExecutionVertex executionVertex = exeVertices.get(i);
-
-        //check current container has enough resources.
-        checkResource(executionVertex.getResources());
-
-        Container targetContainer = resources.getRegisterContainers()
-            .get(resources.getCurrentContainerIndex());
-        List<Slot> targetSlots = targetContainer.getSlots();
-        allocate(executionVertex, targetContainer, targetSlots.get(i % targetSlots.size()));
-      }
-    }
-
-    return resources.getAllocatingMap();
-  }
-
-  private void checkResource(Map<String, Double> requiredResource) {
-    int checkedNum = 0;
-    // if current container does not have enough resource, go to the next one (loop)
-    while (!hasEnoughResource(requiredResource)) {
-      checkedNum++;
-      resources.setCurrentContainerIndex((resources.getCurrentContainerIndex() + 1) %
-          resources.getRegisterContainers().size());
-
-      Preconditions.checkArgument(checkedNum < resources.getRegisterContainers().size(),
-          "No enough resource left, required resource: {}, available resource: {}.",
-          requiredResource, resources.getAllAvailableResource());
-      resources.setCurrentContainerAllocatedActorNum(0);
-    }
-  }
-
-  private boolean hasEnoughResource(Map<String, Double> requiredResource) {
-    LOG.info("Check resource for container, index: {}.", resources.getCurrentContainerIndex());
-
-    if (null == requiredResource) {
-      return true;
-    }
-
-    Container currentContainer = resources.getRegisterContainers()
-        .get(resources.getCurrentContainerIndex());
-    List<Slot> slotActors = resources.getAllocatingMap().get(currentContainer.getContainerId());
-    if (slotActors != null && slotActors.size() > 0) {
-      long allocatedActorNum = slotActors.stream()
-          .map(Slot::getExecutionVertexIds)
-          .mapToLong(List::size)
-          .sum();
-      if (allocatedActorNum  >= resources.getActorPerContainer()) {
-        LOG.info("Container remaining capacity is 0. used: {}, total: {}.", allocatedActorNum,
-            resources.getActorPerContainer());
-        return false;
-      }
-    }
-
-    Map<String, Double> availableResource = currentContainer.getAvailableResource();
-    for (Map.Entry<String, Double> entry : requiredResource.entrySet()) {
-      if (availableResource.containsKey(entry.getKey())) {
-        if (availableResource.get(entry.getKey()) < entry.getValue()) {
-          LOG.warn("No enough resource for container {}. required: {}, available: {}.",
-              currentContainer.getAddress(), requiredResource, availableResource);
-          return false;
-        }
-      }
-    }
-    return true;
-  }
-
-  private void allocate(ExecutionVertex vertex, Container container, Slot slot) {
-    // set slot for execution vertex
-    LOG.info("Set slot {} to vertex {}.", slot, vertex);
-    vertex.setSlotIfNotExist(slot);
-
-    Slot useSlot = resources.getAllocatingMap().get(container.getContainerId())
-        .stream().filter(s -> s.getId() == slot.getId()).findFirst().get();
-    useSlot.getExecutionVertexIds().add(vertex.getId());
-
-    // current container reaches capacity limitation, go to the next one.
-    resources.setCurrentContainerAllocatedActorNum(
-        resources.getCurrentContainerAllocatedActorNum() + 1);
-    if (resources.getCurrentContainerAllocatedActorNum() >= resources.getActorPerContainer()) {
-      resources.setCurrentContainerIndex(
-          (resources.getCurrentContainerIndex() + 1) % resources.getRegisterContainers().size());
-      resources.setCurrentContainerAllocatedActorNum(0);
-    }
-  }
-
-  @Override
-  public String getName() {
-    return SlotAssignStrategyType.PIPELINE_FIRST_STRATEGY.getValue();
-  }
-
-  @Override
-  public void setResources(Resources resources) {
-    this.resources = resources;
-  }
-}

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/RayUtils.java

@@ -0,0 +1,34 @@
+package io.ray.streaming.runtime.util;
+
+import io.ray.api.Ray;
+import io.ray.api.id.UniqueId;
+import io.ray.api.runtimecontext.NodeInfo;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * RayUtils is the utility class to access ray runtime api.
+ */
+public class RayUtils {
+
+  /**
+   * Get all node info from GCS
+   *
+   * @return node info list
+   */
+  public static List<NodeInfo> getAllNodeInfo() {
+    return Ray.getRuntimeContext().getAllNodeInfo();
+  }
+
+  /**
+   * Get all alive node info map
+   *
+   * @return node info map, key is unique node id , value is node info
+   */
+  public static Map<UniqueId, NodeInfo> getAliveNodeInfoMap() {
+    return getAllNodeInfo().stream()
+      .filter(nodeInfo -> nodeInfo.isAlive)
+      .collect(Collectors.toMap(nodeInfo -> nodeInfo.nodeId, nodeInfo -> nodeInfo));
+  }
+}

streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/graph/ExecutionGraphTest.java

@@ -36,7 +36,7 @@ public class ExecutionGraphTest extends BaseUnitTest {
     GraphManager graphManager = new GraphManagerImpl(new JobRuntimeContext(streamingConfig));
     JobGraph jobGraph = buildJobGraph();
     ExecutionGraph executionGraph = buildExecutionGraph(graphManager, jobGraph);
-    List<ExecutionJobVertex> executionJobVertices = executionGraph.getExecutionJobVertexLices();
+    List<ExecutionJobVertex> executionJobVertices = executionGraph.getExecutionJobVertexList();
 
     Assert.assertEquals(executionJobVertices.size(), jobGraph.getJobVertexList().size());
 
@@ -88,6 +88,7 @@ public class ExecutionGraphTest extends BaseUnitTest {
     jobConfig.put("key1", "value1");
     jobConfig.put("key2", "value2");
     jobConfig.put(ResourceConfig.TASK_RESOURCE_CPU, "2.0");
+    jobConfig.put(ResourceConfig.TASK_RESOURCE_MEM, "2.0");
 
     JobGraphBuilder jobGraphBuilder = new JobGraphBuilder(
         Lists.newArrayList(streamSink), "test", jobConfig);

streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/resourcemanager/ResourceManagerTest.java

@@ -1,99 +1,80 @@
 package io.ray.streaming.runtime.resourcemanager;
 
-import io.ray.api.Ray;
-import io.ray.streaming.jobgraph.JobGraph;
-import io.ray.streaming.runtime.BaseUnitTest;
-import io.ray.streaming.runtime.TestHelper;
-import io.ray.streaming.runtime.config.StreamingConfig;
-import io.ray.streaming.runtime.config.global.CommonConfig;
-import io.ray.streaming.runtime.config.master.ResourceConfig;
-import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionGraph;
-import io.ray.streaming.runtime.core.resource.Container;
-import io.ray.streaming.runtime.core.resource.ContainerID;
-import io.ray.streaming.runtime.core.resource.ResourceType;
-import io.ray.streaming.runtime.core.resource.Slot;
-import io.ray.streaming.runtime.graph.ExecutionGraphTest;
-import io.ray.streaming.runtime.master.JobRuntimeContext;
-import io.ray.streaming.runtime.master.graphmanager.GraphManager;
-import io.ray.streaming.runtime.master.graphmanager.GraphManagerImpl;
-import io.ray.streaming.runtime.master.resourcemanager.ResourceManager;
-import io.ray.streaming.runtime.master.resourcemanager.ResourceManagerImpl;
-import io.ray.streaming.runtime.master.scheduler.strategy.SlotAssignStrategy;
-import io.ray.streaming.runtime.master.scheduler.strategy.impl.PipelineFirstStrategy;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import org.aeonbits.owner.util.Collections;
+import org.mockito.MockitoAnnotations;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import io.ray.api.Ray;
+import io.ray.api.id.UniqueId;
+import io.ray.api.runtimecontext.NodeInfo;
+import io.ray.streaming.runtime.config.StreamingConfig;
+import io.ray.streaming.runtime.config.global.CommonConfig;
+import io.ray.streaming.runtime.master.resourcemanager.ResourceManager;
+import io.ray.streaming.runtime.master.resourcemanager.ResourceManagerImpl;
+import io.ray.streaming.runtime.core.resource.Container;
+import io.ray.streaming.runtime.master.JobRuntimeContext;
+import io.ray.streaming.runtime.util.Mockitools;
+import io.ray.streaming.runtime.util.RayUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testng.Assert;
+import org.testng.IObjectFactory;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.ObjectFactory;
 import org.testng.annotations.Test;
 
-public class ResourceManagerTest extends BaseUnitTest {
+@PrepareForTest(RayUtils.class)
+@PowerMockIgnore({"org.slf4j.*", "javax.xml.*"})
+public class ResourceManagerTest {
 
   private static final Logger LOG = LoggerFactory.getLogger(ResourceManagerTest.class);
 
+  private Object rayAsyncContext;
+
+  @ObjectFactory
+  public IObjectFactory getObjectFactory() {
+    return new org.powermock.modules.testng.PowerMockObjectFactory();
+  }
+
   @org.testng.annotations.BeforeClass
   public void setUp() {
-    // ray init
-    Ray.init();
-    TestHelper.setUTFlag();
+    LOG.warn("Do set up");
+    MockitoAnnotations.initMocks(this);
   }
 
   @org.testng.annotations.AfterClass
   public void tearDown() {
-    TestHelper.clearUTFlag();
+    LOG.warn("Do tear down");
+  }
+
+  @BeforeMethod
+  public void mockGscApi() {
+    // ray init
+    Ray.init();
+    rayAsyncContext = Ray.getAsyncContext();
+    Mockitools.mockGscApi();
+  }
+
+  @Test
+  public void testGcsMockedApi() {
+    Map<UniqueId, NodeInfo> nodeInfoMap = RayUtils.getAliveNodeInfoMap();
+    Assert.assertEquals(nodeInfoMap.size(), 5);
   }
 
   @Test
   public void testApi() {
+    Ray.setAsyncContext(rayAsyncContext);
+
     Map<String, String> conf = new HashMap<String, String>();
     conf.put(CommonConfig.JOB_NAME, "testApi");
-    conf.put(ResourceConfig.TASK_RESOURCE_CPU_LIMIT_ENABLE, "true");
-    conf.put(ResourceConfig.TASK_RESOURCE_MEM_LIMIT_ENABLE, "true");
-    conf.put(ResourceConfig.TASK_RESOURCE_MEM, "10");
-    conf.put(ResourceConfig.TASK_RESOURCE_CPU, "2");
     StreamingConfig config = new StreamingConfig(conf);
     JobRuntimeContext jobRuntimeContext = new JobRuntimeContext(config);
     ResourceManager resourceManager = new ResourceManagerImpl(jobRuntimeContext);
 
-    SlotAssignStrategy slotAssignStrategy = resourceManager.getSlotAssignStrategy();
-    Assert.assertTrue(slotAssignStrategy instanceof PipelineFirstStrategy);
-
-    Map<String, Double> containerResource = new HashMap<>();
-    containerResource.put(ResourceType.CPU.name(), 16.0);
-    containerResource.put(ResourceType.MEM.name(), 128.0);
-    Container container1 = new Container(null, "testAddress1", "testHostName1");
-    container1.setAvailableResource(containerResource);
-    Container container2 = new Container(null, "testAddress2", "testHostName2");
-    container2.setAvailableResource(new HashMap<>(containerResource));
-    List<Container> containers = Collections.list(container1, container2);
-    resourceManager.getResources().getRegisterContainers().addAll(containers);
-    Assert.assertEquals(resourceManager.getRegisteredContainers().size(), 2);
-
-    //build ExecutionGraph
-    GraphManager graphManager = new GraphManagerImpl(new JobRuntimeContext(config));
-    JobGraph jobGraph = ExecutionGraphTest.buildJobGraph();
-    ExecutionGraph executionGraph = ExecutionGraphTest.buildExecutionGraph(graphManager, jobGraph);
-
-    int slotNumPerContainer = slotAssignStrategy.getSlotNumPerContainer(containers, executionGraph
-    .getMaxParallelism());
-    Assert.assertEquals(slotNumPerContainer, 1);
-
-    slotAssignStrategy.allocateSlot(containers, slotNumPerContainer);
-
-    Map<ContainerID, List<Slot>> allocatingMap = slotAssignStrategy.assignSlot(executionGraph);
-    Assert.assertEquals(allocatingMap.size(), 2);
-
-    executionGraph.getAllAddedExecutionVertices().forEach(vertex -> {
-      Container container = resourceManager.getResources()
-          .getRegisterContainerByContainerId(vertex.getSlot().getContainerID());
-      Map<String, Double> resource = resourceManager.allocateResource(container, vertex.getResources());
-      Assert.assertNotNull(resource);
-    });
-    Assert.assertEquals(container1.getAvailableResource().get(ResourceType.CPU.name()), 14.0);
-    Assert.assertEquals(container2.getAvailableResource().get(ResourceType.CPU.name()), 14.0);
-    Assert.assertEquals(container1.getAvailableResource().get(ResourceType.MEM.name()), 126.0);
-    Assert.assertEquals(container2.getAvailableResource().get(ResourceType.MEM.name()), 126.0);
+    // test register container
+    List<Container> containers = resourceManager.getRegisteredContainers();
+    Assert.assertEquals(containers.size(), 5);
   }
 }

streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/schedule/strategy/PipelineFirstStrategyTest.java

@@ -4,28 +4,25 @@ import io.ray.api.id.UniqueId;
 import io.ray.streaming.jobgraph.JobGraph;
 import io.ray.streaming.runtime.BaseUnitTest;
 import io.ray.streaming.runtime.config.StreamingConfig;
-import io.ray.streaming.runtime.config.master.ResourceConfig;
+import io.ray.streaming.runtime.config.types.ResourceAssignStrategyType;
 import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionGraph;
 import io.ray.streaming.runtime.core.resource.Container;
-import io.ray.streaming.runtime.core.resource.ContainerID;
 import io.ray.streaming.runtime.core.resource.ResourceType;
-import io.ray.streaming.runtime.core.resource.Resources;
-import io.ray.streaming.runtime.core.resource.Slot;
 import io.ray.streaming.runtime.graph.ExecutionGraphTest;
 import io.ray.streaming.runtime.master.JobRuntimeContext;
 import io.ray.streaming.runtime.master.graphmanager.GraphManager;
 import io.ray.streaming.runtime.master.graphmanager.GraphManagerImpl;
-import io.ray.streaming.runtime.master.scheduler.strategy.SlotAssignStrategy;
-import io.ray.streaming.runtime.master.scheduler.strategy.impl.PipelineFirstStrategy;
+import io.ray.streaming.runtime.master.resourcemanager.ResourceAssignmentView;
+import io.ray.streaming.runtime.master.resourcemanager.strategy.ResourceAssignStrategy;
+import io.ray.streaming.runtime.master.resourcemanager.strategy.impl.PipelineFirstStrategy;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Map.Entry;
-import org.aeonbits.owner.ConfigFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
@@ -33,63 +30,50 @@ public class PipelineFirstStrategyTest extends BaseUnitTest {
 
   private Logger LOG = LoggerFactory.getLogger(PipelineFirstStrategyTest.class);
 
-  private SlotAssignStrategy strategy;
   private List<Container> containers = new ArrayList<>();
-  private JobGraph jobGraph;
-  private ExecutionGraph executionGraph;
-  private int maxParallelism;
+  private ResourceAssignStrategy strategy;
 
   @BeforeClass
   public void setUp() {
     strategy = new PipelineFirstStrategy();
-    Map<String, String> conf = new HashMap<>();
-    ResourceConfig resourceConfig = ConfigFactory.create(ResourceConfig.class, conf);
-    Resources resources = new Resources(resourceConfig);
-
-    Map<String, Double> containerResource = new HashMap<>();
-    containerResource.put(ResourceType.CPU.name(), 16.0);
-    containerResource.put(ResourceType.MEM.name(), 128.0);
     for (int i = 0; i < 2; ++i) {
       UniqueId uniqueId = UniqueId.randomId();
-      Container container = new Container(uniqueId, "1.1.1." + i,  "localhost" + i);
-      container.setAvailableResource(containerResource);
+      Map<String, Double> resource = new HashMap<>();
+      resource.put(ResourceType.CPU.getValue(), 4.0);
+      resource.put(ResourceType.MEM.getValue(), 4.0);
+      Container container = new Container("1.1.1." + i, uniqueId, "localhost" + i, resource);
+      container.getAvailableResources().put(container.getName(), 500.0);
       containers.add(container);
-      resources.getRegisterContainers().add(container);
     }
-    strategy.setResources(resources);
+  }
 
-    //build ExecutionGraph
+  @AfterMethod
+  public void tearDown() {
+    reset();
+  }
+
+  private void reset() {
+    containers = new ArrayList<>();
+    strategy = null;
+  }
+
+  @Test
+  public void testStrategyName() {
+    Assert
+        .assertEquals(ResourceAssignStrategyType.PIPELINE_FIRST_STRATEGY.getName(), strategy.getName());
+  }
+
+  @Test
+  public void testAssignResource() {
+    strategy = new PipelineFirstStrategy();
     Map<String, String> jobConf = new HashMap<>();
     StreamingConfig streamingConfig = new StreamingConfig(jobConf);
     GraphManager graphManager = new GraphManagerImpl(new JobRuntimeContext(streamingConfig));
-    jobGraph = ExecutionGraphTest.buildJobGraph();
-    executionGraph = ExecutionGraphTest.buildExecutionGraph(graphManager, jobGraph);
-    maxParallelism = executionGraph.getMaxParallelism();
+    JobGraph jobGraph = ExecutionGraphTest.buildJobGraph();
+    ExecutionGraph executionGraph = ExecutionGraphTest.buildExecutionGraph(graphManager, jobGraph);
+    ResourceAssignmentView assignmentView = strategy.assignResource(containers, executionGraph);
+    Assert.assertNotNull(assignmentView);
   }
 
-  @Test
-  public int testSlotNumPerContainer() {
-    int slotNumPerContainer = strategy.getSlotNumPerContainer(containers, maxParallelism);
-    Assert.assertEquals(slotNumPerContainer,
-        (int) Math.ceil(Math.max(maxParallelism, containers.size()) * 1.0 / containers.size()));
-    return slotNumPerContainer;
-  }
 
-  @Test
-  public void testAllocateSlot() {
-    int slotNumPerContainer = testSlotNumPerContainer();
-    strategy.allocateSlot(containers, slotNumPerContainer);
-    for (Container container : containers) {
-      Assert.assertEquals(container.getSlots().size(), slotNumPerContainer);
-    }
-  }
-
-  @Test
-  public void testAssignSlot() {
-    Map<ContainerID, List<Slot>> allocatingMap = strategy.assignSlot(executionGraph);
-    for (Entry<ContainerID, List<Slot>> containerSlotEntry : allocatingMap.entrySet()) {
-      containerSlotEntry.getValue()
-          .forEach(slot -> Assert.assertNotEquals(slot.getExecutionVertexIds().size(), 0));
-    }
-  }
 }

streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/util/Mockitools.java

@@ -0,0 +1,82 @@
+package io.ray.streaming.runtime.util;
+
+import io.ray.api.id.UniqueId;
+import io.ray.api.runtimecontext.NodeInfo;
+import io.ray.streaming.runtime.core.resource.ResourceType;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.powermock.api.mockito.PowerMockito;
+
+/**
+ * Mockitools is a tool based on powermock and mokito to mock external service api
+ */
+public class Mockitools {
+
+  /**
+   * Mock GCS get node info api
+   */
+  public static void mockGscApi() {
+    PowerMockito.mockStatic(RayUtils.class);
+    PowerMockito.when(RayUtils.getAliveNodeInfoMap())
+      .thenReturn(mockGetNodeInfoMap(mockGetAllNodeInfo()));
+  }
+
+  /**
+   * Mock get all node info from GCS
+   * @return
+   */
+  public static List<NodeInfo> mockGetAllNodeInfo() {
+    List<NodeInfo> nodeInfos = new LinkedList<>();
+
+    for (int i = 1; i <= 5; i++) {
+      Map<String, Double> resources = new HashMap<>();
+      resources.put("MEM", 16.0);
+      switch (i) {
+        case 1:
+          resources.put(ResourceType.CPU.getValue(), 3.0);
+          break;
+        case 2:
+        case 3:
+        case 4:
+          resources.put(ResourceType.CPU.getValue(), 4.0);
+          break;
+        case 5:
+          resources.put(ResourceType.CPU.getValue(), 2.0);
+          break;
+      }
+
+      nodeInfos.add(mockNodeInfo(i, resources));
+    }
+    return nodeInfos;
+  }
+
+  /**
+   * Mock get node info map
+   * @param nodeInfos all node infos fetched from GCS
+   * @return node info map, key is node unique id, value is node info
+   */
+  public static Map<UniqueId, NodeInfo> mockGetNodeInfoMap(List<NodeInfo> nodeInfos) {
+    return nodeInfos.stream().filter(nodeInfo -> nodeInfo.isAlive).collect(
+      Collectors.toMap(nodeInfo -> nodeInfo.nodeId, nodeInfo -> nodeInfo));
+  }
+
+  private static NodeInfo mockNodeInfo(int i, Map<String, Double> resources) {
+    return new NodeInfo(
+      createNodeId(i),
+      "localhost" + i,
+      "localhost" + i,
+      true,
+      resources);
+  }
+
+  private static UniqueId createNodeId(int id) {
+    byte[] nodeIdBytes = new byte[UniqueId.LENGTH];
+    for (int byteIndex = 0; byteIndex < UniqueId.LENGTH; ++byteIndex) {
+      nodeIdBytes[byteIndex] = String.valueOf(id).getBytes()[0];
+    }
+    return new UniqueId(nodeIdBytes);
+  }
+}