Re: [PR] [FLINK-33875][runtime] Introduce the TASKS for TaskManagerLoadBalanceMode enum [flink]

["1996fanrui (via GitHub)" <gi...@apache.org>]

1996fanrui commented on code in PR #25134:
URL: https://github.com/apache/flink/pull/25134#discussion_r1701252201


##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridge.java:
##########
@@ -202,10 +206,73 @@ protected void onReleaseTaskManager(ResourceCounter previouslyFulfilledRequireme
 
     @VisibleForTesting
     void newSlotsAreAvailable(Collection<? extends PhysicalSlot> newSlots) {
+        if (pendingRequests.isEmpty()) {
+            return;
+        }
+
+        if (slotBatchAllocatable) {
+            newSlotsAvailableForSlotBatchAllocatable(newSlots);
+        } else {
+            newSlotsAvailableForNonSlotBatchAllocatable(newSlots);
+        }
+    }
+
+    private void newSlotsAvailableForSlotBatchAllocatable(
+            Collection<? extends PhysicalSlot> newSlots) {
+        log.debug("Received new available slots: {}", newSlots);
+
+        final FreeSlotInfoTracker freeSlotInfoTracker =
+                getDeclarativeSlotPool().getFreeSlotInfoTracker();
+
+        final int slotsNum = freeSlotInfoTracker.getAvailableSlots().size();
+        if (slotsNum < pendingRequests.size()) {
+            // Do nothing and waiting slots.
+            log.debug(
+                    "The number of available slots: {}, the required number of slots: {}, waiting for more available slots.",
+                    slotsNum,
+                    pendingRequests.size());
+            return;
+        }
+
+        final Collection<PhysicalSlot> availableSlots =
+                freeSlotInfoTracker.getFreeSlotsInformation(PhysicalSlot.class);
+        final Collection<RequestSlotMatchingStrategy.RequestSlotMatch> requestSlotMatches =
+                requestSlotMatchingStrategy.matchRequestsAndSlots(
+                        availableSlots, pendingRequests.values());
+        if (requestSlotMatches.size() == pendingRequests.size()) {
+            reserveMatchedFreeSlots(requestSlotMatches);
+            fulfillMatchedSlots(requestSlotMatches);
+        } else if (requestSlotMatches.size() < pendingRequests.size()) {
+            // Do nothing and waiting slots.
+            log.debug(
+                    "Ignored the matched results: {}, pendingRequests: {}, waiting for more available slots.",
+                    requestSlotMatches,
+                    pendingRequests);
+        } else {
+            // For requestSlotMatches.size() > pendingRequests.size()
+            throw new IllegalStateException(
+                    "The number of matched slots is not equals to the pendingRequests.");
+        }
+    }
+
+    private void newSlotsAvailableForNonSlotBatchAllocatable(

Review Comment:
   ```suggestion
       private void newSlotsAvailableForDirectlyAllocatable(
   ```
   
   how about updating the name to `newSlotsAvailableForDirectlyAllocatable`?



##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridge.java:
##########
@@ -221,7 +288,10 @@ void newSlotsAreAvailable(Collection<? extends PhysicalSlot> newSlots) {
                     slot.getAllocationId(),
                     pendingRequest.getResourceProfile());
         }
+    }
 
+    private void fulfillMatchedSlots(

Review Comment:
   nit: all callers call `reserveMatchedFreeSlots` and `fulfillMatchedSlots` together, why do we need 2 methods?
   
   As I understand, merge them into one method is enough: `reserveAndFulfillMatchedFreeSlots`.



##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/FreeSlotInfoTracker.java:
##########
@@ -58,7 +58,19 @@ public interface FreeSlotInfoTracker {
      * @return a list of {@link SlotInfo} objects about all slots that are currently available in
      *     the slot pool.
      */
-    Collection<SlotInfo> getFreeSlotsInformation();
+    default Collection<SlotInfo> getFreeSlotsInformation() {
+        return getFreeSlotsInformation(SlotInfo.class);
+    }
+
+    /**
+     * Returns a list of {@link SlotInfo} objects about all slots that are currently available in
+     * the slot pool.
+     *
+     * @param klass The class of the type of the elements in the result.
+     * @return a list of {@link SlotInfo} objects about all slots that are currently available in
+     *     the slot pool.
+     */
+    <T extends SlotInfo> Collection<T> getFreeSlotsInformation(Class<T> klass);

Review Comment:
   `PhysicalSlot` extends `SlotInfo`, so could we return `PhysicalSlot` here directly?  It doesn't effect all old code.
   
   If so, we could refactor `FreeSlotInfoTracker` to `FreePhysicalSlotTracker` in this PR(by a separate commit), and update all `SlotInfo` to `PhysicalSlot`. In addition, the comment of `FreeSlotInfoTracker` should be changed together, it not only supports `SlotSelectionStrategy`.
   
   Also, the comment of `getFreeSlotsInformation(Class<T> klass);` is copied from `getFreeSlotsInformation()`, it's a little unsuitable here.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org