org.apache.flink.runtime.jobmaster.slotpool

Class SlotPool

java.lang.Object

org.apache.flink.runtime.rpc.RpcEndpoint

org.apache.flink.runtime.jobmaster.slotpool.SlotPool

All Implemented Interfaces:

AllocatedSlotActions, SlotPoolGateway, RpcGateway

public class SlotPool
extends RpcEndpoint
implements SlotPoolGateway, AllocatedSlotActions

The slot pool serves slot request issued by ExecutionGraph. It will will attempt to acquire new slots from the ResourceManager when it cannot serve a slot request. If no ResourceManager is currently available, or it gets a decline from the ResourceManager, or a request times out, it fails the slot request. The slot pool also holds all the slots that were offered to it and accepted, and can thus provides registered free slots even if the ResourceManager is down. The slots will only be released when they are useless, e.g. when the job is fully running but we still have some free slots.

All the allocation or the slot offering will be identified by self generated AllocationID, we will use it to eliminate ambiguities.

TODO : Make pending requests location preference aware TODO : Make pass location preferences to ResourceManager when sending a slot request

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected static class	SlotPool.AvailableSlots Organize all available slots from different points of view.

Nested classes/interfaces inherited from class org.apache.flink.runtime.rpc.RpcEndpoint

RpcEndpoint.MainThreadExecutor

Field Summary

Fields

Modifier and Type	Field and Description
protected Map<SlotSharingGroupId,SlotSharingManager>	slotSharingManagers Managers for the different slot sharing groups.

Fields inherited from class org.apache.flink.runtime.rpc.RpcEndpoint

log, rpcServer

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	SlotPool(RpcService rpcService, JobID jobId, SchedulingStrategy schedulingStrategy)
	SlotPool(RpcService rpcService, JobID jobId, SchedulingStrategy schedulingStrategy, Clock clock, Time rpcTimeout, Time idleSlotTimeout)

Method Summary

Modifier and Type	Method and Description
CompletableFuture<LogicalSlot>	allocateSlot(SlotRequestId slotRequestId, ScheduledUnit task, SlotProfile slotProfile, boolean allowQueuedScheduling, Time allocationTimeout) Requests to allocate a slot for the given ScheduledUnit.
void	connectToResourceManager(ResourceManagerGateway resourceManagerGateway) Connects the SlotPool to the given ResourceManager.
void	disconnectResourceManager() Disconnects the slot pool from its current Resource Manager.
void	failAllocation(AllocationID allocationID, Exception cause) Fail the specified allocation and release the corresponding slot if we have one.
protected org.apache.flink.runtime.jobmaster.slotpool.SlotPool.AllocatedSlots	getAllocatedSlots()
protected SlotPool.AvailableSlots	getAvailableSlots()
SlotOwner	getSlotOwner() Gets the slot owner implementation for this pool.
SlotProvider	getSlotProvider() Gets the slot provider implementation for this pool.
CompletableFuture<Boolean>	offerSlot(TaskManagerLocation taskManagerLocation, TaskManagerGateway taskManagerGateway, SlotOffer slotOffer) Slot offering by TaskExecutor with AllocationID.
CompletableFuture<Collection<SlotOffer>>	offerSlots(TaskManagerLocation taskManagerLocation, TaskManagerGateway taskManagerGateway, Collection<SlotOffer> offers) Offers multiple slots to the SlotPool.
CompletableFuture<Void>	postStop() User overridable callback.
CompletableFuture<Acknowledge>	registerTaskManager(ResourceID resourceID) Register TaskManager to this pool, only those slots come from registered TaskManager will be considered valid.
CompletableFuture<Acknowledge>	releaseSlot(SlotRequestId slotRequestId, SlotSharingGroupId slotSharingGroupId, Throwable cause) Releases the slot with the given SlotRequestId.
CompletableFuture<Acknowledge>	releaseTaskManager(ResourceID resourceId, Exception cause) Unregister TaskManager from this pool, all the related slots will be released and tasks be canceled.
void	start() Starts the rpc endpoint.
void	start(JobMasterId jobMasterId, String newJobManagerAddress) Start the slot pool to accept RPC calls.
void	suspend() Suspends this pool, meaning it has lost its authority to accept and distribute slots.
protected void	timeoutPendingSlotRequest(SlotRequestId slotRequestId)

Methods inherited from class org.apache.flink.runtime.rpc.RpcEndpoint

callAsync, getAddress, getEndpointId, getHostname, getMainThreadExecutor, getRpcService, getSelfGateway, getTerminationFuture, runAsync, scheduleRunAsync, scheduleRunAsync, shutDown, stop, validateRunsInMainThread

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.apache.flink.runtime.rpc.RpcGateway

getAddress, getHostname

Field Detail

slotSharingManagers

protected final Map<SlotSharingGroupId,SlotSharingManager> slotSharingManagers

Managers for the different slot sharing groups.

Constructor Detail

SlotPool

@VisibleForTesting
protected SlotPool(RpcService rpcService,
                                      JobID jobId,
                                      SchedulingStrategy schedulingStrategy)

SlotPool

public SlotPool(RpcService rpcService,
                JobID jobId,
                SchedulingStrategy schedulingStrategy,
                Clock clock,
                Time rpcTimeout,
                Time idleSlotTimeout)

Method Detail

start

public void start()

Description copied from class: RpcEndpoint

Starts the rpc endpoint. This tells the underlying rpc server that the rpc endpoint is ready to process remote procedure calls.

IMPORTANT: Whenever you override this method, call the parent implementation to enable rpc processing. It is advised to make the parent call last.

Overrides:

start in class RpcEndpoint

start

public void start(JobMasterId jobMasterId,
                  String newJobManagerAddress)
           throws Exception

Start the slot pool to accept RPC calls.

Parameters:

jobMasterId - The necessary leader id for running the job.

newJobManagerAddress - for the slot requests which are sent to the resource manager

Throws:

Exception

postStop

public CompletableFuture<Void> postStop()

Description copied from class: RpcEndpoint

User overridable callback.

This method is called when the RpcEndpoint is being shut down. The method is guaranteed to be executed in the main thread context and can be used to clean up internal state.

IMPORTANT: This method should never be called directly by the user.

Specified by:

postStop in class RpcEndpoint

Returns:

Future which is completed once all post stop actions are completed. If an error occurs this future is completed exceptionally

suspend

public void suspend()

Suspends this pool, meaning it has lost its authority to accept and distribute slots.

Specified by:

suspend in interface SlotPoolGateway

getSlotOwner

public SlotOwner getSlotOwner()

Gets the slot owner implementation for this pool.

This method does not mutate state and can be called directly (no RPC indirection)

Returns:

The slot owner implementation for this pool.

getSlotProvider

public SlotProvider getSlotProvider()

Gets the slot provider implementation for this pool.

This method does not mutate state and can be called directly (no RPC indirection)

Returns:

The slot provider implementation for this pool.

connectToResourceManager

public void connectToResourceManager(ResourceManagerGateway resourceManagerGateway)

Description copied from interface: SlotPoolGateway

Connects the SlotPool to the given ResourceManager. After this method is called, the SlotPool will be able to request resources from the given ResourceManager.

Specified by:

connectToResourceManager in interface SlotPoolGateway

Parameters:

resourceManagerGateway - The RPC gateway for the resource manager.

disconnectResourceManager

public void disconnectResourceManager()

Description copied from interface: SlotPoolGateway

Disconnects the slot pool from its current Resource Manager. After this call, the pool will not be able to request further slots from the Resource Manager, and all currently pending requests to the resource manager will be canceled.

The slot pool will still be able to serve slots from its internal pool.

Specified by:

disconnectResourceManager in interface SlotPoolGateway

allocateSlot

public CompletableFuture<LogicalSlot> allocateSlot(SlotRequestId slotRequestId,
                                                   ScheduledUnit task,
                                                   SlotProfile slotProfile,
                                                   boolean allowQueuedScheduling,
                                                   Time allocationTimeout)

Description copied from interface: SlotPoolGateway

Requests to allocate a slot for the given ScheduledUnit. The request is uniquely identified by the provided SlotRequestId which can also be used to release the slot via AllocatedSlotActions.releaseSlot(SlotRequestId, SlotSharingGroupId, Throwable). The allocated slot will fulfill the requested ResourceProfile and it is tried to place it on one of the location preferences.

If the returned future must not be completed right away (a.k.a. the slot request can be queued), allowQueuedScheduling must be set to true.

Specified by:

allocateSlot in interface SlotPoolGateway

Parameters:

slotRequestId - identifying the requested slot

task - for which to allocate slot

slotProfile - profile that specifies the requirements for the requested slot

allowQueuedScheduling - true if the slot request can be queued (e.g. the returned future must not be completed)

allocationTimeout - for the operation

Returns:

Future which is completed with the allocated LogicalSlot

releaseSlot

public CompletableFuture<Acknowledge> releaseSlot(SlotRequestId slotRequestId,
                                                  @Nullable
                                                  SlotSharingGroupId slotSharingGroupId,
                                                  Throwable cause)

Description copied from interface: AllocatedSlotActions

Releases the slot with the given SlotRequestId. If the slot belonged to a slot sharing group, then the corresponding SlotSharingGroupId has to be provided. Additionally, one can provide a cause for the slot release.

Specified by:

releaseSlot in interface AllocatedSlotActions

Parameters:

slotRequestId - identifying the slot to release

slotSharingGroupId - identifying the slot sharing group to which the slot belongs, null if none

cause - of the slot release, null if none

Returns:

Acknowledge (future) after the slot has been released

offerSlots

public CompletableFuture<Collection<SlotOffer>> offerSlots(TaskManagerLocation taskManagerLocation,
                                                           TaskManagerGateway taskManagerGateway,
                                                           Collection<SlotOffer> offers)

Description copied from interface: SlotPoolGateway

Offers multiple slots to the SlotPool. The slot offerings can be individually accepted or rejected by returning the collection of accepted slot offers.

Specified by:

offerSlots in interface SlotPoolGateway

Parameters:

taskManagerLocation - from which the slot offers originate

taskManagerGateway - to talk to the slot offerer

offers - slot offers which are offered to the SlotPool

Returns:

A collection of accepted slot offers (future). The remaining slot offers are implicitly rejected.

offerSlot

public CompletableFuture<Boolean> offerSlot(TaskManagerLocation taskManagerLocation,
                                            TaskManagerGateway taskManagerGateway,
                                            SlotOffer slotOffer)

Slot offering by TaskExecutor with AllocationID. The AllocationID is originally generated by this pool and transfer through the ResourceManager to TaskManager. We use it to distinguish the different allocation we issued. Slot offering may be rejected if we find something mismatching or there is actually no pending request waiting for this slot (maybe fulfilled by some other returned slot).

Specified by:

offerSlot in interface SlotPoolGateway

Parameters:

taskManagerLocation - location from where the offer comes from

taskManagerGateway - TaskManager gateway

slotOffer - the offered slot

Returns:

True if we accept the offering

failAllocation

public void failAllocation(AllocationID allocationID,
                           Exception cause)

Fail the specified allocation and release the corresponding slot if we have one. This may triggered by JobManager when some slot allocation failed with rpcTimeout. Or this could be triggered by TaskManager, when it finds out something went wrong with the slot, and decided to take it back.

Specified by:

failAllocation in interface SlotPoolGateway

Parameters:

allocationID - Represents the allocation which should be failed

cause - The cause of the failure

registerTaskManager

public CompletableFuture<Acknowledge> registerTaskManager(ResourceID resourceID)

Register TaskManager to this pool, only those slots come from registered TaskManager will be considered valid. Also it provides a way for us to keep "dead" or "abnormal" TaskManagers out of this pool.

Specified by:

registerTaskManager in interface SlotPoolGateway

Parameters:

resourceID - The id of the TaskManager

Returns:

Future acknowledge if th operation was successful

releaseTaskManager

public CompletableFuture<Acknowledge> releaseTaskManager(ResourceID resourceId,
                                                         Exception cause)

Unregister TaskManager from this pool, all the related slots will be released and tasks be canceled. Called when we find some TaskManager becomes "dead" or "abnormal", and we decide to not using slots from it anymore.

Specified by:

releaseTaskManager in interface SlotPoolGateway

Parameters:

resourceId - The id of the TaskManager

cause - for the releasing of the TaskManager

Returns:

Future acknowledge which is completed after the TaskExecutor has been released

timeoutPendingSlotRequest

@VisibleForTesting
protected void timeoutPendingSlotRequest(SlotRequestId slotRequestId)

getAllocatedSlots

@VisibleForTesting
protected org.apache.flink.runtime.jobmaster.slotpool.SlotPool.AllocatedSlots getAllocatedSlots()

getAvailableSlots

@VisibleForTesting
protected SlotPool.AvailableSlots getAvailableSlots()