flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertex.java
/**
	 * Acquires a slot for all the execution vertices of this ExecutionJobVertex. The method returns
	 * pairs of the slots and execution attempts, to ease correlation between vertices and execution
	 * attempts.
	 * 
	 * <p>If this method throws an exception, it makes sure to release all so far requested slots.
	 * 
	 * @param resourceProvider The resource provider from whom the slots are requested.
	 * @param queued if the allocation can be queued
	 * @param locationPreferenceConstraint constraint for the location preferences
	 */
Acquires a slot for all the execution vertices of this ExecutionJobVertex. The method returns
pairs of the slots and execution attempts, to ease correlation between vertices and execution
attempts.

If this method throws an exception, it makes sure to release all so far requested slots.

@param resourceProvider The resource provider from whom the slots are requested.
@param queued if the allocation can be queued
@param locationPreferenceConstraint constraint for the location preferences
Acquires a slot for all the execution vertices of this ExecutionJobVertex . The method returns pairs of the slots and execution attempts to ease correlation between vertices and execution attempts . If this method throws an exception it makes sure to release all so far requested slots . @param resourceProvider The resource provider from whom the slots are requested . @param queued if the allocation can be queued @param locationPreferenceConstraint constraint for the location preferences
public Collection<CompletableFuture<Execution>> allocateResourcesForAll(
			SlotProvider resourceProvider,
			boolean queued,
			LocationPreferenceConstraint locationPreferenceConstraint) {
		final ExecutionVertex[] vertices = this.taskVertices;
		final CompletableFuture<Execution>[] slots = new CompletableFuture[vertices.length];
		for (int i = 0; i < vertices.length; i++) {
			final Execution exec = vertices[i].getCurrentExecutionAttempt();
			final CompletableFuture<Execution> allocationFuture = exec.allocateAndAssignSlotForExecution(
				resourceProvider,
				queued,
				locationPreferenceConstraint);
			slots[i] = allocationFuture;
		}
		return Arrays.asList(slots);
	}
public Collection < CompletableFuture < Execution > > allocateResourcesForAll ( SlotProvider resourceProvider , boolean queued , LocationPreferenceConstraint locationPreferenceConstraint ) { final ExecutionVertex [ ] vertices = this . taskVertices ; final CompletableFuture < Execution > [ ] slots = new CompletableFuture [ vertices . length ] ; for ( int i = 0 ; i < vertices . length ; i ++ ) { final Execution exec = vertices [ i ] . getCurrentExecutionAttempt ( ) ; final CompletableFuture < Execution > allocationFuture = exec . allocateAndAssignSlotForExecution ( resourceProvider , queued , locationPreferenceConstraint ) ; slots [ i ] = allocationFuture ; } return Arrays . asList ( slots ) ; }
Acquires a slot for all the execution vertices of this ExecutionJobVertex.
Acquires a slot for all the execution vertices of this ExecutionJobVertex .
// try to acquire a slot future for each execution. // we store the execution with the future just to be on the safe side // allocate the next slot (future) // all good, we acquired all slots