[FLINK-18751] Implement SlotSharingExecutionSlotAllocator

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Done
• Component/s: Runtime / Coordination
• Labels: pull-request-available
• Assignee: Andrey Zagrebin
• Reporter: Andrey Zagrebin
• Created: 29/Jul/20 07:48
• Updated: 17/Aug/20 08:20
• Resolved: 17/Aug/20 08:20

Description

SlotSharingExecutionSlotAllocator maintains a SharedSlot for each ExecutionSlotSharingGroup. SlotSharingExecutionSlotAllocator allocates physical slots for SharedSlot(s) and then allocates logical slots from it for scheduled tasks. In this way, the slot sharing hints can be respected in the ExecutionSlotAllocator. And we no longer need to rely on the SlotProvider to do the slot sharing matching. Co-location constraints will be respected since co-located subtasks will be in the same ExecutionSlotSharingGroup.

The physical slot will be lazily allocated for a SharedSlot, upon any hosted subtask asking for the SharedSlot. Each subsequent sharing subtask allocates a logical slot from the SharedSlot. The SharedSlot/physical slot can be released only if all the requested logical slots are released or canceled.

Slot Allocation Process

When SlotSharingExecutionSlotAllocator receives a set of tasks to allocate slots for, it should do the following:

1. Map the tasks to ExecutionSlotSharingGroup(s)
2. Check which ExecutionSlotSharingGroup(s) already have SharedSlot(s)
3. For all involved ExecutionSlotSharingGroup(s) which do not have a SharedSlot yet:
   1. Create a SlotProfile future by MergingSharedSlotProfileRetriever and then
   2. Allocate a physical slot from the PhysicalSlotProvider
   3. Create SharedSlot based on the returned physical slot futures
4. Allocate logical slot futures for the tasks from all corresponding SharedSlot(s).
5. If physical slot future fails, cancel its pending logical slot requests within the SharedSlot
6. Generates SlotExecutionVertexAssignment(s)  based on the logical slot futures and returns the results.

Issue Links

• links to: GitHub Pull Request #13071