You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2023/04/25 07:42:38 UTC
[flink] 02/04: [FLINK-31842][runtime] remove redundant class SlotInfoAndResources.
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit baf74c0341307533b11763df8415623c08ec7ba9
Author: Weihua Hu <hu...@gmail.com>
AuthorDate: Sun Apr 23 16:11:01 2023 +0800
[FLINK-31842][runtime] remove redundant class SlotInfoAndResources.
SlotInfoAndResources and slotInfoWithUtilization have the same functions,
we can remove SlotInfoAndResources safely.
---
...ultLocationPreferenceSlotSelectionStrategy.java | 9 ++---
...OutLocationPreferenceSlotSelectionStrategy.java | 15 ++++---
.../LocationPreferenceSlotSelectionStrategy.java | 21 +++++-----
.../slotpool/PhysicalSlotProviderImpl.java | 6 +--
.../PreviousAllocationSlotSelectionStrategy.java | 21 +++++-----
.../jobmaster/slotpool/SlotSelectionStrategy.java | 47 +---------------------
...ocationPreferenceSlotSelectionStrategyTest.java | 6 +--
.../types/SlotSelectionStrategyTestBase.java | 14 +++----
8 files changed, 42 insertions(+), 97 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultLocationPreferenceSlotSelectionStrategy.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultLocationPreferenceSlotSelectionStrategy.java
index 9f32c57f265..b3798c1bc1e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultLocationPreferenceSlotSelectionStrategy.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultLocationPreferenceSlotSelectionStrategy.java
@@ -32,12 +32,11 @@ class DefaultLocationPreferenceSlotSelectionStrategy
@Nonnull
@Override
protected Optional<SlotInfoAndLocality> selectWithoutLocationPreference(
- @Nonnull Collection<SlotInfoAndResources> availableSlots,
+ @Nonnull Collection<SlotInfoWithUtilization> availableSlots,
@Nonnull ResourceProfile resourceProfile) {
- for (SlotInfoAndResources candidate : availableSlots) {
- if (candidate.getRemainingResources().isMatching(resourceProfile)) {
- return Optional.of(
- SlotInfoAndLocality.of(candidate.getSlotInfo(), Locality.UNCONSTRAINED));
+ for (SlotInfoWithUtilization candidate : availableSlots) {
+ if (candidate.getResourceProfile().isMatching(resourceProfile)) {
+ return Optional.of(SlotInfoAndLocality.of(candidate, Locality.UNCONSTRAINED));
}
}
return Optional.empty();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/EvenlySpreadOutLocationPreferenceSlotSelectionStrategy.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/EvenlySpreadOutLocationPreferenceSlotSelectionStrategy.java
index f510da2cabf..4b054a0be8d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/EvenlySpreadOutLocationPreferenceSlotSelectionStrategy.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/EvenlySpreadOutLocationPreferenceSlotSelectionStrategy.java
@@ -32,20 +32,19 @@ class EvenlySpreadOutLocationPreferenceSlotSelectionStrategy
@Nonnull
@Override
protected Optional<SlotInfoAndLocality> selectWithoutLocationPreference(
- @Nonnull Collection<SlotInfoAndResources> availableSlots,
+ @Nonnull Collection<SlotInfoWithUtilization> availableSlots,
@Nonnull ResourceProfile resourceProfile) {
return availableSlots.stream()
.filter(
- slotInfoAndResources ->
- slotInfoAndResources
- .getRemainingResources()
+ slotInfoWithUtilization ->
+ slotInfoWithUtilization
+ .getResourceProfile()
.isMatching(resourceProfile))
- .min(Comparator.comparing(SlotInfoAndResources::getTaskExecutorUtilization))
+ .min(Comparator.comparing(SlotInfoWithUtilization::getTaskExecutorUtilization))
.map(
- slotInfoAndResources ->
+ slotInfoWithUtilization ->
SlotInfoAndLocality.of(
- slotInfoAndResources.getSlotInfo(),
- Locality.UNCONSTRAINED));
+ slotInfoWithUtilization, Locality.UNCONSTRAINED));
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/LocationPreferenceSlotSelectionStrategy.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/LocationPreferenceSlotSelectionStrategy.java
index 903e8345695..27e2c0cde57 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/LocationPreferenceSlotSelectionStrategy.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/LocationPreferenceSlotSelectionStrategy.java
@@ -40,7 +40,7 @@ public abstract class LocationPreferenceSlotSelectionStrategy implements SlotSel
@Override
public Optional<SlotInfoAndLocality> selectBestSlotForProfile(
- @Nonnull Collection<SlotInfoAndResources> availableSlots,
+ @Nonnull Collection<SlotInfoWithUtilization> availableSlots,
@Nonnull SlotProfile slotProfile) {
Collection<TaskManagerLocation> locationPreferences = slotProfile.getPreferredLocations();
@@ -60,7 +60,7 @@ public abstract class LocationPreferenceSlotSelectionStrategy implements SlotSel
@Nonnull
private Optional<SlotInfoAndLocality> selectWithLocationPreference(
- @Nonnull Collection<SlotInfoAndResources> availableSlots,
+ @Nonnull Collection<SlotInfoWithUtilization> availableSlots,
@Nonnull Collection<TaskManagerLocation> locationPreferences,
@Nonnull ResourceProfile resourceProfile) {
@@ -75,25 +75,23 @@ public abstract class LocationPreferenceSlotSelectionStrategy implements SlotSel
preferredFQHostNames.merge(locationPreference.getFQDNHostname(), 1, Integer::sum);
}
- SlotInfoAndResources bestCandidate = null;
+ SlotInfoWithUtilization bestCandidate = null;
Locality bestCandidateLocality = Locality.UNKNOWN;
double bestCandidateScore = Double.NEGATIVE_INFINITY;
- for (SlotInfoAndResources candidate : availableSlots) {
+ for (SlotInfoWithUtilization candidate : availableSlots) {
- if (candidate.getRemainingResources().isMatching(resourceProfile)) {
+ if (candidate.getResourceProfile().isMatching(resourceProfile)) {
// this gets candidate is local-weigh
int localWeigh =
preferredResourceIDs.getOrDefault(
- candidate.getSlotInfo().getTaskManagerLocation().getResourceID(),
- 0);
+ candidate.getTaskManagerLocation().getResourceID(), 0);
// this gets candidate is host-local-weigh
int hostLocalWeigh =
preferredFQHostNames.getOrDefault(
- candidate.getSlotInfo().getTaskManagerLocation().getFQDNHostname(),
- 0);
+ candidate.getTaskManagerLocation().getFQDNHostname(), 0);
double candidateScore =
calculateCandidateScore(
@@ -111,14 +109,13 @@ public abstract class LocationPreferenceSlotSelectionStrategy implements SlotSel
// at the end of the iteration, we return the candidate with best possible locality or null.
return bestCandidate != null
- ? Optional.of(
- SlotInfoAndLocality.of(bestCandidate.getSlotInfo(), bestCandidateLocality))
+ ? Optional.of(SlotInfoAndLocality.of(bestCandidate, bestCandidateLocality))
: Optional.empty();
}
@Nonnull
protected abstract Optional<SlotInfoAndLocality> selectWithoutLocationPreference(
- @Nonnull Collection<SlotInfoAndResources> availableSlots,
+ @Nonnull Collection<SlotInfoWithUtilization> availableSlots,
@Nonnull ResourceProfile resourceProfile);
protected abstract double calculateCandidateScore(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotProviderImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotProviderImpl.java
index 270b8749cf4..507aeef3473 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotProviderImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotProviderImpl.java
@@ -29,7 +29,6 @@ import org.slf4j.LoggerFactory;
import java.util.Collection;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-import java.util.stream.Collectors;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -86,10 +85,7 @@ public class PhysicalSlotProviderImpl implements PhysicalSlotProvider {
private Optional<PhysicalSlot> tryAllocateFromAvailable(
SlotRequestId slotRequestId, SlotProfile slotProfile) {
- Collection<SlotSelectionStrategy.SlotInfoAndResources> slotInfoList =
- slotPool.getAvailableSlotsInformation().stream()
- .map(SlotSelectionStrategy.SlotInfoAndResources::fromSingleSlot)
- .collect(Collectors.toList());
+ Collection<SlotInfoWithUtilization> slotInfoList = slotPool.getAvailableSlotsInformation();
Optional<SlotSelectionStrategy.SlotInfoAndLocality> selectedAvailableSlot =
slotSelectionStrategy.selectBestSlotForProfile(slotInfoList, slotProfile);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/PreviousAllocationSlotSelectionStrategy.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/PreviousAllocationSlotSelectionStrategy.java
index d9f507e651a..66e6b9da4fa 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/PreviousAllocationSlotSelectionStrategy.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/PreviousAllocationSlotSelectionStrategy.java
@@ -51,7 +51,7 @@ public class PreviousAllocationSlotSelectionStrategy implements SlotSelectionStr
@Override
public Optional<SlotInfoAndLocality> selectBestSlotForProfile(
- @Nonnull Collection<SlotInfoAndResources> availableSlots,
+ @Nonnull Collection<SlotInfoWithUtilization> availableSlots,
@Nonnull SlotProfile slotProfile) {
LOG.debug("Select best slot for profile {}.", slotProfile);
@@ -60,35 +60,34 @@ public class PreviousAllocationSlotSelectionStrategy implements SlotSelectionStr
// First, if there was a prior allocation try to schedule to the same/old slot
if (!priorAllocations.isEmpty()) {
- for (SlotInfoAndResources availableSlot : availableSlots) {
- if (priorAllocations.contains(availableSlot.getSlotInfo().getAllocationId())) {
- return Optional.of(
- SlotInfoAndLocality.of(availableSlot.getSlotInfo(), Locality.LOCAL));
+ for (SlotInfoWithUtilization availableSlot : availableSlots) {
+ if (priorAllocations.contains(availableSlot.getAllocationId())) {
+ return Optional.of(SlotInfoAndLocality.of(availableSlot, Locality.LOCAL));
}
}
}
// Second, select based on location preference, excluding blacklisted allocations
Set<AllocationID> blackListedAllocations = slotProfile.getReservedAllocations();
- Collection<SlotInfoAndResources> availableAndAllowedSlots =
+ Collection<SlotInfoWithUtilization> availableAndAllowedSlots =
computeWithoutBlacklistedSlots(availableSlots, blackListedAllocations);
return fallbackSlotSelectionStrategy.selectBestSlotForProfile(
availableAndAllowedSlots, slotProfile);
}
@Nonnull
- private Collection<SlotInfoAndResources> computeWithoutBlacklistedSlots(
- @Nonnull Collection<SlotInfoAndResources> availableSlots,
+ private Collection<SlotInfoWithUtilization> computeWithoutBlacklistedSlots(
+ @Nonnull Collection<SlotInfoWithUtilization> availableSlots,
@Nonnull Set<AllocationID> blacklistedAllocations) {
if (blacklistedAllocations.isEmpty()) {
return Collections.unmodifiableCollection(availableSlots);
}
- ArrayList<SlotInfoAndResources> availableAndAllowedSlots =
+ ArrayList<SlotInfoWithUtilization> availableAndAllowedSlots =
new ArrayList<>(availableSlots.size());
- for (SlotInfoAndResources availableSlot : availableSlots) {
- if (!blacklistedAllocations.contains(availableSlot.getSlotInfo().getAllocationId())) {
+ for (SlotInfoWithUtilization availableSlot : availableSlots) {
+ if (!blacklistedAllocations.contains(availableSlot.getAllocationId())) {
availableAndAllowedSlots.add(availableSlot);
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotSelectionStrategy.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotSelectionStrategy.java
index 59a415b5cfe..36b001016ae 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotSelectionStrategy.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotSelectionStrategy.java
@@ -18,7 +18,6 @@
package org.apache.flink.runtime.jobmaster.slotpool;
-import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.clusterframework.types.SlotProfile;
import org.apache.flink.runtime.jobmanager.scheduler.Locality;
import org.apache.flink.runtime.jobmaster.SlotInfo;
@@ -42,53 +41,9 @@ public interface SlotSelectionStrategy {
* @return the selected slot info with the corresponding locality hint.
*/
Optional<SlotInfoAndLocality> selectBestSlotForProfile(
- @Nonnull Collection<SlotInfoAndResources> availableSlots,
+ @Nonnull Collection<SlotInfoWithUtilization> availableSlots,
@Nonnull SlotProfile slotProfile);
- /**
- * This class is a value type that combines a {@link SlotInfo} with its remaining {@link
- * ResourceProfile}.
- */
- final class SlotInfoAndResources {
-
- @Nonnull private final SlotInfo slotInfo;
-
- @Nonnull private final ResourceProfile remainingResources;
-
- private final double taskExecutorUtilization;
-
- public SlotInfoAndResources(
- @Nonnull SlotInfo slotInfo,
- @Nonnull ResourceProfile remainingResources,
- double taskExecutorUtilization) {
- this.slotInfo = slotInfo;
- this.remainingResources = remainingResources;
- this.taskExecutorUtilization = taskExecutorUtilization;
- }
-
- @Nonnull
- public SlotInfo getSlotInfo() {
- return slotInfo;
- }
-
- @Nonnull
- public ResourceProfile getRemainingResources() {
- return remainingResources;
- }
-
- public double getTaskExecutorUtilization() {
- return taskExecutorUtilization;
- }
-
- public static SlotInfoAndResources fromSingleSlot(
- @Nonnull SlotInfoWithUtilization slotInfoWithUtilization) {
- return new SlotInfoAndResources(
- slotInfoWithUtilization,
- slotInfoWithUtilization.getResourceProfile(),
- slotInfoWithUtilization.getTaskExecutorUtilization());
- }
- }
-
/** This class is a value type that combines a {@link SlotInfo} with a {@link Locality} hint. */
final class SlotInfoAndLocality {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/LocationPreferenceSlotSelectionStrategyTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/LocationPreferenceSlotSelectionStrategyTest.java
index c090b375ab7..bbae2412d44 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/LocationPreferenceSlotSelectionStrategyTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/LocationPreferenceSlotSelectionStrategyTest.java
@@ -177,15 +177,15 @@ class LocationPreferenceSlotSelectionStrategyTest extends SlotSelectionStrategyT
protected static void assertMatchingSlotLocalityAndInCandidates(
Optional<SlotSelectionStrategy.SlotInfoAndLocality> matchingSlot,
Locality locality,
- Set<SlotSelectionStrategy.SlotInfoAndResources> candidates) {
+ Set<SlotInfoWithUtilization> candidates) {
assertThat(matchingSlot)
.hasValueSatisfying(
slotInfoAndLocality -> {
assertThat(slotInfoAndLocality.getLocality()).isEqualTo(locality);
assertThat(candidates)
.anySatisfy(
- slotInfoAndResources ->
- assertThat(slotInfoAndResources.getSlotInfo())
+ slotInfoWithUtilization ->
+ assertThat(slotInfoWithUtilization)
.isEqualTo(
slotInfoAndLocality
.getSlotInfo()));
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/SlotSelectionStrategyTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/SlotSelectionStrategyTestBase.java
index 53a80750881..eaea531e00f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/SlotSelectionStrategyTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/SlotSelectionStrategyTestBase.java
@@ -70,17 +70,17 @@ abstract class SlotSelectionStrategyTestBase {
SlotInfoWithUtilization.from(
new SimpleSlotContext(aid4, tml4, 4, taskManagerGateway, resourceProfile), 0);
- protected final Set<SlotSelectionStrategy.SlotInfoAndResources> candidates =
+ protected final Set<SlotInfoWithUtilization> candidates =
Collections.unmodifiableSet(createCandidates());
protected SlotSelectionStrategy selectionStrategy;
- private Set<SlotSelectionStrategy.SlotInfoAndResources> createCandidates() {
- Set<SlotSelectionStrategy.SlotInfoAndResources> candidates = new HashSet<>(4);
- candidates.add(SlotSelectionStrategy.SlotInfoAndResources.fromSingleSlot(slotInfo1));
- candidates.add(SlotSelectionStrategy.SlotInfoAndResources.fromSingleSlot(slotInfo2));
- candidates.add(SlotSelectionStrategy.SlotInfoAndResources.fromSingleSlot(slotInfo3));
- candidates.add(SlotSelectionStrategy.SlotInfoAndResources.fromSingleSlot(slotInfo4));
+ private Set<SlotInfoWithUtilization> createCandidates() {
+ Set<SlotInfoWithUtilization> candidates = new HashSet<>(4);
+ candidates.add(slotInfo1);
+ candidates.add(slotInfo2);
+ candidates.add(slotInfo3);
+ candidates.add(slotInfo4);
return candidates;
}