You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2015/05/27 17:01:19 UTC
[1/7] flink git commit: [FLINK-1952] [jobmanager] Rework and fix slot
sharing scheduler
Repository: flink
Updated Branches:
refs/heads/master 8e35febcc -> 96bbdbd70
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SharedSlotsTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SharedSlotsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SharedSlotsTest.java
new file mode 100644
index 0000000..e49f19f
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SharedSlotsTest.java
@@ -0,0 +1,666 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.instance;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobmanager.scheduler.CoLocationConstraint;
+import org.apache.flink.runtime.jobmanager.scheduler.CoLocationGroup;
+import org.apache.flink.runtime.jobmanager.scheduler.Locality;
+import org.apache.flink.runtime.jobmanager.scheduler.SchedulerTestUtils;
+import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
+import org.apache.flink.util.AbstractID;
+
+import org.junit.Test;
+
+import java.util.Collections;
+
+import static org.junit.Assert.*;
+
+/**
+ * Tests for the allocation, properties, and release of shared slots.
+ */
+public class SharedSlotsTest {
+
+ @Test
+ public void allocateAndReleaseEmptySlot() {
+ try {
+ JobID jobId = new JobID();
+ JobVertexID vertexId = new JobVertexID();
+
+ SlotSharingGroup sharingGroup = new SlotSharingGroup(vertexId);
+ SlotSharingGroupAssignment assignment = sharingGroup.getTaskAssignment();
+
+ assertEquals(0, assignment.getNumberOfSlots());
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vertexId));
+
+ Instance instance = SchedulerTestUtils.getRandomInstance(2);
+
+ assertEquals(2, instance.getTotalNumberOfSlots());
+ assertEquals(0, instance.getNumberOfAllocatedSlots());
+ assertEquals(2, instance.getNumberOfAvailableSlots());
+
+ // allocate a shared slot
+ SharedSlot slot = instance.allocateSharedSlot(jobId, assignment);
+ assertEquals(2, instance.getTotalNumberOfSlots());
+ assertEquals(1, instance.getNumberOfAllocatedSlots());
+ assertEquals(1, instance.getNumberOfAvailableSlots());
+
+ // check that the new slot is fresh
+ assertTrue(slot.isAlive());
+ assertFalse(slot.isCanceled());
+ assertFalse(slot.isReleased());
+ assertEquals(0, slot.getNumberLeaves());
+ assertFalse(slot.hasChildren());
+ assertTrue(slot.isRootAndEmpty());
+ assertNotNull(slot.toString());
+ assertTrue(slot.getSubSlots().isEmpty());
+ assertEquals(0, slot.getSlotNumber());
+ assertEquals(0, slot.getRootSlotNumber());
+
+ // release the slot immediately.
+ slot.releaseSlot();
+
+ assertTrue(slot.isCanceled());
+ assertTrue(slot.isReleased());
+
+ // the slot sharing group and instance should not
+ assertEquals(2, instance.getTotalNumberOfSlots());
+ assertEquals(0, instance.getNumberOfAllocatedSlots());
+ assertEquals(2, instance.getNumberOfAvailableSlots());
+
+ assertEquals(0, assignment.getNumberOfSlots());
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vertexId));
+
+ // we should not be able to allocate any children from this released slot
+ assertNull(slot.allocateSharedSlot(new AbstractID()));
+ assertNull(slot.allocateSubSlot(new AbstractID()));
+
+ // we cannot add this slot to the assignment group
+ assertNull(assignment.addSharedSlotAndAllocateSubSlot(slot, Locality.NON_LOCAL, vertexId));
+ assertEquals(0, assignment.getNumberOfSlots());
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void allocateSimpleSlotsAndReleaseFromRoot() {
+ try {
+ JobID jobId = new JobID();
+ JobVertexID vid1 = new JobVertexID();
+ JobVertexID vid2 = new JobVertexID();
+ JobVertexID vid3 = new JobVertexID();
+ JobVertexID vid4 = new JobVertexID();
+
+ SlotSharingGroup sharingGroup = new SlotSharingGroup(vid1, vid2, vid3, vid4);
+ SlotSharingGroupAssignment assignment = sharingGroup.getTaskAssignment();
+
+ Instance instance = SchedulerTestUtils.getRandomInstance(1);
+
+ // allocate a shared slot
+ SharedSlot sharedSlot = instance.allocateSharedSlot(jobId, assignment);
+
+ // allocate a series of sub slots
+
+ SimpleSlot sub1 = assignment.addSharedSlotAndAllocateSubSlot(sharedSlot, Locality.LOCAL, vid1);
+ assertNotNull(sub1);
+
+ assertNull(sub1.getExecutedVertex());
+ assertEquals(Locality.LOCAL, sub1.getLocality());
+ assertEquals(1, sub1.getNumberLeaves());
+ assertEquals(vid1, sub1.getGroupID());
+ assertEquals(instance, sub1.getInstance());
+ assertEquals(jobId, sub1.getJobID());
+ assertEquals(sharedSlot, sub1.getParent());
+ assertEquals(sharedSlot, sub1.getRoot());
+ assertEquals(0, sub1.getRootSlotNumber());
+ assertEquals(0, sub1.getSlotNumber());
+
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid1));
+ assertEquals(1, assignment.getNumberOfAvailableSlotsForGroup(vid2));
+ assertEquals(1, assignment.getNumberOfAvailableSlotsForGroup(vid3));
+ assertEquals(1, assignment.getNumberOfAvailableSlotsForGroup(vid4));
+
+ SimpleSlot sub2 = assignment.getSlotForTask(vid2, Collections.<Instance>emptySet());
+ assertNotNull(sub2);
+
+ assertNull(sub2.getExecutedVertex());
+ assertEquals(Locality.UNCONSTRAINED, sub2.getLocality());
+ assertEquals(1, sub2.getNumberLeaves());
+ assertEquals(vid2, sub2.getGroupID());
+ assertEquals(instance, sub2.getInstance());
+ assertEquals(jobId, sub2.getJobID());
+ assertEquals(sharedSlot, sub2.getParent());
+ assertEquals(sharedSlot, sub2.getRoot());
+ assertEquals(0, sub2.getRootSlotNumber());
+ assertEquals(1, sub2.getSlotNumber());
+
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid1));
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid2));
+ assertEquals(1, assignment.getNumberOfAvailableSlotsForGroup(vid3));
+ assertEquals(1, assignment.getNumberOfAvailableSlotsForGroup(vid4));
+
+ SimpleSlot sub3 = assignment.getSlotForTask(vid3, Collections.singleton(instance));
+ assertNotNull(sub3);
+
+ assertNull(sub3.getExecutedVertex());
+ assertEquals(Locality.LOCAL, sub3.getLocality());
+ assertEquals(1, sub3.getNumberLeaves());
+ assertEquals(vid3, sub3.getGroupID());
+ assertEquals(instance, sub3.getInstance());
+ assertEquals(jobId, sub3.getJobID());
+ assertEquals(sharedSlot, sub3.getParent());
+ assertEquals(sharedSlot, sub3.getRoot());
+ assertEquals(0, sub3.getRootSlotNumber());
+ assertEquals(2, sub3.getSlotNumber());
+
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid1));
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid2));
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid3));
+ assertEquals(1, assignment.getNumberOfAvailableSlotsForGroup(vid4));
+
+ SimpleSlot sub4 = assignment.getSlotForTask(vid4,
+ Collections.singleton(SchedulerTestUtils.getRandomInstance(1)));
+ assertNotNull(sub4);
+
+ assertNull(sub4.getExecutedVertex());
+ assertEquals(Locality.NON_LOCAL, sub4.getLocality());
+ assertEquals(1, sub4.getNumberLeaves());
+ assertEquals(vid4, sub4.getGroupID());
+ assertEquals(instance, sub4.getInstance());
+ assertEquals(jobId, sub4.getJobID());
+ assertEquals(sharedSlot, sub4.getParent());
+ assertEquals(sharedSlot, sub4.getRoot());
+ assertEquals(0, sub4.getRootSlotNumber());
+ assertEquals(3, sub4.getSlotNumber());
+
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid1));
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid2));
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid3));
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid4));
+
+ // release from the root.
+ sharedSlot.releaseSlot();
+
+ assertTrue(sharedSlot.isReleased());
+ assertTrue(sub1.isReleased());
+ assertTrue(sub2.isReleased());
+ assertTrue(sub3.isReleased());
+ assertTrue(sub4.isReleased());
+
+ assertEquals(0, sharedSlot.getNumberLeaves());
+ assertFalse(sharedSlot.hasChildren());
+
+ assertEquals(1, instance.getNumberOfAvailableSlots());
+ assertEquals(0, assignment.getNumberOfSlots());
+
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid1));
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid2));
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid3));
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid4));
+
+ assertNull(sharedSlot.allocateSharedSlot(new AbstractID()));
+ assertNull(sharedSlot.allocateSubSlot(new AbstractID()));
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void allocateSimpleSlotsAndReleaseFromleaves() {
+ try {
+ JobID jobId = new JobID();
+ JobVertexID vid1 = new JobVertexID();
+ JobVertexID vid2 = new JobVertexID();
+ JobVertexID vid3 = new JobVertexID();
+
+ SlotSharingGroup sharingGroup = new SlotSharingGroup(vid1, vid2, vid3);
+ SlotSharingGroupAssignment assignment = sharingGroup.getTaskAssignment();
+
+ Instance instance = SchedulerTestUtils.getRandomInstance(1);
+
+ // allocate a shared slot
+ SharedSlot sharedSlot = instance.allocateSharedSlot(jobId, assignment);
+
+ // allocate a series of sub slots
+
+ SimpleSlot sub1 = assignment.addSharedSlotAndAllocateSubSlot(sharedSlot, Locality.UNCONSTRAINED, vid1);
+ SimpleSlot sub2 = assignment.getSlotForTask(vid2, Collections.<Instance>emptySet());
+ SimpleSlot sub3 = assignment.getSlotForTask(vid3, Collections.<Instance>emptySet());
+
+ assertNotNull(sub1);
+ assertNotNull(sub2);
+ assertNotNull(sub3);
+
+ assertEquals(3, sharedSlot.getNumberLeaves());
+
+ assertEquals(1, assignment.getNumberOfSlots());
+
+ // release from the leaves.
+
+ sub2.releaseSlot();
+
+ assertTrue(sharedSlot.isAlive());
+ assertTrue(sub1.isAlive());
+ assertTrue(sub2.isReleased());
+ assertTrue(sub3.isAlive());
+
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid1));
+ assertEquals(1, assignment.getNumberOfAvailableSlotsForGroup(vid2));
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid3));
+ assertEquals(1, assignment.getNumberOfSlots());
+
+ assertEquals(2, sharedSlot.getNumberLeaves());
+
+
+ sub1.releaseSlot();
+
+ assertTrue(sharedSlot.isAlive());
+ assertTrue(sub1.isReleased());
+ assertTrue(sub2.isReleased());
+ assertTrue(sub3.isAlive());
+
+ assertEquals(1, assignment.getNumberOfAvailableSlotsForGroup(vid1));
+ assertEquals(1, assignment.getNumberOfAvailableSlotsForGroup(vid2));
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid3));
+ assertEquals(1, assignment.getNumberOfSlots());
+
+ assertEquals(1, sharedSlot.getNumberLeaves());
+
+ sub3.releaseSlot();
+
+ assertTrue(sharedSlot.isReleased());
+ assertTrue(sub1.isReleased());
+ assertTrue(sub2.isReleased());
+ assertTrue(sub3.isReleased());
+
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid1));
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid2));
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid3));
+ assertEquals(0, assignment.getNumberOfSlots());
+
+ assertEquals(1, instance.getNumberOfAvailableSlots());
+ assertEquals(0, assignment.getNumberOfSlots());
+
+ assertNull(sharedSlot.allocateSharedSlot(new AbstractID()));
+ assertNull(sharedSlot.allocateSubSlot(new AbstractID()));
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void allocateAndReleaseInMixedOrder() {
+ try {
+ JobID jobId = new JobID();
+ JobVertexID vid1 = new JobVertexID();
+ JobVertexID vid2 = new JobVertexID();
+ JobVertexID vid3 = new JobVertexID();
+
+ SlotSharingGroup sharingGroup = new SlotSharingGroup(vid1, vid2, vid3);
+ SlotSharingGroupAssignment assignment = sharingGroup.getTaskAssignment();
+
+ Instance instance = SchedulerTestUtils.getRandomInstance(1);
+
+ // allocate a shared slot
+ SharedSlot sharedSlot = instance.allocateSharedSlot(jobId, assignment);
+
+ // allocate a series of sub slots
+
+ SimpleSlot sub1 = assignment.addSharedSlotAndAllocateSubSlot(sharedSlot, Locality.UNCONSTRAINED, vid1);
+ SimpleSlot sub2 = assignment.getSlotForTask(vid2, Collections.<Instance>emptySet());
+
+ assertNotNull(sub1);
+ assertNotNull(sub2);
+
+ assertEquals(2, sharedSlot.getNumberLeaves());
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid1));
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid2));
+ assertEquals(1, assignment.getNumberOfAvailableSlotsForGroup(vid3));
+ assertEquals(1, assignment.getNumberOfSlots());
+
+
+ sub2.releaseSlot();
+
+ assertEquals(1, sharedSlot.getNumberLeaves());
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid1));
+ assertEquals(1, assignment.getNumberOfAvailableSlotsForGroup(vid2));
+ assertEquals(1, assignment.getNumberOfAvailableSlotsForGroup(vid3));
+ assertEquals(1, assignment.getNumberOfSlots());
+
+
+ SimpleSlot sub3 = assignment.getSlotForTask(vid3, Collections.<Instance>emptySet());
+ assertNotNull(sub3);
+
+ assertEquals(2, sharedSlot.getNumberLeaves());
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid1));
+ assertEquals(1, assignment.getNumberOfAvailableSlotsForGroup(vid2));
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid3));
+ assertEquals(1, assignment.getNumberOfSlots());
+
+ sub3.releaseSlot();
+ sub1.releaseSlot();
+
+ assertTrue(sharedSlot.isReleased());
+ assertEquals(0, sharedSlot.getNumberLeaves());
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid1));
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid2));
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(vid3));
+ assertEquals(0, assignment.getNumberOfSlots());
+
+ assertEquals(1, instance.getNumberOfAvailableSlots());
+ assertEquals(0, assignment.getNumberOfSlots());
+
+ assertNull(sharedSlot.allocateSharedSlot(new AbstractID()));
+ assertNull(sharedSlot.allocateSubSlot(new AbstractID()));
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ /**
+ * We allocate and release the structure below, starting by allocating a simple slot in the
+ * shared slot and finishing by releasing a simple slot.
+ *
+ * <pre>
+ * Shared(0)(root)
+ * |
+ * +-- Simple(2)(sink)
+ * |
+ * +-- Shared(1)(co-location-group)
+ * | |
+ * | +-- Simple(0)(tail)
+ * | +-- Simple(1)(head)
+ * |
+ * +-- Simple(0)(source)
+ * </pre>
+ */
+ @Test
+ public void testAllocateAndReleaseTwoLevels() {
+ try {
+ JobVertexID sourceId = new JobVertexID();
+ JobVertexID headId = new JobVertexID();
+ JobVertexID tailId = new JobVertexID();
+ JobVertexID sinkId = new JobVertexID();
+
+ AbstractJobVertex headVertex = new AbstractJobVertex("head", headId);
+ AbstractJobVertex tailVertex = new AbstractJobVertex("tail", tailId);
+
+ SlotSharingGroup sharingGroup = new SlotSharingGroup(sourceId, headId, tailId, sinkId);
+ SlotSharingGroupAssignment assignment = sharingGroup.getTaskAssignment();
+ assertEquals(0, assignment.getNumberOfSlots());
+
+ CoLocationGroup coLocationGroup = new CoLocationGroup(headVertex, tailVertex);
+ CoLocationConstraint constraint = coLocationGroup.getLocationConstraint(0);
+ assertFalse(constraint.isAssigned());
+
+ Instance instance = SchedulerTestUtils.getRandomInstance(1);
+
+ // allocate a shared slot
+ SharedSlot sharedSlot = instance.allocateSharedSlot(new JobID(), assignment);
+
+ // get the first simple slot
+ SimpleSlot sourceSlot = assignment.addSharedSlotAndAllocateSubSlot(sharedSlot, Locality.LOCAL, sourceId);
+
+ assertEquals(1, sharedSlot.getNumberLeaves());
+
+ // get the first slot in the nested shared slot from the co-location constraint
+ SimpleSlot headSlot = assignment.getSlotForTask(constraint, Collections.<Instance>emptySet());
+ assertEquals(2, sharedSlot.getNumberLeaves());
+
+ assertNotNull(constraint.getSharedSlot());
+ assertTrue(constraint.getSharedSlot().isAlive());
+ assertFalse(constraint.isAssigned());
+
+ // we do not immediately lock the location
+ headSlot.releaseSlot();
+ assertEquals(1, sharedSlot.getNumberLeaves());
+
+ assertNotNull(constraint.getSharedSlot());
+ assertTrue(constraint.getSharedSlot().isReleased());
+ assertFalse(constraint.isAssigned());
+
+ // re-allocate the head slot
+ headSlot = assignment.getSlotForTask(constraint, Collections.<Instance>emptySet());
+
+ constraint.lockLocation();
+ assertNotNull(constraint.getSharedSlot());
+ assertTrue(constraint.isAssigned());
+ assertTrue(constraint.isAssignedAndAlive());
+ assertEquals(instance, constraint.getLocation());
+
+ SimpleSlot tailSlot = assignment.getSlotForTask(constraint, Collections.<Instance>emptySet());
+
+ assertEquals(constraint.getSharedSlot(), headSlot.getParent());
+ assertEquals(constraint.getSharedSlot(), tailSlot.getParent());
+
+ SimpleSlot sinkSlot = assignment.getSlotForTask(sinkId, Collections.<Instance>emptySet());
+ assertEquals(4, sharedSlot.getNumberLeaves());
+
+ // we release our co-location constraint tasks
+ headSlot.releaseSlot();
+ tailSlot.releaseSlot();
+
+ assertEquals(2, sharedSlot.getNumberLeaves());
+ assertTrue(headSlot.isReleased());
+ assertTrue(tailSlot.isReleased());
+ assertTrue(constraint.isAssigned());
+ assertFalse(constraint.isAssignedAndAlive());
+ assertEquals(instance, constraint.getLocation());
+
+ // we should have resources again for the co-location constraint
+ assertEquals(1, assignment.getNumberOfAvailableSlotsForGroup(constraint.getGroupId()));
+
+ // re-allocate head and tail from the constraint
+ headSlot = assignment.getSlotForTask(constraint, Collections.<Instance>emptySet());
+ tailSlot = assignment.getSlotForTask(constraint, Collections.<Instance>emptySet());
+
+ assertEquals(4, sharedSlot.getNumberLeaves());
+ assertEquals(0, assignment.getNumberOfAvailableSlotsForGroup(constraint.getGroupId()));
+
+ // verify some basic properties of the slots
+ assertEquals(instance, sourceSlot.getInstance());
+ assertEquals(instance, headSlot.getInstance());
+ assertEquals(instance, tailSlot.getInstance());
+ assertEquals(instance, sinkSlot.getInstance());
+
+ assertEquals(sourceId, sourceSlot.getGroupID());
+ assertEquals(sinkId, sinkSlot.getGroupID());
+ assertNull(headSlot.getGroupID());
+ assertNull(tailSlot.getGroupID());
+ assertEquals(constraint.getGroupId(), constraint.getSharedSlot().getGroupID());
+
+ // release all
+ sourceSlot.releaseSlot();
+ headSlot.releaseSlot();
+ tailSlot.releaseSlot();
+ sinkSlot.releaseSlot();
+
+ assertTrue(sharedSlot.isReleased());
+ assertTrue(sourceSlot.isReleased());
+ assertTrue(headSlot.isReleased());
+ assertTrue(tailSlot.isReleased());
+ assertTrue(sinkSlot.isReleased());
+ assertTrue(constraint.getSharedSlot().isReleased());
+
+ assertTrue(constraint.isAssigned());
+ assertFalse(constraint.isAssignedAndAlive());
+
+ assertEquals(1, instance.getNumberOfAvailableSlots());
+ assertEquals(0, assignment.getNumberOfSlots());
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ /**
+ * We allocate and the structure below and release it from the root.
+ *
+ * <pre>
+ * Shared(0)(root)
+ * |
+ * +-- Simple(2)(sink)
+ * |
+ * +-- Shared(1)(co-location-group)
+ * | |
+ * | +-- Simple(0)(tail)
+ * | +-- Simple(1)(head)
+ * |
+ * +-- Simple(0)(source)
+ * </pre>
+ */
+ @Test
+ public void testReleaseTwoLevelsFromRoot() {
+ try {
+ JobVertexID sourceId = new JobVertexID();
+ JobVertexID headId = new JobVertexID();
+ JobVertexID tailId = new JobVertexID();
+ JobVertexID sinkId = new JobVertexID();
+
+ AbstractJobVertex headVertex = new AbstractJobVertex("head", headId);
+ AbstractJobVertex tailVertex = new AbstractJobVertex("tail", tailId);
+
+ SlotSharingGroup sharingGroup = new SlotSharingGroup(sourceId, headId, tailId, sinkId);
+ SlotSharingGroupAssignment assignment = sharingGroup.getTaskAssignment();
+ assertEquals(0, assignment.getNumberOfSlots());
+
+ CoLocationGroup coLocationGroup = new CoLocationGroup(headVertex, tailVertex);
+ CoLocationConstraint constraint = coLocationGroup.getLocationConstraint(0);
+ assertFalse(constraint.isAssigned());
+
+ Instance instance = SchedulerTestUtils.getRandomInstance(1);
+
+ // allocate a shared slot
+ SharedSlot sharedSlot = instance.allocateSharedSlot(new JobID(), assignment);
+
+ // get the first simple slot
+ SimpleSlot sourceSlot = assignment.addSharedSlotAndAllocateSubSlot(sharedSlot, Locality.LOCAL, sourceId);
+
+ SimpleSlot headSlot = assignment.getSlotForTask(constraint, Collections.<Instance>emptySet());
+ constraint.lockLocation();
+ SimpleSlot tailSlot = assignment.getSlotForTask(constraint, Collections.<Instance>emptySet());
+
+ SimpleSlot sinkSlot = assignment.getSlotForTask(sinkId, Collections.<Instance>emptySet());
+
+ assertEquals(4, sharedSlot.getNumberLeaves());
+
+ // release all
+ sourceSlot.releaseSlot();
+ headSlot.releaseSlot();
+ tailSlot.releaseSlot();
+ sinkSlot.releaseSlot();
+
+ assertTrue(sharedSlot.isReleased());
+ assertTrue(sourceSlot.isReleased());
+ assertTrue(headSlot.isReleased());
+ assertTrue(tailSlot.isReleased());
+ assertTrue(sinkSlot.isReleased());
+ assertTrue(constraint.getSharedSlot().isReleased());
+
+ assertTrue(constraint.isAssigned());
+ assertFalse(constraint.isAssignedAndAlive());
+
+ assertEquals(1, instance.getNumberOfAvailableSlots());
+ assertEquals(0, instance.getNumberOfAllocatedSlots());
+
+ assertEquals(0, assignment.getNumberOfSlots());
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+
+ @Test
+ public void testImmediateReleaseOneLevel() {
+ try {
+ JobID jobId = new JobID();
+ JobVertexID vid = new JobVertexID();
+
+ SlotSharingGroup sharingGroup = new SlotSharingGroup(vid);
+ SlotSharingGroupAssignment assignment = sharingGroup.getTaskAssignment();
+
+ Instance instance = SchedulerTestUtils.getRandomInstance(1);
+
+ SharedSlot sharedSlot = instance.allocateSharedSlot(jobId, assignment);
+
+ SimpleSlot sub = assignment.addSharedSlotAndAllocateSubSlot(sharedSlot, Locality.UNCONSTRAINED, vid);
+ sub.releaseSlot();
+
+ assertTrue(sub.isReleased());
+ assertTrue(sharedSlot.isReleased());
+
+ assertEquals(1, instance.getNumberOfAvailableSlots());
+ assertEquals(0, instance.getNumberOfAllocatedSlots());
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testImmediateReleaseTwoLevel() {
+ try {
+ JobID jobId = new JobID();
+ JobVertexID vid = new JobVertexID();
+ AbstractJobVertex vertex = new AbstractJobVertex("vertex", vid);
+
+ SlotSharingGroup sharingGroup = new SlotSharingGroup(vid);
+ SlotSharingGroupAssignment assignment = sharingGroup.getTaskAssignment();
+
+ CoLocationGroup coLocationGroup = new CoLocationGroup(vertex);
+ CoLocationConstraint constraint = coLocationGroup.getLocationConstraint(0);
+
+ Instance instance = SchedulerTestUtils.getRandomInstance(1);
+
+ SharedSlot sharedSlot = instance.allocateSharedSlot(jobId, assignment);
+
+ SimpleSlot sub = assignment.addSharedSlotAndAllocateSubSlot(sharedSlot, Locality.UNCONSTRAINED, constraint);
+
+ assertNull(sub.getGroupID());
+ assertEquals(constraint.getSharedSlot(), sub.getParent());
+
+ sub.releaseSlot();
+
+ assertTrue(sub.isReleased());
+ assertTrue(sharedSlot.isReleased());
+
+ assertEquals(1, instance.getNumberOfAvailableSlots());
+ assertEquals(0, instance.getNumberOfAllocatedSlots());
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SimpleSlotTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SimpleSlotTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SimpleSlotTest.java
index f6c0c9f..e075ab6 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SimpleSlotTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SimpleSlotTest.java
@@ -24,9 +24,12 @@ import static org.junit.Assert.*;
import java.net.InetAddress;
import akka.actor.ActorRef;
+
import org.apache.flink.runtime.executiongraph.Execution;
import org.apache.flink.api.common.JobID;
+
import org.junit.Test;
+
import org.mockito.Matchers;
public class SimpleSlotTest {
@@ -34,28 +37,28 @@ public class SimpleSlotTest {
@Test
public void testStateTransitions() {
try {
- // cancel, then release
+ // release immediately
{
SimpleSlot slot = getSlot();
assertTrue(slot.isAlive());
-
- slot.cancel();
- assertFalse(slot.isAlive());
- assertTrue(slot.isCanceled());
- assertFalse(slot.isReleased());
-
+
slot.releaseSlot();
assertFalse(slot.isAlive());
assertTrue(slot.isCanceled());
assertTrue(slot.isReleased());
}
-
- // release immediately
+
+ // state transitions manually
{
SimpleSlot slot = getSlot();
assertTrue(slot.isAlive());
-
- slot.releaseSlot();
+
+ slot.markCancelled();
+ assertFalse(slot.isAlive());
+ assertTrue(slot.isCanceled());
+ assertFalse(slot.isReleased());
+
+ slot.markReleased();
assertFalse(slot.isAlive());
assertTrue(slot.isCanceled());
assertTrue(slot.isReleased());
@@ -66,41 +69,51 @@ public class SimpleSlotTest {
fail(e.getMessage());
}
}
-
+
@Test
public void testSetExecutionVertex() {
try {
Execution ev = mock(Execution.class);
Execution ev_2 = mock(Execution.class);
-
+
// assign to alive slot
{
SimpleSlot slot = getSlot();
-
+
assertTrue(slot.setExecutedVertex(ev));
- assertEquals(ev, slot.getExecution());
-
+ assertEquals(ev, slot.getExecutedVertex());
+
// try to add another one
assertFalse(slot.setExecutedVertex(ev_2));
- assertEquals(ev, slot.getExecution());
+ assertEquals(ev, slot.getExecutedVertex());
}
-
+
// assign to canceled slot
{
SimpleSlot slot = getSlot();
- slot.cancel();
-
+ assertTrue(slot.markCancelled());
+
assertFalse(slot.setExecutedVertex(ev));
- assertNull(slot.getExecution());
+ assertNull(slot.getExecutedVertex());
+ }
+
+ // assign to released marked slot
+ {
+ SimpleSlot slot = getSlot();
+ assertTrue(slot.markCancelled());
+ assertTrue(slot.markReleased());
+
+ assertFalse(slot.setExecutedVertex(ev));
+ assertNull(slot.getExecutedVertex());
}
// assign to released
{
SimpleSlot slot = getSlot();
slot.releaseSlot();
-
+
assertFalse(slot.setExecutedVertex(ev));
- assertNull(slot.getExecution());
+ assertNull(slot.getExecutedVertex());
}
}
catch (Exception e) {
@@ -108,20 +121,20 @@ public class SimpleSlotTest {
fail(e.getMessage());
}
}
-
+
@Test
public void testReleaseCancelsVertex() {
try {
Execution ev = mock(Execution.class);
-
+
SimpleSlot slot = getSlot();
assertTrue(slot.setExecutedVertex(ev));
- assertEquals(ev, slot.getExecution());
-
- slot.cancel();
+ assertEquals(ev, slot.getExecutedVertex());
+
slot.releaseSlot();
- slot.cancel();
-
+ slot.releaseSlot();
+ slot.releaseSlot();
+
verify(ev, times(1)).fail(Matchers.any(Throwable.class));
}
catch (Exception e) {
@@ -129,13 +142,13 @@ public class SimpleSlotTest {
fail(e.getMessage());
}
}
-
+
public static SimpleSlot getSlot() throws Exception {
HardwareDescription hardwareDescription = new HardwareDescription(4, 2L*1024*1024*1024, 1024*1024*1024, 512*1024*1024);
InetAddress address = InetAddress.getByName("127.0.0.1");
InstanceConnectionInfo connection = new InstanceConnectionInfo(address, 10001);
-
+
Instance instance = new Instance(ActorRef.noSender(), connection, new InstanceID(), hardwareDescription, 1);
return instance.allocateSimpleSlot(new JobID());
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SlotTestUtils.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SlotTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SlotTestUtils.java
deleted file mode 100644
index 81becd2..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SlotTestUtils.java
+++ /dev/null
@@ -1,23 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.instance;
-
-public class SlotTestUtils {
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationConstraintTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationConstraintTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationConstraintTest.java
new file mode 100644
index 0000000..94dee71
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationConstraintTest.java
@@ -0,0 +1,177 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.jobmanager.scheduler;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.instance.Instance;
+import org.apache.flink.runtime.instance.SharedSlot;
+import org.apache.flink.runtime.instance.SlotSharingGroupAssignment;
+import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.util.AbstractID;
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+/**
+ * Tests for the {@link CoLocationConstraint}.
+ */
+public class CoLocationConstraintTest {
+
+ @Test
+ public void testCreateConstraints() {
+ try {
+ JobVertexID id1 = new JobVertexID();
+ JobVertexID id2 = new JobVertexID();
+
+ AbstractJobVertex vertex1 = new AbstractJobVertex("vertex1", id1);
+ vertex1.setParallelism(2);
+
+ AbstractJobVertex vertex2 = new AbstractJobVertex("vertex2", id2);
+ vertex2.setParallelism(3);
+
+ CoLocationGroup group = new CoLocationGroup(vertex1, vertex2);
+
+ AbstractID groupId = group.getId();
+ assertNotNull(groupId);
+
+ CoLocationConstraint constraint1 = group.getLocationConstraint(0);
+ CoLocationConstraint constraint2 = group.getLocationConstraint(1);
+ CoLocationConstraint constraint3 = group.getLocationConstraint(2);
+
+ assertFalse(constraint1 == constraint2);
+ assertFalse(constraint1 == constraint3);
+ assertFalse(constraint2 == constraint3);
+
+ assertEquals(groupId, constraint1.getGroupId());
+ assertEquals(groupId, constraint2.getGroupId());
+ assertEquals(groupId, constraint3.getGroupId());
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testAssignSlotAndLockLocation() {
+ try {
+ JobID jid = new JobID();
+
+ AbstractJobVertex vertex = new AbstractJobVertex("vertex");
+ vertex.setParallelism(1);
+
+ SlotSharingGroup sharingGroup = new SlotSharingGroup(vertex.getID());
+ SlotSharingGroupAssignment assignment = sharingGroup.getTaskAssignment();
+
+ CoLocationGroup constraintGroup = new CoLocationGroup(vertex);
+ CoLocationConstraint constraint = constraintGroup.getLocationConstraint(0);
+
+ // constraint is completely unassigned
+ assertFalse(constraint.isAssigned());
+ assertFalse(constraint.isAssignedAndAlive());
+
+ Instance instance1 = SchedulerTestUtils.getRandomInstance(2);
+ Instance instance2 = SchedulerTestUtils.getRandomInstance(2);
+
+ SharedSlot slot1_1 = instance1.allocateSharedSlot(jid, assignment);
+ SharedSlot slot1_2 = instance1.allocateSharedSlot(jid, assignment);
+ SharedSlot slot2_1 = instance2.allocateSharedSlot(jid, assignment);
+ SharedSlot slot2_2 = instance2.allocateSharedSlot(jid, assignment);
+
+ // constraint is still completely unassigned
+ assertFalse(constraint.isAssigned());
+ assertFalse(constraint.isAssignedAndAlive());
+
+ // set the slot, but do not lock the location yet
+ constraint.setSharedSlot(slot1_1);
+
+ assertFalse(constraint.isAssigned());
+ assertFalse(constraint.isAssignedAndAlive());
+
+ // try to get the location
+ try {
+ constraint.getLocation();
+ fail("should throw an IllegalStateException");
+ }
+ catch (IllegalStateException e) {
+ // as expected
+ }
+ catch (Exception e) {
+ fail("wrong exception, should be IllegalStateException");
+ }
+
+ // check that we can reassign the slot as long as the location is not locked
+ constraint.setSharedSlot(slot2_1);
+
+ // the previous slot should have been released now
+ assertTrue(slot1_1.isReleased());
+
+ // still the location is not assigned
+ assertFalse(constraint.isAssigned());
+ assertFalse(constraint.isAssignedAndAlive());
+
+ // we can do an identity re-assign
+ constraint.setSharedSlot(slot2_1);
+ assertFalse(slot2_1.isReleased());
+
+ // still the location is not assigned
+ assertFalse(constraint.isAssigned());
+ assertFalse(constraint.isAssignedAndAlive());
+
+ constraint.lockLocation();
+
+ // now, the location is assigned and we have a location
+ assertTrue(constraint.isAssigned());
+ assertTrue(constraint.isAssignedAndAlive());
+ assertEquals(instance2, constraint.getLocation());
+
+ // release the slot
+ slot2_1.releaseSlot();
+
+ // we should still have a location
+ assertTrue(constraint.isAssigned());
+ assertFalse(constraint.isAssignedAndAlive());
+ assertEquals(instance2, constraint.getLocation());
+
+ // we can not assign a different location
+ try {
+ constraint.setSharedSlot(slot1_2);
+ fail("should throw an IllegalArgumentException");
+ }
+ catch (IllegalArgumentException e) {
+ // as expected
+ }
+ catch (Exception e) {
+ fail("wrong exception, should be IllegalArgumentException");
+ }
+
+ // assign a new slot with the same location
+ constraint.setSharedSlot(slot2_2);
+
+ assertTrue(constraint.isAssigned());
+ assertTrue(constraint.isAssignedAndAlive());
+ assertEquals(instance2, constraint.getLocation());
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/ScheduleWithCoLocationHintTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/ScheduleWithCoLocationHintTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/ScheduleWithCoLocationHintTest.java
index 272a911..a15e477 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/ScheduleWithCoLocationHintTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/ScheduleWithCoLocationHintTest.java
@@ -21,17 +21,20 @@ package org.apache.flink.runtime.jobmanager.scheduler;
import static org.apache.flink.runtime.jobmanager.scheduler.SchedulerTestUtils.getRandomInstance;
import static org.apache.flink.runtime.jobmanager.scheduler.SchedulerTestUtils.getTestVertex;
import static org.apache.flink.runtime.jobmanager.scheduler.SchedulerTestUtils.getTestVertexWithLocation;
+
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-import org.apache.flink.runtime.instance.SimpleSlot;
import akka.actor.ActorSystem;
import akka.testkit.JavaTestKit;
+
+import org.apache.flink.runtime.instance.SimpleSlot;
import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.testingUtils.TestingUtils;
+
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -563,8 +566,8 @@ public class ScheduleWithCoLocationHintTest {
assertEquals(2, scheduler.getNumberOfAvailableSlots());
assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfSlots());
- assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForJid(jid1));
- assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForJid(jid2));
+ assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForGroup(jid1));
+ assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForGroup(jid2));
}
catch (Exception e) {
e.printStackTrace();
@@ -612,8 +615,8 @@ public class ScheduleWithCoLocationHintTest {
assertEquals(2, scheduler.getNumberOfAvailableSlots());
assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfSlots());
- assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForJid(jid1));
- assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForJid(jid2));
+ assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForGroup(jid1));
+ assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForGroup(jid2));
}
catch (Exception e) {
e.printStackTrace();
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerSlotSharingTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerSlotSharingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerSlotSharingTest.java
index d1c6938..f987e07 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerSlotSharingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerSlotSharingTest.java
@@ -212,8 +212,8 @@ public class SchedulerSlotSharingTest {
s4.releaseSlot();
assertEquals(4, sharingGroup.getTaskAssignment().getNumberOfSlots());
- assertEquals(2, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForJid(jid1));
- assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForJid(jid2));
+ assertEquals(2, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForGroup(jid1));
+ assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForGroup(jid2));
// we can still not schedule anything from the second group of vertices
try {
@@ -232,8 +232,8 @@ public class SchedulerSlotSharingTest {
assertNotNull(s5);
assertEquals(4, sharingGroup.getTaskAssignment().getNumberOfSlots());
- assertEquals(1, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForJid(jid1));
- assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForJid(jid2));
+ assertEquals(1, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForGroup(jid1));
+ assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForGroup(jid2));
// now we release a slot from the second vertex group and schedule another task from that group
@@ -285,8 +285,8 @@ public class SchedulerSlotSharingTest {
SimpleSlot s4 = scheduler.scheduleImmediately(new ScheduledUnit(getTestVertex(jid1, 3, 4), sharingGroup));
assertEquals(4, sharingGroup.getTaskAssignment().getNumberOfSlots());
- assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForJid(jid1));
- assertEquals(4, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForJid(jid2));
+ assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForGroup(jid1));
+ assertEquals(4, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForGroup(jid2));
s1.releaseSlot();
s2.releaseSlot();
@@ -294,8 +294,8 @@ public class SchedulerSlotSharingTest {
s4.releaseSlot();
assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfSlots());
- assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForJid(jid1));
- assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForJid(jid2));
+ assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForGroup(jid1));
+ assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForGroup(jid2));
// schedule some tasks from the second ID group
SimpleSlot s1_2 = scheduler.scheduleImmediately(new ScheduledUnit(getTestVertex(jid2, 0, 4), sharingGroup));
@@ -304,8 +304,8 @@ public class SchedulerSlotSharingTest {
SimpleSlot s4_2 = scheduler.scheduleImmediately(new ScheduledUnit(getTestVertex(jid2, 3, 4), sharingGroup));
assertEquals(4, sharingGroup.getTaskAssignment().getNumberOfSlots());
- assertEquals(4, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForJid(jid1));
- assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForJid(jid2));
+ assertEquals(4, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForGroup(jid1));
+ assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForGroup(jid2));
s1_2.releaseSlot();
s2_2.releaseSlot();
@@ -313,8 +313,8 @@ public class SchedulerSlotSharingTest {
s4_2.releaseSlot();
assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfSlots());
- assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForJid(jid1));
- assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForJid(jid2));
+ assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForGroup(jid1));
+ assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForGroup(jid2));
// test that everything is released
assertEquals(4, scheduler.getNumberOfAvailableSlots());
@@ -463,8 +463,8 @@ public class SchedulerSlotSharingTest {
assertNotNull(s2_2);
assertEquals(2, sharingGroup.getTaskAssignment().getNumberOfSlots());
- assertEquals(1, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForJid(jid1));
- assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForJid(jid2));
+ assertEquals(1, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForGroup(jid1));
+ assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForGroup(jid2));
// release the two from the second
s2_1.releaseSlot();
@@ -476,8 +476,8 @@ public class SchedulerSlotSharingTest {
assertNotNull(sx);
assertEquals(1, sharingGroup.getTaskAssignment().getNumberOfSlots());
- assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForJid(jid1));
- assertEquals(1, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForJid(jid2));
+ assertEquals(0, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForGroup(jid1));
+ assertEquals(1, sharingGroup.getTaskAssignment().getNumberOfAvailableSlotsForGroup(jid2));
// check the scheduler's bookkeeping
assertEquals(0, scheduler.getNumberOfLocalizedAssignments());
@@ -1007,6 +1007,7 @@ public class SchedulerSlotSharingTest {
executor.execute(deploy0);
// wait until all tasks have finished
+ //noinspection SynchronizationOnLocalVariableOrMethodParameter
synchronized (completed) {
while (!failed.get() && completed.get() < 13) {
completed.wait(1000);
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerTestUtils.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerTestUtils.java
index 776184c..694b88b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerTestUtils.java
@@ -25,6 +25,7 @@ import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.atomic.AtomicInteger;
@@ -53,7 +54,8 @@ public class SchedulerTestUtils {
InetAddress address;
try {
address = InetAddress.getByName("127.0.0.1");
- } catch (UnknownHostException e) {
+ }
+ catch (UnknownHostException e) {
throw new RuntimeException("Test could not create IP address for localhost loopback.");
}
@@ -133,9 +135,7 @@ public class SchedulerTestUtils {
}
HashSet<Object> set = new HashSet<Object>();
- for (Object o : obj) {
- set.add(o);
- }
+ Collections.addAll(set, obj);
return set.size() == obj.length;
}
@@ -154,5 +154,4 @@ public class SchedulerTestUtils {
return set.isEmpty();
}
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SharedSlotsTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SharedSlotsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SharedSlotsTest.java
deleted file mode 100644
index 42a3702..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SharedSlotsTest.java
+++ /dev/null
@@ -1,145 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.jobmanager.scheduler;
-
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.doAnswer;
-import static org.mockito.Matchers.any;
-import static org.junit.Assert.*;
-
-import org.apache.flink.runtime.instance.SharedSlot;
-import org.apache.flink.runtime.instance.SimpleSlot;
-import org.junit.Test;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
-import org.apache.flink.runtime.instance.Instance;
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.runtime.jobgraph.JobVertexID;
-
-public class SharedSlotsTest {
-
- @Test
- public void createAndDoNotRelease() {
- try {
- SlotSharingGroupAssignment assignment = mock(SlotSharingGroupAssignment.class);
- doAnswer(new Answer<Void>() {
- @Override
- public Void answer(InvocationOnMock invocation) throws Throwable {
- final SimpleSlot simpleSlot = (SimpleSlot) invocation.getArguments()[0];
- final SharedSlot sharedSlot = simpleSlot.getParent();
-
- sharedSlot.freeSubSlot(simpleSlot);
-
- return null;
- }
-
- }).when(assignment).releaseSimpleSlot(any(SimpleSlot.class));
-
- JobVertexID id1 = new JobVertexID();
-
- Instance instance = SchedulerTestUtils.getRandomInstance(1);
-
- SharedSlot slot = instance.allocateSharedSlot(new JobID(), assignment, id1);
- assertFalse(slot.isDead());
-
- SimpleSlot ss1 = slot.allocateSubSlot(id1);
- assertNotNull(ss1);
-
- // verify resources
- assertEquals(instance, ss1.getInstance());
- assertEquals(0, ss1.getSlotNumber());
- assertEquals(slot.getJobID(), ss1.getJobID());
-
- SimpleSlot ss2 = slot.allocateSubSlot(new JobVertexID());
- assertNotNull(ss2);
-
- assertEquals(2, slot.getNumberLeaves());
-
- // release first slot, should not trigger release
- ss1.releaseSlot();
- assertFalse(slot.isDead());
-
- ss2.releaseSlot();
- assertFalse(slot.isDead());
-
- // the shared slot should now dispose itself
- assertEquals(0, slot.getNumberLeaves());
- }
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
- }
-
- @Test
- public void createAndRelease() {
- try {
- SlotSharingGroupAssignment assignment = mock(SlotSharingGroupAssignment.class);
- doAnswer(new Answer<Boolean>() {
- @Override
- public Boolean answer(InvocationOnMock invocation) throws Throwable {
- final SimpleSlot slot = (SimpleSlot) invocation.getArguments()[0];
- final SharedSlot shared = slot.getParent();
- if (shared.freeSubSlot(slot) == 0) {
- shared.markDead();
- return true;
- }
- return false;
- }
-
- }).when(assignment).releaseSimpleSlot(any(SimpleSlot.class));
-
- JobVertexID id1 = new JobVertexID();
-
- Instance instance = SchedulerTestUtils.getRandomInstance(1);
-
- SharedSlot slot = instance.allocateSharedSlot(new JobID(), assignment, id1);
- assertFalse(slot.isDead());
-
- SimpleSlot ss1 = slot.allocateSubSlot(id1);
- assertNotNull(ss1);
-
- // verify resources
- assertEquals(instance, ss1.getInstance());
- assertEquals(0, ss1.getSlotNumber());
- assertEquals(slot.getJobID(), ss1.getJobID());
-
- SimpleSlot ss2 = slot.allocateSubSlot(new JobVertexID());
- assertNotNull(ss2);
-
- assertEquals(2, slot.getNumberLeaves());
-
- // release first slot, should not trigger release
- ss1.releaseSlot();
- assertFalse(slot.isDead());
-
- ss2.releaseSlot();
- assertTrue(slot.isDead());
-
- // the shared slot should now dispose itself
- assertEquals(0, slot.getNumberLeaves());
-
- assertNull(slot.allocateSubSlot(new JobVertexID()));
- }
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
- }
-}
[6/7] flink git commit: [tests] Minor warning cleanups
Posted by se...@apache.org.
[tests] Minor warning cleanups
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/96bbdbd7
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/96bbdbd7
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/96bbdbd7
Branch: refs/heads/master
Commit: 96bbdbd70cfe55c4741ce7513cdb051883ef6f48
Parents: 3061eb1
Author: Stephan Ewen <se...@apache.org>
Authored: Wed May 27 12:44:56 2015 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed May 27 16:36:02 2015 +0200
----------------------------------------------------------------------
.../java/org/apache/flink/api/common/ExecutionConfigTest.java | 7 ++++---
1 file changed, 4 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/96bbdbd7/flink-core/src/test/java/org/apache/flink/api/common/ExecutionConfigTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/ExecutionConfigTest.java b/flink-core/src/test/java/org/apache/flink/api/common/ExecutionConfigTest.java
index ad3ad91..39b3b71 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/ExecutionConfigTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/ExecutionConfigTest.java
@@ -18,19 +18,20 @@
package org.apache.flink.api.common;
-import org.junit.Test;
import static org.junit.Assert.*;
import java.util.Arrays;
import java.util.List;
+import org.junit.Test;
+
public class ExecutionConfigTest {
@Test
public void testDoubleTypeRegistration() {
ExecutionConfig config = new ExecutionConfig();
- List<Class<?>> types = Arrays.asList((Class<?>)Double.class, Integer.class, Double.class);
- List<Class<?>> expectedTypes = Arrays.asList((Class<?>)Double.class, Integer.class);
+ List<Class<?>> types = Arrays.<Class<?>>asList(Double.class, Integer.class, Double.class);
+ List<Class<?>> expectedTypes = Arrays.<Class<?>>asList(Double.class, Integer.class);
for(Class<?> tpe: types) {
config.registerKryoType(tpe);
[2/7] flink git commit: [FLINK-1952] [jobmanager] Rework and fix slot
sharing scheduler
Posted by se...@apache.org.
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/Scheduler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/Scheduler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/Scheduler.java
index 9b2000b..579a6b4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/Scheduler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/Scheduler.java
@@ -36,27 +36,39 @@ import akka.dispatch.Futures;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
+
import org.apache.flink.runtime.akka.AkkaUtils;
-import org.apache.flink.util.AbstractID;
+import org.apache.flink.runtime.instance.SlotSharingGroupAssignment;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.instance.SharedSlot;
import org.apache.flink.runtime.instance.SimpleSlot;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.apache.flink.runtime.executiongraph.ExecutionVertex;
import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.instance.InstanceDiedException;
import org.apache.flink.runtime.instance.InstanceListener;
import org.apache.flink.util.ExceptionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
/**
- * The scheduler is responsible for distributing the ready-to-run tasks and assigning them to instances and
- * slots.
+ * The scheduler is responsible for distributing the ready-to-run tasks among instances and slots.
+ *
+ * <p>The scheduler supports two scheduling modes:</p>
+ * <ul>
+ * <li>Immediate scheduling: A request for a task slot immediately returns a task slot, if one is
+ * available, or throws a {@link NoResourceAvailableException}</li>.
+ * <li>Queued Scheduling: A request for a task slot is queued and returns a future that will be
+ * fulfilled as soon as a slot becomes available.</li>
+ * </ul>
*/
public class Scheduler implements InstanceListener, SlotAvailabilityListener {
- static final Logger LOG = LoggerFactory.getLogger(Scheduler.class);
+ /** Scheduler-wide logger */
+ private static final Logger LOG = LoggerFactory.getLogger(Scheduler.class);
+ /** All modifications to the scheduler structures are performed under a global scheduler lock */
private final Object globalLock = new Object();
/** All instances that the scheduler can deploy to */
@@ -71,20 +83,24 @@ public class Scheduler implements InstanceListener, SlotAvailabilityListener {
/** All tasks pending to be scheduled */
private final Queue<QueuedTask> taskQueue = new ArrayDeque<QueuedTask>();
- private final BlockingQueue<Instance> newlyAvailableInstances;
+ private final BlockingQueue<Instance> newlyAvailableInstances = new LinkedBlockingQueue<Instance>();
+ /** The number of slot allocations that had no location preference */
private int unconstrainedAssignments;
-
+
+ /** The number of slot allocations where locality could be respected */
private int localizedAssignments;
-
+
+ /** The number of slot allocations where locality could not be respected */
private int nonLocalizedAssignments;
-
-
- public Scheduler() {
- this.newlyAvailableInstances = new LinkedBlockingQueue<Instance>();
- }
-
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * Creates a new scheduler.
+ */
+ public Scheduler() {}
/**
* Shuts the scheduler down. After shut down no more tasks can be added to the scheduler.
@@ -102,9 +118,9 @@ public class Scheduler implements InstanceListener, SlotAvailabilityListener {
}
}
- // --------------------------------------------------------------------------------------------
+ // ------------------------------------------------------------------------
// Scheduling
- // --------------------------------------------------------------------------------------------
+ // ------------------------------------------------------------------------
public SimpleSlot scheduleImmediately(ScheduledUnit task) throws NoResourceAvailableException {
Object ret = scheduleTask(task, false);
@@ -130,13 +146,12 @@ public class Scheduler implements InstanceListener, SlotAvailabilityListener {
}
/**
- * Returns either an {@link org.apache.flink.runtime.instance.SimpleSlot}, or an {@link SlotAllocationFuture}.
+ * Returns either a {@link org.apache.flink.runtime.instance.SimpleSlot}, or a {@link SlotAllocationFuture}.
*/
private Object scheduleTask(ScheduledUnit task, boolean queueIfNoResource) throws NoResourceAvailableException {
if (task == null) {
- throw new IllegalArgumentException();
+ throw new NullPointerException();
}
-
if (LOG.isDebugEnabled()) {
LOG.debug("Scheduling task " + task);
}
@@ -148,14 +163,16 @@ public class Scheduler implements InstanceListener, SlotAvailabilityListener {
preferredLocations != null && preferredLocations.iterator().hasNext();
synchronized (globalLock) {
-
- // 1) === If the task has a slot sharing group, schedule with shared slots ===
SlotSharingGroup sharingUnit = task.getSlotSharingGroup();
+
if (sharingUnit != null) {
+
+ // 1) === If the task has a slot sharing group, schedule with shared slots ===
if (queueIfNoResource) {
- throw new IllegalArgumentException("A task with a vertex sharing group was scheduled in a queued fashion.");
+ throw new IllegalArgumentException(
+ "A task with a vertex sharing group was scheduled in a queued fashion.");
}
final SlotSharingGroupAssignment assignment = sharingUnit.getTaskAssignment();
@@ -163,12 +180,12 @@ public class Scheduler implements InstanceListener, SlotAvailabilityListener {
// sanity check that we do not use an externally forced location and a co-location constraint together
if (constraint != null && forceExternalLocation) {
- throw new IllegalArgumentException("The scheduling cannot be contrained simultaneously by a "
- + "co-location constriaint and an external location constraint.");
+ throw new IllegalArgumentException("The scheduling cannot be constrained simultaneously by a "
+ + "co-location constraint and an external location constraint.");
}
// get a slot from the group, if the group has one for us (and can fulfill the constraint)
- SimpleSlot slotFromGroup;
+ final SimpleSlot slotFromGroup;
if (constraint == null) {
slotFromGroup = assignment.getSlotForTask(vertex);
}
@@ -182,74 +199,87 @@ public class Scheduler implements InstanceListener, SlotAvailabilityListener {
// the following needs to make sure any allocated slot is released in case of an error
try {
- // check whether the slot from the group is already what we want
- if (slotFromGroup != null) {
- // local (or unconstrained in the current group)
- if (slotFromGroup.getLocality() != Locality.NON_LOCAL) {
- updateLocalityCounters(slotFromGroup.getLocality());
- return slotFromGroup;
+ // check whether the slot from the group is already what we want.
+ // any slot that is local, or where the assignment was unconstrained is good!
+ if (slotFromGroup != null && slotFromGroup.getLocality() != Locality.NON_LOCAL) {
+
+ // if this is the first slot for the co-location constraint, we lock
+ // the location, because we are quite happy with the slot
+ if (constraint != null && !constraint.isAssigned()) {
+ constraint.lockLocation();
}
+
+ updateLocalityCounters(slotFromGroup.getLocality(), vertex, slotFromGroup.getInstance());
+ return slotFromGroup;
}
- final Iterable<Instance> locations = (constraint == null || constraint.isUnassigned()) ?
- vertex.getPreferredLocations() : Collections.singleton(constraint.getLocation());
+ // the group did not have a local slot for us. see if we can one (or a better one)
+
+ // our location preference is either determined by the location constraint, or by the
+ // vertex's preferred locations
+ final Iterable<Instance> locations;
+ final boolean localOnly;
+ if (constraint != null && constraint.isAssigned()) {
+ locations = Collections.singleton(constraint.getLocation());
+ localOnly = true;
+ }
+ else {
+ locations = vertex.getPreferredLocations();
+ localOnly = forceExternalLocation;
+ }
- // get a new slot, since we could not place it into the group, or we could not place it locally
- newSlot = getFreeSubSlotForTask(vertex, locations, assignment, constraint, forceExternalLocation);
+ newSlot = getNewSlotForSharingGroup(vertex, locations, assignment, constraint, localOnly);
if (newSlot == null) {
if (slotFromGroup == null) {
- // both null
- if (constraint == null || constraint.isUnassigned()) {
- if (forceExternalLocation) {
- // could not satisfy the external location constraint
- String hosts = getHostnamesFromInstances(preferredLocations);
- throw new NoResourceAvailableException("Could not schedule task " + vertex
- + " to any of the required hosts: " + hosts);
- }
- else {
- // simply nothing is available
- throw new NoResourceAvailableException(task, getNumberOfAvailableInstances(),
- getTotalNumberOfSlots(), getNumberOfAvailableSlots());
- }
+ // both null, which means there is nothing available at all
+
+ if (constraint != null && constraint.isAssigned()) {
+ // nothing is available on the node where the co-location constraint forces us to
+ throw new NoResourceAvailableException("Could not allocate a slot on instance " +
+ constraint.getLocation() + ", as required by the co-location constraint.");
+ }
+ else if (forceExternalLocation) {
+ // could not satisfy the external location constraint
+ String hosts = getHostnamesFromInstances(preferredLocations);
+ throw new NoResourceAvailableException("Could not schedule task " + vertex
+ + " to any of the required hosts: " + hosts);
}
else {
- // nothing is available on the node where the co-location constraint pushes us
- throw new NoResourceAvailableException("Could not allocate a slot on instance " +
- constraint.getLocation() + ", as required by the co-location constraint.");
+ // simply nothing is available
+ throw new NoResourceAvailableException(task, getNumberOfAvailableInstances(),
+ getTotalNumberOfSlots(), getNumberOfAvailableSlots());
}
- } else {
- // got a non-local from the group, and no new one
+ }
+ else {
+ // got a non-local from the group, and no new one, so we use the non-local
+ // slot from the sharing group
toUse = slotFromGroup;
}
}
- else if (slotFromGroup == null || newSlot.getLocality() == Locality.LOCAL) {
- // new slot is preferable
+ else if (slotFromGroup == null || !slotFromGroup.isAlive() || newSlot.getLocality() == Locality.LOCAL) {
+ // if there is no slot from the group, or the new slot is local,
+ // then we use the new slot
if (slotFromGroup != null) {
slotFromGroup.releaseSlot();
}
-
toUse = newSlot;
}
else {
- // both are available and usable. neither is local
+ // both are available and usable. neither is local. in that case, we may
+ // as well use the slot from the sharing group, to minimize the number of
+ // instances that the job occupies
newSlot.releaseSlot();
toUse = slotFromGroup;
}
-
- // assign to the co-location hint, if we have one and it is unassigned
- // if it was assigned before and the new one is not local, it is a fail
- if (constraint != null) {
- if (constraint.isUnassigned() || toUse.getLocality() == Locality.LOCAL) {
- constraint.setSharedSlot(toUse.getParent());
- } else {
- // the fail
- throw new NoResourceAvailableException("Could not allocate a slot on instance " +
- constraint.getLocation() + ", as required by the co-location constraint.");
- }
+
+ // if this is the first slot for the co-location constraint, we lock
+ // the location, because we are going to use that slot
+ if (constraint != null && !constraint.isAssigned()) {
+ constraint.lockLocation();
}
- updateLocalityCounters(toUse.getLocality());
+ updateLocalityCounters(toUse.getLocality(), vertex, toUse.getInstance());
}
catch (NoResourceAvailableException e) {
throw e;
@@ -266,11 +296,14 @@ public class Scheduler implements InstanceListener, SlotAvailabilityListener {
}
return toUse;
- } else {
+ }
+ else {
+
// 2) === schedule without hints and sharing ===
+
SimpleSlot slot = getFreeSlotForTask(vertex, preferredLocations, forceExternalLocation);
if (slot != null) {
- updateLocalityCounters(slot.getLocality());
+ updateLocalityCounters(slot.getLocality(), vertex, slot.getInstance());
return slot;
}
else {
@@ -286,30 +319,14 @@ public class Scheduler implements InstanceListener, SlotAvailabilityListener {
+ " to any of the required hosts: " + hosts);
}
else {
- throw new NoResourceAvailableException(getNumberOfAvailableInstances(), getTotalNumberOfSlots(), getNumberOfAvailableSlots());
+ throw new NoResourceAvailableException(getNumberOfAvailableInstances(),
+ getTotalNumberOfSlots(), getNumberOfAvailableSlots());
}
}
}
}
}
- private String getHostnamesFromInstances(Iterable<Instance> instances) {
- StringBuilder bld = new StringBuilder();
-
- for (Instance i : instances) {
- bld.append(i.getInstanceConnectionInfo().getHostname());
- bld.append(", ");
- }
-
- if (bld.length() == 0) {
- return "";
- }
- else {
- bld.setLength(bld.length() - 2);
- return bld.toString();
- }
- }
-
/**
* Gets a suitable instance to schedule the vertex execution to.
* <p>
@@ -318,8 +335,9 @@ public class Scheduler implements InstanceListener, SlotAvailabilityListener {
* @param vertex The task to run.
* @return The instance to run the vertex on, it {@code null}, if no instance is available.
*/
- protected SimpleSlot getFreeSlotForTask(ExecutionVertex vertex, Iterable<Instance> requestedLocations, boolean localOnly) {
-
+ protected SimpleSlot getFreeSlotForTask(ExecutionVertex vertex,
+ Iterable<Instance> requestedLocations,
+ boolean localOnly) {
// we need potentially to loop multiple times, because there may be false positives
// in the set-with-available-instances
while (true) {
@@ -332,18 +350,8 @@ public class Scheduler implements InstanceListener, SlotAvailabilityListener {
Instance instanceToUse = instanceLocalityPair.getLeft();
Locality locality = instanceLocalityPair.getRight();
- if (LOG.isDebugEnabled()){
- if(locality == Locality.LOCAL){
- LOG.debug("Local assignment: " + vertex.getSimpleName() + " --> " + instanceToUse);
- }else if(locality == Locality.NON_LOCAL){
- LOG.debug("Non-local assignment: " + vertex.getSimpleName() + " --> " + instanceToUse);
- }else if(locality == Locality.UNCONSTRAINED) {
- LOG.debug("Unconstrained assignment: " + vertex.getSimpleName() + " --> " + instanceToUse);
- }
- }
-
try {
- SimpleSlot slot = instanceToUse.allocateSimpleSlot(vertex.getJobId(), vertex.getJobvertexId());
+ SimpleSlot slot = instanceToUse.allocateSimpleSlot(vertex.getJobId());
// if the instance has further available slots, re-add it to the set of available resources.
if (instanceToUse.hasResourcesAvailable()) {
@@ -365,54 +373,64 @@ public class Scheduler implements InstanceListener, SlotAvailabilityListener {
}
}
- protected SimpleSlot getFreeSubSlotForTask(ExecutionVertex vertex,
- Iterable<Instance> requestedLocations,
- SlotSharingGroupAssignment groupAssignment,
- CoLocationConstraint constraint,
- boolean localOnly) {
+ /**
+ * Tries to allocate a new slot for a vertex that is part of a slot sharing group. If one
+ * of the instances has a slot available, the method will allocate it as a shared slot, add that
+ * shared slot to the sharing group, and allocate a simple slot from that shared slot.
+ *
+ * <p>This method will try to allocate a slot from one of the local instances, and fall back to
+ * non-local instances, if permitted.</p>
+ *
+ * @param vertex The vertex to allocate the slot for.
+ * @param requestedLocations The locations that are considered local. May be null or empty, if the
+ * vertex has no location preferences.
+ * @param groupAssignment The slot sharing group of the vertex. Mandatory parameter.
+ * @param constraint The co-location constraint of the vertex. May be null.
+ * @param localOnly Flag to indicate if non-local choices are acceptable.
+ *
+ * @return A sub-slot for the given vertex, or {@code null}, if no slot is available.
+ */
+ protected SimpleSlot getNewSlotForSharingGroup(ExecutionVertex vertex,
+ Iterable<Instance> requestedLocations,
+ SlotSharingGroupAssignment groupAssignment,
+ CoLocationConstraint constraint,
+ boolean localOnly)
+ {
// we need potentially to loop multiple times, because there may be false positives
// in the set-with-available-instances
while (true) {
Pair<Instance, Locality> instanceLocalityPair = findInstance(requestedLocations, localOnly);
-
+
if (instanceLocalityPair == null) {
+ // nothing is available
return null;
}
- Instance instanceToUse = instanceLocalityPair.getLeft();
- Locality locality = instanceLocalityPair.getRight();
-
- if (LOG.isDebugEnabled()) {
- if (locality == Locality.LOCAL) {
- LOG.debug("Local assignment: " + vertex.getSimpleName() + " --> " + instanceToUse);
- } else if(locality == Locality.NON_LOCAL) {
- LOG.debug("Non-local assignment: " + vertex.getSimpleName() + " --> " + instanceToUse);
- } else if(locality == Locality.UNCONSTRAINED) {
- LOG.debug("Unconstrained assignment: " + vertex.getSimpleName() + " --> " + instanceToUse);
- }
- }
+ final Instance instanceToUse = instanceLocalityPair.getLeft();
+ final Locality locality = instanceLocalityPair.getRight();
try {
- AbstractID groupID = constraint == null ? vertex.getJobvertexId() : constraint.getGroupId();
-
- // root SharedSlot
- SharedSlot sharedSlot = instanceToUse.allocateSharedSlot(vertex.getJobId(), groupAssignment, groupID);
+ JobVertexID groupID = vertex.getJobvertexId();
+
+ // allocate a shared slot from the instance
+ SharedSlot sharedSlot = instanceToUse.allocateSharedSlot(vertex.getJobId(), groupAssignment);
// if the instance has further available slots, re-add it to the set of available resources.
if (instanceToUse.hasResourcesAvailable()) {
this.instancesWithAvailableResources.add(instanceToUse);
}
- if(sharedSlot != null){
- // If constraint != null, then slot nested in a SharedSlot nested in sharedSlot
- // If constraint == null, then slot nested in sharedSlot
- SimpleSlot slot = groupAssignment.addSharedSlotAndAllocateSubSlot(sharedSlot,
- locality, groupID, constraint);
+ if (sharedSlot != null) {
+ // add the shared slot to the assignment group and allocate a sub-slot
+ SimpleSlot slot = constraint == null ?
+ groupAssignment.addSharedSlotAndAllocateSubSlot(sharedSlot, locality, groupID) :
+ groupAssignment.addSharedSlotAndAllocateSubSlot(sharedSlot, locality, constraint);
- if(slot != null){
+ if (slot != null) {
return slot;
- } else {
- // release shared slot
+ }
+ else {
+ // could not add and allocate the sub-slot, so release shared slot
sharedSlot.releaseSlot();
}
}
@@ -428,58 +446,56 @@ public class Scheduler implements InstanceListener, SlotAvailabilityListener {
}
/**
- * NOTE: This method is not thread-safe, it needs to be synchronized by the caller.
- *
* Tries to find a requested instance. If no such instance is available it will return a non-
* local instance. If no such instance exists (all slots occupied), then return null.
+ *
+ * <p><b>NOTE:</b> This method is not thread-safe, it needs to be synchronized by the caller.</p>
*
- * @param requestedLocations
+ * @param requestedLocations The list of preferred instances. May be null or empty, which indicates that
+ * no locality preference exists.
+ * @param localOnly Flag to indicate whether only one of the exact local instances can be chosen.
*/
private Pair<Instance, Locality> findInstance(Iterable<Instance> requestedLocations, boolean localOnly){
- if (this.instancesWithAvailableResources.isEmpty()) {
- // check if the asynchronous calls did not yet return the queues
+ // drain the queue of newly available instances
+ while (this.newlyAvailableInstances.size() > 0) {
Instance queuedInstance = this.newlyAvailableInstances.poll();
- if (queuedInstance == null) {
- return null;
- } else {
+ if (queuedInstance != null) {
this.instancesWithAvailableResources.add(queuedInstance);
}
}
+
+ // if nothing is available at all, return null
+ if (this.instancesWithAvailableResources.isEmpty()) {
+ return null;
+ }
Iterator<Instance> locations = requestedLocations == null ? null : requestedLocations.iterator();
- Instance instanceToUse = null;
- Locality locality = Locality.UNCONSTRAINED;
-
if (locations != null && locations.hasNext()) {
// we have a locality preference
while (locations.hasNext()) {
Instance location = locations.next();
-
if (location != null && this.instancesWithAvailableResources.remove(location)) {
- instanceToUse = location;
- locality = Locality.LOCAL;
- break;
+ return new ImmutablePair<Instance, Locality>(location, Locality.LOCAL);
}
}
-
- if (instanceToUse == null) {
- if (localOnly) {
- return null;
- }
- else {
- instanceToUse = this.instancesWithAvailableResources.poll();
- locality = Locality.NON_LOCAL;
- }
+
+ // no local instance available
+ if (localOnly) {
+ return null;
+ }
+ else {
+ Instance instanceToUse = this.instancesWithAvailableResources.poll();
+ return new ImmutablePair<Instance, Locality>(instanceToUse, Locality.NON_LOCAL);
}
}
else {
- instanceToUse = this.instancesWithAvailableResources.poll();
+ // no location preference, so use some instance
+ Instance instanceToUse = this.instancesWithAvailableResources.poll();
+ return new ImmutablePair<Instance, Locality>(instanceToUse, Locality.UNCONSTRAINED);
}
-
- return new ImmutablePair<Instance, Locality>(instanceToUse, locality);
}
@Override
@@ -524,7 +540,7 @@ public class Scheduler implements InstanceListener, SlotAvailabilityListener {
ExecutionVertex vertex = task.getTaskToExecute().getVertex();
try {
- SimpleSlot newSlot = instance.allocateSimpleSlot(vertex.getJobId(), vertex.getJobvertexId());
+ SimpleSlot newSlot = instance.allocateSimpleSlot(vertex.getJobId());
if (newSlot != null) {
// success, remove from the task queue and notify the future
@@ -554,7 +570,7 @@ public class Scheduler implements InstanceListener, SlotAvailabilityListener {
}
}
- private void updateLocalityCounters(Locality locality) {
+ private void updateLocalityCounters(Locality locality, ExecutionVertex vertex, Instance location) {
switch (locality) {
case UNCONSTRAINED:
this.unconstrainedAssignments++;
@@ -568,12 +584,22 @@ public class Scheduler implements InstanceListener, SlotAvailabilityListener {
default:
throw new RuntimeException(locality.name());
}
+
+ if (LOG.isDebugEnabled()) {
+ switch (locality) {
+ case UNCONSTRAINED:
+ LOG.debug("Unconstrained assignment: " + vertex.getTaskNameWithSubtaskIndex() + " --> " + location);
+ break;
+ case LOCAL:
+ LOG.debug("Local assignment: " + vertex.getTaskNameWithSubtaskIndex() + " --> " + location);
+ break;
+ case NON_LOCAL:
+ LOG.debug("Non-local assignment: " + vertex.getTaskNameWithSubtaskIndex() + " --> " + location);
+ break;
+ }
+ }
}
-
-
-
-
// --------------------------------------------------------------------------------------------
// Instance Availability
// --------------------------------------------------------------------------------------------
@@ -646,7 +672,7 @@ public class Scheduler implements InstanceListener, SlotAvailabilityListener {
if (instance == null) {
throw new NullPointerException();
}
-
+
allInstances.remove(instance);
instancesWithAvailableResources.remove(instance);
@@ -755,7 +781,36 @@ public class Scheduler implements InstanceListener, SlotAvailabilityListener {
}
}
}
+
+
+ // ------------------------------------------------------------------------
+ // Utilities
+ // ------------------------------------------------------------------------
+
+ private static String getHostnamesFromInstances(Iterable<Instance> instances) {
+ StringBuilder bld = new StringBuilder();
+
+ boolean successive = false;
+ for (Instance i : instances) {
+ if (successive) {
+ bld.append(", ");
+ } else {
+ successive = true;
+ }
+ bld.append(i.getInstanceConnectionInfo().getHostname());
+ }
+
+ return bld.toString();
+ }
+ // ------------------------------------------------------------------------
+ // Nested members
+ // ------------------------------------------------------------------------
+
+ /**
+ * An entry in the queue of schedule requests. Contains the task to be scheduled and
+ * the future that tracks the completion.
+ */
private static final class QueuedTask {
private final ScheduledUnit task;
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/SlotSharingGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/SlotSharingGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/SlotSharingGroup.java
index dcde6b2..0fa1362 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/SlotSharingGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/SlotSharingGroup.java
@@ -22,6 +22,7 @@ import java.util.Collections;
import java.util.Set;
import java.util.TreeSet;
+import org.apache.flink.runtime.instance.SlotSharingGroupAssignment;
import org.apache.flink.runtime.jobgraph.JobVertexID;
/**
@@ -47,7 +48,8 @@ public class SlotSharingGroup implements java.io.Serializable {
this.ids.add(id);
}
}
-
+
+ // --------------------------------------------------------------------------------------------
public void addVertexToGroup(JobVertexID id) {
this.ids.add(id);
@@ -79,7 +81,9 @@ public class SlotSharingGroup implements java.io.Serializable {
this.taskAssignment = null;
}
- // --------------------------------------------------------------------------------------------
+ // ------------------------------------------------------------------------
+ // Utilities
+ // ------------------------------------------------------------------------
@Override
public String toString() {
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/SlotSharingGroupAssignment.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/SlotSharingGroupAssignment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/SlotSharingGroupAssignment.java
deleted file mode 100644
index 3c292f7..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/SlotSharingGroupAssignment.java
+++ /dev/null
@@ -1,485 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.jobmanager.scheduler;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.LinkedHashMap;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.commons.lang3.tuple.ImmutablePair;
-import org.apache.commons.lang3.tuple.Pair;
-import org.apache.flink.util.AbstractID;
-import org.apache.flink.runtime.executiongraph.ExecutionVertex;
-import org.apache.flink.runtime.instance.Instance;
-import org.apache.flink.runtime.instance.SharedSlot;
-import org.apache.flink.runtime.instance.SimpleSlot;
-import org.apache.flink.runtime.instance.Slot;
-import org.apache.flink.runtime.jobgraph.JobVertexID;
-import org.slf4j.Logger;
-
-
-public class SlotSharingGroupAssignment {
-
- private static final Logger LOG = Scheduler.LOG;
-
- private final Object lock = new Object();
-
- /** All slots currently allocated to this sharing group */
- private final Set<SharedSlot> allSlots = new LinkedHashSet<SharedSlot>();
-
- /** The slots available per vertex type (jid), keyed by instance, to make them locatable */
- private final Map<AbstractID, Map<Instance, List<SharedSlot>>> availableSlotsPerJid = new LinkedHashMap<AbstractID, Map<Instance, List<SharedSlot>>>();
-
- // --------------------------------------------------------------------------------------------
-
- public SimpleSlot addSharedSlotAndAllocateSubSlot(SharedSlot sharedSlot, Locality locality,
- AbstractID groupId, CoLocationConstraint constraint) {
-
- final Instance location = sharedSlot.getInstance();
-
- synchronized (lock) {
- // add to the total bookkeeping
- allSlots.add(sharedSlot);
-
- SimpleSlot subSlot = null;
-
- if (constraint == null) {
- // allocate us a sub slot to return
- subSlot = sharedSlot.allocateSubSlot(groupId);
- } else {
- // we need a colocation slot --> a SimpleSlot nested in a SharedSlot to host other colocated tasks
- SharedSlot constraintGroupSlot = sharedSlot.allocateSharedSlot(groupId);
-
- if(constraintGroupSlot == null) {
- subSlot = null;
- } else {
- subSlot = constraintGroupSlot.allocateSubSlot(null);
-
- // could not create a sub slot --> release constraintGroupSlot
- if(subSlot == null){
- constraintGroupSlot.releaseSlot();
- }
- }
- }
-
- // if sharedSlot is dead, but this should never happen since we just created a fresh
- // SharedSlot in the caller
- if(subSlot == null) {
- LOG.warn("Could not allocate a sub slot.");
-
- return null;
- } else {
- // preserve the locality information
- subSlot.setLocality(locality);
-
- boolean entryForNewJidExists = false;
-
- // let the other vertex types know about this one as well
- for (Map.Entry<AbstractID, Map<Instance, List<SharedSlot>>> entry : availableSlotsPerJid.entrySet()) {
-
- if (entry.getKey().equals(groupId)) {
- entryForNewJidExists = true;
- continue;
- }
-
- Map<Instance, List<SharedSlot>> available = entry.getValue();
- putIntoMultiMap(available, location, sharedSlot);
- }
-
- // make sure an empty entry exists for this group, if no other entry exists
- if (!entryForNewJidExists) {
- availableSlotsPerJid.put(groupId, new LinkedHashMap<Instance, List<SharedSlot>>());
- }
-
- return subSlot;
- }
- }
- }
-
- /**
- * Gets a slot suitable for the given task vertex. This method will prefer slots that are local
- * (with respect to {@link ExecutionVertex#getPreferredLocations()}), but will return non local
- * slots if no local slot is available. The method returns null, when no slot is available for the
- * given JobVertexID at all.
- *
- * @param vertex
- *
- * @return A task vertex for a task with the given JobVertexID, or null, if none is available.
- */
- public SimpleSlot getSlotForTask(ExecutionVertex vertex) {
- synchronized (lock) {
- Pair<SharedSlot, Locality> p = getSlotForTaskInternal(vertex.getJobvertexId(), vertex, vertex.getPreferredLocations(), false);
-
- if (p != null) {
- SharedSlot ss = p.getLeft();
- SimpleSlot slot = ss.allocateSubSlot(vertex.getJobvertexId());
- slot.setLocality(p.getRight());
- return slot;
- }
- else {
- return null;
- }
- }
-
- }
-
- public SimpleSlot getSlotForTask(ExecutionVertex vertex, CoLocationConstraint constraint) {
-
- synchronized (lock) {
- SharedSlot shared = constraint.getSharedSlot();
-
- if (shared != null && !shared.isDead()) {
- // initialized and set
- SimpleSlot subslot = shared.allocateSubSlot(null);
- subslot.setLocality(Locality.LOCAL);
- return subslot;
- }
- else if (shared == null) {
- // not initialized, grab a new slot. preferred locations are defined by the vertex
- // we only associate the slot with the constraint, if it was a local match
- Pair<SharedSlot, Locality> p = getSlotForTaskInternal(constraint.getGroupId(), vertex, vertex.getPreferredLocations(), false);
-
- if (p == null) {
- return null;
- } else {
- shared = p.getLeft();
- Locality l = p.getRight();
-
- // we need a colocation slot --> SimpleSlot nested in a SharedSlot to host other colocated tasks
- SharedSlot constraintGroupSlot = shared.allocateSharedSlot(constraint.getGroupId());
- // Depth=3 => groupID==null
- SimpleSlot sub = constraintGroupSlot.allocateSubSlot(null);
- sub.setLocality(l);
-
- if (l != Locality.NON_LOCAL) {
- constraint.setSharedSlot(constraintGroupSlot);
- }
- return sub;
- }
- }
- else {
- // disposed. get a new slot on the same instance
- Instance location = shared.getInstance();
- Pair<SharedSlot, Locality> p = getSlotForTaskInternal(constraint.getGroupId(), vertex, Collections.singleton(location), true);
-
- if (p == null) {
- return null;
- } else {
- shared = p.getLeft();
- // we need colocation slot --> SimpleSlot nested in a SharedSlot to host other colocated tasks
- SharedSlot constraintGroupSlot = shared.allocateSharedSlot(constraint.getGroupId());
- constraint.setSharedSlot(constraintGroupSlot);
- SimpleSlot subSlot = constraintGroupSlot.allocateSubSlot(null);
- subSlot.setLocality(Locality.LOCAL);
- return subSlot;
- }
- }
- }
- }
-
- /**
- * NOTE: This method is not synchronized by itself, needs to be synchronized externally.
- *
- * @return An allocated sub slot, or {@code null}, if no slot is available.
- */
- private Pair<SharedSlot, Locality> getSlotForTaskInternal(AbstractID groupId, ExecutionVertex vertex, Iterable<Instance> preferredLocations, boolean localOnly) {
- Map<Instance, List<SharedSlot>> slotsForGroup = availableSlotsPerJid.get(groupId);
-
- if (allSlots.isEmpty()) {
- return null;
- }
-
- // get the available slots for the group
- if (slotsForGroup == null) {
- // no task is yet scheduled for that group, so all slots are available
- slotsForGroup = new LinkedHashMap<Instance, List<SharedSlot>>();
- availableSlotsPerJid.put(groupId, slotsForGroup);
-
- for (SharedSlot availableSlot : allSlots) {
- putIntoMultiMap(slotsForGroup, availableSlot.getInstance(), availableSlot);
- }
- }
- else if (slotsForGroup.isEmpty()) {
- return null;
- }
-
- // check whether we can schedule the task to a preferred location
- boolean didNotGetPreferred = false;
-
- if (preferredLocations != null) {
- for (Instance location : preferredLocations) {
-
- // set the flag that we failed a preferred location. If one will be found,
- // we return early anyways and skip the flag evaluation
- didNotGetPreferred = true;
-
- SharedSlot slot = removeFromMultiMap(slotsForGroup, location);
- if (slot != null && !slot.isDead()) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Local assignment in shared group : " + vertex + " --> " + slot);
- }
-
- return new ImmutablePair<SharedSlot, Locality>(slot, Locality.LOCAL);
- }
- }
- }
-
- // if we want only local assignments, exit now with a "not found" result
- if (didNotGetPreferred && localOnly) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("No local assignment in shared possible for " + vertex);
- }
- return null;
- }
-
- // schedule the task to any available location
- SharedSlot slot = pollFromMultiMap(slotsForGroup);
- if (slot != null && !slot.isDead()) {
- if (LOG.isDebugEnabled()) {
- LOG.debug((didNotGetPreferred ? "Non-local" : "Unconstrained") + " assignment in shared group : " + vertex + " --> " + slot);
- }
-
- return new ImmutablePair<SharedSlot, Locality>(slot, didNotGetPreferred ? Locality.NON_LOCAL : Locality.UNCONSTRAINED);
- }
- else {
- return null;
- }
- }
-
- /**
- * Removes the shared slot from the assignment group.
- *
- * @param sharedSlot
- */
- private void removeSharedSlot(SharedSlot sharedSlot){
- if (!allSlots.contains(sharedSlot)) {
- throw new IllegalArgumentException("Slot was not associated with this SlotSharingGroup before.");
- }
-
- allSlots.remove(sharedSlot);
-
- Instance location = sharedSlot.getInstance();
-
- for(Map.Entry<AbstractID, Map<Instance, List<SharedSlot>>> mapEntry: availableSlotsPerJid.entrySet()){
- Map<Instance, List<SharedSlot>> map = mapEntry.getValue();
-
- List<SharedSlot> list = map.get(location);
-
- if(list == null || !list.remove(sharedSlot)){
- throw new IllegalStateException("Bug: SharedSlot was not available to another vertex type that it was not allocated for before.");
- }
-
- if(list.isEmpty()){
- map.remove(location);
- }
- }
-
- sharedSlot.markCancelled();
-
- returnAllocatedSlot(sharedSlot);
- }
-
- /**
- * Releases the shared slot from the assignment group.
- * @param sharedSlot The SharedSlot to be released
- */
- public void releaseSharedSlot(SharedSlot sharedSlot){
- synchronized (lock) {
- Set<Slot> subSlots = sharedSlot.getSubSlots();
-
- for(Slot subSlot: subSlots) {
-
- subSlot.markDisposed();
-
- if(subSlot instanceof SharedSlot){
- releaseSharedSlot((SharedSlot) subSlot);
- }else if(subSlot instanceof SimpleSlot){
- releaseSimpleSlot((SimpleSlot) subSlot);
- }
- }
-
- subSlots.clear();
-
- returnSlot(sharedSlot);
- }
- }
-
- /**
- * Releases the simple slot from the assignment group.
- * @param simpleSlot The SimpleSlot to be released
- */
- public void releaseSimpleSlot(SimpleSlot simpleSlot){
- synchronized (lock) {
- simpleSlot.cancel();
-
- returnSlot(simpleSlot);
- }
-
- }
-
- /**
- * Removes the given slot from the assignment group. If the slot is a root object, then it has
- * to be a SharedSlot and it is removed from the availableSlotsPerJid field and the slot is
- * returned to the instance. If the slot is a sub slot of the root slot, then this sub slot
- * is marked available again for tasks of the same group. Otherwise, the slot is simply removed
- * from its parent if it is not already marked as disposed. If a slot is already marked to be
- * disposed, then the releasing was called from a parent slot which will take care of the
- * disposal.
- *
- * IMPORTANT: The method is not synchronized. The caller is responsible for that.
- *
- * @param slot The slot to be returned.
- */
- private void returnSlot(Slot slot){
- // each slot can only be returned once, if a slot is returned then it should no longer be used --> markDead
- if(slot.markDead()) {
- // slot is a root slot
- if(slot.getParent() == null){
- // only SharedSlots are allowed to be root slots in a SlotSharingGroupAssignment
- if(slot instanceof SharedSlot){
- removeSharedSlot((SharedSlot) slot);
- } else {
- throw new IllegalStateException("Simple slot cannot be returned from SlotSharingGroupAssignment.");
- }
- } else {
- AbstractID groupID = slot.getGroupID();
- SharedSlot parent = slot.getParent();
-
- // Only colocation constraint slots (SimpleSlot nested in a SharedSlot nested in a SharedSlot) have a groupID==null
- // One can also say, all nested slots deeper than 2 have a groupID==null
- if(groupID != null){
- if (!allSlots.contains(parent)) {
- throw new IllegalArgumentException("Slot was not associated with this SlotSharingGroup before.");
- }
-
- // make the shared slot available to tasks within the group it available to
- Map<Instance, List<SharedSlot>> slotsForJid = availableSlotsPerJid.get(groupID);
-
- // sanity check
- if (slotsForJid == null) {
- throw new IllegalStateException("Trying to return a slot for group " + groupID +
- " when available slots indicated that all slots were available.");
- }
-
- putIntoMultiMap(slotsForJid, parent.getInstance(), parent);
- }
-
- // if no one else takes care of disposal, then remove the slot from the parent
- if(slot.markDisposed()) {
- if (slot.getParent().freeSubSlot(slot) == 0) {
- releaseSharedSlot(slot.getParent());
- }
- }
- }
- }
- }
-
- private void returnAllocatedSlot(SharedSlot slot){
- slot.getInstance().returnAllocatedSlot(slot);
- }
-
- // --------------------------------------------------------------------------------------------
- // State
- // --------------------------------------------------------------------------------------------
-
- public int getNumberOfSlots() {
- return allSlots.size();
- }
-
- public int getNumberOfAvailableSlotsForJid(JobVertexID jid) {
- synchronized (lock) {
- Map<Instance, List<SharedSlot>> available = availableSlotsPerJid.get(jid);
-
- if (available != null) {
- Set<SharedSlot> set = new HashSet<SharedSlot>();
-
- for (List<SharedSlot> list : available.values()) {
- for (SharedSlot slot : list) {
- set.add(slot);
- }
- }
-
- return set.size();
- } else {
- return allSlots.size();
- }
- }
- }
-
- // --------------------------------------------------------------------------------------------
-
-
-
- // --------------------------------------------------------------------------------------------
- // Utilities
- // --------------------------------------------------------------------------------------------
-
- private static final void putIntoMultiMap(Map<Instance, List<SharedSlot>> map, Instance location, SharedSlot slot) {
- List<SharedSlot> slotsForInstance = map.get(location);
- if (slotsForInstance == null) {
- slotsForInstance = new ArrayList<SharedSlot>();
- map.put(location, slotsForInstance);
- }
- slotsForInstance.add(slot);
- }
-
- private static final SharedSlot removeFromMultiMap(Map<Instance, List<SharedSlot>> map, Instance location) {
- List<SharedSlot> slotsForLocation = map.get(location);
-
- if (slotsForLocation == null) {
- return null;
- }
- else {
- SharedSlot slot = slotsForLocation.remove(slotsForLocation.size() - 1);
- if (slotsForLocation.isEmpty()) {
- map.remove(location);
- }
-
- return slot;
- }
- }
-
- private static final SharedSlot pollFromMultiMap(Map<Instance, List<SharedSlot>> map) {
- Iterator<Map.Entry<Instance, List<SharedSlot>>> iter = map.entrySet().iterator();
-
- while (iter.hasNext()) {
- List<SharedSlot> slots = iter.next().getValue();
-
- if (slots.isEmpty()) {
- iter.remove();
- }
- else if (slots.size() == 1) {
- SharedSlot slot = slots.remove(0);
- iter.remove();
- return slot;
- }
- else {
- return slots.remove(slots.size() - 1);
- }
- }
-
- return null;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java
index 42c3f84..a53c318 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java
@@ -348,4 +348,4 @@ public class ExecutionGraphDeploymentTest {
throw new Exception();
}
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java
index 1e78ac5..733ad11 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java
@@ -60,31 +60,31 @@ public class ExecutionStateProgressTest {
try {
final JobID jid = new JobID();
final JobVertexID vid = new JobVertexID();
-
+
AbstractJobVertex ajv = new AbstractJobVertex("TestVertex", vid);
ajv.setParallelism(3);
ajv.setInvokableClass(mock(AbstractInvokable.class).getClass());
-
+
ExecutionGraph graph = new ExecutionGraph(jid, "test job", new Configuration(),
AkkaUtils.getDefaultTimeout());
graph.attachJobGraph(Arrays.asList(ajv));
-
+
setGraphStatus(graph, JobStatus.RUNNING);
-
+
ExecutionJobVertex ejv = graph.getJobVertex(vid);
-
+
// mock resources and mock taskmanager
ActorRef taskManager = system.actorOf(Props.create(SimpleAcknowledgingTaskManager.class));
for (ExecutionVertex ee : ejv.getTaskVertices()) {
SimpleSlot slot = getInstance(taskManager).allocateSimpleSlot(jid);
ee.deployToSlot(slot);
}
-
+
// finish all
for (ExecutionVertex ee : ejv.getTaskVertices()) {
ee.executionFinished();
}
-
+
assertTrue(ejv.isInFinalState());
assertEquals(JobStatus.FINISHED, graph.getState());
}
@@ -93,4 +93,4 @@ public class ExecutionStateProgressTest {
fail(e.getMessage());
}
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java
index 757976d..eb0aab4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java
@@ -45,13 +45,14 @@ import org.apache.flink.runtime.messages.Messages;
import org.apache.flink.runtime.messages.TaskMessages;
import org.apache.flink.runtime.messages.TaskMessages.TaskOperationResult;
import org.apache.flink.runtime.testingUtils.TestingUtils;
+
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
@SuppressWarnings("serial")
public class ExecutionVertexCancelTest {
-
+
private static ActorSystem system;
@BeforeClass
@@ -68,24 +69,24 @@ public class ExecutionVertexCancelTest {
// --------------------------------------------------------------------------------------------
// Canceling in different states
// --------------------------------------------------------------------------------------------
-
+
@Test
public void testCancelFromCreated() {
try {
final JobVertexID jid = new JobVertexID();
final ExecutionJobVertex ejv = getExecutionVertex(jid);
-
+
final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
AkkaUtils.getDefaultTimeout());
-
+
assertEquals(ExecutionState.CREATED, vertex.getExecutionState());
-
+
vertex.cancel();
-
+
assertEquals(ExecutionState.CANCELED, vertex.getExecutionState());
-
+
assertNull(vertex.getFailureCause());
-
+
assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELING) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELED) > 0);
@@ -95,25 +96,25 @@ public class ExecutionVertexCancelTest {
fail(e.getMessage());
}
}
-
+
@Test
public void testCancelFromScheduled() {
try {
final JobVertexID jid = new JobVertexID();
final ExecutionJobVertex ejv = getExecutionVertex(jid);
-
+
final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
AkkaUtils.getDefaultTimeout());
-
+
setVertexState(vertex, ExecutionState.SCHEDULED);
assertEquals(ExecutionState.SCHEDULED, vertex.getExecutionState());
-
+
vertex.cancel();
-
+
assertEquals(ExecutionState.CANCELED, vertex.getExecutionState());
-
+
assertNull(vertex.getFailureCause());
-
+
assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELING) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELED) > 0);
@@ -123,7 +124,7 @@ public class ExecutionVertexCancelTest {
fail(e.getMessage());
}
}
-
+
@Test
public void testCancelConcurrentlyToDeploying_CallsNotOvertaking() {
new JavaTestKit(system){{
@@ -148,7 +149,7 @@ public class ExecutionVertexCancelTest {
new TaskOperationResult(execId, false))));
Instance instance = getInstance(taskManager);
- SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
+ SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
vertex.deployToSlot(slot);
@@ -187,15 +188,17 @@ public class ExecutionVertexCancelTest {
assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELING) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELED) > 0);
- }catch(Exception e){
+ }
+ catch(Exception e) {
e.printStackTrace();
fail(e.getMessage());
- }finally{
+ }
+ finally {
TestingUtils.setGlobalExecutionContext();
}
}};
}
-
+
@Test
public void testCancelConcurrentlyToDeploying_CallsOvertaking() {
new JavaTestKit(system){
@@ -272,7 +275,7 @@ public class ExecutionVertexCancelTest {
}
};
}
-
+
@Test
public void testCancelFromRunning() {
new JavaTestKit(system) {
@@ -291,7 +294,7 @@ public class ExecutionVertexCancelTest {
TaskOperationResult(execId, true))));
Instance instance = getInstance(taskManager);
- SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
+ SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
setVertexState(vertex, ExecutionState.RUNNING);
setVertexResource(vertex, slot);
@@ -319,7 +322,7 @@ public class ExecutionVertexCancelTest {
}
};
}
-
+
@Test
public void testRepeatedCancelFromRunning() {
new JavaTestKit(system) {
@@ -339,7 +342,7 @@ public class ExecutionVertexCancelTest {
TaskOperationResult(execId, true))));
Instance instance = getInstance(taskManager);
- SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
+ SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
setVertexState(vertex, ExecutionState.RUNNING);
setVertexResource(vertex, slot);
@@ -375,7 +378,7 @@ public class ExecutionVertexCancelTest {
}
};
}
-
+
@Test
public void testCancelFromRunningDidNotFindTask() {
// this may happen when the task finished or failed while the call was in progress
@@ -395,7 +398,7 @@ public class ExecutionVertexCancelTest {
TaskOperationResult(execId, false))));
Instance instance = getInstance(taskManager);
- SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
+ SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
setVertexState(vertex, ExecutionState.RUNNING);
setVertexResource(vertex, slot);
@@ -419,7 +422,7 @@ public class ExecutionVertexCancelTest {
}
};
}
-
+
@Test
public void testCancelCallFails() {
new JavaTestKit(system) {
@@ -436,7 +439,7 @@ public class ExecutionVertexCancelTest {
CancelSequenceTaskManagerCreator()));
Instance instance = getInstance(taskManager);
- SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
+ SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
setVertexState(vertex, ExecutionState.RUNNING);
setVertexResource(vertex, slot);
@@ -463,7 +466,7 @@ public class ExecutionVertexCancelTest {
}
};
}
-
+
@Test
public void testSendCancelAndReceiveFail() {
new JavaTestKit(system) {
@@ -482,7 +485,7 @@ public class ExecutionVertexCancelTest {
)));
Instance instance = getInstance(taskManager);
- SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
+ SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
setVertexState(vertex, ExecutionState.RUNNING);
setVertexResource(vertex, slot);
@@ -507,11 +510,11 @@ public class ExecutionVertexCancelTest {
}
};
}
-
+
// --------------------------------------------------------------------------------------------
// Actions after a vertex has been canceled or while canceling
// --------------------------------------------------------------------------------------------
-
+
@Test
public void testScheduleOrDeployAfterCancel() {
try {
@@ -521,26 +524,26 @@ public class ExecutionVertexCancelTest {
final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
AkkaUtils.getDefaultTimeout());
setVertexState(vertex, ExecutionState.CANCELED);
-
+
assertEquals(ExecutionState.CANCELED, vertex.getExecutionState());
-
+
// 1)
// scheduling after being created should be tolerated (no exception) because
// it can occur as the result of races
{
Scheduler scheduler = mock(Scheduler.class);
vertex.scheduleForExecution(scheduler, false);
-
+
assertEquals(ExecutionState.CANCELED, vertex.getExecutionState());
}
-
+
// 2)
// deploying after canceling from CREATED needs to raise an exception, because
// the scheduler (or any caller) needs to know that the slot should be released
try {
Instance instance = getInstance(ActorRef.noSender());
SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
-
+
vertex.deployToSlot(slot);
fail("Method should throw an exception");
}
@@ -553,60 +556,62 @@ public class ExecutionVertexCancelTest {
fail(e.getMessage());
}
}
-
+
@Test
public void testActionsWhileCancelling() {
-
+
try {
final JobVertexID jid = new JobVertexID();
final ExecutionJobVertex ejv = getExecutionVertex(jid);
-
+
// scheduling while canceling is an illegal state transition
try {
ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
AkkaUtils.getDefaultTimeout());
setVertexState(vertex, ExecutionState.CANCELING);
-
+
Scheduler scheduler = mock(Scheduler.class);
vertex.scheduleForExecution(scheduler, false);
}
catch (Exception e) {
fail("should not throw an exception");
}
-
-
+
+
// deploying while in canceling state is illegal (should immediately go to canceled)
try {
ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
AkkaUtils.getDefaultTimeout());
setVertexState(vertex, ExecutionState.CANCELING);
-
+
Instance instance = getInstance(ActorRef.noSender());
SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
-
+
vertex.deployToSlot(slot);
fail("Method should throw an exception");
}
- catch (IllegalStateException e) {}
-
-
+ catch (IllegalStateException e) {
+ // that is what we expect
+ }
+
+
// fail while canceling
{
ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
AkkaUtils.getDefaultTimeout());
-
+
Instance instance = getInstance(ActorRef.noSender());
SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
-
+
setVertexResource(vertex, slot);
setVertexState(vertex, ExecutionState.CANCELING);
-
+
Exception failureCause = new Exception("test exception");
-
+
vertex.fail(failureCause);
assertEquals(ExecutionState.FAILED, vertex.getExecutionState());
assertEquals(failureCause, vertex.getFailureCause());
-
+
assertTrue(slot.isReleased());
}
}
@@ -651,4 +656,4 @@ public class ExecutionVertexCancelTest {
}
}
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexDeploymentTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexDeploymentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexDeploymentTest.java
index c08ae01..eadf328 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexDeploymentTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexDeploymentTest.java
@@ -28,19 +28,21 @@ import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.testkit.JavaTestKit;
import akka.testkit.TestActorRef;
+
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.instance.SimpleSlot;
-import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.messages.TaskMessages.TaskOperationResult;
import org.apache.flink.runtime.testingUtils.TestingUtils;
+
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
public class ExecutionVertexDeploymentTest {
+
private static ActorSystem system;
@BeforeClass
@@ -61,41 +63,43 @@ public class ExecutionVertexDeploymentTest {
TestingUtils.setCallingThreadDispatcher(system);
ActorRef tm = TestActorRef.create(system, Props.create(SimpleAcknowledgingTaskManager
.class));
-
+
+ final ExecutionJobVertex ejv = getExecutionVertex(jid);
+
// mock taskmanager to simply accept the call
Instance instance = getInstance(tm);
+ final SimpleSlot slot = instance.allocateSimpleSlot(ejv.getJobId());
- final SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
-
- final ExecutionJobVertex ejv = getExecutionVertex(jid);
-
final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
AkkaUtils.getDefaultTimeout());
-
+
assertEquals(ExecutionState.CREATED, vertex.getExecutionState());
vertex.deployToSlot(slot);
assertEquals(ExecutionState.DEPLOYING, vertex.getExecutionState());
-
+
// no repeated scheduling
try {
vertex.deployToSlot(slot);
fail("Scheduled from wrong state");
}
- catch (IllegalStateException e) {}
-
+ catch (IllegalStateException e) {
+ // as expected
+ }
+
assertNull(vertex.getFailureCause());
-
+
assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.DEPLOYING) > 0);
}
catch (Exception e) {
e.printStackTrace();
fail(e.getMessage());
- }finally{
+ }
+ finally {
TestingUtils.setGlobalExecutionContext();
}
}
-
+
@Test
public void testDeployWithSynchronousAnswer() {
try {
@@ -105,31 +109,32 @@ public class ExecutionVertexDeploymentTest {
final TestActorRef<? extends Actor> simpleTaskManager = TestActorRef.create(system,
Props.create(SimpleAcknowledgingTaskManager.class));
-
- final Instance instance = getInstance(simpleTaskManager);
- final SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
-
final ExecutionJobVertex ejv = getExecutionVertex(jid);
-
+
+ final Instance instance = getInstance(simpleTaskManager);
+ final SimpleSlot slot = instance.allocateSimpleSlot(ejv.getJobId());
+
final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
AkkaUtils.getDefaultTimeout());
assertEquals(ExecutionState.CREATED, vertex.getExecutionState());
-
+
vertex.deployToSlot(slot);
-
+
assertEquals(ExecutionState.DEPLOYING, vertex.getExecutionState());
-
+
// no repeated scheduling
try {
vertex.deployToSlot(slot);
fail("Scheduled from wrong state");
}
- catch (IllegalStateException e) {}
-
+ catch (IllegalStateException e) {
+ // as expected
+ }
+
assertNull(vertex.getFailureCause());
-
+
assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.DEPLOYING) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.RUNNING) == 0);
@@ -137,48 +142,51 @@ public class ExecutionVertexDeploymentTest {
catch (Exception e) {
e.printStackTrace();
fail(e.getMessage());
- }finally{
+ }
+ finally {
TestingUtils.setGlobalExecutionContext();
}
}
-
+
@Test
public void testDeployWithAsynchronousAnswer() {
try {
final JobVertexID jid = new JobVertexID();
+ final ExecutionJobVertex ejv = getExecutionVertex(jid);
+
+ final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
+ AkkaUtils.getDefaultTimeout());
final TestActorRef<? extends Actor> simpleTaskManager = TestActorRef.create(system,
Props.create(SimpleAcknowledgingTaskManager.class));
-
- final Instance instance = getInstance(simpleTaskManager);
- final SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
-
- final ExecutionJobVertex ejv = getExecutionVertex(jid);
-
- final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
- AkkaUtils.getDefaultTimeout());
+ final Instance instance = getInstance(simpleTaskManager);
+ final SimpleSlot slot = instance.allocateSimpleSlot(ejv.getJobId());
assertEquals(ExecutionState.CREATED, vertex.getExecutionState());
-
+
vertex.deployToSlot(slot);
-
+
// no repeated scheduling
try {
vertex.deployToSlot(slot);
fail("Scheduled from wrong state");
}
- catch (IllegalStateException e) {}
+ catch (IllegalStateException e) {
+ // as expected
+ }
assertEquals(ExecutionState.DEPLOYING, vertex.getExecutionState());
-
+
// no repeated scheduling
try {
vertex.deployToSlot(slot);
fail("Scheduled from wrong state");
}
- catch (IllegalStateException e) {}
-
+ catch (IllegalStateException e) {
+ // as expected
+ }
+
assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.DEPLOYING) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.RUNNING) == 0);
@@ -188,33 +196,32 @@ public class ExecutionVertexDeploymentTest {
fail(e.getMessage());
}
}
-
+
@Test
public void testDeployFailedSynchronous() {
try {
TestingUtils.setCallingThreadDispatcher(system);
final JobVertexID jid = new JobVertexID();
+ final ExecutionJobVertex ejv = getExecutionVertex(jid);
+
+ final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
+ AkkaUtils.getDefaultTimeout());
final TestActorRef<? extends Actor> simpleTaskManager = TestActorRef.create(system,
Props.create(SimpleFailingTaskManager.class));
-
+
final Instance instance = getInstance(simpleTaskManager);
- final SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
-
- final ExecutionJobVertex ejv = getExecutionVertex(jid);
-
- final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
- AkkaUtils.getDefaultTimeout());
+ final SimpleSlot slot = instance.allocateSimpleSlot(ejv.getJobId());
assertEquals(ExecutionState.CREATED, vertex.getExecutionState());
-
+
vertex.deployToSlot(slot);
-
+
assertEquals(ExecutionState.FAILED, vertex.getExecutionState());
assertNotNull(vertex.getFailureCause());
assertTrue(vertex.getFailureCause().getMessage().contains(ERROR_MESSAGE));
-
+
assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.DEPLOYING) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.FAILED) > 0);
@@ -222,31 +229,30 @@ public class ExecutionVertexDeploymentTest {
catch (Exception e) {
e.printStackTrace();
fail(e.getMessage());
- }finally{
+ }
+ finally {
TestingUtils.setGlobalExecutionContext();
}
}
-
+
@Test
public void testDeployFailedAsynchronously() {
try {
final JobVertexID jid = new JobVertexID();
+ final ExecutionJobVertex ejv = getExecutionVertex(jid);
+ final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
+ AkkaUtils.getDefaultTimeout());
final TestActorRef<? extends Actor> simpleTaskManager = TestActorRef.create(system,
Props.create(SimpleFailingTaskManager.class));
-
- final Instance instance = getInstance(simpleTaskManager);
- final SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
-
- final ExecutionJobVertex ejv = getExecutionVertex(jid);
- final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
- AkkaUtils.getDefaultTimeout());
+ final Instance instance = getInstance(simpleTaskManager);
+ final SimpleSlot slot = instance.allocateSimpleSlot(ejv.getJobId());
assertEquals(ExecutionState.CREATED, vertex.getExecutionState());
-
+
vertex.deployToSlot(slot);
-
+
// wait until the state transition must be done
for (int i = 0; i < 100; i++) {
if (vertex.getExecutionState() == ExecutionState.FAILED && vertex.getFailureCause() != null) {
@@ -255,11 +261,11 @@ public class ExecutionVertexDeploymentTest {
Thread.sleep(10);
}
}
-
+
assertEquals(ExecutionState.FAILED, vertex.getExecutionState());
assertNotNull(vertex.getFailureCause());
assertTrue(vertex.getFailureCause().getMessage().contains(ERROR_MESSAGE));
-
+
assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.DEPLOYING) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.FAILED) > 0);
@@ -269,11 +275,15 @@ public class ExecutionVertexDeploymentTest {
fail(e.getMessage());
}
}
-
+
@Test
public void testFailExternallyDuringDeploy() {
try {
final JobVertexID jid = new JobVertexID();
+ final ExecutionJobVertex ejv = getExecutionVertex(jid);
+
+ final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
+ AkkaUtils.getDefaultTimeout());
final ActionQueue queue = new ActionQueue();
final TestingUtils.QueuedActionExecutionContext ec = new TestingUtils
@@ -284,12 +294,7 @@ public class ExecutionVertexDeploymentTest {
final TestActorRef<? extends Actor> simpleTaskManager = TestActorRef.create(system,
Props.create(SimpleAcknowledgingTaskManager.class));
final Instance instance = getInstance(simpleTaskManager);
- final SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
-
- final ExecutionJobVertex ejv = getExecutionVertex(jid);
-
- final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
- AkkaUtils.getDefaultTimeout());
+ final SimpleSlot slot = instance.allocateSimpleSlot(ejv.getJobId());
assertEquals(ExecutionState.CREATED, vertex.getExecutionState());
vertex.deployToSlot(slot);
@@ -306,17 +311,19 @@ public class ExecutionVertexDeploymentTest {
assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.DEPLOYING) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.FAILED) > 0);
- } catch (Exception e) {
+ }
+ catch (Exception e) {
e.printStackTrace();
fail(e.getMessage());
- }finally{
+ }
+ finally {
TestingUtils.setGlobalExecutionContext();
}
}
-
+
@Test
public void testFailCallOvertakesDeploymentAnswer() {
-
+
try {
ActionQueue queue = new ActionQueue();
TestingUtils.QueuedActionExecutionContext context = new TestingUtils
@@ -329,6 +336,7 @@ public class ExecutionVertexDeploymentTest {
final ExecutionJobVertex ejv = getExecutionVertex(jid);
final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
AkkaUtils.getDefaultTimeout());
+
final ExecutionAttemptID eid = vertex.getCurrentExecutionAttempt().getAttemptId();
final TestActorRef<? extends Actor> simpleTaskManager = TestActorRef.create(system, Props.create(new
@@ -336,16 +344,16 @@ public class ExecutionVertexDeploymentTest {
TaskOperationResult(eid, false), new TaskOperationResult(eid, true))));
final Instance instance = getInstance(simpleTaskManager);
- final SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
+ final SimpleSlot slot = instance.allocateSimpleSlot(ejv.getJobId());
assertEquals(ExecutionState.CREATED, vertex.getExecutionState());
vertex.deployToSlot(slot);
assertEquals(ExecutionState.DEPLOYING, vertex.getExecutionState());
-
+
Exception testError = new Exception("test error");
vertex.fail(testError);
-
+
assertEquals(ExecutionState.FAILED, vertex.getExecutionState());
// cancel call overtakes deploy call
@@ -361,7 +369,7 @@ public class ExecutionVertexDeploymentTest {
assertEquals(ExecutionState.FAILED, vertex.getExecutionState());
assertEquals(testError, vertex.getFailureCause());
-
+
assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.DEPLOYING) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.FAILED) > 0);
@@ -371,8 +379,9 @@ public class ExecutionVertexDeploymentTest {
catch (Exception e) {
e.printStackTrace();
fail(e.getMessage());
- }finally{
+ }
+ finally {
TestingUtils.setGlobalExecutionContext();
}
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexSchedulingTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexSchedulingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexSchedulingTest.java
index 06f0e9d..1e9c30b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexSchedulingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexSchedulingTest.java
@@ -28,17 +28,17 @@ import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.testkit.JavaTestKit;
import akka.testkit.TestActorRef;
+
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.instance.SimpleSlot;
-import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobmanager.scheduler.Scheduler;
import org.apache.flink.runtime.jobmanager.scheduler.ScheduledUnit;
import org.apache.flink.runtime.jobmanager.scheduler.SlotAllocationFuture;
-
import org.apache.flink.runtime.testingUtils.TestingUtils;
+
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -46,6 +46,7 @@ import org.junit.Test;
import org.mockito.Matchers;
public class ExecutionVertexSchedulingTest {
+
private static ActorSystem system;
@BeforeClass
@@ -58,93 +59,91 @@ public class ExecutionVertexSchedulingTest {
JavaTestKit.shutdownActorSystem(system);
system = null;
}
-
+
@Test
public void testSlotReleasedWhenScheduledImmediately() {
-
try {
- // a slot than cannot be deployed to
- final Instance instance = getInstance(ActorRef.noSender());
- final SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
- slot.cancel();
- assertFalse(slot.isReleased());
-
final ExecutionJobVertex ejv = getExecutionVertex(new JobVertexID());
final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
AkkaUtils.getDefaultTimeout());
+
+ // a slot than cannot be deployed to
+ final Instance instance = getInstance(ActorRef.noSender());
+ final SimpleSlot slot = instance.allocateSimpleSlot(ejv.getJobId());
+ slot.releaseSlot();
+ assertTrue(slot.isReleased());
+
Scheduler scheduler = mock(Scheduler.class);
when(scheduler.scheduleImmediately(Matchers.any(ScheduledUnit.class))).thenReturn(slot);
-
+
assertEquals(ExecutionState.CREATED, vertex.getExecutionState());
// try to deploy to the slot
vertex.scheduleForExecution(scheduler, false);
-
+
// will have failed
assertEquals(ExecutionState.FAILED, vertex.getExecutionState());
- assertTrue(slot.isReleased());
}
catch (Exception e) {
e.printStackTrace();
fail(e.getMessage());
}
}
-
+
@Test
public void testSlotReleasedWhenScheduledQueued() {
-
try {
- // a slot than cannot be deployed to
- final Instance instance = getInstance(ActorRef.noSender());
- final SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
- slot.cancel();
- assertFalse(slot.isReleased());
-
- final SlotAllocationFuture future = new SlotAllocationFuture();
-
final ExecutionJobVertex ejv = getExecutionVertex(new JobVertexID());
final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
AkkaUtils.getDefaultTimeout());
-
+
+ // a slot than cannot be deployed to
+ final Instance instance = getInstance(ActorRef.noSender());
+ final SimpleSlot slot = instance.allocateSimpleSlot(ejv.getJobId());
+
+ slot.releaseSlot();
+ assertTrue(slot.isReleased());
+
+ final SlotAllocationFuture future = new SlotAllocationFuture();
+
Scheduler scheduler = mock(Scheduler.class);
when(scheduler.scheduleQueued(Matchers.any(ScheduledUnit.class))).thenReturn(future);
-
+
assertEquals(ExecutionState.CREATED, vertex.getExecutionState());
// try to deploy to the slot
vertex.scheduleForExecution(scheduler, true);
-
+
// future has not yet a slot
assertEquals(ExecutionState.SCHEDULED, vertex.getExecutionState());
-
+
future.setSlot(slot);
-
+
// will have failed
assertEquals(ExecutionState.FAILED, vertex.getExecutionState());
- assertTrue(slot.isReleased());
}
catch (Exception e) {
e.printStackTrace();
fail(e.getMessage());
}
}
-
+
@Test
public void testScheduleToDeploying() {
try {
+ final ExecutionJobVertex ejv = getExecutionVertex(new JobVertexID());
+ final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
+ AkkaUtils.getDefaultTimeout());
+
TestingUtils.setCallingThreadDispatcher(system);
ActorRef tm = TestActorRef.create(system, Props.create(ExecutionGraphTestUtils
.SimpleAcknowledgingTaskManager.class));
final Instance instance = getInstance(tm);
- final SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
-
- final ExecutionJobVertex ejv = getExecutionVertex(new JobVertexID());
- final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
- AkkaUtils.getDefaultTimeout());
-
+ final SimpleSlot slot = instance.allocateSimpleSlot(ejv.getJobId());
+
Scheduler scheduler = mock(Scheduler.class);
when(scheduler.scheduleImmediately(Matchers.any(ScheduledUnit.class))).thenReturn(slot);
-
+
assertEquals(ExecutionState.CREATED, vertex.getExecutionState());
// try to deploy to the slot
@@ -158,4 +157,4 @@ public class ExecutionVertexSchedulingTest {
TestingUtils.setGlobalExecutionContext();
}
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/test/java/org/apache/flink/runtime/instance/InstanceTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/instance/InstanceTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/instance/InstanceTest.java
index c0ed629..595ac7e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/instance/InstanceTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/instance/InstanceTest.java
@@ -38,57 +38,52 @@ public class InstanceTest {
HardwareDescription hardwareDescription = new HardwareDescription(4, 2L*1024*1024*1024, 1024*1024*1024, 512*1024*1024);
InetAddress address = InetAddress.getByName("127.0.0.1");
InstanceConnectionInfo connection = new InstanceConnectionInfo(address, 10001);
-
+
Instance instance = new Instance(ActorRef.noSender(), connection, new InstanceID(), hardwareDescription, 4);
-
+
assertEquals(4, instance.getTotalNumberOfSlots());
assertEquals(4, instance.getNumberOfAvailableSlots());
assertEquals(0, instance.getNumberOfAllocatedSlots());
-
+
SimpleSlot slot1 = instance.allocateSimpleSlot(new JobID());
SimpleSlot slot2 = instance.allocateSimpleSlot(new JobID());
SimpleSlot slot3 = instance.allocateSimpleSlot(new JobID());
SimpleSlot slot4 = instance.allocateSimpleSlot(new JobID());
-
+
assertNotNull(slot1);
assertNotNull(slot2);
assertNotNull(slot3);
assertNotNull(slot4);
-
+
assertEquals(0, instance.getNumberOfAvailableSlots());
assertEquals(4, instance.getNumberOfAllocatedSlots());
- assertEquals(6, slot1.getSlotNumber() + slot2.getSlotNumber() +
+ assertEquals(6, slot1.getSlotNumber() + slot2.getSlotNumber() +
slot3.getSlotNumber() + slot4.getSlotNumber());
-
+
// no more slots
assertNull(instance.allocateSimpleSlot(new JobID()));
try {
instance.returnAllocatedSlot(slot2);
fail("instance accepted a non-cancelled slot.");
- } catch (IllegalArgumentException e) {
+ }
+ catch (IllegalArgumentException e) {
// good
}
-
+
// release the slots. this returns them to the instance
slot1.releaseSlot();
slot2.releaseSlot();
- assertFalse(instance.returnAllocatedSlot(slot1));
- assertFalse(instance.returnAllocatedSlot(slot2));
-
- // cancel some slots. this does not release them, yet
- slot3.cancel();
- slot4.cancel();
- assertTrue(instance.returnAllocatedSlot(slot3));
- assertTrue(instance.returnAllocatedSlot(slot4));
-
+ slot3.releaseSlot();
+ slot4.releaseSlot();
+
assertEquals(4, instance.getNumberOfAvailableSlots());
assertEquals(0, instance.getNumberOfAllocatedSlots());
-
+
assertFalse(instance.returnAllocatedSlot(slot1));
assertFalse(instance.returnAllocatedSlot(slot2));
assertFalse(instance.returnAllocatedSlot(slot3));
assertFalse(instance.returnAllocatedSlot(slot4));
-
+
assertEquals(4, instance.getNumberOfAvailableSlots());
assertEquals(0, instance.getNumberOfAllocatedSlots());
}
@@ -97,27 +92,27 @@ public class InstanceTest {
fail(e.getMessage());
}
}
-
+
@Test
public void testInstanceDies() {
try {
HardwareDescription hardwareDescription = new HardwareDescription(4, 2L*1024*1024*1024, 1024*1024*1024, 512*1024*1024);
InetAddress address = InetAddress.getByName("127.0.0.1");
InstanceConnectionInfo connection = new InstanceConnectionInfo(address, 10001);
-
+
Instance instance = new Instance(ActorRef.noSender(), connection, new InstanceID(), hardwareDescription, 3);
-
+
assertEquals(3, instance.getNumberOfAvailableSlots());
-
+
SimpleSlot slot1 = instance.allocateSimpleSlot(new JobID());
SimpleSlot slot2 = instance.allocateSimpleSlot(new JobID());
SimpleSlot slot3 = instance.allocateSimpleSlot(new JobID());
-
+
instance.markDead();
-
+
assertEquals(0, instance.getNumberOfAllocatedSlots());
assertEquals(0, instance.getNumberOfAvailableSlots());
-
+
assertTrue(slot1.isCanceled());
assertTrue(slot2.isCanceled());
assertTrue(slot3.isCanceled());
@@ -127,26 +122,26 @@ public class InstanceTest {
fail(e.getMessage());
}
}
-
+
@Test
public void testCancelAllSlots() {
try {
HardwareDescription hardwareDescription = new HardwareDescription(4, 2L*1024*1024*1024, 1024*1024*1024, 512*1024*1024);
InetAddress address = InetAddress.getByName("127.0.0.1");
InstanceConnectionInfo connection = new InstanceConnectionInfo(address, 10001);
-
+
Instance instance = new Instance(ActorRef.noSender(), connection, new InstanceID(), hardwareDescription, 3);
-
+
assertEquals(3, instance.getNumberOfAvailableSlots());
-
+
SimpleSlot slot1 = instance.allocateSimpleSlot(new JobID());
SimpleSlot slot2 = instance.allocateSimpleSlot(new JobID());
SimpleSlot slot3 = instance.allocateSimpleSlot(new JobID());
-
+
instance.cancelAndReleaseAllSlots();
-
+
assertEquals(3, instance.getNumberOfAvailableSlots());
-
+
assertTrue(slot1.isCanceled());
assertTrue(slot2.isCanceled());
assertTrue(slot3.isCanceled());
@@ -156,7 +151,7 @@ public class InstanceTest {
fail(e.getMessage());
}
}
-
+
/**
* It is crucial for some portions of the code that instance objects do not override equals and
* are only considered equal, if the references are equal.
@@ -172,4 +167,4 @@ public class InstanceTest {
fail(e.getMessage());
}
}
-}
+}
\ No newline at end of file
[4/7] flink git commit: [tests] Add a "not-so-mini-cluster" test to
validate scheduler for iterative jobs on many TaskManagers
Posted by se...@apache.org.
[tests] Add a "not-so-mini-cluster" test to validate scheduler for iterative jobs on many TaskManagers
This closes #731
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e96f2790
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e96f2790
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e96f2790
Branch: refs/heads/master
Commit: e96f279055123a6dd04fd1af2258fb839e56bd98
Parents: bf94a34
Author: Stephan Ewen <se...@apache.org>
Authored: Tue May 19 19:08:25 2015 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed May 27 16:36:02 2015 +0200
----------------------------------------------------------------------
.../java/org/apache/flink/test/misc/CCTest.java | 115 +++++++++++++++++++
1 file changed, 115 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/e96f2790/flink-tests/src/test/java/org/apache/flink/test/misc/CCTest.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/misc/CCTest.java b/flink-tests/src/test/java/org/apache/flink/test/misc/CCTest.java
new file mode 100644
index 0000000..cca6461
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/misc/CCTest.java
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.misc;
+
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.aggregation.Aggregations;
+import org.apache.flink.api.java.io.DiscardingOutputFormat;
+import org.apache.flink.api.java.operators.DeltaIteration;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.examples.java.graph.ConnectedComponents;
+import org.apache.flink.examples.java.graph.util.ConnectedComponentsData;
+import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster;
+
+import static org.junit.Assert.fail;
+
+public class CCTest {
+
+ private static final int PARALLELISM = 100;
+
+ public static void main(String[] args) {
+ if ((Runtime.getRuntime().maxMemory() >>> 20) < 5000) {
+ throw new RuntimeException("This est program needs to run with at least 5GB of heap space.");
+ }
+
+
+ LocalFlinkMiniCluster cluster = null;
+
+ try {
+ Configuration config = new Configuration();
+ config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, PARALLELISM);
+ config.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, 8);
+ config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 1);
+ config.setInteger(ConfigConstants.TASK_MANAGER_NETWORK_NUM_BUFFERS_KEY, 1000);
+ config.setInteger(ConfigConstants.TASK_MANAGER_NETWORK_BUFFER_SIZE_KEY, 8 * 1024);
+
+
+ config.setInteger("taskmanager.net.server.numThreads", 1);
+ config.setInteger("taskmanager.net.client.numThreads", 1);
+
+ cluster = new LocalFlinkMiniCluster(config, false);
+
+ runConnectedComponents(cluster.getJobManagerRPCPort());
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ finally {
+ if (cluster != null) {
+ cluster.shutdown();
+ }
+ }
+ }
+
+ private static void runConnectedComponents(int jmPort) throws Exception {
+
+ ExecutionEnvironment env = ExecutionEnvironment.createRemoteEnvironment("localhost", jmPort);
+ env.setParallelism(PARALLELISM);
+ env.getConfig().disableSysoutLogging();
+
+ // read vertex and edge data
+ DataSet<Long> vertices = ConnectedComponentsData.getDefaultVertexDataSet(env)
+ .rebalance();
+
+ DataSet<Tuple2<Long, Long>> edges = ConnectedComponentsData.getDefaultEdgeDataSet(env)
+ .rebalance()
+ .flatMap(new ConnectedComponents.UndirectEdge());
+
+ // assign the initial components (equal to the vertex id)
+ DataSet<Tuple2<Long, Long>> verticesWithInitialId = vertices
+ .map(new ConnectedComponents.DuplicateValue<Long>());
+
+ // open a delta iteration
+ DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iteration =
+ verticesWithInitialId.iterateDelta(verticesWithInitialId, 100, 0);
+
+ // apply the step logic: join with the edges, select the minimum neighbor,
+ // update if the component of the candidate is smaller
+ DataSet<Tuple2<Long, Long>> changes = iteration.getWorkset().join(edges)
+ .where(0).equalTo(0)
+ .with(new ConnectedComponents.NeighborWithComponentIDJoin())
+
+ .groupBy(0).aggregate(Aggregations.MIN, 1)
+
+ .join(iteration.getSolutionSet())
+ .where(0).equalTo(0)
+ .with(new ConnectedComponents.ComponentIdFilter());
+
+ // close the delta iteration (delta and new workset are identical)
+ DataSet<Tuple2<Long, Long>> result = iteration.closeWith(changes, changes);
+
+ result.output(new DiscardingOutputFormat<Tuple2<Long, Long>>());
+
+ env.execute();
+ }
+}
[7/7] flink git commit: [tests] Fix AvroExternalJarProgramITCase
logging
Posted by se...@apache.org.
[tests] Fix AvroExternalJarProgramITCase logging
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/bf94a341
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/bf94a341
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/bf94a341
Branch: refs/heads/master
Commit: bf94a341f81ade0b386d729df5cd4c89b36b1168
Parents: 017b4f0
Author: Stephan Ewen <se...@apache.org>
Authored: Tue May 26 22:56:50 2015 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed May 27 16:36:02 2015 +0200
----------------------------------------------------------------------
.../org/apache/flink/api/avro/AvroExternalJarProgramITCase.java | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/bf94a341/flink-staging/flink-avro/src/test/java/org/apache/flink/api/avro/AvroExternalJarProgramITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-avro/src/test/java/org/apache/flink/api/avro/AvroExternalJarProgramITCase.java b/flink-staging/flink-avro/src/test/java/org/apache/flink/api/avro/AvroExternalJarProgramITCase.java
index a955682..528e871 100644
--- a/flink-staging/flink-avro/src/test/java/org/apache/flink/api/avro/AvroExternalJarProgramITCase.java
+++ b/flink-staging/flink-avro/src/test/java/org/apache/flink/api/avro/AvroExternalJarProgramITCase.java
@@ -53,6 +53,7 @@ public class AvroExternalJarProgramITCase {
Client c = new Client(new InetSocketAddress("localhost", testMiniCluster.getJobManagerRPCPort()),
new Configuration(), program.getUserCodeClassLoader(), -1);
+ c.setPrintStatusDuringExecution(false);
c.run(program, 4, true);
}
catch (Throwable t) {
[5/7] flink git commit: [tests] Consolidate manually triggered tests
into a dedicated package
Posted by se...@apache.org.
[tests] Consolidate manually triggered tests into a dedicated package
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3061eb12
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3061eb12
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3061eb12
Branch: refs/heads/master
Commit: 3061eb126a69319a4ad45995f66f98c58dc23fdc
Parents: e96f279
Author: Stephan Ewen <se...@apache.org>
Authored: Wed May 27 11:27:56 2015 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed May 27 16:36:02 2015 +0200
----------------------------------------------------------------------
.../sort/MassiveStringSortingITCase.java | 388 ------------------
.../sort/MassiveStringValueSortingITCase.java | 402 ------------------
.../flink/test/manual/MassiveStringSorting.java | 391 ++++++++++++++++++
.../test/manual/MassiveStringValueSorting.java | 403 +++++++++++++++++++
.../test/manual/NotSoMiniClusterIterations.java | 117 ++++++
.../apache/flink/test/manual/package-info.java | 24 ++
.../java/org/apache/flink/test/misc/CCTest.java | 115 ------
7 files changed, 935 insertions(+), 905 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3061eb12/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/MassiveStringSortingITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/MassiveStringSortingITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/MassiveStringSortingITCase.java
deleted file mode 100644
index b96fdf2..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/MassiveStringSortingITCase.java
+++ /dev/null
@@ -1,388 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.operators.sort;
-
-import java.io.BufferedReader;
-import java.io.BufferedWriter;
-import java.io.File;
-import java.io.FileReader;
-import java.io.FileWriter;
-import java.io.IOException;
-import java.util.Random;
-
-import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.typeutils.TypeComparator;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.common.typeutils.base.StringComparator;
-import org.apache.flink.api.common.typeutils.base.StringSerializer;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.typeutils.TupleTypeInfo;
-import org.apache.flink.api.java.typeutils.TypeInfoParser;
-import org.apache.flink.api.java.typeutils.runtime.RuntimeSerializerFactory;
-import org.apache.flink.runtime.io.disk.iomanager.IOManager;
-import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
-import org.apache.flink.runtime.memorymanager.DefaultMemoryManager;
-import org.apache.flink.runtime.memorymanager.MemoryManager;
-import org.apache.flink.runtime.operators.testutils.DummyInvokable;
-import org.apache.flink.util.MutableObjectIterator;
-import org.junit.Assert;
-
-public class MassiveStringSortingITCase {
-
- private static final long SEED = 347569784659278346L;
-
-
- public void testStringSorting() {
- File input = null;
- File sorted = null;
-
- try {
- // the source file
- input = generateFileWithStrings(300000, "http://some-uri.com/that/is/a/common/prefix/to/all");
-
- // the sorted file
- sorted = File.createTempFile("sorted_strings", "txt");
-
- String[] command = {"/bin/bash","-c","export LC_ALL=\"C\" && cat \"" + input.getAbsolutePath() + "\" | sort > \"" + sorted.getAbsolutePath() + "\""};
-
- Process p = null;
- try {
- p = Runtime.getRuntime().exec(command);
- int retCode = p.waitFor();
- if (retCode != 0) {
- throw new Exception("Command failed with return code " + retCode);
- }
- p = null;
- } finally {
- if (p != null) {
- p.destroy();
- }
- }
-
- // sort the data
- UnilateralSortMerger<String> sorter = null;
- BufferedReader reader = null;
- BufferedReader verifyReader = null;
-
- try {
- MemoryManager mm = new DefaultMemoryManager(1024 * 1024, 1);
- IOManager ioMan = new IOManagerAsync();
-
- TypeSerializer<String> serializer = StringSerializer.INSTANCE;
- TypeComparator<String> comparator = new StringComparator(true);
-
- reader = new BufferedReader(new FileReader(input));
- MutableObjectIterator<String> inputIterator = new StringReaderMutableObjectIterator(reader);
-
- sorter = new UnilateralSortMerger<String>(mm, ioMan, inputIterator, new DummyInvokable(),
- new RuntimeSerializerFactory<String>(serializer, String.class), comparator, 1.0, 4, 0.8f);
-
- MutableObjectIterator<String> sortedData = sorter.getIterator();
-
- reader.close();
-
- // verify
- verifyReader = new BufferedReader(new FileReader(sorted));
- String next;
-
- while ((next = verifyReader.readLine()) != null) {
- String nextFromStratoSort = sortedData.next("");
-
- Assert.assertNotNull(nextFromStratoSort);
- Assert.assertEquals(next, nextFromStratoSort);
- }
- }
- finally {
- if (reader != null) {
- reader.close();
- }
- if (verifyReader != null) {
- verifyReader.close();
- }
- if (sorter != null) {
- sorter.close();
- }
- }
- }
- catch (Exception e) {
- System.err.println(e.getMessage());
- e.printStackTrace();
- Assert.fail(e.getMessage());
- }
- finally {
- if (input != null) {
- input.delete();
- }
- if (sorted != null) {
- sorted.delete();
- }
- }
- }
-
- @SuppressWarnings("unchecked")
- public void testStringTuplesSorting() {
- final int NUM_STRINGS = 300000;
-
- File input = null;
- File sorted = null;
-
- try {
- // the source file
- input = generateFileWithStringTuples(NUM_STRINGS, "http://some-uri.com/that/is/a/common/prefix/to/all");
-
- // the sorted file
- sorted = File.createTempFile("sorted_strings", "txt");
-
- String[] command = {"/bin/bash","-c","export LC_ALL=\"C\" && cat \"" + input.getAbsolutePath() + "\" | sort > \"" + sorted.getAbsolutePath() + "\""};
-
- Process p = null;
- try {
- p = Runtime.getRuntime().exec(command);
- int retCode = p.waitFor();
- if (retCode != 0) {
- throw new Exception("Command failed with return code " + retCode);
- }
- p = null;
- } finally {
- if (p != null) {
- p.destroy();
- }
- }
-
- // sort the data
- UnilateralSortMerger<Tuple2<String, String[]>> sorter = null;
- BufferedReader reader = null;
- BufferedReader verifyReader = null;
-
- try {
- MemoryManager mm = new DefaultMemoryManager(1024 * 1024, 1);
- IOManager ioMan = new IOManagerAsync();
-
- TupleTypeInfo<Tuple2<String, String[]>> typeInfo = (TupleTypeInfo<Tuple2<String, String[]>>) (TupleTypeInfo<?>) TypeInfoParser.parse("Tuple2<String, String[]>");
-
- TypeSerializer<Tuple2<String, String[]>> serializer = typeInfo.createSerializer(new ExecutionConfig());
- TypeComparator<Tuple2<String, String[]>> comparator = typeInfo.createComparator(new int[] { 0 }, new boolean[] { true }, 0, new ExecutionConfig());
-
- reader = new BufferedReader(new FileReader(input));
- MutableObjectIterator<Tuple2<String, String[]>> inputIterator = new StringTupleReaderMutableObjectIterator(reader);
-
- sorter = new UnilateralSortMerger<Tuple2<String, String[]>>(mm, ioMan, inputIterator, new DummyInvokable(),
- new RuntimeSerializerFactory<Tuple2<String, String[]>>(serializer, (Class<Tuple2<String, String[]>>) (Class<?>) Tuple2.class), comparator, 1.0, 4, 0.8f);
-
-
-
- // use this part to verify that all if good when sorting in memory
-
-// List<MemorySegment> memory = mm.allocatePages(new DummyInvokable(), mm.computeNumberOfPages(1024*1024*1024));
-// NormalizedKeySorter<Tuple2<String, String[]>> nks = new NormalizedKeySorter<Tuple2<String,String[]>>(serializer, comparator, memory);
-//
-// {
-// Tuple2<String, String[]> wi = new Tuple2<String, String[]>("", new String[0]);
-// while ((wi = inputIterator.next(wi)) != null) {
-// Assert.assertTrue(nks.write(wi));
-// }
-//
-// new QuickSort().sort(nks);
-// }
-//
-// MutableObjectIterator<Tuple2<String, String[]>> sortedData = nks.getIterator();
-
- MutableObjectIterator<Tuple2<String, String[]>> sortedData = sorter.getIterator();
- reader.close();
-
- // verify
- verifyReader = new BufferedReader(new FileReader(sorted));
- MutableObjectIterator<Tuple2<String, String[]>> verifyIterator = new StringTupleReaderMutableObjectIterator(verifyReader);
-
- Tuple2<String, String[]> next = new Tuple2<String, String[]>("", new String[0]);
- Tuple2<String, String[]> nextFromStratoSort = new Tuple2<String, String[]>("", new String[0]);
-
- int num = 0;
-
- while ((next = verifyIterator.next(next)) != null) {
- num++;
-
- nextFromStratoSort = sortedData.next(nextFromStratoSort);
- Assert.assertNotNull(nextFromStratoSort);
-
- Assert.assertEquals(next.f0, nextFromStratoSort.f0);
- Assert.assertArrayEquals(next.f1, nextFromStratoSort.f1);
- }
-
- Assert.assertNull(sortedData.next(nextFromStratoSort));
- Assert.assertEquals(NUM_STRINGS, num);
-
- }
- finally {
- if (reader != null) {
- reader.close();
- }
- if (verifyReader != null) {
- verifyReader.close();
- }
- if (sorter != null) {
- sorter.close();
- }
- }
- }
- catch (Exception e) {
- System.err.println(e.getMessage());
- e.printStackTrace();
- Assert.fail(e.getMessage());
- }
- finally {
- if (input != null) {
- input.delete();
- }
- if (sorted != null) {
- sorted.delete();
- }
- }
- }
-
- // --------------------------------------------------------------------------------------------
-
- private static final class StringReaderMutableObjectIterator implements MutableObjectIterator<String> {
-
- private final BufferedReader reader;
-
- public StringReaderMutableObjectIterator(BufferedReader reader) {
- this.reader = reader;
- }
-
- @Override
- public String next(String reuse) throws IOException {
- return reader.readLine();
- }
-
- @Override
- public String next() throws IOException {
- return reader.readLine();
- }
- }
-
- private static final class StringTupleReaderMutableObjectIterator implements MutableObjectIterator<Tuple2<String, String[]>> {
-
- private final BufferedReader reader;
-
- public StringTupleReaderMutableObjectIterator(BufferedReader reader) {
- this.reader = reader;
- }
-
- @Override
- public Tuple2<String, String[]> next(Tuple2<String, String[]> reuse) throws IOException {
- String line = reader.readLine();
- if (line == null) {
- return null;
- }
-
- String[] parts = line.split(" ");
- reuse.f0 = parts[0];
- reuse.f1 = parts;
- return reuse;
- }
-
- @Override
- public Tuple2<String, String[]> next() throws IOException {
- return next(new Tuple2<String, String[]>());
- }
- }
-
- // --------------------------------------------------------------------------------------------
-
- private File generateFileWithStrings(int numStrings, String prefix) throws IOException {
- final Random rnd = new Random(SEED);
-
- final StringBuilder bld = new StringBuilder();
- final int resetValue = prefix.length();
-
- bld.append(prefix);
-
- File f = File.createTempFile("strings", "txt");
- BufferedWriter wrt = null;
- try {
- wrt = new BufferedWriter(new FileWriter(f));
-
- for (int i = 0 ; i < numStrings; i++) {
- bld.setLength(resetValue);
-
- int len = rnd.nextInt(20) + 300;
- for (int k = 0; k < len; k++) {
- char c = (char) (rnd.nextInt(80) + 40);
- bld.append(c);
- }
-
- String str = bld.toString();
- wrt.write(str);
- wrt.newLine();
- }
- } finally {
- wrt.close();
- }
-
- return f;
- }
-
- private File generateFileWithStringTuples(int numStrings, String prefix) throws IOException {
- final Random rnd = new Random(SEED);
-
- final StringBuilder bld = new StringBuilder();
-
- File f = File.createTempFile("strings", "txt");
- BufferedWriter wrt = null;
- try {
- wrt = new BufferedWriter(new FileWriter(f));
-
- for (int i = 0 ; i < numStrings; i++) {
- bld.setLength(0);
-
- int numComps = rnd.nextInt(5) + 1;
-
- for (int z = 0; z < numComps; z++) {
- if (z > 0) {
- bld.append(' ');
- }
- bld.append(prefix);
-
- int len = rnd.nextInt(20) + 10;
- for (int k = 0; k < len; k++) {
- char c = (char) (rnd.nextInt(80) + 40);
- bld.append(c);
- }
- }
-
- String str = bld.toString();
-
- wrt.write(str);
- wrt.newLine();
- }
- } finally {
- wrt.close();
- }
-
- return f;
- }
-
- // --------------------------------------------------------------------------------------------
-
- public static void main(String[] args) {
- new MassiveStringSortingITCase().testStringSorting();
- new MassiveStringSortingITCase().testStringTuplesSorting();
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/3061eb12/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/MassiveStringValueSortingITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/MassiveStringValueSortingITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/MassiveStringValueSortingITCase.java
deleted file mode 100644
index 5265cdb..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/MassiveStringValueSortingITCase.java
+++ /dev/null
@@ -1,402 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.operators.sort;
-
-import java.io.BufferedReader;
-import java.io.BufferedWriter;
-import java.io.File;
-import java.io.FileReader;
-import java.io.FileWriter;
-import java.io.IOException;
-import java.util.Random;
-
-import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.typeutils.TypeComparator;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.typeutils.TupleTypeInfo;
-import org.apache.flink.api.java.typeutils.TypeInfoParser;
-import org.apache.flink.api.java.typeutils.runtime.CopyableValueComparator;
-import org.apache.flink.api.java.typeutils.runtime.CopyableValueSerializer;
-import org.apache.flink.api.java.typeutils.runtime.RuntimeSerializerFactory;
-import org.apache.flink.runtime.io.disk.iomanager.IOManager;
-import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
-import org.apache.flink.runtime.memorymanager.DefaultMemoryManager;
-import org.apache.flink.runtime.memorymanager.MemoryManager;
-import org.apache.flink.runtime.operators.testutils.DummyInvokable;
-import org.apache.flink.types.StringValue;
-import org.apache.flink.util.MutableObjectIterator;
-import org.junit.Assert;
-
-public class MassiveStringValueSortingITCase {
-
- private static final long SEED = 347569784659278346L;
-
- public void testStringValueSorting() {
- File input = null;
- File sorted = null;
-
- try {
- // the source file
- input = generateFileWithStrings(300000, "http://some-uri.com/that/is/a/common/prefix/to/all");
-
- // the sorted file
- sorted = File.createTempFile("sorted_strings", "txt");
-
- String[] command = {"/bin/bash","-c","export LC_ALL=\"C\" && cat \"" + input.getAbsolutePath() + "\" | sort > \"" + sorted.getAbsolutePath() + "\""};
-
- Process p = null;
- try {
- p = Runtime.getRuntime().exec(command);
- int retCode = p.waitFor();
- if (retCode != 0) {
- throw new Exception("Command failed with return code " + retCode);
- }
- p = null;
- } finally {
- if (p != null) {
- p.destroy();
- }
- }
-
- // sort the data
- UnilateralSortMerger<StringValue> sorter = null;
- BufferedReader reader = null;
- BufferedReader verifyReader = null;
-
- try {
- MemoryManager mm = new DefaultMemoryManager(1024 * 1024, 1);
- IOManager ioMan = new IOManagerAsync();
-
- TypeSerializer<StringValue> serializer = new CopyableValueSerializer<StringValue>(StringValue.class);
- TypeComparator<StringValue> comparator = new CopyableValueComparator<StringValue>(true, StringValue.class);
-
- reader = new BufferedReader(new FileReader(input));
- MutableObjectIterator<StringValue> inputIterator = new StringValueReaderMutableObjectIterator(reader);
-
- sorter = new UnilateralSortMerger<StringValue>(mm, ioMan, inputIterator, new DummyInvokable(),
- new RuntimeSerializerFactory<StringValue>(serializer, StringValue.class), comparator, 1.0, 4, 0.8f);
-
- MutableObjectIterator<StringValue> sortedData = sorter.getIterator();
-
- reader.close();
-
- // verify
- verifyReader = new BufferedReader(new FileReader(sorted));
- String nextVerify;
- StringValue nextFromFlinkSort = new StringValue();
-
- while ((nextVerify = verifyReader.readLine()) != null) {
- nextFromFlinkSort = sortedData.next(nextFromFlinkSort);
-
- Assert.assertNotNull(nextFromFlinkSort);
- Assert.assertEquals(nextVerify, nextFromFlinkSort.getValue());
- }
- }
- finally {
- if (reader != null) {
- reader.close();
- }
- if (verifyReader != null) {
- verifyReader.close();
- }
- if (sorter != null) {
- sorter.close();
- }
- }
- }
- catch (Exception e) {
- System.err.println(e.getMessage());
- e.printStackTrace();
- Assert.fail(e.getMessage());
- }
- finally {
- if (input != null) {
- input.delete();
- }
- if (sorted != null) {
- sorted.delete();
- }
- }
- }
-
- @SuppressWarnings("unchecked")
- public void testStringValueTuplesSorting() {
- final int NUM_STRINGS = 300000;
-
- File input = null;
- File sorted = null;
-
- try {
- // the source file
- input = generateFileWithStringTuples(NUM_STRINGS, "http://some-uri.com/that/is/a/common/prefix/to/all");
-
- // the sorted file
- sorted = File.createTempFile("sorted_strings", "txt");
-
- String[] command = {"/bin/bash","-c","export LC_ALL=\"C\" && cat \"" + input.getAbsolutePath() + "\" | sort > \"" + sorted.getAbsolutePath() + "\""};
-
- Process p = null;
- try {
- p = Runtime.getRuntime().exec(command);
- int retCode = p.waitFor();
- if (retCode != 0) {
- throw new Exception("Command failed with return code " + retCode);
- }
- p = null;
- } finally {
- if (p != null) {
- p.destroy();
- }
- }
-
- // sort the data
- UnilateralSortMerger<Tuple2<StringValue, StringValue[]>> sorter = null;
- BufferedReader reader = null;
- BufferedReader verifyReader = null;
-
- try {
- MemoryManager mm = new DefaultMemoryManager(1024 * 1024, 1);
- IOManager ioMan = new IOManagerAsync();
-
- TupleTypeInfo<Tuple2<StringValue, StringValue[]>> typeInfo = (TupleTypeInfo<Tuple2<StringValue, StringValue[]>>) (TupleTypeInfo<?>)
- TypeInfoParser.parse("Tuple2<org.apache.flink.types.StringValue, org.apache.flink.types.StringValue[]>");
-
- TypeSerializer<Tuple2<StringValue, StringValue[]>> serializer = typeInfo.createSerializer(new ExecutionConfig());
- TypeComparator<Tuple2<StringValue, StringValue[]>> comparator = typeInfo.createComparator(new int[] { 0 }, new boolean[] { true }, 0, new ExecutionConfig());
-
- reader = new BufferedReader(new FileReader(input));
- MutableObjectIterator<Tuple2<StringValue, StringValue[]>> inputIterator = new StringValueTupleReaderMutableObjectIterator(reader);
-
- sorter = new UnilateralSortMerger<Tuple2<StringValue, StringValue[]>>(mm, ioMan, inputIterator, new DummyInvokable(),
- new RuntimeSerializerFactory<Tuple2<StringValue, StringValue[]>>(serializer, (Class<Tuple2<StringValue, StringValue[]>>) (Class<?>) Tuple2.class), comparator, 1.0, 4, 0.8f);
-
-
-
- // use this part to verify that all if good when sorting in memory
-
-// List<MemorySegment> memory = mm.allocatePages(new DummyInvokable(), mm.computeNumberOfPages(1024*1024*1024));
-// NormalizedKeySorter<Tuple2<String, String[]>> nks = new NormalizedKeySorter<Tuple2<String,String[]>>(serializer, comparator, memory);
-//
-// {
-// Tuple2<String, String[]> wi = new Tuple2<String, String[]>("", new String[0]);
-// while ((wi = inputIterator.next(wi)) != null) {
-// Assert.assertTrue(nks.write(wi));
-// }
-//
-// new QuickSort().sort(nks);
-// }
-//
-// MutableObjectIterator<Tuple2<String, String[]>> sortedData = nks.getIterator();
-
- MutableObjectIterator<Tuple2<StringValue, StringValue[]>> sortedData = sorter.getIterator();
- reader.close();
-
- // verify
- verifyReader = new BufferedReader(new FileReader(sorted));
- MutableObjectIterator<Tuple2<StringValue, StringValue[]>> verifyIterator = new StringValueTupleReaderMutableObjectIterator(verifyReader);
-
- Tuple2<StringValue, StringValue[]> nextVerify = new Tuple2<StringValue, StringValue[]>(new StringValue(), new StringValue[0]);
- Tuple2<StringValue, StringValue[]> nextFromFlinkSort = new Tuple2<StringValue, StringValue[]>(new StringValue(), new StringValue[0]);
-
- int num = 0;
-
- while ((nextVerify = verifyIterator.next(nextVerify)) != null) {
- num++;
-
- nextFromFlinkSort = sortedData.next(nextFromFlinkSort);
- Assert.assertNotNull(nextFromFlinkSort);
-
- Assert.assertEquals(nextVerify.f0, nextFromFlinkSort.f0);
- Assert.assertArrayEquals(nextVerify.f1, nextFromFlinkSort.f1);
- }
-
- Assert.assertNull(sortedData.next(nextFromFlinkSort));
- Assert.assertEquals(NUM_STRINGS, num);
-
- }
- finally {
- if (reader != null) {
- reader.close();
- }
- if (verifyReader != null) {
- verifyReader.close();
- }
- if (sorter != null) {
- sorter.close();
- }
- }
- }
- catch (Exception e) {
- System.err.println(e.getMessage());
- e.printStackTrace();
- Assert.fail(e.getMessage());
- }
- finally {
- if (input != null) {
- input.delete();
- }
- if (sorted != null) {
- sorted.delete();
- }
- }
- }
-
- // --------------------------------------------------------------------------------------------
-
- private static final class StringValueReaderMutableObjectIterator implements MutableObjectIterator<StringValue> {
-
- private final BufferedReader reader;
-
- public StringValueReaderMutableObjectIterator(BufferedReader reader) {
- this.reader = reader;
- }
-
- @Override
- public StringValue next(StringValue reuse) throws IOException {
- String line = reader.readLine();
-
- if (line == null) {
- return null;
- }
-
- reuse.setValue(line);
- return reuse;
- }
-
- @Override
- public StringValue next() throws IOException {
- return next(new StringValue());
- }
- }
-
- private static final class StringValueTupleReaderMutableObjectIterator implements MutableObjectIterator<Tuple2<StringValue, StringValue[]>> {
-
- private final BufferedReader reader;
-
- public StringValueTupleReaderMutableObjectIterator(BufferedReader reader) {
- this.reader = reader;
- }
-
- @Override
- public Tuple2<StringValue, StringValue[]> next(Tuple2<StringValue, StringValue[]> reuse) throws IOException {
- String line = reader.readLine();
- if (line == null) {
- return null;
- }
-
- String[] parts = line.split(" ");
- reuse.f0.setValue(parts[0]);
- reuse.f1 = new StringValue[parts.length];
-
- for (int i = 0; i < parts.length; i++) {
- reuse.f1[i] = new StringValue(parts[i]);
- }
-
- return reuse;
- }
-
- @Override
- public Tuple2<StringValue, StringValue[]> next() throws IOException {
- return next(new Tuple2<StringValue, StringValue[]>(new StringValue(), new StringValue[0]));
- }
- }
-
- // --------------------------------------------------------------------------------------------
-
- private File generateFileWithStrings(int numStrings, String prefix) throws IOException {
- final Random rnd = new Random(SEED);
-
- final StringBuilder bld = new StringBuilder();
- final int resetValue = prefix.length();
-
- bld.append(prefix);
-
- File f = File.createTempFile("strings", "txt");
- BufferedWriter wrt = null;
- try {
- wrt = new BufferedWriter(new FileWriter(f));
-
- for (int i = 0 ; i < numStrings; i++) {
- bld.setLength(resetValue);
-
- int len = rnd.nextInt(20) + 300;
- for (int k = 0; k < len; k++) {
- char c = (char) (rnd.nextInt(80) + 40);
- bld.append(c);
- }
-
- String str = bld.toString();
- wrt.write(str);
- wrt.newLine();
- }
- } finally {
- wrt.close();
- }
-
- return f;
- }
-
- private File generateFileWithStringTuples(int numStrings, String prefix) throws IOException {
- final Random rnd = new Random(SEED);
-
- final StringBuilder bld = new StringBuilder();
-
- File f = File.createTempFile("strings", "txt");
- BufferedWriter wrt = null;
- try {
- wrt = new BufferedWriter(new FileWriter(f));
-
- for (int i = 0 ; i < numStrings; i++) {
- bld.setLength(0);
-
- int numComps = rnd.nextInt(5) + 1;
-
- for (int z = 0; z < numComps; z++) {
- if (z > 0) {
- bld.append(' ');
- }
- bld.append(prefix);
-
- int len = rnd.nextInt(20) + 10;
- for (int k = 0; k < len; k++) {
- char c = (char) (rnd.nextInt(80) + 40);
- bld.append(c);
- }
- }
-
- String str = bld.toString();
-
- wrt.write(str);
- wrt.newLine();
- }
- } finally {
- wrt.close();
- }
-
- return f;
- }
-
- // --------------------------------------------------------------------------------------------
-
- public static void main(String[] args) {
- new MassiveStringValueSortingITCase().testStringValueSorting();
- new MassiveStringValueSortingITCase().testStringValueTuplesSorting();
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/3061eb12/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringSorting.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringSorting.java b/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringSorting.java
new file mode 100644
index 0000000..2da29e8
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringSorting.java
@@ -0,0 +1,391 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.manual;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileReader;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.Random;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.StringComparator;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.java.typeutils.TypeInfoParser;
+import org.apache.flink.api.java.typeutils.runtime.RuntimeSerializerFactory;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
+import org.apache.flink.runtime.memorymanager.DefaultMemoryManager;
+import org.apache.flink.runtime.memorymanager.MemoryManager;
+import org.apache.flink.runtime.operators.sort.UnilateralSortMerger;
+import org.apache.flink.runtime.operators.testutils.DummyInvokable;
+import org.apache.flink.util.MutableObjectIterator;
+import org.junit.Assert;
+
+public class MassiveStringSorting {
+
+ private static final long SEED = 347569784659278346L;
+
+
+ public void testStringSorting() {
+ File input = null;
+ File sorted = null;
+
+ try {
+ // the source file
+ input = generateFileWithStrings(300000, "http://some-uri.com/that/is/a/common/prefix/to/all");
+
+ // the sorted file
+ sorted = File.createTempFile("sorted_strings", "txt");
+
+ String[] command = {"/bin/bash","-c","export LC_ALL=\"C\" && cat \"" + input.getAbsolutePath() + "\" | sort > \"" + sorted.getAbsolutePath() + "\""};
+
+ Process p = null;
+ try {
+ p = Runtime.getRuntime().exec(command);
+ int retCode = p.waitFor();
+ if (retCode != 0) {
+ throw new Exception("Command failed with return code " + retCode);
+ }
+ p = null;
+ } finally {
+ if (p != null) {
+ p.destroy();
+ }
+ }
+
+ // sort the data
+ UnilateralSortMerger<String> sorter = null;
+ BufferedReader reader = null;
+ BufferedReader verifyReader = null;
+
+ try {
+ MemoryManager mm = new DefaultMemoryManager(1024 * 1024, 1);
+ IOManager ioMan = new IOManagerAsync();
+
+ TypeSerializer<String> serializer = StringSerializer.INSTANCE;
+ TypeComparator<String> comparator = new StringComparator(true);
+
+ reader = new BufferedReader(new FileReader(input));
+ MutableObjectIterator<String> inputIterator = new StringReaderMutableObjectIterator(reader);
+
+ sorter = new UnilateralSortMerger<String>(mm, ioMan, inputIterator, new DummyInvokable(),
+ new RuntimeSerializerFactory<String>(serializer, String.class), comparator, 1.0, 4, 0.8f);
+
+ MutableObjectIterator<String> sortedData = sorter.getIterator();
+
+ reader.close();
+
+ // verify
+ verifyReader = new BufferedReader(new FileReader(sorted));
+ String next;
+
+ while ((next = verifyReader.readLine()) != null) {
+ String nextFromStratoSort = sortedData.next("");
+
+ Assert.assertNotNull(nextFromStratoSort);
+ Assert.assertEquals(next, nextFromStratoSort);
+ }
+ }
+ finally {
+ if (reader != null) {
+ reader.close();
+ }
+ if (verifyReader != null) {
+ verifyReader.close();
+ }
+ if (sorter != null) {
+ sorter.close();
+ }
+ }
+ }
+ catch (Exception e) {
+ System.err.println(e.getMessage());
+ e.printStackTrace();
+ Assert.fail(e.getMessage());
+ }
+ finally {
+ if (input != null) {
+ input.delete();
+ }
+ if (sorted != null) {
+ sorted.delete();
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void testStringTuplesSorting() {
+ final int NUM_STRINGS = 300000;
+
+ File input = null;
+ File sorted = null;
+
+ try {
+ // the source file
+ input = generateFileWithStringTuples(NUM_STRINGS, "http://some-uri.com/that/is/a/common/prefix/to/all");
+
+ // the sorted file
+ sorted = File.createTempFile("sorted_strings", "txt");
+
+ String[] command = {"/bin/bash","-c","export LC_ALL=\"C\" && cat \"" + input.getAbsolutePath() + "\" | sort > \"" + sorted.getAbsolutePath() + "\""};
+
+ Process p = null;
+ try {
+ p = Runtime.getRuntime().exec(command);
+ int retCode = p.waitFor();
+ if (retCode != 0) {
+ throw new Exception("Command failed with return code " + retCode);
+ }
+ p = null;
+ } finally {
+ if (p != null) {
+ p.destroy();
+ }
+ }
+
+ // sort the data
+ UnilateralSortMerger<Tuple2<String, String[]>> sorter = null;
+ BufferedReader reader = null;
+ BufferedReader verifyReader = null;
+
+ try {
+ MemoryManager mm = new DefaultMemoryManager(1024 * 1024, 1);
+ IOManager ioMan = new IOManagerAsync();
+
+ TupleTypeInfo<Tuple2<String, String[]>> typeInfo = (TupleTypeInfo<Tuple2<String, String[]>>)
+ TypeInfoParser.<Tuple2<String, String[]>>parse("Tuple2<String, String[]>");
+
+ TypeSerializer<Tuple2<String, String[]>> serializer = typeInfo.createSerializer(new ExecutionConfig());
+ TypeComparator<Tuple2<String, String[]>> comparator = typeInfo.createComparator(new int[] { 0 }, new boolean[] { true }, 0, new ExecutionConfig());
+
+ reader = new BufferedReader(new FileReader(input));
+ MutableObjectIterator<Tuple2<String, String[]>> inputIterator = new StringTupleReaderMutableObjectIterator(reader);
+
+ sorter = new UnilateralSortMerger<Tuple2<String, String[]>>(mm, ioMan, inputIterator, new DummyInvokable(),
+ new RuntimeSerializerFactory<Tuple2<String, String[]>>(serializer, (Class<Tuple2<String, String[]>>) (Class<?>) Tuple2.class), comparator, 1.0, 4, 0.8f);
+
+
+
+ // use this part to verify that all if good when sorting in memory
+
+// List<MemorySegment> memory = mm.allocatePages(new DummyInvokable(), mm.computeNumberOfPages(1024*1024*1024));
+// NormalizedKeySorter<Tuple2<String, String[]>> nks = new NormalizedKeySorter<Tuple2<String,String[]>>(serializer, comparator, memory);
+//
+// {
+// Tuple2<String, String[]> wi = new Tuple2<String, String[]>("", new String[0]);
+// while ((wi = inputIterator.next(wi)) != null) {
+// Assert.assertTrue(nks.write(wi));
+// }
+//
+// new QuickSort().sort(nks);
+// }
+//
+// MutableObjectIterator<Tuple2<String, String[]>> sortedData = nks.getIterator();
+
+ MutableObjectIterator<Tuple2<String, String[]>> sortedData = sorter.getIterator();
+ reader.close();
+
+ // verify
+ verifyReader = new BufferedReader(new FileReader(sorted));
+ MutableObjectIterator<Tuple2<String, String[]>> verifyIterator = new StringTupleReaderMutableObjectIterator(verifyReader);
+
+ Tuple2<String, String[]> next = new Tuple2<String, String[]>("", new String[0]);
+ Tuple2<String, String[]> nextFromStratoSort = new Tuple2<String, String[]>("", new String[0]);
+
+ int num = 0;
+
+ while ((next = verifyIterator.next(next)) != null) {
+ num++;
+
+ nextFromStratoSort = sortedData.next(nextFromStratoSort);
+ Assert.assertNotNull(nextFromStratoSort);
+
+ Assert.assertEquals(next.f0, nextFromStratoSort.f0);
+ Assert.assertArrayEquals(next.f1, nextFromStratoSort.f1);
+ }
+
+ Assert.assertNull(sortedData.next(nextFromStratoSort));
+ Assert.assertEquals(NUM_STRINGS, num);
+
+ }
+ finally {
+ if (reader != null) {
+ reader.close();
+ }
+ if (verifyReader != null) {
+ verifyReader.close();
+ }
+ if (sorter != null) {
+ sorter.close();
+ }
+ }
+ }
+ catch (Exception e) {
+ System.err.println(e.getMessage());
+ e.printStackTrace();
+ Assert.fail(e.getMessage());
+ }
+ finally {
+ if (input != null) {
+ input.delete();
+ }
+ if (sorted != null) {
+ sorted.delete();
+ }
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private static final class StringReaderMutableObjectIterator implements MutableObjectIterator<String> {
+
+ private final BufferedReader reader;
+
+ public StringReaderMutableObjectIterator(BufferedReader reader) {
+ this.reader = reader;
+ }
+
+ @Override
+ public String next(String reuse) throws IOException {
+ return reader.readLine();
+ }
+
+ @Override
+ public String next() throws IOException {
+ return reader.readLine();
+ }
+ }
+
+ private static final class StringTupleReaderMutableObjectIterator implements MutableObjectIterator<Tuple2<String, String[]>> {
+
+ private final BufferedReader reader;
+
+ public StringTupleReaderMutableObjectIterator(BufferedReader reader) {
+ this.reader = reader;
+ }
+
+ @Override
+ public Tuple2<String, String[]> next(Tuple2<String, String[]> reuse) throws IOException {
+ String line = reader.readLine();
+ if (line == null) {
+ return null;
+ }
+
+ String[] parts = line.split(" ");
+ reuse.f0 = parts[0];
+ reuse.f1 = parts;
+ return reuse;
+ }
+
+ @Override
+ public Tuple2<String, String[]> next() throws IOException {
+ return next(new Tuple2<String, String[]>());
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private File generateFileWithStrings(int numStrings, String prefix) throws IOException {
+ final Random rnd = new Random(SEED);
+
+ final StringBuilder bld = new StringBuilder();
+ final int resetValue = prefix.length();
+
+ bld.append(prefix);
+
+ File f = File.createTempFile("strings", "txt");
+ BufferedWriter wrt = null;
+ try {
+ wrt = new BufferedWriter(new FileWriter(f));
+
+ for (int i = 0 ; i < numStrings; i++) {
+ bld.setLength(resetValue);
+
+ int len = rnd.nextInt(20) + 300;
+ for (int k = 0; k < len; k++) {
+ char c = (char) (rnd.nextInt(80) + 40);
+ bld.append(c);
+ }
+
+ String str = bld.toString();
+ wrt.write(str);
+ wrt.newLine();
+ }
+ } finally {
+ wrt.close();
+ }
+
+ return f;
+ }
+
+ private File generateFileWithStringTuples(int numStrings, String prefix) throws IOException {
+ final Random rnd = new Random(SEED);
+
+ final StringBuilder bld = new StringBuilder();
+
+ File f = File.createTempFile("strings", "txt");
+ BufferedWriter wrt = null;
+ try {
+ wrt = new BufferedWriter(new FileWriter(f));
+
+ for (int i = 0 ; i < numStrings; i++) {
+ bld.setLength(0);
+
+ int numComps = rnd.nextInt(5) + 1;
+
+ for (int z = 0; z < numComps; z++) {
+ if (z > 0) {
+ bld.append(' ');
+ }
+ bld.append(prefix);
+
+ int len = rnd.nextInt(20) + 10;
+ for (int k = 0; k < len; k++) {
+ char c = (char) (rnd.nextInt(80) + 40);
+ bld.append(c);
+ }
+ }
+
+ String str = bld.toString();
+
+ wrt.write(str);
+ wrt.newLine();
+ }
+ } finally {
+ wrt.close();
+ }
+
+ return f;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ public static void main(String[] args) {
+ new MassiveStringSorting().testStringSorting();
+ new MassiveStringSorting().testStringTuplesSorting();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/3061eb12/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringValueSorting.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringValueSorting.java b/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringValueSorting.java
new file mode 100644
index 0000000..bc65ab9
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringValueSorting.java
@@ -0,0 +1,403 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.manual;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileReader;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.Random;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.java.typeutils.TypeInfoParser;
+import org.apache.flink.api.java.typeutils.runtime.CopyableValueComparator;
+import org.apache.flink.api.java.typeutils.runtime.CopyableValueSerializer;
+import org.apache.flink.api.java.typeutils.runtime.RuntimeSerializerFactory;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
+import org.apache.flink.runtime.memorymanager.DefaultMemoryManager;
+import org.apache.flink.runtime.memorymanager.MemoryManager;
+import org.apache.flink.runtime.operators.sort.UnilateralSortMerger;
+import org.apache.flink.runtime.operators.testutils.DummyInvokable;
+import org.apache.flink.types.StringValue;
+import org.apache.flink.util.MutableObjectIterator;
+import org.junit.Assert;
+
+public class MassiveStringValueSorting {
+
+ private static final long SEED = 347569784659278346L;
+
+ public void testStringValueSorting() {
+ File input = null;
+ File sorted = null;
+
+ try {
+ // the source file
+ input = generateFileWithStrings(300000, "http://some-uri.com/that/is/a/common/prefix/to/all");
+
+ // the sorted file
+ sorted = File.createTempFile("sorted_strings", "txt");
+
+ String[] command = {"/bin/bash","-c","export LC_ALL=\"C\" && cat \"" + input.getAbsolutePath() + "\" | sort > \"" + sorted.getAbsolutePath() + "\""};
+
+ Process p = null;
+ try {
+ p = Runtime.getRuntime().exec(command);
+ int retCode = p.waitFor();
+ if (retCode != 0) {
+ throw new Exception("Command failed with return code " + retCode);
+ }
+ p = null;
+ } finally {
+ if (p != null) {
+ p.destroy();
+ }
+ }
+
+ // sort the data
+ UnilateralSortMerger<StringValue> sorter = null;
+ BufferedReader reader = null;
+ BufferedReader verifyReader = null;
+
+ try {
+ MemoryManager mm = new DefaultMemoryManager(1024 * 1024, 1);
+ IOManager ioMan = new IOManagerAsync();
+
+ TypeSerializer<StringValue> serializer = new CopyableValueSerializer<StringValue>(StringValue.class);
+ TypeComparator<StringValue> comparator = new CopyableValueComparator<StringValue>(true, StringValue.class);
+
+ reader = new BufferedReader(new FileReader(input));
+ MutableObjectIterator<StringValue> inputIterator = new StringValueReaderMutableObjectIterator(reader);
+
+ sorter = new UnilateralSortMerger<StringValue>(mm, ioMan, inputIterator, new DummyInvokable(),
+ new RuntimeSerializerFactory<StringValue>(serializer, StringValue.class), comparator, 1.0, 4, 0.8f);
+
+ MutableObjectIterator<StringValue> sortedData = sorter.getIterator();
+
+ reader.close();
+
+ // verify
+ verifyReader = new BufferedReader(new FileReader(sorted));
+ String nextVerify;
+ StringValue nextFromFlinkSort = new StringValue();
+
+ while ((nextVerify = verifyReader.readLine()) != null) {
+ nextFromFlinkSort = sortedData.next(nextFromFlinkSort);
+
+ Assert.assertNotNull(nextFromFlinkSort);
+ Assert.assertEquals(nextVerify, nextFromFlinkSort.getValue());
+ }
+ }
+ finally {
+ if (reader != null) {
+ reader.close();
+ }
+ if (verifyReader != null) {
+ verifyReader.close();
+ }
+ if (sorter != null) {
+ sorter.close();
+ }
+ }
+ }
+ catch (Exception e) {
+ System.err.println(e.getMessage());
+ e.printStackTrace();
+ Assert.fail(e.getMessage());
+ }
+ finally {
+ if (input != null) {
+ input.delete();
+ }
+ if (sorted != null) {
+ sorted.delete();
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void testStringValueTuplesSorting() {
+ final int NUM_STRINGS = 300000;
+
+ File input = null;
+ File sorted = null;
+
+ try {
+ // the source file
+ input = generateFileWithStringTuples(NUM_STRINGS, "http://some-uri.com/that/is/a/common/prefix/to/all");
+
+ // the sorted file
+ sorted = File.createTempFile("sorted_strings", "txt");
+
+ String[] command = {"/bin/bash","-c","export LC_ALL=\"C\" && cat \"" + input.getAbsolutePath() + "\" | sort > \"" + sorted.getAbsolutePath() + "\""};
+
+ Process p = null;
+ try {
+ p = Runtime.getRuntime().exec(command);
+ int retCode = p.waitFor();
+ if (retCode != 0) {
+ throw new Exception("Command failed with return code " + retCode);
+ }
+ p = null;
+ } finally {
+ if (p != null) {
+ p.destroy();
+ }
+ }
+
+ // sort the data
+ UnilateralSortMerger<Tuple2<StringValue, StringValue[]>> sorter = null;
+ BufferedReader reader = null;
+ BufferedReader verifyReader = null;
+
+ try {
+ MemoryManager mm = new DefaultMemoryManager(1024 * 1024, 1);
+ IOManager ioMan = new IOManagerAsync();
+
+ TupleTypeInfo<Tuple2<StringValue, StringValue[]>> typeInfo = (TupleTypeInfo<Tuple2<StringValue, StringValue[]>>)
+ TypeInfoParser.<Tuple2<StringValue, StringValue[]>>parse("Tuple2<org.apache.flink.types.StringValue, org.apache.flink.types.StringValue[]>");
+
+ TypeSerializer<Tuple2<StringValue, StringValue[]>> serializer = typeInfo.createSerializer(new ExecutionConfig());
+ TypeComparator<Tuple2<StringValue, StringValue[]>> comparator = typeInfo.createComparator(new int[] { 0 }, new boolean[] { true }, 0, new ExecutionConfig());
+
+ reader = new BufferedReader(new FileReader(input));
+ MutableObjectIterator<Tuple2<StringValue, StringValue[]>> inputIterator = new StringValueTupleReaderMutableObjectIterator(reader);
+
+ sorter = new UnilateralSortMerger<Tuple2<StringValue, StringValue[]>>(mm, ioMan, inputIterator, new DummyInvokable(),
+ new RuntimeSerializerFactory<Tuple2<StringValue, StringValue[]>>(serializer, (Class<Tuple2<StringValue, StringValue[]>>) (Class<?>) Tuple2.class), comparator, 1.0, 4, 0.8f);
+
+
+
+ // use this part to verify that all if good when sorting in memory
+
+// List<MemorySegment> memory = mm.allocatePages(new DummyInvokable(), mm.computeNumberOfPages(1024*1024*1024));
+// NormalizedKeySorter<Tuple2<String, String[]>> nks = new NormalizedKeySorter<Tuple2<String,String[]>>(serializer, comparator, memory);
+//
+// {
+// Tuple2<String, String[]> wi = new Tuple2<String, String[]>("", new String[0]);
+// while ((wi = inputIterator.next(wi)) != null) {
+// Assert.assertTrue(nks.write(wi));
+// }
+//
+// new QuickSort().sort(nks);
+// }
+//
+// MutableObjectIterator<Tuple2<String, String[]>> sortedData = nks.getIterator();
+
+ MutableObjectIterator<Tuple2<StringValue, StringValue[]>> sortedData = sorter.getIterator();
+ reader.close();
+
+ // verify
+ verifyReader = new BufferedReader(new FileReader(sorted));
+ MutableObjectIterator<Tuple2<StringValue, StringValue[]>> verifyIterator = new StringValueTupleReaderMutableObjectIterator(verifyReader);
+
+ Tuple2<StringValue, StringValue[]> nextVerify = new Tuple2<StringValue, StringValue[]>(new StringValue(), new StringValue[0]);
+ Tuple2<StringValue, StringValue[]> nextFromFlinkSort = new Tuple2<StringValue, StringValue[]>(new StringValue(), new StringValue[0]);
+
+ int num = 0;
+
+ while ((nextVerify = verifyIterator.next(nextVerify)) != null) {
+ num++;
+
+ nextFromFlinkSort = sortedData.next(nextFromFlinkSort);
+ Assert.assertNotNull(nextFromFlinkSort);
+
+ Assert.assertEquals(nextVerify.f0, nextFromFlinkSort.f0);
+ Assert.assertArrayEquals(nextVerify.f1, nextFromFlinkSort.f1);
+ }
+
+ Assert.assertNull(sortedData.next(nextFromFlinkSort));
+ Assert.assertEquals(NUM_STRINGS, num);
+
+ }
+ finally {
+ if (reader != null) {
+ reader.close();
+ }
+ if (verifyReader != null) {
+ verifyReader.close();
+ }
+ if (sorter != null) {
+ sorter.close();
+ }
+ }
+ }
+ catch (Exception e) {
+ System.err.println(e.getMessage());
+ e.printStackTrace();
+ Assert.fail(e.getMessage());
+ }
+ finally {
+ if (input != null) {
+ input.delete();
+ }
+ if (sorted != null) {
+ sorted.delete();
+ }
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private static final class StringValueReaderMutableObjectIterator implements MutableObjectIterator<StringValue> {
+
+ private final BufferedReader reader;
+
+ public StringValueReaderMutableObjectIterator(BufferedReader reader) {
+ this.reader = reader;
+ }
+
+ @Override
+ public StringValue next(StringValue reuse) throws IOException {
+ String line = reader.readLine();
+
+ if (line == null) {
+ return null;
+ }
+
+ reuse.setValue(line);
+ return reuse;
+ }
+
+ @Override
+ public StringValue next() throws IOException {
+ return next(new StringValue());
+ }
+ }
+
+ private static final class StringValueTupleReaderMutableObjectIterator implements MutableObjectIterator<Tuple2<StringValue, StringValue[]>> {
+
+ private final BufferedReader reader;
+
+ public StringValueTupleReaderMutableObjectIterator(BufferedReader reader) {
+ this.reader = reader;
+ }
+
+ @Override
+ public Tuple2<StringValue, StringValue[]> next(Tuple2<StringValue, StringValue[]> reuse) throws IOException {
+ String line = reader.readLine();
+ if (line == null) {
+ return null;
+ }
+
+ String[] parts = line.split(" ");
+ reuse.f0.setValue(parts[0]);
+ reuse.f1 = new StringValue[parts.length];
+
+ for (int i = 0; i < parts.length; i++) {
+ reuse.f1[i] = new StringValue(parts[i]);
+ }
+
+ return reuse;
+ }
+
+ @Override
+ public Tuple2<StringValue, StringValue[]> next() throws IOException {
+ return next(new Tuple2<StringValue, StringValue[]>(new StringValue(), new StringValue[0]));
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private File generateFileWithStrings(int numStrings, String prefix) throws IOException {
+ final Random rnd = new Random(SEED);
+
+ final StringBuilder bld = new StringBuilder();
+ final int resetValue = prefix.length();
+
+ bld.append(prefix);
+
+ File f = File.createTempFile("strings", "txt");
+ BufferedWriter wrt = null;
+ try {
+ wrt = new BufferedWriter(new FileWriter(f));
+
+ for (int i = 0 ; i < numStrings; i++) {
+ bld.setLength(resetValue);
+
+ int len = rnd.nextInt(20) + 300;
+ for (int k = 0; k < len; k++) {
+ char c = (char) (rnd.nextInt(80) + 40);
+ bld.append(c);
+ }
+
+ String str = bld.toString();
+ wrt.write(str);
+ wrt.newLine();
+ }
+ } finally {
+ wrt.close();
+ }
+
+ return f;
+ }
+
+ private File generateFileWithStringTuples(int numStrings, String prefix) throws IOException {
+ final Random rnd = new Random(SEED);
+
+ final StringBuilder bld = new StringBuilder();
+
+ File f = File.createTempFile("strings", "txt");
+ BufferedWriter wrt = null;
+ try {
+ wrt = new BufferedWriter(new FileWriter(f));
+
+ for (int i = 0 ; i < numStrings; i++) {
+ bld.setLength(0);
+
+ int numComps = rnd.nextInt(5) + 1;
+
+ for (int z = 0; z < numComps; z++) {
+ if (z > 0) {
+ bld.append(' ');
+ }
+ bld.append(prefix);
+
+ int len = rnd.nextInt(20) + 10;
+ for (int k = 0; k < len; k++) {
+ char c = (char) (rnd.nextInt(80) + 40);
+ bld.append(c);
+ }
+ }
+
+ String str = bld.toString();
+
+ wrt.write(str);
+ wrt.newLine();
+ }
+ } finally {
+ wrt.close();
+ }
+
+ return f;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ public static void main(String[] args) {
+ new MassiveStringValueSorting().testStringValueSorting();
+ new MassiveStringValueSorting().testStringValueTuplesSorting();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/3061eb12/flink-tests/src/test/java/org/apache/flink/test/manual/NotSoMiniClusterIterations.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/manual/NotSoMiniClusterIterations.java b/flink-tests/src/test/java/org/apache/flink/test/manual/NotSoMiniClusterIterations.java
new file mode 100644
index 0000000..d340f57
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/manual/NotSoMiniClusterIterations.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.manual;
+
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.aggregation.Aggregations;
+import org.apache.flink.api.java.io.DiscardingOutputFormat;
+import org.apache.flink.api.java.operators.DeltaIteration;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.examples.java.graph.ConnectedComponents;
+import org.apache.flink.examples.java.graph.util.ConnectedComponentsData;
+import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster;
+
+import static org.junit.Assert.fail;
+
+/**
+ * This test starts a mini cluster with 100 task managers and runs connected components
+ * with a parallelism of 100.
+ */
+public class NotSoMiniClusterIterations {
+
+ private static final int PARALLELISM = 100;
+
+ public static void main(String[] args) {
+ if ((Runtime.getRuntime().maxMemory() >>> 20) < 5000) {
+ throw new RuntimeException("This test program needs to run with at least 5GB of heap space.");
+ }
+
+ LocalFlinkMiniCluster cluster = null;
+
+ try {
+ Configuration config = new Configuration();
+ config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, PARALLELISM);
+ config.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, 8);
+ config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 1);
+ config.setInteger(ConfigConstants.TASK_MANAGER_NETWORK_NUM_BUFFERS_KEY, 1000);
+ config.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SEGMENT_SIZE_KEY, 8 * 1024);
+
+ config.setInteger("taskmanager.net.server.numThreads", 1);
+ config.setInteger("taskmanager.net.client.numThreads", 1);
+
+ cluster = new LocalFlinkMiniCluster(config, false);
+
+ runConnectedComponents(cluster.getJobManagerRPCPort());
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ finally {
+ if (cluster != null) {
+ cluster.shutdown();
+ }
+ }
+ }
+
+ private static void runConnectedComponents(int jmPort) throws Exception {
+
+ ExecutionEnvironment env = ExecutionEnvironment.createRemoteEnvironment("localhost", jmPort);
+ env.setParallelism(PARALLELISM);
+ env.getConfig().disableSysoutLogging();
+
+ // read vertex and edge data
+ DataSet<Long> vertices = ConnectedComponentsData.getDefaultVertexDataSet(env)
+ .rebalance();
+
+ DataSet<Tuple2<Long, Long>> edges = ConnectedComponentsData.getDefaultEdgeDataSet(env)
+ .rebalance()
+ .flatMap(new ConnectedComponents.UndirectEdge());
+
+ // assign the initial components (equal to the vertex id)
+ DataSet<Tuple2<Long, Long>> verticesWithInitialId = vertices
+ .map(new ConnectedComponents.DuplicateValue<Long>());
+
+ // open a delta iteration
+ DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iteration =
+ verticesWithInitialId.iterateDelta(verticesWithInitialId, 100, 0);
+
+ // apply the step logic: join with the edges, select the minimum neighbor,
+ // update if the component of the candidate is smaller
+ DataSet<Tuple2<Long, Long>> changes = iteration.getWorkset().join(edges)
+ .where(0).equalTo(0)
+ .with(new ConnectedComponents.NeighborWithComponentIDJoin())
+
+ .groupBy(0).aggregate(Aggregations.MIN, 1)
+
+ .join(iteration.getSolutionSet())
+ .where(0).equalTo(0)
+ .with(new ConnectedComponents.ComponentIdFilter());
+
+ // close the delta iteration (delta and new workset are identical)
+ DataSet<Tuple2<Long, Long>> result = iteration.closeWith(changes, changes);
+
+ result.output(new DiscardingOutputFormat<Tuple2<Long, Long>>());
+
+ env.execute();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/3061eb12/flink-tests/src/test/java/org/apache/flink/test/manual/package-info.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/manual/package-info.java b/flink-tests/src/test/java/org/apache/flink/test/manual/package-info.java
new file mode 100644
index 0000000..893f3cc
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/manual/package-info.java
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * This package contains various tests that are not automatically executed, but
+ * need to be manually invoked, because they are extremely heavy of require larger-than-usual
+ * JVMs.
+ */
+package org.apache.flink.test.manual;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/3061eb12/flink-tests/src/test/java/org/apache/flink/test/misc/CCTest.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/misc/CCTest.java b/flink-tests/src/test/java/org/apache/flink/test/misc/CCTest.java
deleted file mode 100644
index cca6461..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/misc/CCTest.java
+++ /dev/null
@@ -1,115 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.test.misc;
-
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.aggregation.Aggregations;
-import org.apache.flink.api.java.io.DiscardingOutputFormat;
-import org.apache.flink.api.java.operators.DeltaIteration;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.configuration.ConfigConstants;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.examples.java.graph.ConnectedComponents;
-import org.apache.flink.examples.java.graph.util.ConnectedComponentsData;
-import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster;
-
-import static org.junit.Assert.fail;
-
-public class CCTest {
-
- private static final int PARALLELISM = 100;
-
- public static void main(String[] args) {
- if ((Runtime.getRuntime().maxMemory() >>> 20) < 5000) {
- throw new RuntimeException("This est program needs to run with at least 5GB of heap space.");
- }
-
-
- LocalFlinkMiniCluster cluster = null;
-
- try {
- Configuration config = new Configuration();
- config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, PARALLELISM);
- config.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, 8);
- config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 1);
- config.setInteger(ConfigConstants.TASK_MANAGER_NETWORK_NUM_BUFFERS_KEY, 1000);
- config.setInteger(ConfigConstants.TASK_MANAGER_NETWORK_BUFFER_SIZE_KEY, 8 * 1024);
-
-
- config.setInteger("taskmanager.net.server.numThreads", 1);
- config.setInteger("taskmanager.net.client.numThreads", 1);
-
- cluster = new LocalFlinkMiniCluster(config, false);
-
- runConnectedComponents(cluster.getJobManagerRPCPort());
- }
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
- finally {
- if (cluster != null) {
- cluster.shutdown();
- }
- }
- }
-
- private static void runConnectedComponents(int jmPort) throws Exception {
-
- ExecutionEnvironment env = ExecutionEnvironment.createRemoteEnvironment("localhost", jmPort);
- env.setParallelism(PARALLELISM);
- env.getConfig().disableSysoutLogging();
-
- // read vertex and edge data
- DataSet<Long> vertices = ConnectedComponentsData.getDefaultVertexDataSet(env)
- .rebalance();
-
- DataSet<Tuple2<Long, Long>> edges = ConnectedComponentsData.getDefaultEdgeDataSet(env)
- .rebalance()
- .flatMap(new ConnectedComponents.UndirectEdge());
-
- // assign the initial components (equal to the vertex id)
- DataSet<Tuple2<Long, Long>> verticesWithInitialId = vertices
- .map(new ConnectedComponents.DuplicateValue<Long>());
-
- // open a delta iteration
- DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iteration =
- verticesWithInitialId.iterateDelta(verticesWithInitialId, 100, 0);
-
- // apply the step logic: join with the edges, select the minimum neighbor,
- // update if the component of the candidate is smaller
- DataSet<Tuple2<Long, Long>> changes = iteration.getWorkset().join(edges)
- .where(0).equalTo(0)
- .with(new ConnectedComponents.NeighborWithComponentIDJoin())
-
- .groupBy(0).aggregate(Aggregations.MIN, 1)
-
- .join(iteration.getSolutionSet())
- .where(0).equalTo(0)
- .with(new ConnectedComponents.ComponentIdFilter());
-
- // close the delta iteration (delta and new workset are identical)
- DataSet<Tuple2<Long, Long>> result = iteration.closeWith(changes, changes);
-
- result.output(new DiscardingOutputFormat<Tuple2<Long, Long>>());
-
- env.execute();
- }
-}
[3/7] flink git commit: [FLINK-1952] [jobmanager] Rework and fix slot
sharing scheduler
Posted by se...@apache.org.
[FLINK-1952] [jobmanager] Rework and fix slot sharing scheduler
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/017b4f0f
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/017b4f0f
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/017b4f0f
Branch: refs/heads/master
Commit: 017b4f0f90c3a40c0f5557916c6ad5c4612606a3
Parents: 8e35feb
Author: Stephan Ewen <se...@apache.org>
Authored: Wed May 20 11:37:56 2015 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed May 27 16:36:01 2015 +0200
----------------------------------------------------------------------
.../apache/flink/runtime/instance/Instance.java | 160 +++--
.../flink/runtime/instance/SharedSlot.java | 203 ++++--
.../flink/runtime/instance/SimpleSlot.java | 125 +++-
.../org/apache/flink/runtime/instance/Slot.java | 177 +++--
.../instance/SlotSharingGroupAssignment.java | 676 +++++++++++++++++++
.../scheduler/CoLocationConstraint.java | 138 +++-
.../jobmanager/scheduler/CoLocationGroup.java | 35 +-
.../scheduler/NoResourceAvailableException.java | 12 +-
.../runtime/jobmanager/scheduler/Scheduler.java | 389 ++++++-----
.../jobmanager/scheduler/SlotSharingGroup.java | 8 +-
.../scheduler/SlotSharingGroupAssignment.java | 485 -------------
.../ExecutionGraphDeploymentTest.java | 2 +-
.../ExecutionStateProgressTest.java | 16 +-
.../ExecutionVertexCancelTest.java | 113 ++--
.../ExecutionVertexDeploymentTest.java | 171 ++---
.../ExecutionVertexSchedulingTest.java | 75 +-
.../flink/runtime/instance/InstanceTest.java | 67 +-
.../flink/runtime/instance/SharedSlotsTest.java | 666 ++++++++++++++++++
.../flink/runtime/instance/SimpleSlotTest.java | 79 ++-
.../flink/runtime/instance/SlotTestUtils.java | 23 -
.../scheduler/CoLocationConstraintTest.java | 177 +++++
.../ScheduleWithCoLocationHintTest.java | 13 +-
.../scheduler/SchedulerSlotSharingTest.java | 33 +-
.../scheduler/SchedulerTestUtils.java | 9 +-
.../jobmanager/scheduler/SharedSlotsTest.java | 145 ----
25 files changed, 2639 insertions(+), 1358 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Instance.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Instance.java b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Instance.java
index 324629f..7b48693 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Instance.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Instance.java
@@ -26,29 +26,28 @@ import java.util.Queue;
import java.util.Set;
import akka.actor.ActorRef;
-import org.apache.flink.util.AbstractID;
+
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.jobmanager.scheduler.SlotAvailabilityListener;
-import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroupAssignment;
/**
* An instance represents a {@link org.apache.flink.runtime.taskmanager.TaskManager}
* registered at a JobManager and ready to receive work.
*/
public class Instance {
-
+
/** The lock on which to synchronize allocations and failure state changes */
private final Object instanceLock = new Object();
-
+
/** The actor ref to the task manager represented by this taskManager. */
private final ActorRef taskManager;
/** The instance connection information for the data transfer. */
private final InstanceConnectionInfo connectionInfo;
-
+
/** A description of the resources of the task manager */
private final HardwareDescription resources;
-
+
/** The ID identifying the taskManager. */
private final InstanceID instanceId;
@@ -57,27 +56,27 @@ public class Instance {
/** A list of available slot positions */
private final Queue<Integer> availableSlots;
-
+
/** Allocated slots on this taskManager */
private final Set<Slot> allocatedSlots = new HashSet<Slot>();
/** A listener to be notified upon new slot availability */
private SlotAvailabilityListener slotAvailabilityListener;
-
+
/** Time when last heat beat has been received from the task manager running on this taskManager. */
private volatile long lastReceivedHeartBeat = System.currentTimeMillis();
private byte[] lastMetricsReport;
-
+
/** Flag marking the instance as alive or as dead. */
private volatile boolean isDead;
// --------------------------------------------------------------------------------------------
-
+
/**
* Constructs an instance reflecting a registered TaskManager.
- *
+ *
* @param taskManager The actor reference of the represented task manager.
* @param connectionInfo The remote connection where the task manager receives requests.
* @param id The id under which the taskManager is registered.
@@ -91,7 +90,7 @@ public class Instance {
this.instanceId = id;
this.resources = resources;
this.numberOfSlots = numberOfSlots;
-
+
this.availableSlots = new ArrayDeque<Integer>(numberOfSlots);
for (int i = 0; i < numberOfSlots; i++) {
this.availableSlots.add(i);
@@ -101,32 +100,32 @@ public class Instance {
// --------------------------------------------------------------------------------------------
// Properties
// --------------------------------------------------------------------------------------------
-
+
public InstanceID getId() {
return instanceId;
}
-
+
public HardwareDescription getResources() {
return this.resources;
}
-
+
public int getTotalNumberOfSlots() {
return numberOfSlots;
}
-
+
// --------------------------------------------------------------------------------------------
// Life and Death
// --------------------------------------------------------------------------------------------
-
+
public boolean isAlive() {
return !isDead;
}
public void markDead() {
-
+
// create a copy of the slots to avoid concurrent modification exceptions
List<Slot> slots;
-
+
synchronized (instanceLock) {
if (isDead) {
return;
@@ -135,9 +134,9 @@ public class Instance {
// no more notifications for the slot releasing
this.slotAvailabilityListener = null;
-
+
slots = new ArrayList<Slot>(allocatedSlots);
-
+
allocatedSlots.clear();
availableSlots.clear();
}
@@ -156,16 +155,16 @@ public class Instance {
// --------------------------------------------------------------------------------------------
// Heartbeats
// --------------------------------------------------------------------------------------------
-
+
/**
* Gets the timestamp of the last heartbeat.
- *
+ *
* @return The timestamp of the last heartbeat.
*/
public long getLastHeartBeat() {
return this.lastReceivedHeartBeat;
}
-
+
/**
* Updates the time of last received heart beat to the current system time.
*/
@@ -184,7 +183,7 @@ public class Instance {
/**
* Checks whether the last heartbeat occurred within the last {@code n} milliseconds
* before the given timestamp {@code now}.
- *
+ *
* @param now The timestamp representing the current time.
* @param cleanUpInterval The maximum time (in msecs) that the last heartbeat may lie in the past.
* @return True, if this taskManager is considered alive, false otherwise.
@@ -192,37 +191,58 @@ public class Instance {
public boolean isStillAlive(long now, long cleanUpInterval) {
return this.lastReceivedHeartBeat + cleanUpInterval > now;
}
-
+
// --------------------------------------------------------------------------------------------
// Resource allocation
// --------------------------------------------------------------------------------------------
-
+
+ /**
+ * Allocates a simple slot on this TaskManager instance. This method returns {@code null}, if no slot
+ * is available at the moment.
+ *
+ * @param jobID The ID of the job that the slot is allocated for.
+ *
+ * @return A simple slot that represents a task slot on this TaskManager instance, or null, if the
+ * TaskManager instance has no more slots available.
+ *
+ * @throws InstanceDiedException Thrown if the instance is no longer alive by the time the
+ * slot is allocated.
+ */
public SimpleSlot allocateSimpleSlot(JobID jobID) throws InstanceDiedException {
- return allocateSimpleSlot(jobID, new AbstractID());
- }
-
- public SimpleSlot allocateSimpleSlot(JobID jobID, AbstractID groupID) throws InstanceDiedException {
if (jobID == null) {
throw new IllegalArgumentException();
}
-
+
synchronized (instanceLock) {
if (isDead) {
throw new InstanceDiedException(this);
}
-
+
Integer nextSlot = availableSlots.poll();
if (nextSlot == null) {
return null;
- } else {
- SimpleSlot slot = new SimpleSlot(jobID, this, nextSlot, null, groupID);
+ }
+ else {
+ SimpleSlot slot = new SimpleSlot(jobID, this, nextSlot);
allocatedSlots.add(slot);
return slot;
}
}
}
- public SharedSlot allocateSharedSlot(JobID jobID, SlotSharingGroupAssignment sharingGroupAssignment, AbstractID groupID)
+ /**
+ * Allocates a shared slot on this TaskManager instance. This method returns {@code null}, if no slot
+ * is available at the moment. The shared slot will be managed by the given SlotSharingGroupAssignment.
+ *
+ * @param jobID The ID of the job that the slot is allocated for.
+ * @param sharingGroupAssignment The assignment group that manages this shared slot.
+ *
+ * @return A shared slot that represents a task slot on this TaskManager instance and can hold other
+ * (shared) slots, or null, if the TaskManager instance has no more slots available.
+ *
+ * @throws InstanceDiedException Thrown if the instance is no longer alive by the time the slot is allocated.
+ */
+ public SharedSlot allocateSharedSlot(JobID jobID, SlotSharingGroupAssignment sharingGroupAssignment)
throws InstanceDiedException
{
// the slot needs to be in the returned to taskManager state
@@ -238,53 +258,65 @@ public class Instance {
Integer nextSlot = availableSlots.poll();
if (nextSlot == null) {
return null;
- } else {
- SharedSlot slot = new SharedSlot(jobID, this, nextSlot, sharingGroupAssignment, null, groupID);
+ }
+ else {
+ SharedSlot slot = new SharedSlot(jobID, this, nextSlot, sharingGroupAssignment);
allocatedSlots.add(slot);
return slot;
}
}
}
+ /**
+ * Returns a slot that has been allocated from this instance. The slot needs have been canceled
+ * prior to calling this method.
+ *
+ * <p>The method will transition the slot to the "released" state. If the slot is already in state
+ * "released", this method will do nothing.</p>
+ *
+ * @param slot The slot to return.
+ * @return True, if the slot was returned, false if not.
+ */
public boolean returnAllocatedSlot(Slot slot) {
if (slot == null || slot.getInstance() != this) {
- throw new IllegalArgumentException("Slot is null or belongs to the wrong taskManager.");
+ throw new IllegalArgumentException("Slot is null or belongs to the wrong TaskManager.");
}
if (slot.isAlive()) {
throw new IllegalArgumentException("Slot is still alive");
}
-
- if (slot.markReleased()) {
+
+ if (slot.markReleased()) {
synchronized (instanceLock) {
if (isDead) {
return false;
}
-
+
if (this.allocatedSlots.remove(slot)) {
this.availableSlots.add(slot.getSlotNumber());
-
+
if (this.slotAvailabilityListener != null) {
this.slotAvailabilityListener.newSlotAvailable(this);
}
-
+
return true;
- } else {
- throw new IllegalArgumentException("Slot was not allocated from the taskManager.");
+ }
+ else {
+ throw new IllegalArgumentException("Slot was not allocated from this TaskManager.");
}
}
- } else {
+ }
+ else {
return false;
}
}
-
+
public void cancelAndReleaseAllSlots() {
-
// we need to do this copy because of concurrent modification exceptions
List<Slot> copy;
synchronized (instanceLock) {
copy = new ArrayList<Slot>(this.allocatedSlots);
}
-
+
for (Slot slot : copy) {
slot.releaseSlot();
}
@@ -301,23 +333,28 @@ public class Instance {
public InstanceConnectionInfo getInstanceConnectionInfo() {
return connectionInfo;
}
-
+
public int getNumberOfAvailableSlots() {
return this.availableSlots.size();
}
-
+
public int getNumberOfAllocatedSlots() {
return this.allocatedSlots.size();
}
-
+
public boolean hasResourcesAvailable() {
return !isDead && getNumberOfAvailableSlots() > 0;
}
-
+
// --------------------------------------------------------------------------------------------
// Listeners
// --------------------------------------------------------------------------------------------
-
+
+ /**
+ * Sets the listener that receives notifications for slot availability.
+ *
+ * @param slotAvailabilityListener The listener.
+ */
public void setSlotAvailabilityListener(SlotAvailabilityListener slotAvailabilityListener) {
synchronized (instanceLock) {
if (this.slotAvailabilityListener != null) {
@@ -327,22 +364,23 @@ public class Instance {
}
}
}
-
+
+ /**
+ * Removes the listener that receives notifications for slot availability.
+ */
public void removeSlotListener() {
synchronized (instanceLock) {
this.slotAvailabilityListener = null;
}
}
-
+
// --------------------------------------------------------------------------------------------
// Standard Utilities
// --------------------------------------------------------------------------------------------
-
+
@Override
public String toString() {
return String.format("%s @ %s - %d slots - URL: %s", instanceId, connectionInfo.getHostname(),
numberOfSlots, (taskManager != null ? taskManager.path() : "ActorRef.noSender"));
}
-
-
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/main/java/org/apache/flink/runtime/instance/SharedSlot.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/SharedSlot.java b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/SharedSlot.java
index 7bd70a7..ef62910 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/SharedSlot.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/SharedSlot.java
@@ -20,135 +20,202 @@ package org.apache.flink.runtime.instance;
import org.apache.flink.util.AbstractID;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroupAssignment;
+import java.util.ConcurrentModificationException;
import java.util.HashSet;
import java.util.Set;
/**
* This class represents a shared slot. A shared slot can have multiple
- * {@link org.apache.flink.runtime.instance.SimpleSlot} instances within itself. This allows to
- * schedule multiple tasks simultaneously, enabling Flink's streaming capabilities.
+ * {@link SimpleSlot} instances within itself. This allows to
+ * schedule multiple tasks simultaneously to the same resource. Sharing a resource with multiple
+ * tasks is crucial for simple pipelined / streamed execution, where both the sender and the receiver
+ * are typically active at the same time.
*
- * IMPORTANT: This class contains no synchronization. Thus, the caller has to guarantee proper
+ * <p><b>IMPORTANT:</b> This class contains no synchronization. Thus, the caller has to guarantee proper
* synchronization. In the current implementation, all concurrently modifying operations are
* passed through a {@link SlotSharingGroupAssignment} object which is responsible for
- * synchronization.
- *
+ * synchronization.</p>
*/
public class SharedSlot extends Slot {
+ /** The assignment group os shared slots that manages the availability and release of the slots */
private final SlotSharingGroupAssignment assignmentGroup;
+ /** The set os sub-slots allocated from this shared slot */
private final Set<Slot> subSlots;
- public SharedSlot(JobID jobID, Instance instance, int slotNumber,
- SlotSharingGroupAssignment assignmentGroup, SharedSlot parent,
- AbstractID groupID) {
- super(jobID, instance, slotNumber, parent, groupID);
-
- this.assignmentGroup = assignmentGroup;
- this.subSlots = new HashSet<Slot>();
- }
- public Set<Slot> getSubSlots() {
- return subSlots;
+ /**
+ * Creates a new shared slot that has no parent (is a root slot) and does not belong to any task group.
+ * This constructor is used to create a slot directly from an instance.
+ *
+ * @param jobID The ID of the job that the slot is created for.
+ * @param instance The instance that holds the slot.
+ * @param slotNumber The number of the slot.
+ * @param assignmentGroup The assignment group that this shared slot belongs to.
+ */
+ public SharedSlot(JobID jobID, Instance instance, int slotNumber, SlotSharingGroupAssignment assignmentGroup) {
+ this(jobID, instance, slotNumber, assignmentGroup, null, null);
}
/**
- * Removes the simple slot from the {@link org.apache.flink.runtime.instance.SharedSlot}. Should
- * only be called through the
- * {@link org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroupAssignment} attribute
- * assignmnetGroup.
- *
- * @param slot slot to be removed from the set of sub slots.
- * @return Number of remaining sub slots
+ * Creates a new shared slot that has is a sub-slot of the given parent shared slot, and that belongs
+ * to the given task group.
+ *
+ * @param jobID The ID of the job that the slot is created for.
+ * @param instance The instance that holds the slot.
+ * @param slotNumber The number of the slot.
+ * @param assignmentGroup The assignment group that this shared slot belongs to.
*/
- public int freeSubSlot(Slot slot){
- if(!subSlots.remove(slot)){
- throw new IllegalArgumentException("Wrong shared slot for sub slot.");
- }
+ public SharedSlot(JobID jobID, Instance instance, int slotNumber,
+ SlotSharingGroupAssignment assignmentGroup, SharedSlot parent, AbstractID groupId) {
+ super(jobID, instance, slotNumber, parent, groupId);
- return subSlots.size();
+ this.assignmentGroup = assignmentGroup;
+ this.subSlots = new HashSet<Slot>();
}
+ // ------------------------------------------------------------------------
+ // Properties
+ // ------------------------------------------------------------------------
+
@Override
public int getNumberLeaves() {
- int result = 0;
-
- for(Slot slot: subSlots){
- result += slot.getNumberLeaves();
+ while (true) {
+ try {
+ int result = 0;
+ for (Slot slot: subSlots){
+ result += slot.getNumberLeaves();
+ }
+ return result;
+ }
+ catch (ConcurrentModificationException e) {
+ // ignore and retry
+ }
}
+ }
- return result;
+ /**
+ * Checks whether this slot is a root slot that has not yet added any child slots.
+ *
+ * @return True, if this slot is a root slot and has not yet added any children, false otherwise.
+ */
+ public boolean isRootAndEmpty() {
+ return getParent() == null && subSlots.isEmpty();
}
+ /**
+ * Checks whether this shared slot has any sub slots.
+ *
+ * @return True, if the shared slot has sub slots, false otherwise.
+ */
+ public boolean hasChildren() {
+ return subSlots.size() > 0;
+ }
+
@Override
- public void cancel() {
- // Guarantee that the operation is only executed once
- if (markCancelled()) {
- assignmentGroup.releaseSharedSlot(this);
+ public void releaseSlot() {
+ assignmentGroup.releaseSharedSlot(this);
+
+ if (!(isReleased() && subSlots.isEmpty())) {
+ throw new IllegalStateException("Bug: SharedSlot is not empty and released after call to releaseSlot()");
}
}
/**
- * Release this shared slot. In order to do this:
- *
- * 1. Cancel and release all sub slots atomically with respect to the assigned assignment group.
- * 2. Set the state of the shared slot to be cancelled.
- * 3. Dispose the shared slot (returning the slot to the instance).
+ * Gets the set of all slots allocated as sub-slots of this shared slot.
*
- * After cancelAndReleaseSubSlots, the shared slot is marked to be dead. This prevents further
- * sub slot creation by the scheduler.
+ * @return All sub-slots allocated from this shared slot.
*/
- @Override
- public void releaseSlot() {
- assignmentGroup.releaseSharedSlot(this);
+ Set<Slot> getSubSlots() {
+ return subSlots;
}
+ // ------------------------------------------------------------------------
+ // INTERNAL : TO BE CALLED ONLY BY THE assignmentGroup - Allocating sub-slots
+ // ------------------------------------------------------------------------
+
/**
* Creates a new sub slot if the slot is not dead, yet. This method should only be called from
* the assignment group instance to guarantee synchronization.
+ *
+ * <b>NOTE:</b> This method is not synchronized and must only be called from
+ * the slot's assignment group.
*
- * @param jID id to identify tasks which can be deployed in this sub slot
- * @return new sub slot if the shared slot is still alive, otherwise null
+ * @param groupId The ID to identify tasks which can be deployed in this sub slot.
+ * @return The new sub slot if the shared slot is still alive, otherwise null.
*/
- public SimpleSlot allocateSubSlot(AbstractID jID){
- if(isDead()){
- return null;
- } else {
- SimpleSlot slot = new SimpleSlot(getJobID(), getInstance(), subSlots.size(), this, jID);
+ SimpleSlot allocateSubSlot(AbstractID groupId) {
+ if (isAlive()) {
+ SimpleSlot slot = new SimpleSlot(getJobID(), getInstance(), subSlots.size(), this, groupId);
subSlots.add(slot);
-
return slot;
}
+ else {
+ return null;
+ }
}
- public SharedSlot allocateSharedSlot(AbstractID jID){
- if(isDead()){
- return null;
- } else {
- SharedSlot slot = new SharedSlot(getJobID(), getInstance(), subSlots.size(), assignmentGroup, this, jID);
+ /**
+ * Creates a new sub slot if the slot is not dead, yet. This method should only be called from
+ * the assignment group instance to guarantee synchronization.
+ *
+ * NOTE: This method should only be called from the slot's assignment group.
+ *
+ * @param groupId The ID to identify tasks which can be deployed in this sub slot.
+ * @return The new sub slot if the shared slot is still alive, otherwise null.
+ */
+ SharedSlot allocateSharedSlot(AbstractID groupId){
+ if (isAlive()) {
+ SharedSlot slot = new SharedSlot(getJobID(), getInstance(), subSlots.size(), assignmentGroup, this, groupId);
subSlots.add(slot);
-
return slot;
}
+ else {
+ return null;
+ }
}
+ // ------------------------------------------------------------------------
+ // INTERNAL : TO BE CALLED ONLY BY THE assignmentGroup - releasing slots
+ // ------------------------------------------------------------------------
+
/**
- * Disposes the given sub slot. This
- * is done by the means of the assignmentGroup in order to synchronize the method. If the
- * disposed slot was the last sub slot, then the shared slot is marked to be cancelled and is
- * disposed/returned to the owning instance.
+ * Disposes the given sub slot. This method is called by the child simple slot to tell this
+ * shared slot to release it.
+ *
+ * The releasing process itself is done by the {@link SlotSharingGroupAssignment}, which controls
+ * all the modifications in this shared slot.
*
- * @param slot sub slot which shall be removed from the shared slot
+ * NOTE: This method must not modify the shared slot directly !!!
+ *
+ * @param slot The sub-slot which shall be removed from the shared slot.
*/
- public void disposeChild(SimpleSlot slot){
+ void releaseChild(SimpleSlot slot) {
assignmentGroup.releaseSimpleSlot(slot);
}
+
+ /**
+ * Removes the given slot from this shared slot. This method Should only be called
+ * through this shared slot's {@link SlotSharingGroupAssignment}
+ *
+ * @param slot slot to be removed from the set of sub slots.
+ * @return Number of remaining sub slots
+ */
+ int removeDisposedChildSlot(Slot slot) {
+ if (!slot.isReleased() || !subSlots.remove(slot)) {
+ throw new IllegalArgumentException();
+ }
+ return subSlots.size();
+ }
+
+ // ------------------------------------------------------------------------
+ // Utilities
+ // ------------------------------------------------------------------------
@Override
public String toString() {
return "Shared " + super.toString();
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/main/java/org/apache/flink/runtime/instance/SimpleSlot.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/SimpleSlot.java b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/SimpleSlot.java
index 601cd28..9bc977d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/SimpleSlot.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/SimpleSlot.java
@@ -18,60 +18,91 @@
package org.apache.flink.runtime.instance;
-import org.apache.flink.util.AbstractID;
import org.apache.flink.runtime.executiongraph.Execution;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.jobmanager.scheduler.Locality;
+import org.apache.flink.util.AbstractID;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
/**
- * Class which represents a single slot on a machine or within a shared slot. If this slot is part
- * of a [[SharedSlot]], then its parent attribute is set to this instance. If not, then the parent
- * attribute is null.
+ * A SimpleSlot represents a single slot on a TaskManager instance, or a slot within a shared slot.
*
- * IMPORTANT: This class has no synchronization. Thus it has to be synchronized by the calling
- * object.
+ * <p>If this slot is part of a {@link SharedSlot}, then the parent attribute will point to that shared slot.
+ * If not, then the parent attribute is null.</p>
*/
public class SimpleSlot extends Slot {
+ /** The updater used to atomically swap in the execution */
private static final AtomicReferenceFieldUpdater<SimpleSlot, Execution> VERTEX_UPDATER =
AtomicReferenceFieldUpdater.newUpdater(SimpleSlot.class, Execution.class, "executedTask");
+ // ------------------------------------------------------------------------
+
/** Task being executed in the slot. Volatile to force a memory barrier and allow for correct double-checking */
private volatile Execution executedTask;
+ /** The locality attached to the slot, defining whether the slot was allocated at the desired location. */
private Locality locality = Locality.UNCONSTRAINED;
- public SimpleSlot(JobID jobID, Instance instance, int slotNumber, SharedSlot parent, AbstractID groupID){
+
+ /**
+ * Creates a new simple slot that stands alone and does not belong to shared slot.
+ *
+ * @param jobID The ID of the job that the slot is allocated for.
+ * @param instance The instance that the slot belongs to.
+ * @param slotNumber The number of the task slot on the instance.
+ */
+ public SimpleSlot(JobID jobID, Instance instance, int slotNumber) {
+ super(jobID, instance, slotNumber, null, null);
+ }
+
+ /**
+ * Creates a new simple slot that belongs to the given shared slot and
+ * is identified by the given ID..
+ *
+ * @param jobID The ID of the job that the slot is allocated for.
+ * @param instance The instance that the slot belongs to.
+ * @param slotNumber The number of the simple slot in its parent shared slot.
+ * @param parent The parent shared slot.
+ * @param groupID The ID that identifies the group that the slot belongs to.
+ */
+ public SimpleSlot(JobID jobID, Instance instance, int slotNumber, SharedSlot parent, AbstractID groupID) {
super(jobID, instance, slotNumber, parent, groupID);
}
+ // ------------------------------------------------------------------------
+ // Properties
+ // ------------------------------------------------------------------------
+
@Override
public int getNumberLeaves() {
return 1;
}
-
- public Execution getExecution() {
+ /**
+ * Gets the task execution attempt currently executed in this slot. This may return null, if no
+ * task execution attempt has been placed into this slot.
+ *
+ * @return The slot's task execution attempt, or null, if no task is executed in this slot, yet.
+ */
+ public Execution getExecutedVertex() {
return executedTask;
}
- public Locality getLocality() {
- return locality;
- }
-
- public void setLocality(Locality locality) {
- this.locality = locality;
- }
-
+ /**
+ * Atomically sets the executed vertex, if no vertex has been assigned to this slot so far.
+ *
+ * @param executedVertex The vertex to assign to this slot.
+ * @return True, if the vertex was assigned, false, otherwise.
+ */
public boolean setExecutedVertex(Execution executedVertex) {
if (executedVertex == null) {
throw new NullPointerException();
}
// check that we can actually run in this slot
- if (getStatus() != ALLOCATED_AND_ALIVE) {
+ if (isCanceled()) {
return false;
}
@@ -81,7 +112,7 @@ public class SimpleSlot extends Slot {
}
// we need to do a double check that we were not cancelled in the meantime
- if (getStatus() != ALLOCATED_AND_ALIVE) {
+ if (isCanceled()) {
this.executedTask = null;
return false;
}
@@ -89,36 +120,60 @@ public class SimpleSlot extends Slot {
return true;
}
+ /**
+ * Gets the locality information attached to this slot.
+ * @return The locality attached to the slot.
+ */
+ public Locality getLocality() {
+ return locality;
+ }
+
+ /**
+ * Attached locality information to this slot.
+ * @param locality The locality attached to the slot.
+ */
+ public void setLocality(Locality locality) {
+ this.locality = locality;
+ }
+
+ // ------------------------------------------------------------------------
+ // Cancelling & Releasing
+ // ------------------------------------------------------------------------
+
@Override
- public void cancel() {
+ public void releaseSlot() {
+
+ // try to transition to the CANCELED state. That state marks
+ // that the releasing is in progress
if (markCancelled()) {
+
// kill all tasks currently running in this slot
Execution exec = this.executedTask;
if (exec != null && !exec.isFinished()) {
- exec.fail(new Exception("The slot in which the task was scheduled has been killed (probably loss of TaskManager). Instance:"+getInstance()));
+ exec.fail(new Exception(
+ "The slot in which the task was executed has been released. Probably loss of TaskManager "
+ + getInstance()));
}
- }
- }
- @Override
- public void releaseSlot() {
- // cancel everything, if there is something. since this is atomically status based,
- // it will not happen twice if another attempt happened before or concurrently
- try {
- cancel();
- } finally {
- if (getParent() != null) {
- // we have to ask our parent to dispose us
- getParent().disposeChild(this);
- } else {
+ // release directly (if we are directly allocated),
+ // otherwise release through the parent shared slot
+ if (getParent() == null) {
// we have to give back the slot to the owning instance
getInstance().returnAllocatedSlot(this);
}
+ else {
+ // we have to ask our parent to dispose us
+ getParent().releaseChild(this);
+ }
}
}
+ // ------------------------------------------------------------------------
+ // Utilities
+ // ------------------------------------------------------------------------
+
@Override
public String toString() {
return "SimpleSlot " + super.toString();
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Slot.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Slot.java b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Slot.java
index 082fbf2..730e08a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Slot.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Slot.java
@@ -24,27 +24,40 @@ import org.apache.flink.api.common.JobID;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
/**
- * Base class for task slots. TaskManagers offer one or more task slots, they define how many
- * parallel tasks or task groups a TaskManager executes.
+ * Base class for task slots. TaskManagers offer one or more task slots, which define a slice of
+ * their resources.
+ *
+ * <p>In the simplest case, a slot holds a single task ({@link SimpleSlot}). In the more complex
+ * case, a slot is shared ({@link SharedSlot}) and contains a set of tasks. Shared slots may contain
+ * other shared slots which in turn can hold simple slots. That way, a shared slot may define a tree
+ * of slots that belong to it.</p>
*/
public abstract class Slot {
-
+
+ /** Updater for atomic state transitions */
private static final AtomicIntegerFieldUpdater<Slot> STATUS_UPDATER =
AtomicIntegerFieldUpdater.newUpdater(Slot.class, "status");
- protected static final int ALLOCATED_AND_ALIVE = 0; // tasks may be added and might be running
- protected static final int CANCELLED = 1; // no more tasks may run
- protected static final int RELEASED = 2; // has been given back to the instance
+ /** State where slot is fresh and alive. Tasks may be added to the slot. */
+ private static final int ALLOCATED_AND_ALIVE = 0;
+
+ /** State where the slot has been canceled and is in the process of being released */
+ private static final int CANCELLED = 1;
+
+ /** State where all tasks in this slot have been canceled and the slot been given back to the instance */
+ private static final int RELEASED = 2;
+
+ // ------------------------------------------------------------------------
/** The ID of the job this slice belongs to. */
private final JobID jobID;
- /** The id of the group that this slot is allocated to */
+ /** The id of the group that this slot is allocated to. May be null. */
private final AbstractID groupID;
-
+
/** The instance on which the slot is allocated */
private final Instance instance;
-
+
/** The parent of this slot in the hierarchy, or null, if this is the parent */
private final SharedSlot parent;
@@ -53,14 +66,18 @@ public abstract class Slot {
/** The state of the vertex, only atomically updated */
private volatile int status = ALLOCATED_AND_ALIVE;
-
- /** Indicates whether this slot was marked dead by the system */
- private volatile boolean dead = false;
-
- private volatile boolean disposed = false;
-
-
- public Slot(JobID jobID, Instance instance, int slotNumber, SharedSlot parent, AbstractID groupID) {
+
+ /**
+ * Base constructor for slots.
+ *
+ * @param jobID The ID of the job that this slot is allocated for.
+ * @param instance The instance from which this slot is allocated.
+ * @param slotNumber The number of this slot.
+ * @param parent The parent slot that contains this slot. May be null, if this slot is the root.
+ * @param groupID The ID that identifies the task group for which this slot is allocated. May be null
+ * if the slot does not belong to any task group.
+ */
+ protected Slot(JobID jobID, Instance instance, int slotNumber, SharedSlot parent, AbstractID groupID) {
if (jobID == null || instance == null || slotNumber < 0) {
throw new IllegalArgumentException();
}
@@ -83,93 +100,143 @@ public abstract class Slot {
return this.jobID;
}
+ /**
+ * Gets the instance from which the slot was allocated.
+ *
+ * @return The instance from which the slot was allocated.
+ */
public Instance getInstance() {
return instance;
}
+ /**
+ * Gets the number of the slot. For a simple slot, that is the number of the slot
+ * on its instance. For a non-root slot, this returns the number of the slot in the
+ * list amongst its siblings in the tree.
+ *
+ * @return The number of the slot on the instance or amongst its siblings that share the same slot.
+ */
public int getSlotNumber() {
return slotNumber;
}
+ /**
+ * Gets the number of the root slot. This code behaves equal to {@code getRoot().getSlotNumber()}.
+ * If this slot is the root of the tree of shared slots, then this method returns the same
+ * value as {@link #getSlotNumber()}.
+ *
+ * @return The slot number of the root slot.
+ */
+ public int getRootSlotNumber() {
+ if (parent == null) {
+ return slotNumber;
+ } else {
+ return parent.getRootSlotNumber();
+ }
+ }
+
+ /**
+ * Gets the ID that identifies the logical group to which this slot belongs:
+ * <ul>
+ * <li>If the slot does not belong to any group in particular, this field is null.</li>
+ * <li>If this slot was allocated as a sub-slot of a
+ * {@link org.apache.flink.runtime.instance.SlotSharingGroupAssignment},
+ * then this ID will be the JobVertexID of the vertex whose task the slot
+ * holds in its shared slot.</li>
+ * <li>In case that the slot represents the shared slot of a co-location constraint, this ID will be the
+ * ID of the co-location constraint.</li>
+ * </ul>
+ *
+ * @return The ID identifying the logical group of slots.
+ */
public AbstractID getGroupID() {
return groupID;
}
+ /**
+ * Gets the parent slot of this slot. Returns null, if this slot has no parent.
+ *
+ * @return The parent slot, or null, if no this slot has no parent.
+ */
public SharedSlot getParent() {
return parent;
}
public Slot getRoot() {
- if(parent == null){
+ if (parent == null) {
return this;
} else {
return parent.getRoot();
}
}
-
- public int getStatus() {
- return status;
- }
+ /**
+ * Gets the number of simple slots that are at the leaves of the tree of slots.
+ *
+ * @return The number of simple slots at the leaves.
+ */
public abstract int getNumberLeaves();
// --------------------------------------------------------------------------------------------
// Status and life cycle
// --------------------------------------------------------------------------------------------
+ /**
+ * Checks of the slot is still alive, i.e. in state {@link #ALLOCATED_AND_ALIVE}.
+ *
+ * @return True if the slot is alive, false otherwise.
+ */
public boolean isAlive() {
return status == ALLOCATED_AND_ALIVE;
}
+ /**
+ * Checks of the slot has been cancelled. Note that a released slot is also cancelled.
+ *
+ * @return True if the slot is cancelled or released, false otherwise.
+ */
public boolean isCanceled() {
return status != ALLOCATED_AND_ALIVE;
}
+ /**
+ * Checks of the slot has been released.
+ *
+ * @return True if the slot is released, false otherwise.
+ */
public boolean isReleased() {
return status == RELEASED;
}
- public abstract void cancel();
-
- public abstract void releaseSlot();
-
- public boolean markReleased() {
- return STATUS_UPDATER.compareAndSet(this, CANCELLED, RELEASED);
- }
-
- public boolean markCancelled() {
+ /**
+ * Atomically marks the slot as cancelled, if it was alive before.
+ *
+ * @return True, if the state change was successful, false otherwise.
+ */
+ final boolean markCancelled() {
return STATUS_UPDATER.compareAndSet(this, ALLOCATED_AND_ALIVE, CANCELLED);
}
/**
- * Marks this shared slot to be dead. Returns if the slot was alive before. Should only
- * be called through the {@link org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroupAssignment} attribute assignmentGroup.
+ * Atomically marks the slot as released, if it was cancelled before.
*
- * @return if the slot was alive before
+ * @return True, if the state change was successful, false otherwise.
*/
- public boolean markDead() {
- boolean result = !dead;
-
- dead = true;
-
- return result;
- }
-
- public boolean isDead() {
- return dead;
+ final boolean markReleased() {
+ return STATUS_UPDATER.compareAndSet(this, CANCELLED, RELEASED);
}
- public boolean markDisposed() {
- boolean result = !disposed;
-
- disposed = true;
-
- return result;
- }
+ /**
+ * This method cancels and releases the slot and all its sub-slots.
+ *
+ * After this method completed successfully, the slot will be in state "released", and the
+ * {@link #isReleased()} method will return {@code true}.
+ *
+ * If this slot is a simple slot, it will be returned to its instance. If it is a shared slot,
+ * it will release all of its sub-slots and release itself.
+ */
+ public abstract void releaseSlot();
- public boolean isDisposed() {
- return disposed;
- }
// --------------------------------------------------------------------------------------------
// Utilities
@@ -196,4 +263,4 @@ public abstract class Slot {
return "(unknown)";
}
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/main/java/org/apache/flink/runtime/instance/SlotSharingGroupAssignment.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/SlotSharingGroupAssignment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/SlotSharingGroupAssignment.java
new file mode 100644
index 0000000..f2b7dba
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/SlotSharingGroupAssignment.java
@@ -0,0 +1,676 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.instance;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.flink.runtime.jobmanager.scheduler.CoLocationConstraint;
+import org.apache.flink.runtime.jobmanager.scheduler.Locality;
+import org.apache.flink.util.AbstractID;
+import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+
+
+/**
+ * The SlotSharingGroupAssignment manages a set of shared slots, which are shared between
+ * tasks of a {@link org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup}.
+ *
+ * <p>The assignments shares tasks by allowing a shared slot to hold one vertex per
+ * JobVertexID. For example, consider a program consisting of job vertices "source", "map",
+ * "reduce", and "sink". If the slot sharing group spans all four job vertices, then
+ * each shared slot can hold one parallel subtask of the source, the map, the reduce, and the
+ * sink vertex. Each shared slot holds the actual subtasks in child slots, which are (at the leaf level),
+ * the {@link SimpleSlot}s.</p>
+ *
+ * <p>An exception are the co-location-constraints, that define that the i-th subtask of one
+ * vertex needs to be scheduled strictly together with the i-th subtasks of of the vertices
+ * that share the co-location-constraint. To manage that, a co-location-constraint gets its
+ * own shared slot inside the shared slots of a sharing group.</p>
+ *
+ * <p>Consider a job set up like this:</p>
+ *
+ * <pre>
+ * +-------------- Slot Sharing Group --------------+
+ * | |
+ * | +-- Co Location Group --+ |
+ * | | | |
+ * | (source) ---> (head) ---> (tail) ---> (sink) |
+ * | | | |
+ * | +-----------------------+ |
+ * +------------------------------------------------+
+ * </pre>
+ *
+ * <p>The slot hierarchy in the slot sharing group will look like the following</p>
+ *
+ * <pre>
+ * Shared(0)(root)
+ * |
+ * +-- Simple(2)(sink)
+ * |
+ * +-- Shared(1)(co-location-group)
+ * | |
+ * | +-- Simple(0)(tail)
+ * | +-- Simple(1)(head)
+ * |
+ * +-- Simple(0)(source)
+ * </pre>
+ */
+public class SlotSharingGroupAssignment {
+
+ /** The lock globally guards against concurrent modifications in the data structures */
+ private final Object lock = new Object();
+
+ /** All slots currently allocated to this sharing group */
+ private final Set<SharedSlot> allSlots = new LinkedHashSet<SharedSlot>();
+
+ /** The slots available per vertex type (jid), keyed by instance, to make them locatable */
+ private final Map<AbstractID, Map<Instance, List<SharedSlot>>> availableSlotsPerJid =
+ new LinkedHashMap<AbstractID, Map<Instance, List<SharedSlot>>>();
+
+
+ // --------------------------------------------------------------------------------------------
+ // Accounting
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Gets the number of slots that are currently governed by this assignment group.
+ * This refers to the slots allocated from an {@link org.apache.flink.runtime.instance.Instance},
+ * and not the sub-slots given out as children of those shared slots.
+ *
+ * @return The number of resource slots managed by this assignment group.
+ */
+ public int getNumberOfSlots() {
+ return allSlots.size();
+ }
+
+ /**
+ * Gets the number of shared slots into which the given group can place subtasks or
+ * nested task groups.
+ *
+ * @param groupId The ID of the group.
+ * @return The number of shared slots available to the given job vertex.
+ */
+ public int getNumberOfAvailableSlotsForGroup(AbstractID groupId) {
+ synchronized (lock) {
+ Map<Instance, List<SharedSlot>> available = availableSlotsPerJid.get(groupId);
+
+ if (available != null) {
+ Set<SharedSlot> set = new HashSet<SharedSlot>();
+
+ for (List<SharedSlot> list : available.values()) {
+ for (SharedSlot slot : list) {
+ set.add(slot);
+ }
+ }
+
+ return set.size();
+ }
+ else {
+ // if no entry exists for a JobVertexID so far, then the vertex with that ID can
+ // add a subtask into each shared slot of this group. Consequently, all
+ // of them are available for that JobVertexID.
+ return allSlots.size();
+ }
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Slot allocation
+ // ------------------------------------------------------------------------
+
+ /**
+ *
+ * @param sharedSlot
+ * @param locality
+ * @param groupId
+ * @return
+ */
+ public SimpleSlot addSharedSlotAndAllocateSubSlot(SharedSlot sharedSlot, Locality locality, JobVertexID groupId) {
+ return addSharedSlotAndAllocateSubSlot(sharedSlot, locality, groupId, null);
+ }
+
+ /**
+ *
+ * @param sharedSlot
+ * @param locality
+ * @param constraint
+ * @return
+ */
+ public SimpleSlot addSharedSlotAndAllocateSubSlot(SharedSlot sharedSlot, Locality locality,
+ CoLocationConstraint constraint) {
+ return addSharedSlotAndAllocateSubSlot(sharedSlot, locality, null, constraint);
+ }
+
+ private SimpleSlot addSharedSlotAndAllocateSubSlot(SharedSlot sharedSlot, Locality locality,
+ JobVertexID groupId, CoLocationConstraint constraint) {
+ // sanity checks
+ if (!sharedSlot.isRootAndEmpty()) {
+ throw new IllegalArgumentException("The given slot is not an empty root slot.");
+ }
+
+ final Instance location = sharedSlot.getInstance();
+
+ synchronized (lock) {
+ // early out in case that the slot died (instance disappeared)
+ if (!sharedSlot.isAlive()) {
+ return null;
+ }
+
+ // add to the total bookkeeping
+ if (!allSlots.add(sharedSlot)) {
+ throw new IllegalArgumentException("Slot was already contained in the assignment group");
+ }
+
+ SimpleSlot subSlot;
+ AbstractID groupIdForMap;
+
+ if (constraint == null) {
+ // allocate us a sub slot to return
+ subSlot = sharedSlot.allocateSubSlot(groupId);
+ groupIdForMap = groupId;
+ }
+ else {
+ // sanity check
+ if (constraint.isAssignedAndAlive()) {
+ throw new IllegalStateException(
+ "Trying to add a shared slot to a co-location constraint that has a life slot.");
+ }
+
+ // we need a co-location slot --> a SimpleSlot nested in a SharedSlot to
+ // host other co-located tasks
+ SharedSlot constraintGroupSlot = sharedSlot.allocateSharedSlot(constraint.getGroupId());
+ groupIdForMap = constraint.getGroupId();
+
+ if (constraintGroupSlot != null) {
+ // the sub-slots in the co-location constraint slot have no own group IDs
+ subSlot = constraintGroupSlot.allocateSubSlot(null);
+ if (subSlot != null) {
+ // all went well, we can give the constraint its slot
+ constraint.setSharedSlot(constraintGroupSlot);
+
+ // NOTE: Do not lock the location constraint, because we don't yet know whether we will
+ // take the slot here
+ }
+ else {
+ // if we could not create a sub slot, release the co-location slot
+ // note that this does implicitly release the slot we have just added
+ // as well, because we release its last child slot. That is expected
+ // and desired.
+ constraintGroupSlot.releaseSlot();
+ }
+ }
+ else {
+ // this should not happen, as we are under the lock that also
+ // guards slot disposals. Keep the check to be on the safe side
+ subSlot = null;
+ }
+ }
+
+ if (subSlot != null) {
+ // preserve the locality information
+ subSlot.setLocality(locality);
+
+ // let the other groups know that this slot exists and that they
+ // can place a task into this slot.
+ boolean entryForNewJidExists = false;
+
+ for (Map.Entry<AbstractID, Map<Instance, List<SharedSlot>>> entry : availableSlotsPerJid.entrySet()) {
+ // there is already an entry for this groupID
+ if (entry.getKey().equals(groupIdForMap)) {
+ entryForNewJidExists = true;
+ continue;
+ }
+
+ Map<Instance, List<SharedSlot>> available = entry.getValue();
+ putIntoMultiMap(available, location, sharedSlot);
+ }
+
+ // make sure an empty entry exists for this group, if no other entry exists
+ if (!entryForNewJidExists) {
+ availableSlotsPerJid.put(groupIdForMap, new LinkedHashMap<Instance, List<SharedSlot>>());
+ }
+
+ return subSlot;
+ }
+ else {
+ // if sharedSlot is releases, abort.
+ // This should be a rare case, since this method is called with a fresh slot.
+ return null;
+ }
+ }
+ // end synchronized (lock)
+ }
+
+ /**
+ * Gets a slot suitable for the given task vertex. This method will prefer slots that are local
+ * (with respect to {@link ExecutionVertex#getPreferredLocations()}), but will return non local
+ * slots if no local slot is available. The method returns null, when this sharing group has
+ * no slot is available for the given JobVertexID.
+ *
+ * @param vertex The vertex to allocate a slot for.
+ *
+ * @return A slot to execute the given ExecutionVertex in, or null, if none is available.
+ */
+ public SimpleSlot getSlotForTask(ExecutionVertex vertex) {
+ return getSlotForTask(vertex.getJobvertexId(), vertex.getPreferredLocations());
+ }
+
+ /**
+ *
+ * @param vertexID
+ * @param locationPreferences
+ * @return
+ */
+ SimpleSlot getSlotForTask(JobVertexID vertexID, Iterable<Instance> locationPreferences) {
+ synchronized (lock) {
+ Pair<SharedSlot, Locality> p = getSlotForTaskInternal(vertexID, locationPreferences, false);
+
+ if (p != null) {
+ SharedSlot ss = p.getLeft();
+ SimpleSlot slot = ss.allocateSubSlot(vertexID);
+ slot.setLocality(p.getRight());
+ return slot;
+ }
+ else {
+ return null;
+ }
+ }
+ }
+
+ /**
+ * Gets a slot for a task that has a co-location constraint. This method tries to grab
+ * a slot form the location-constraint's shared slot. If that slot has not been initialized,
+ * then the method tries to grab another slot that is available for the location-constraint-group.
+ *
+ * <p>In cases where the co-location constraint has not yet been initialized with a slot,
+ * or where that slot has been disposed in the meantime, this method tries to allocate a shared
+ * slot for the co-location constraint (inside on of the other available slots).</p>
+ *
+ * <p>If a suitable shared slot is available, this method allocates a simple slot within that
+ * shared slot and returns it. If no suitable shared slot could be found, this method
+ * returns null.</p>
+ *
+ * @param vertex The execution vertex to find a slot for.
+ * @param constraint The co-location constraint for the placement of the execution vertex.
+ *
+ * @return A simple slot allocate within a suitable shared slot, or {@code null}, if no suitable
+ * shared slot is available.
+ */
+ public SimpleSlot getSlotForTask(ExecutionVertex vertex, CoLocationConstraint constraint) {
+ return getSlotForTask(constraint, vertex.getPreferredLocations());
+ }
+
+ SimpleSlot getSlotForTask(CoLocationConstraint constraint, Iterable<Instance> locationPreferences) {
+ synchronized (lock) {
+ if (constraint.isAssignedAndAlive()) {
+ // the shared slot of the co-location group is initialized and set we allocate a sub-slot
+ final SharedSlot shared = constraint.getSharedSlot();
+ SimpleSlot subslot = shared.allocateSubSlot(null);
+ subslot.setLocality(Locality.LOCAL);
+ return subslot;
+ }
+ else if (constraint.isAssigned()) {
+ // we had an assignment before.
+
+ SharedSlot previous = constraint.getSharedSlot();
+ if (previous == null) {
+ throw new IllegalStateException("Bug: Found assigned co-location constraint without a slot.");
+ }
+
+ Instance location = previous.getInstance();
+ Pair<SharedSlot, Locality> p = getSlotForTaskInternal(constraint.getGroupId(),
+ Collections.singleton(location), true);
+ if (p == null) {
+ return null;
+ }
+ else {
+ SharedSlot newSharedSlot = p.getLeft();
+
+ // allocate the co-location group slot inside the shared slot
+ SharedSlot constraintGroupSlot = newSharedSlot.allocateSharedSlot(constraint.getGroupId());
+ if (constraintGroupSlot != null) {
+ constraint.setSharedSlot(constraintGroupSlot);
+
+ // the sub slots in the co location constraint slot have no group that they belong to
+ // (other than the co-location-constraint slot)
+ SimpleSlot subSlot = constraintGroupSlot.allocateSubSlot(null);
+ subSlot.setLocality(Locality.LOCAL);
+ return subSlot;
+ }
+ else {
+ // could not allocate the co-location-constraint shared slot
+ return null;
+ }
+ }
+ }
+ else {
+ // the location constraint has not been associated with a shared slot, yet.
+ // grab a new slot and initialize the constraint with that one.
+ // preferred locations are defined by the vertex
+ Pair<SharedSlot, Locality> p =
+ getSlotForTaskInternal(constraint.getGroupId(), locationPreferences, false);
+ if (p == null) {
+ // could not get a shared slot for this co-location-group
+ return null;
+ }
+ else {
+ final SharedSlot availableShared = p.getLeft();
+ final Locality l = p.getRight();
+
+ // allocate the co-location group slot inside the shared slot
+ SharedSlot constraintGroupSlot = availableShared.allocateSharedSlot(constraint.getGroupId());
+
+ // IMPORTANT: We do not lock the location, yet, since we cannot be sure that the
+ // caller really sticks with the slot we picked!
+ constraint.setSharedSlot(constraintGroupSlot);
+
+ // the sub slots in the co location constraint slot have no group that they belong to
+ // (other than the co-location-constraint slot)
+ SimpleSlot sub = constraintGroupSlot.allocateSubSlot(null);
+ sub.setLocality(l);
+ return sub;
+ }
+ }
+ }
+ }
+
+
+ private Pair<SharedSlot, Locality> getSlotForTaskInternal(AbstractID groupId,
+ Iterable<Instance> preferredLocations,
+ boolean localOnly)
+ {
+ // check if there is anything at all in this group assignment
+ if (allSlots.isEmpty()) {
+ return null;
+ }
+
+ // get the available slots for the group
+ Map<Instance, List<SharedSlot>> slotsForGroup = availableSlotsPerJid.get(groupId);
+
+ if (slotsForGroup == null) {
+ // we have a new group, so all slots are available
+ slotsForGroup = new LinkedHashMap<Instance, List<SharedSlot>>();
+ availableSlotsPerJid.put(groupId, slotsForGroup);
+
+ for (SharedSlot availableSlot : allSlots) {
+ putIntoMultiMap(slotsForGroup, availableSlot.getInstance(), availableSlot);
+ }
+ }
+ else if (slotsForGroup.isEmpty()) {
+ // the group exists, but nothing is available for that group
+ return null;
+ }
+
+ // check whether we can schedule the task to a preferred location
+ boolean didNotGetPreferred = false;
+
+ if (preferredLocations != null) {
+ for (Instance location : preferredLocations) {
+
+ // set the flag that we failed a preferred location. If one will be found,
+ // we return early anyways and skip the flag evaluation
+ didNotGetPreferred = true;
+
+ SharedSlot slot = removeFromMultiMap(slotsForGroup, location);
+ if (slot != null && slot.isAlive()) {
+ return new ImmutablePair<SharedSlot, Locality>(slot, Locality.LOCAL);
+ }
+ }
+ }
+
+ // if we want only local assignments, exit now with a "not found" result
+ if (didNotGetPreferred && localOnly) {
+ return null;
+ }
+
+ Locality locality = didNotGetPreferred ? Locality.NON_LOCAL : Locality.UNCONSTRAINED;
+
+ // schedule the task to any available location
+ SharedSlot slot;
+ while ((slot = pollFromMultiMap(slotsForGroup)) != null) {
+ if (slot.isAlive()) {
+ return new ImmutablePair<SharedSlot, Locality>(slot, locality);
+ }
+ }
+
+ // nothing available after all, all slots were dead
+ return null;
+ }
+
+ // ------------------------------------------------------------------------
+ // Slot releasing
+ // ------------------------------------------------------------------------
+
+ /**
+ * Releases the simple slot from the assignment group.
+ *
+ * @param simpleSlot The SimpleSlot to be released
+ */
+ void releaseSimpleSlot(SimpleSlot simpleSlot) {
+ synchronized (lock) {
+ // sanity checks
+ if (simpleSlot.isAlive()) {
+ throw new IllegalStateException("slot is still alive");
+ }
+
+ // check whether the slot is already released
+ if (simpleSlot.markReleased()) {
+
+ AbstractID groupID = simpleSlot.getGroupID();
+ SharedSlot parent = simpleSlot.getParent();
+
+ // if we have a group ID, then our parent slot is tracked here
+ if (groupID != null && !allSlots.contains(parent)) {
+ throw new IllegalArgumentException("Slot was not associated with this SlotSharingGroup before.");
+ }
+
+ int parentRemaining = parent.removeDisposedChildSlot(simpleSlot);
+
+ if (parentRemaining > 0) {
+ // the parent shared slot is still alive. make sure we make it
+ // available again to the group of the just released slot
+
+ if (groupID != null) {
+ // if we have a group ID, then our parent becomes available
+ // for that group again. otherwise, the slot is part of a
+ // co-location group and nothing becomes immediately available
+
+ Map<Instance, List<SharedSlot>> slotsForJid = availableSlotsPerJid.get(groupID);
+
+ // sanity check
+ if (slotsForJid == null) {
+ throw new IllegalStateException("Trying to return a slot for group " + groupID +
+ " when available slots indicated that all slots were available.");
+ }
+
+ putIntoMultiMap(slotsForJid, parent.getInstance(), parent);
+ }
+ }
+ else {
+ // the parent shared slot is now empty and can be released
+ parent.markCancelled();
+ internalDisposeEmptySharedSlot(parent);
+ }
+ }
+ }
+ }
+
+ /**
+ * Called from {@link org.apache.flink.runtime.instance.SharedSlot#releaseSlot()}.
+ *
+ * @param sharedSlot The slot to be released.
+ */
+ void releaseSharedSlot(SharedSlot sharedSlot) {
+ synchronized (lock) {
+ if (sharedSlot.markCancelled()) {
+ // we are releasing this slot
+
+ if (sharedSlot.hasChildren()) {
+ // by simply releasing all children, we should eventually release this slot.
+ Set<Slot> children = sharedSlot.getSubSlots();
+ while (children.size() > 0) {
+ children.iterator().next().releaseSlot();
+ }
+ }
+ else {
+ // if there are no children that trigger the release, we trigger it directly
+ internalDisposeEmptySharedSlot(sharedSlot);
+ }
+ }
+ }
+ }
+
+ /**
+ *
+ * <p><b>NOTE: This method must be called from within a scope that holds the lock.</b></p>
+ *
+ * @param sharedSlot
+ */
+ private void internalDisposeEmptySharedSlot(SharedSlot sharedSlot) {
+ // sanity check
+ if (sharedSlot.isAlive() | !sharedSlot.getSubSlots().isEmpty()) {
+ throw new IllegalArgumentException();
+ }
+
+ final SharedSlot parent = sharedSlot.getParent();
+ final AbstractID groupID = sharedSlot.getGroupID();
+
+ // 1) If we do not have a parent, we are a root slot.
+ // 2) If we are not a root slot, we are a slot with a groupID and our parent
+ // becomes available for that group
+
+ if (parent == null) {
+ // root slot, return to the instance.
+ sharedSlot.getInstance().returnAllocatedSlot(sharedSlot);
+
+ // also, make sure we remove this slot from everywhere
+ allSlots.remove(sharedSlot);
+ removeSlotFromAllEntries(availableSlotsPerJid, sharedSlot);
+ }
+ else if (groupID != null) {
+ // we remove ourselves from our parent slot
+
+ if (sharedSlot.markReleased()) {
+ int parentRemaining = parent.removeDisposedChildSlot(sharedSlot);
+
+ if (parentRemaining > 0) {
+ // the parent becomes available for the group again
+ Map<Instance, List<SharedSlot>> slotsForGroup = availableSlotsPerJid.get(groupID);
+
+ // sanity check
+ if (slotsForGroup == null) {
+ throw new IllegalStateException("Trying to return a slot for group " + groupID +
+ " when available slots indicated that all slots were available.");
+ }
+
+ putIntoMultiMap(slotsForGroup, parent.getInstance(), parent);
+
+ }
+ else {
+ // this was the last child of the parent. release the parent.
+ parent.markCancelled();
+ internalDisposeEmptySharedSlot(parent);
+ }
+ }
+ }
+ else {
+ throw new IllegalStateException(
+ "Found a shared slot that is neither a root slot, nor associated with a vertex group.");
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Utilities
+ // ------------------------------------------------------------------------
+
+ private static void putIntoMultiMap(Map<Instance, List<SharedSlot>> map, Instance location, SharedSlot slot) {
+ List<SharedSlot> slotsForInstance = map.get(location);
+ if (slotsForInstance == null) {
+ slotsForInstance = new ArrayList<SharedSlot>();
+ map.put(location, slotsForInstance);
+ }
+ slotsForInstance.add(slot);
+ }
+
+ private static SharedSlot removeFromMultiMap(Map<Instance, List<SharedSlot>> map, Instance location) {
+ List<SharedSlot> slotsForLocation = map.get(location);
+
+ if (slotsForLocation == null) {
+ return null;
+ }
+ else {
+ SharedSlot slot = slotsForLocation.remove(slotsForLocation.size() - 1);
+ if (slotsForLocation.isEmpty()) {
+ map.remove(location);
+ }
+
+ return slot;
+ }
+ }
+
+ private static SharedSlot pollFromMultiMap(Map<Instance, List<SharedSlot>> map) {
+ Iterator<Map.Entry<Instance, List<SharedSlot>>> iter = map.entrySet().iterator();
+
+ while (iter.hasNext()) {
+ List<SharedSlot> slots = iter.next().getValue();
+
+ if (slots.isEmpty()) {
+ iter.remove();
+ }
+ else if (slots.size() == 1) {
+ SharedSlot slot = slots.remove(0);
+ iter.remove();
+ return slot;
+ }
+ else {
+ return slots.remove(slots.size() - 1);
+ }
+ }
+
+ return null;
+ }
+
+ private static void removeSlotFromAllEntries(Map<AbstractID, Map<Instance, List<SharedSlot>>> availableSlots,
+ SharedSlot slot)
+ {
+ final Instance instance = slot.getInstance();
+
+ for (Map.Entry<AbstractID, Map<Instance, List<SharedSlot>>> entry : availableSlots.entrySet()) {
+ Map<Instance, List<SharedSlot>> map = entry.getValue();
+
+ List<SharedSlot> list = map.get(instance);
+ if (list != null) {
+ list.remove(slot);
+ if (list.isEmpty()) {
+ map.remove(instance);
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationConstraint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationConstraint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationConstraint.java
index 89220be..9ff56a4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationConstraint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationConstraint.java
@@ -24,58 +24,146 @@ import org.apache.flink.runtime.instance.Instance;
import com.google.common.base.Preconditions;
import org.apache.flink.runtime.instance.SharedSlot;
+/**
+ * A CoLocationConstraint manages the location of a set of tasks
+ * (Execution Vertices). In co-location groups, the different subtasks of
+ * different JobVertices need to be executed on the same {@link Instance}.
+ * This is realized by creating a special shared slot that holds these tasks.
+ *
+ * <p>This class tracks the location and the shared slot for this set of tasks.</p>
+ */
public class CoLocationConstraint {
private final CoLocationGroup group;
private volatile SharedSlot sharedSlot;
+ private volatile boolean locationLocked;
+
CoLocationConstraint(CoLocationGroup group) {
Preconditions.checkNotNull(group);
this.group = group;
}
-
+ // ------------------------------------------------------------------------
+ // Status & Properties
+ // ------------------------------------------------------------------------
+
+ /**
+ * Gets the shared slot into which this constraint's tasks are places.
+ *
+ * @return The shared slot into which this constraint's tasks are places.
+ */
public SharedSlot getSharedSlot() {
return sharedSlot;
}
-
+
+ /**
+ * Gets the ID that identifies the co-location group.
+ *
+ * @return The ID that identifies the co-location group.
+ */
+ public AbstractID getGroupId() {
+ return this.group.getId();
+ }
+
+ /**
+ * Checks whether the location of this constraint has been assigned.
+ * The location is assigned once a slot has been set, via the
+ * {@link #setSharedSlot(org.apache.flink.runtime.instance.SharedSlot)} method,
+ * and the location is locked via the {@link #lockLocation()} method.
+ *
+ * @return True if the location has been assigned, false otherwise.
+ */
+ public boolean isAssigned() {
+ return locationLocked;
+ }
+
+ /**
+ * Checks whether the location of this constraint has been assigned
+ * (as defined in the {@link #isAssigned()} method, and the current
+ * shared slot is alive.
+ *
+ * @return True if the location has been assigned and the shared slot is alive,
+ * false otherwise.
+ */
+ public boolean isAssignedAndAlive() {
+ return locationLocked && sharedSlot.isAlive();
+ }
+
+ /**
+ * Gets the location assigned to this slot. This method only succeeds after
+ * the location has been locked via the {@link #lockLocation()} method and
+ * {@link #isAssigned()} returns true.
+ *
+ * @return The instance describing the location for the tasks of this constraint.
+ * @throws IllegalStateException Thrown if the location has not been assigned, yet.
+ */
public Instance getLocation() {
- if (sharedSlot != null) {
+ if (locationLocked) {
return sharedSlot.getInstance();
} else {
- throw new IllegalStateException("Not assigned");
+ throw new IllegalStateException("Location not yet locked");
}
}
-
- public void setSharedSlot(SharedSlot sharedSlot) {
- if (this.sharedSlot == sharedSlot) {
- return;
+
+ // ------------------------------------------------------------------------
+ // Assigning resources and location
+ // ------------------------------------------------------------------------
+
+ /**
+ * Assigns a new shared slot to this co-location constraint. The shared slot
+ * will hold the subtasks that are executed under this co-location constraint.
+ * If the constraint's location is assigned, then this slot needs to be from
+ * the same location (instance) as the one assigned to this constraint.
+ *
+ * <p>If the constraint already has a slot, the current one will be released.</p>
+ *
+ * @param newSlot The new shared slot to assign to this constraint.
+ * @throws IllegalArgumentException If the constraint's location has been assigned and
+ * the new slot is from a different location.
+ */
+ public void setSharedSlot(SharedSlot newSlot) {
+ if (this.sharedSlot == null) {
+ this.sharedSlot = newSlot;
}
- else if (this.sharedSlot == null || this.sharedSlot.isDead()) {
- this.sharedSlot = sharedSlot;
- } else {
- throw new IllegalStateException("Overriding shared slot that is still alive.");
+ else if (newSlot != this.sharedSlot){
+ if (locationLocked && this.sharedSlot.getInstance() != newSlot.getInstance()) {
+ throw new IllegalArgumentException(
+ "Cannot assign different location to a constraint whose location is locked.");
+ }
+ if (this.sharedSlot.isAlive()) {
+ this.sharedSlot.releaseSlot();
+ }
+
+ this.sharedSlot = newSlot;
}
}
-
- public boolean isUnassigned() {
- return this.sharedSlot == null;
- }
-
- public boolean isUnassignedOrDisposed() {
- return this.sharedSlot == null || this.sharedSlot.isDead();
- }
-
- public AbstractID getGroupId() {
- return this.group.getId();
+
+ /**
+ * Locks the location of this slot. The location can be locked only once
+ * and only after a shared slot has been assigned.
+ *
+ * @throws IllegalStateException Thrown, if the location is already locked,
+ * or is no slot has been set, yet.
+ */
+ public void lockLocation() throws IllegalStateException {
+ if (locationLocked) {
+ throw new IllegalStateException("Location is already locked");
+ }
+ if (sharedSlot == null) {
+ throw new IllegalStateException("Cannot lock location without a slot.");
+ }
+ locationLocked = true;
}
+ // ------------------------------------------------------------------------
+ // Utilities
+ // ------------------------------------------------------------------------
+
@Override
public String toString() {
return "CoLocation constraint id " + getGroupId() + " shared slot " + sharedSlot;
}
-
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroup.java
index 74662d5..a88c89d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroup.java
@@ -26,16 +26,26 @@ import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
import com.google.common.base.Preconditions;
+/**
+ * A Co-location group is a group of JobVertices, where the <i>i-th</i> subtask of one vertex
+ * has to be executed on the same TaskManager as the <i>i-th</i> subtask of all
+ * other JobVertices in the same group.
+ *
+ * <p>The co-location group is used for example to make sure that the i-th subtasks for iteration
+ * head and iteration tail are scheduled to the same TaskManager.</p>
+ */
public class CoLocationGroup implements java.io.Serializable {
private static final long serialVersionUID = -2605819490401895297L;
- // we use a job vertex ID, because the co location group acts as a unit inside which exclusive sharing of
- // slots is used
+
+ /** The ID that describes the slot co-location-constraint as a group */
private final AbstractID id = new AbstractID();
+ /** The vertices participating in the co-location group */
private final List<AbstractJobVertex> vertices = new ArrayList<AbstractJobVertex>();
+ /** The constraints, which hold the shared slots for the co-located operators */
private transient ArrayList<CoLocationConstraint> constraints;
// --------------------------------------------------------------------------------------------
@@ -88,15 +98,28 @@ public class CoLocationGroup implements java.io.Serializable {
}
}
}
-
+
+ /**
+ * Gets the ID that identifies this co-location group.
+ *
+ * @return The ID that identifies this co-location group.
+ */
public AbstractID getId() {
return id;
}
-
+
+ /**
+ * Resets this co-location group, meaning that future calls to {@link #getLocationConstraint(int)}
+ * will give out new CoLocationConstraints.
+ *
+ * <p>This method can only be called when no tasks from any of the CoLocationConstraints are
+ * executed any more.</p>
+ */
public void resetConstraints() {
for (CoLocationConstraint c : this.constraints) {
- if (!c.isUnassignedOrDisposed()) {
- throw new IllegalStateException("Cannot reset co-location group: some constraints still have executing vertices.");
+ if (c.isAssignedAndAlive()) {
+ throw new IllegalStateException(
+ "Cannot reset co-location group: some constraints still have live tasks");
}
}
this.constraints.clear();
http://git-wip-us.apache.org/repos/asf/flink/blob/017b4f0f/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/NoResourceAvailableException.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/NoResourceAvailableException.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/NoResourceAvailableException.java
index e5cfebb..e45747b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/NoResourceAvailableException.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/NoResourceAvailableException.java
@@ -58,16 +58,10 @@ public class NoResourceAvailableException extends JobException {
// --------------------------------------------------------------------------------------------
@Override
- public boolean equals(Object obj){
- if(obj == null){
- return false;
- }
+ public boolean equals(Object obj) {
+ return obj instanceof NoResourceAvailableException &&
+ getMessage().equals(((NoResourceAvailableException) obj).getMessage());
- if (!(obj instanceof NoResourceAvailableException)) {
- return false;
- } else {
- return getMessage().equals(((NoResourceAvailableException)obj).getMessage());
- }
}
@Override