...

• For tasks with specified resource requirements, we add up resource requirements of all the tasks in the slot sharing group, and request a slot with the sum resources.
• For tasks with unknown resource requirements, we request a slot with default resources.

Implementation Steps

Step 1. Change pipelined edges between different slot sharing groups to blocking

StreamGraphGenerator should make this change at compiling time.

This step introduce behavior changes, which does not affect functionality and is necessary a cost to prevent potential deadlocks.

Step 2. Introduce option allSourcesInSamePipelinedRegion in ExecutionConfig

• Introduce option allSourcesInSamePipelinedRegion in ExecutionConfig
• Set it to false by default
• Set it to true for streaming jobs

This step should not introduce any behavior changes. 

Step 3. Set slot sharing groups according to pipelined regions

StreamGraphGenerator set slot sharing group for operators at compiling time.

• Identify pipelined regions
  • after change pipelined edges between slot sharing groups to blocking
  • with respect to allSourcesInSamePipelinedRegion
• Set slot sharing groups according to pipelined regions 
  • By default, each pipelined region should go into a separate slot sharing group
  • If the user sets operators in multiple pipelined regions into same slot sharing group, it should be respected

This step should not introduce any behavior changes, given that later scheduled pipelined regions can reuse slots from previous scheduled pipelined regions. 

Step 4. Introduce managed memory fractions to StreamConfig

Introduce fracManagedMemOnHeap and fracManagedMemOffHeap in StreamConfig, so they can be set by StreamGraphGenerator and used by operators in runtime. 

This step should not introduce any behavior changes.

Step 5. Set managed memory fractions according to slot sharing groups

• For operators with specified ResourceSpecs, calculate fractions according to operators ResourceSpecs
• For operators with unknown ResourceSpecs, calculate fractions according to number of operators using managed memory

This step should not introduce any behavior changes.

Step 6. Operators use fractions to decide how many managed memory to allocate

• Operators allocate memory segments with the amount returned by MemoryManager#computeNumberOfPages.
• Operators reserve memory with the amount returned by MemoryManager#computeMemorySize. 

This step activates the new fraction based managed memory.

Compatibility, Deprecation, and Migration Plan

...