[FLINK-7956] [flip6] Add support for queued scheduling with slot sharing to SlotPool #5091

tillrohrmann · 2017-11-27T17:54:53Z

What is the purpose of the change

This commit adds support for queued scheduling with slot sharing to the
SlotPool. The idea of slot sharing is that multiple tasks can run in the
same slot. Moreover, queued scheduling means that a slot request must not
be completed right away but at a later point in time. This allows to
start new TaskExecutors in case that there are no more slots left.

The main component responsible for the management of shared slots is the
SlotSharingManager. The SlotSharingManager maintains internally a tree-like
structure which stores the SlotContext future of the underlying
AllocatedSlot. Whenever this future is completed potentially pending
LogicalSlot instantiations are executed and sent to the slot requester.

A shared slot is represented by a MultiTaskSlot which can harbour multiple
TaskSlots. A TaskSlot can either be a MultiTaskSlot or a SingleTaskSlot.

In order to represent co-location constraints, we first obtain a root
MultiTaskSlot and then allocate a nested MultiTaskSlot in which the
co-located tasks are allocated. The corresponding SlotRequestID is assigned
to the CoLocationConstraint in order to make the TaskSlot retrievable for
other tasks assigned to the same CoLocationConstraint.

This PR also moves the SlotPool components to o.a.f.runtime.jobmaster.slotpool.

This PR is based on #5090

Brief change log

Add SlotSharingManager to manage shared slots
Rework SlotPool to use SlotSharingManager
Add SlotPool#allocateMultiTaskSlot to allocate a shared slot
Add SlotPool#allocateCoLocatedMultiTaskSlot to allocate a co-located slot
Move SlotPool components to o.a.f.runtime.jobmaster.slotpool

Verifying this change

Port SchedulerSlotSharingTest, SchedulerIsolatedTasksTest and
ScheduleWithCoLocationHintTest to run with SlotPool
Add SlotSharingManagerTest, SlotPoolSlotSharingTest and
SlotPoolCoLocationTest

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)
If yes, how is the feature documented? (not applicable)

CC: @GJL

…ing to SlotPool This commit adds support for queued scheduling with slot sharing to the SlotPool. The idea of slot sharing is that multiple tasks can run in the same slot. Moreover, queued scheduling means that a slot request must not be completed right away but at a later point in time. This allows to start new TaskExecutors in case that there are no more slots left. The main component responsible for the management of shared slots is the SlotSharingManager. The SlotSharingManager maintains internally a tree-like structure which stores the SlotContext future of the underlying AllocatedSlot. Whenever this future is completed potentially pending LogicalSlot instantiations are executed and sent to the slot requester. A shared slot is represented by a MultiTaskSlot which can harbour multiple TaskSlots. A TaskSlot can either be a MultiTaskSlot or a SingleTaskSlot. In order to represent co-location constraints, we first obtain a root MultiTaskSlot and then allocate a nested MultiTaskSlot in which the co-located tasks are allocated. The corresponding SlotRequestID is assigned to the CoLocationConstraint in order to make the TaskSlot retrievable for other tasks assigned to the same CoLocationConstraint. Port SchedulerSlotSharingTest, SchedulerIsolatedTasksTest and ScheduleWithCoLocationHintTest to run with SlotPool. Restructure SlotPool components. Add SlotSharingManagerTest, SlotPoolSlotSharingTest and SlotPoolCoLocationTest. This closes apache#5091.

GJL · 2017-12-07T12:04:44Z

...e/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/TestingAllocatedSlotActions.java

+		Consumer<Tuple3<SlotRequestId, SlotSharingGroupId, Throwable>> currentReleaseSlotConsumer = this.releaseSlotConsumer;
+
+		if (currentReleaseSlotConsumer != null) {
+			currentReleaseSlotConsumer.accept(Tuple3.of(slotRequestId, slotSharingGroupId, cause	));


nit: whitespace after cause

... cause ));

Will fix it.

GJL · 2017-12-07T12:09:40Z

flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerTest.java

+				fail("Scheduler accepted dead instance");
+			}
+			catch (IllegalArgumentException e) {
+				// stimmt


GJL · 2017-12-07T12:10:23Z

flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerTest.java

+			assertFalse(i2.isAlive());
+			assertFalse(i3.isAlive());
+		}
+		catch (Exception e) {


Better propagate the exception but I guess this file was just copy pasted.

jup, but I'll remove it.

GJL · 2017-12-07T12:26:40Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotSharingManager.java

+		}
+
+		private void progressToNextElement() {
+			while(baseIterator.hasNext() && ! currentIterator.hasNext()) {


nit: Missing space between while and (. Excess space after !.

Good catch. Will correct it.

GJL · 2017-12-07T12:28:21Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotSharingManager.java

+import org.apache.flink.runtime.jobmaster.SlotContext;
+import org.apache.flink.runtime.jobmaster.SlotOwner;
+import org.apache.flink.runtime.jobmaster.SlotRequestId;
+import org.apache.flink.runtime.instance.SlotSharingGroupId;


nit: wrong import order (not sorted lexicographically)

import org.apache.flink.runtime.instance.SlotSharingGroupId; import org.apache.flink.runtime.jobmanager.scheduler.Locality;

items should appear before LogicalSlot

Will correct it.

GJL · 2017-12-07T13:29:29Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotSharingManager.java

+	 */
+	@VisibleForTesting
+	public Collection<MultiTaskSlot> getResolvedRootSlots() {
+		ResolvedRootSlotValues vs = resolvedMultiTaskSlotValues;


Since this is only for testing, do we really need to cache the instance in resolvedMultiTaskSlotValues? It is not obvious that resolvedMultiTaskSlotValues is only used for testing. How about just

public Collection<MultiTaskSlot> getResolvedRootSlots() { return new ResolvedRootSlotValues(); }

Yes this is better. Will change it.

GJL · 2017-12-07T13:34:12Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotSharingManager.java

+
+	private final SlotSharingGroupId slotSharingGroupId;
+
+	// needed to release allocated slots after a complete multi task slot hierarchy has been released


nit: All fields are commented with non-javadoc comments. Normally comments on fields are also done in Javadoc style, e.g., SlotPool. Javadoc comments on fields are displayed by IntelliJ (Ctrl + J).

good point. Will change it.

GJL · 2017-12-07T14:04:06Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotSharingManager.java

+					final MultiTaskSlot resolvedRootNode = unresolvedRootSlots.remove(slotRequestId);
+
+					if (resolvedRootNode != null) {
+						final Set<MultiTaskSlot> innerCollection = resolvedRootSlots.computeIfAbsent(


In theory the signature of this method allows concurrent modifications on resolvedRootSlots. Maybe a comment why this cannot happen would be nice; perhaps with@GuardedBy.

True. I'll add the necessary synchronization because it's not guaranteed that the future completion won't change in the future.

GJL · 2017-12-07T15:17:40Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/LogicalSlot.java

@@ -32,6 +34,20 @@
 */
 public interface LogicalSlot {

+	 Payload TERMINATED_PAYLOAD = new Payload() {
+
+		private final CompletableFuture<?> COMPLETED_TERMINATION_FUTURE = CompletableFuture.completedFuture(null);


nit: COMPLETED_TERMINATION_FUTURE should be camel cased because is not actually a constant (not static).

GJL · 2017-12-07T17:37:26Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java

+			// sanity check
+			Preconditions.checkState(!multiTaskSlotFuture.getMultiTaskSlot().contains(task.getJobVertexId()));
+
+			final SlotSharingManager.SingleTaskSlot leave = multiTaskSlotFuture.getMultiTaskSlot().allocateSingleTaskSlot(


nit: variable name should be leaf

https://www.dict.cc/?s=leaf

jup, we wouldn't want the single task slot to leave after all.

GJL · 2017-12-07T18:36:49Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java

+					this,
+					providerAndOwner));
+
+			final SlotSharingManager.MultiTaskSlotLocality multiTaskSlotFuture;


The variable name is confusing. multiTaskSlotFuture is not of type Future.

True, will rename it.

GJL · 2017-12-07T18:39:50Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/LogicalSlot.java

+
+	 Payload TERMINATED_PAYLOAD = new Payload() {
+
+		private final CompletableFuture<?> COMPLETED_TERMINATION_FUTURE = CompletableFuture.completedFuture(null);


nit: COMPLETED_TERMINATION_FUTURE should be camel cased because is not actually a constant (not static).

Good point. Will fix it.

GJL · 2017-12-08T10:56:56Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java

+		// lock the co-location constraint once we have obtained the allocated slot
+		coLocationSlot.getSlotContextFuture().whenComplete(
+			(SlotContext slotContext, Throwable throwable) -> {
+				if (throwable == null) {


Is it acceptable to swallow the Throwable here?

I will log the throwable here.

GJL · 2017-12-08T11:12:14Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java

+
+				multiTaskSlotFuture = slotSharingManager.createRootSlot(
+					multiTaskSlotRequestId,
+					futureSlot.thenApply(Function.identity()),


nit: If you change the signature to

public MultiTaskSlot createRootSlot( SlotRequestId slotRequestId, CompletableFuture<? extends SlotContext> slotContextFuture, SlotRequestId allocatedSlotRequestId)

you won't need this trick to satisfy the compiler (PECS rule). Unfortunately more changes in the code would be needed, e.g., in MultiTaskSlot.

Good point. Will change it.

GJL · 2017-12-08T11:33:42Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotSharingManager.java

+ *
+ * <p>The {@link TaskSlot} hierarchy is implemented by {@link MultiTaskSlot} and
+ * {@link SingleTaskSlot}. The former class represents inner nodes which can contain
+ * a number of other {@link TaskSlot} and the latter class represents the leave nodes.


nit: leaf nodes

good catch.

GJL · 2017-12-08T12:40:47Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotSharingManager.java

+				Locality locality) {
+			Preconditions.checkState(!super.contains(groupId));
+
+			final SingleTaskSlot leave = new SingleTaskSlot(


nit: leaf

will change.

GJL · 2017-12-08T12:48:34Z

...untime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotSharingManagerTest.java

+ */
+public class SlotSharingManagerTest extends TestLogger {
+
+	private static final SlotSharingGroupId slotSharingGroupId = new SlotSharingGroupId();


Should be SLOT_SHARING_GROUP_ID since it is a constant.

GJL · 2017-12-08T12:49:01Z

...untime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotSharingManagerTest.java

+	@Test
+	public void testRootSlotRelease() throws ExecutionException, InterruptedException {
+		final CompletableFuture<SlotRequestId> slotReleasedFuture = new CompletableFuture<>();
+		final TestingAllocatedSlotActions testingAllocatedSlotActions = new TestingAllocatedSlotActions();


You already have an instance of TestingAllocatedSlotActions defined in line 57.

Indeed. Will change it.

GJL · 2017-12-08T12:50:02Z

...untime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotSharingManagerTest.java

+
+	private static final DummySlotOwner slotOwner = new DummySlotOwner();
+
+	private static final TestingAllocatedSlotActions allocatedSlotActions = new TestingAllocatedSlotActions();


This instance is mutable... should not be static

Will make it non static.

ifndef-SleePy · 2017-12-05T08:09:06Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/ScheduledUnit.java

+
+	public ScheduledUnit(
+		Execution task,
+		JobVertexID jobVertexId,


We can get JobVertexID from Execution. Do we need this in Constructor?

Yes because task can be null. Will add the missing @Nullable annotation.

ifndef-SleePy · 2017-12-13T10:48:12Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java

+					null,
+					new FlinkException("Locality constraint is not better fulfilled by allocated slot."));
+			}
+			return multiTaskSlotLocality;


If multiTaskSlotLocality != null and multiTaskSlotLocality.getLocality() != LOCAL and slotAndLocality == null
It will return multiTaskSlotLocality without LOCAL locality. Is this expectable?

Yes because it could not find a slot which is LOCAL. The alternative would be to ask the ResourceManager but it will only give you a slot with an UNKNOWN locality because it is uncompleted.

The cluster entrypoints start the ResourceManager with the web interface URL. This URL is used to set the correct tracking URL in Yarn when registering the Yarn application. This closes apache#5128.

This commit introduces the SlotContext which is an abstraction for the SimpleSlot to obtain the relevant slot information to do the communication with the TaskManager without relying on the AllocatedSlot which is now only used by the SlotPool. This closes apache#5088.

Before logical slots like the SimpleSlot and SharedSlot where associated to the actually allocated slot via the AllocationID. This, however, was sub-optimal because allocated slots can be re-used to fulfill also other slot requests (logical slots). Therefore, we should bind the logical slots to the right id with the right lifecycle which is the slot request id. This closes apache#5089.

Not only check for a slot request with the right allocation id but also check whether we can fulfill other pending slot requests with an unclaimed offered slot before adding it to the list of available slots. This closes apache#5090.

…ing to SlotPool This commit adds support for queued scheduling with slot sharing to the SlotPool. The idea of slot sharing is that multiple tasks can run in the same slot. Moreover, queued scheduling means that a slot request must not be completed right away but at a later point in time. This allows to start new TaskExecutors in case that there are no more slots left. The main component responsible for the management of shared slots is the SlotSharingManager. The SlotSharingManager maintains internally a tree-like structure which stores the SlotContext future of the underlying AllocatedSlot. Whenever this future is completed potentially pending LogicalSlot instantiations are executed and sent to the slot requester. A shared slot is represented by a MultiTaskSlot which can harbour multiple TaskSlots. A TaskSlot can either be a MultiTaskSlot or a SingleTaskSlot. In order to represent co-location constraints, we first obtain a root MultiTaskSlot and then allocate a nested MultiTaskSlot in which the co-located tasks are allocated. The corresponding SlotRequestID is assigned to the CoLocationConstraint in order to make the TaskSlot retrievable for other tasks assigned to the same CoLocationConstraint. Port SchedulerSlotSharingTest, SchedulerIsolatedTasksTest and ScheduleWithCoLocationHintTest to run with SlotPool. Restructure SlotPool components. Add SlotSharingManagerTest, SlotPoolSlotSharingTest and SlotPoolCoLocationTest. This closes apache#5091.

tillrohrmann · 2017-12-14T16:41:02Z

Thanks for the review @GJL and @ifndef-SleePy. Travis passed locally. Merging this PR.

tillrohrmann mentioned this pull request Nov 29, 2017

[FLINK-8171] [flip6] Remove work arounds from Flip6LocalStreamEnvironment #5101

Merged

tillrohrmann force-pushed the slotPoolSlots branch from 6489c67 to c4fbaf0 Compare December 1, 2017 23:01

ifndef-SleePy mentioned this pull request Dec 4, 2017

[Flink-7871] [flip6] SlotPool should release unused slots to RM #5048

Closed

GJL reviewed Dec 7, 2017

View reviewed changes

GJL reviewed Dec 8, 2017

View reviewed changes

ifndef-SleePy reviewed Dec 14, 2017

View reviewed changes

tillrohrmann added 5 commits December 14, 2017 14:32

[FLINK-8120] [flip6] Register Yarn application with correct tracking URL

627bcda

The cluster entrypoints start the ResourceManager with the web interface URL. This URL is used to set the correct tracking URL in Yarn when registering the Yarn application. This closes apache#5128.

[hotfix] Speed up RecoveryITCase

331ce82

tillrohrmann force-pushed the slotPoolSlots branch from c4fbaf0 to 3fdb5c6 Compare December 14, 2017 14:25

tillrohrmann added 2 commits December 14, 2017 15:26

[hotfix] [tests] Speed up queryable state IT tests by removing sleep

917fbcb

tillrohrmann force-pushed the slotPoolSlots branch from 3fdb5c6 to 917fbcb Compare December 14, 2017 14:26

asfgit merged commit 917fbcb into apache:master Dec 14, 2017

tillrohrmann deleted the slotPoolSlots branch February 13, 2018 09:11

rmetzger added the component=Runtime/Coordination label Mar 18, 2019


		private final SlotSharingGroupId slotSharingGroupId;

		// needed to release allocated slots after a complete multi task slot hierarchy has been released


		Payload TERMINATED_PAYLOAD = new Payload() {

		private final CompletableFuture<?> COMPLETED_TERMINATION_FUTURE = CompletableFuture.completedFuture(null);


		private static final DummySlotOwner slotOwner = new DummySlotOwner();

		private static final TestingAllocatedSlotActions allocatedSlotActions = new TestingAllocatedSlotActions();

[FLINK-7956] [flip6] Add support for queued scheduling with slot sharing to SlotPool #5091

[FLINK-7956] [flip6] Add support for queued scheduling with slot sharing to SlotPool #5091

Conversation

tillrohrmann commented Nov 27, 2017

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

GJL Dec 7, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

GJL Dec 7, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

GJL Dec 7, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

GJL Dec 8, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

GJL Dec 8, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tillrohrmann commented Dec 14, 2017

GJL Dec 7, 2017 •

edited

GJL Dec 7, 2017 •

edited

GJL Dec 7, 2017 •

edited

GJL Dec 8, 2017 •

edited

GJL Dec 8, 2017 •

edited