, 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();
}
}
diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/ExecutionGraph.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/ExecutionGraph.java
index bf13f4528..7e1006883 100644
--- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/ExecutionGraph.java
+++ b/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.
*
* 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;
diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionGraph.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionGraph.java
index bff6c7909..9d1f1bb3e 100644
--- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionGraph.java
+++ b/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 getExecutionJobVertexLices() {
+ public List getExecutionJobVertexList() {
return new ArrayList(executionJobVertexMap.values());
}
diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionVertex.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionVertex.java
index cee20610d..1c7768972 100644
--- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionVertex.java
+++ b/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 workerActor;
private List inputEdges = new ArrayList<>();
private List 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();
}
diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/resource/Container.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/resource/Container.java
index bb690060a..a1641cc9b 100644
--- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/resource/Container.java
+++ b/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 availableResource = new HashMap<>();
- private List slots = new ArrayList<>();
+ /**
+ * Container unique id fetched from raylet
+ */
+ private UniqueId nodeId;
+
+ /**
+ * Container available resources
+ */
+ private Map availableResources = new HashMap<>();
+
+ /**
+ * List of {@link ExecutionVertex} ids
+ * belong to the container.
+ */
+ private List 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 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 getAvailableResource() {
- return availableResource;
+ public Map getAvailableResources() {
+ return availableResources;
}
- public void setAvailableResource(Map availableResource) {
- this.availableResource = availableResource;
+ public int getCapacity() {
+ return capacity;
}
- public List 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 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 getExecutionVertexIds() {
+ return executionVertexIds;
+ }
+
+ private void decreaseResource(Map 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 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();
}
}
\ No newline at end of file
diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/resource/Resources.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/resource/Resources.java
index e97df9d6f..763bf8a15 100644
--- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/resource/Resources.java
+++ b/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 registerContainers = new ArrayList<>();
- /**
- * Mapping of allocated container to slots.
- */
- private Map> 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 getRegisteredContainers() {
+ return ImmutableList.copyOf(registerContainers);
}
- public List 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 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> getAllocatingMap() {
- return allocatingMap;
- }
-
- public void setAllocatingMap(
- Map> allocatingMap) {
- this.allocatingMap = allocatingMap;
- }
-
- public Map> getAllAvailableResource() {
- Map> availableResource = new HashMap<>();
- for (Container container : registerContainers) {
- availableResource.put(container.getNodeId(), container.getAvailableResource());
+ public void unRegisterContainer(List deletedUniqueIds) {
+ Iterator 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 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();
}
}
diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/resource/Slot.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/resource/Slot.java
deleted file mode 100644
index 189f9d12c..000000000
--- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/resource/Slot.java
+++ /dev/null
@@ -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 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 getExecutionVertexIds() {
- return executionVertexIds;
- }
-
- public void setExecutionVertexIds(List executionVertexIds) {
- this.executionVertexIds = executionVertexIds;
- }
-
- @Override
- public String toString() {
- return MoreObjects.toStringHelper(this)
- .add("id", id)
- .add("containerID", containerID)
- .add("actorCount", actorCount)
- .toString();
- }
-}
diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/ResourceAssignmentView.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/ResourceAssignmentView.java
new file mode 100644
index 000000000..8bb9a4b9d
--- /dev/null
+++ b/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> {
+
+ public static ResourceAssignmentView of(Map> assignmentView) {
+ ResourceAssignmentView view = new ResourceAssignmentView();
+ view.putAll(assignmentView);
+ return view;
+ }
+}
diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/ResourceManager.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/ResourceManager.java
index d427dede6..c885ffcea 100644
--- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/ResourceManager.java
+++ b/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 getRegisteredContainers();
+ ImmutableList getRegisteredContainers();
- /**
- * Allocate resource to actor.
- *
- * @param container Specify the container to be allocated.
- * @param requireResource Resource size to be requested.
- * @return Allocated resource.
- */
- Map allocateResource(final Container container,
- final Map 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 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();
}
\ No newline at end of file
diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/ResourceManagerImpl.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/ResourceManagerImpl.java
index 714cfdb2d..b704fbd61 100644
--- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/ResourceManagerImpl.java
+++ b/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 allocateResource(final Container container,
- final Map requireResource) {
- LOG.info("Start to allocate resource for actor with container: {}.", container);
-
- // allocate resource to actor
- Map resources = new HashMap<>();
- Map containResource = container.getAvailableResource();
- for (Map.Entry 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 containers,
+ ExecutionGraph executionGraph) {
+ return resourceAssignStrategy.assignResource(containers, executionGraph);
}
@Override
- public void deallocateResource(final Container container,
- final Map releaseResource) {
- LOG.info("Deallocating resource for container {}.", container);
-
- Map containResource = container.getAvailableResource();
- for (Map.Entry 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 getRegisteredContainers() {
- return new ArrayList<>(resources.getRegisterContainers());
- }
-
- @Override
- public SlotAssignStrategy getSlotAssignStrategy() {
- return slotAssignStrategy;
- }
-
- @Override
- public Resources getResources() {
- return this.resources;
+ public ImmutableList 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 latestNodeInfos = Ray.getRuntimeContext().getAllNodeInfo();
+ private void checkAndUpdateResource() {
+ //Get add&del nodes(node -> container)
+ Map latestNodeInfos = RayUtils.getAliveNodeInfoMap();
- List addNodes = latestNodeInfos.stream().filter(nodeInfo -> {
- for (Container container : resources.getRegisterContainers()) {
- if (container.getNodeId().equals(nodeInfo.nodeId)) {
- return false;
- }
- }
- return true;
- }).collect(Collectors.toList());
+ List addNodes = latestNodeInfos.keySet().stream()
+ .filter(this::isAddedNode).collect(Collectors.toList());
- List deleteContainers = resources.getRegisterContainers().stream()
- .filter(container -> {
- for (NodeInfo nodeInfo : latestNodeInfos) {
- if (nodeInfo.nodeId.equals(container.getNodeId())) {
- return false;
- }
- }
- return true;
- }).collect(Collectors.toList());
+ List 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 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 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);
+ }
+
}
diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/ViewBuilder.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/ViewBuilder.java
new file mode 100644
index 000000000..16eba4f45
--- /dev/null
+++ b/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 containers) {
+ Map> assignmentView = containers.stream()
+ .collect(java.util.stream.Collectors.toMap(Container::getId,
+ Container::getExecutionVertexIds));
+
+ return ResourceAssignmentView.of(assignmentView);
+ }
+}
diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/strategy/ResourceAssignStrategy.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/strategy/ResourceAssignStrategy.java
new file mode 100644
index 000000000..34eae669f
--- /dev/null
+++ b/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 containers, ExecutionGraph executionGraph);
+
+ /**
+ * Get container assign strategy name
+ */
+ String getName();
+
+}
diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/strategy/ResourceAssignStrategyFactory.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/strategy/ResourceAssignStrategyFactory.java
new file mode 100644
index 000000000..dddb1fed9
--- /dev/null
+++ b/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;
+ }
+}
diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/strategy/impl/PipelineFirstStrategy.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/strategy/impl/PipelineFirstStrategy.java
new file mode 100644
index 000000000..4fd93e5c7
--- /dev/null
+++ b/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:
+ *
+ * container_0
+ * |- source_1
+ * |- sink_1
+ * container_1
+ * |- source_2
+ * |- sink_2
+ *
+ */
+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 containers,
+ ExecutionGraph executionGraph) {
+
+ Map vertices = executionGraph.getExecutionJobVertexMap();
+ Map 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 exeVertices = jobVertex.getExecutionVertices();
+ // current job vertex assign finished
+ if (exeVertices.size() <= i) {
+ continue;
+ }
+ ExecutionVertex executionVertex = exeVertices.get(i);
+ Map 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 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 requiredResource,
+ List 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 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 availableResource = container.getAvailableResources();
+ for (Map.Entry 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 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 containers) {
+ return containers.get(currentContainerIndex);
+ }
+}
diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/ScheduleException.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/ScheduleException.java
new file mode 100644
index 000000000..d26bc16d7
--- /dev/null
+++ b/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);
+ }
+}
diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/strategy/SlotAssignStrategy.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/strategy/SlotAssignStrategy.java
deleted file mode 100644
index 65df0c94a..000000000
--- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/strategy/SlotAssignStrategy.java
+++ /dev/null
@@ -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 containers, int maxParallelism);
-
- /**
- * Allocate slot to container
- */
- void allocateSlot(final List containers, final int slotNumPerContainer);
-
- /**
- * Assign slot to execution vertex
- */
- Map> assignSlot(ExecutionGraph executionGraph);
-
- /**
- * Get slot assign strategy name
- */
- String getName();
-
- /**
- * Set resources.
- */
- void setResources(Resources resources);
-}
diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/strategy/SlotAssignStrategyFactory.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/strategy/SlotAssignStrategyFactory.java
deleted file mode 100644
index 6cef25a68..000000000
--- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/strategy/SlotAssignStrategyFactory.java
+++ /dev/null
@@ -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;
- }
-}
diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/strategy/impl/PipelineFirstStrategy.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/strategy/impl/PipelineFirstStrategy.java
deleted file mode 100644
index df89847fa..000000000
--- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/strategy/impl/PipelineFirstStrategy.java
+++ /dev/null
@@ -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:
- *
- * container_0
- * |- source_1
- * |- sink_1
- * container_1
- * |- source_2
- * |- sink_2
- *
- */
-public class PipelineFirstStrategy implements SlotAssignStrategy {
-
- public static final Logger LOG = LoggerFactory.getLogger(PipelineFirstStrategy.class);
-
- protected Resources resources;
-
- @Override
- public int getSlotNumPerContainer(List 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:
- *
- * container_0
- * |- slot_0
- * |- slot_2
- * |- slot_4
- * container_1
- * |- slot_1
- * |- slot_3
- * |- slot_5
- *
- */
- @Override
- public void allocateSlot(List 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 slots = c.getSlots();
- resources.getAllocatingMap().put(c.getContainerId(), slots);
- });
-
- LOG.info("Allocate slot result: {}.", resources.getAllocatingMap());
- }
-
- @Override
- public Map> assignSlot(ExecutionGraph executionGraph) {
- LOG.info("Container available resources: {}.", resources.getAllAvailableResource());
- Map vertices = executionGraph.getExecutionJobVertexMap();
- Map 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 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 targetSlots = targetContainer.getSlots();
- allocate(executionVertex, targetContainer, targetSlots.get(i % targetSlots.size()));
- }
- }
-
- return resources.getAllocatingMap();
- }
-
- private void checkResource(Map 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 requiredResource) {
- LOG.info("Check resource for container, index: {}.", resources.getCurrentContainerIndex());
-
- if (null == requiredResource) {
- return true;
- }
-
- Container currentContainer = resources.getRegisterContainers()
- .get(resources.getCurrentContainerIndex());
- List 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 availableResource = currentContainer.getAvailableResource();
- for (Map.Entry 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;
- }
-}
diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/RayUtils.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/RayUtils.java
new file mode 100644
index 000000000..5f688eabb
--- /dev/null
+++ b/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 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 getAliveNodeInfoMap() {
+ return getAllNodeInfo().stream()
+ .filter(nodeInfo -> nodeInfo.isAlive)
+ .collect(Collectors.toMap(nodeInfo -> nodeInfo.nodeId, nodeInfo -> nodeInfo));
+ }
+}
diff --git a/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/graph/ExecutionGraphTest.java b/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/graph/ExecutionGraphTest.java
index f76e91f61..c046bb177 100644
--- a/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/graph/ExecutionGraphTest.java
+++ b/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 executionJobVertices = executionGraph.getExecutionJobVertexLices();
+ List 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);
diff --git a/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/resourcemanager/ResourceManagerTest.java b/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/resourcemanager/ResourceManagerTest.java
index 0d1b9ece8..c751c413f 100644
--- a/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/resourcemanager/ResourceManagerTest.java
+++ b/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 nodeInfoMap = RayUtils.getAliveNodeInfoMap();
+ Assert.assertEquals(nodeInfoMap.size(), 5);
}
@Test
public void testApi() {
+ Ray.setAsyncContext(rayAsyncContext);
+
Map conf = new HashMap();
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 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 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> allocatingMap = slotAssignStrategy.assignSlot(executionGraph);
- Assert.assertEquals(allocatingMap.size(), 2);
-
- executionGraph.getAllAddedExecutionVertices().forEach(vertex -> {
- Container container = resourceManager.getResources()
- .getRegisterContainerByContainerId(vertex.getSlot().getContainerID());
- Map 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 containers = resourceManager.getRegisteredContainers();
+ Assert.assertEquals(containers.size(), 5);
}
}
diff --git a/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/schedule/strategy/PipelineFirstStrategyTest.java b/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/schedule/strategy/PipelineFirstStrategyTest.java
index de66df82f..40f3cc769 100644
--- a/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/schedule/strategy/PipelineFirstStrategyTest.java
+++ b/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 containers = new ArrayList<>();
- private JobGraph jobGraph;
- private ExecutionGraph executionGraph;
- private int maxParallelism;
+ private ResourceAssignStrategy strategy;
@BeforeClass
public void setUp() {
strategy = new PipelineFirstStrategy();
- Map conf = new HashMap<>();
- ResourceConfig resourceConfig = ConfigFactory.create(ResourceConfig.class, conf);
- Resources resources = new Resources(resourceConfig);
-
- Map 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 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 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> allocatingMap = strategy.assignSlot(executionGraph);
- for (Entry> containerSlotEntry : allocatingMap.entrySet()) {
- containerSlotEntry.getValue()
- .forEach(slot -> Assert.assertNotEquals(slot.getExecutionVertexIds().size(), 0));
- }
- }
}
diff --git a/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/util/Mockitools.java b/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/util/Mockitools.java
new file mode 100644
index 000000000..9a2f27589
--- /dev/null
+++ b/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 mockGetAllNodeInfo() {
+ List nodeInfos = new LinkedList<>();
+
+ for (int i = 1; i <= 5; i++) {
+ Map 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 mockGetNodeInfoMap(List nodeInfos) {
+ return nodeInfos.stream().filter(nodeInfo -> nodeInfo.isAlive).collect(
+ Collectors.toMap(nodeInfo -> nodeInfo.nodeId, nodeInfo -> nodeInfo));
+ }
+
+ private static NodeInfo mockNodeInfo(int i, Map 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);
+ }
+}