Skip to content

Commit

Permalink
xds: replace random with atomic sequence in WRR (#10458)
Browse files Browse the repository at this point in the history
  • Loading branch information
@tonyjongyoonan
tonyjongyoonan authored Aug 7, 2023
1 parent 4049f89 commit 40bff67
Show file tree
Hide file tree
Showing 2 changed files with 39 additions and 24 deletions.
12 changes: 5 additions & 7 deletions xds/src/main/java/io/grpc/xds/WeightedRoundRobinLoadBalancer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ final class WeightedRoundRobinLoadBalancer extends RoundRobinLoadBalancer {
private final ScheduledExecutorService timeService;
private ScheduledHandle weightUpdateTimer;
private final Runnable updateWeightTask;
private final Random random;
private final AtomicInteger sequence;
private final long infTime;
private final Ticker ticker;

Expand All @@ -81,7 +81,7 @@ public WeightedRoundRobinLoadBalancer(WrrHelper helper, Ticker ticker, Random ra
this.syncContext = checkNotNull(helper.getSynchronizationContext(), "syncContext");
this.timeService = checkNotNull(helper.getScheduledExecutorService(), "timeService");
this.updateWeightTask = new UpdateWeightTask();
this.random = random;
this.sequence = new AtomicInteger(random.nextInt());
log.log(Level.FINE, "weighted_round_robin LB created");
}

Expand Down Expand Up @@ -294,9 +294,7 @@ private void updateWeight() {
double newWeight = subchannel.getWeight();
newWeights[i] = newWeight > 0 ? (float) newWeight : 0.0f;
}

StaticStrideScheduler scheduler = new StaticStrideScheduler(newWeights, random);
this.scheduler = scheduler;
this.scheduler = new StaticStrideScheduler(newWeights, sequence);
}

@Override
Expand Down Expand Up @@ -353,7 +351,7 @@ static final class StaticStrideScheduler {
private final AtomicInteger sequence;
private static final int K_MAX_WEIGHT = 0xFFFF;

StaticStrideScheduler(float[] weights, Random random) {
StaticStrideScheduler(float[] weights, AtomicInteger sequence) {
checkArgument(weights.length >= 1, "Couldn't build scheduler: requires at least one weight");
int numChannels = weights.length;
int numWeightedChannels = 0;
Expand Down Expand Up @@ -386,7 +384,7 @@ static final class StaticStrideScheduler {
}

this.scaledWeights = scaledWeights;
this.sequence = new AtomicInteger(random.nextInt());
this.sequence = sequence;

}

Expand Down
51 changes: 34 additions & 17 deletions xds/src/test/java/io/grpc/xds/WeightedRoundRobinLoadBalancerTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -836,15 +836,17 @@ public void wrrConfig_BooleanValueNonNull() {
public void emptyWeights() {
float[] weights = {};
Random random = new Random();
StaticStrideScheduler sss = new StaticStrideScheduler(weights, random);
StaticStrideScheduler sss = new StaticStrideScheduler(weights,
new AtomicInteger(random.nextInt()));
sss.pick();
}

@Test
public void testPicksEqualsWeights() {
float[] weights = {1.0f, 2.0f, 3.0f};
Random random = new Random();
StaticStrideScheduler sss = new StaticStrideScheduler(weights, random);
StaticStrideScheduler sss = new StaticStrideScheduler(weights,
new AtomicInteger(random.nextInt()));
int[] expectedPicks = new int[] {1, 2, 3};
int[] picks = new int[3];
for (int i = 0; i < 6; i++) {
Expand All @@ -857,7 +859,8 @@ public void testPicksEqualsWeights() {
public void testContainsZeroWeightUseMean() {
float[] weights = {3.0f, 0.0f, 1.0f};
Random random = new Random();
StaticStrideScheduler sss = new StaticStrideScheduler(weights, random);
StaticStrideScheduler sss = new StaticStrideScheduler(weights,
new AtomicInteger(random.nextInt()));
int[] expectedPicks = new int[] {3, 2, 1};
int[] picks = new int[3];
for (int i = 0; i < 6; i++) {
Expand All @@ -870,7 +873,8 @@ public void testContainsZeroWeightUseMean() {
public void testContainsNegativeWeightUseMean() {
float[] weights = {3.0f, -1.0f, 1.0f};
Random random = new Random();
StaticStrideScheduler sss = new StaticStrideScheduler(weights, random);
StaticStrideScheduler sss = new StaticStrideScheduler(weights,
new AtomicInteger(random.nextInt()));
int[] expectedPicks = new int[] {3, 2, 1};
int[] picks = new int[3];
for (int i = 0; i < 6; i++) {
Expand All @@ -883,7 +887,8 @@ public void testContainsNegativeWeightUseMean() {
public void testAllSameWeights() {
float[] weights = {1.0f, 1.0f, 1.0f};
Random random = new Random();
StaticStrideScheduler sss = new StaticStrideScheduler(weights, random);
StaticStrideScheduler sss = new StaticStrideScheduler(weights,
new AtomicInteger(random.nextInt()));
int[] expectedPicks = new int[] {2, 2, 2};
int[] picks = new int[3];
for (int i = 0; i < 6; i++) {
Expand All @@ -896,7 +901,8 @@ public void testAllSameWeights() {
public void testAllZeroWeightsUseOne() {
float[] weights = {0.0f, 0.0f, 0.0f};
Random random = new Random();
StaticStrideScheduler sss = new StaticStrideScheduler(weights, random);
StaticStrideScheduler sss = new StaticStrideScheduler(weights,
new AtomicInteger(random.nextInt()));
int[] expectedPicks = new int[] {2, 2, 2};
int[] picks = new int[3];
for (int i = 0; i < 6; i++) {
Expand All @@ -909,7 +915,8 @@ public void testAllZeroWeightsUseOne() {
public void testAllInvalidWeightsUseOne() {
float[] weights = {-3.1f, -0.0f, 0.0f};
Random random = new Random();
StaticStrideScheduler sss = new StaticStrideScheduler(weights, random);
StaticStrideScheduler sss = new StaticStrideScheduler(weights,
new AtomicInteger(random.nextInt()));
int[] expectedPicks = new int[] {2, 2, 2};
int[] picks = new int[3];
for (int i = 0; i < 6; i++) {
Expand All @@ -923,7 +930,8 @@ public void testLargestWeightIndexPickedEveryGeneration() {
float[] weights = {1.0f, 2.0f, 3.0f};
int largestWeightIndex = 2;
Random random = new Random();
StaticStrideScheduler sss = new StaticStrideScheduler(weights, random);
StaticStrideScheduler sss = new StaticStrideScheduler(weights,
new AtomicInteger(random.nextInt()));
int largestWeightPickCount = 0;
int kMaxWeight = 65535;
for (int i = 0; i < largestWeightIndex * kMaxWeight; i++) {
Expand All @@ -938,7 +946,8 @@ public void testLargestWeightIndexPickedEveryGeneration() {
public void testStaticStrideSchedulerNonIntegers1() {
float[] weights = {2.0f, (float) (10.0 / 3.0), 1.0f};
Random random = new Random();
StaticStrideScheduler sss = new StaticStrideScheduler(weights, random);
StaticStrideScheduler sss = new StaticStrideScheduler(weights,
new AtomicInteger(random.nextInt()));
double totalWeight = 2 + 10.0 / 3.0 + 1.0;
Map<Integer, Integer> pickCount = new HashMap<>();
for (int i = 0; i < 1000; i++) {
Expand All @@ -955,7 +964,8 @@ public void testStaticStrideSchedulerNonIntegers1() {
public void testStaticStrideSchedulerNonIntegers2() {
float[] weights = {0.5f, 0.3f, 1.0f};
Random random = new Random();
StaticStrideScheduler sss = new StaticStrideScheduler(weights, random);
StaticStrideScheduler sss = new StaticStrideScheduler(weights,
new AtomicInteger(random.nextInt()));
double totalWeight = 1.8;
Map<Integer, Integer> pickCount = new HashMap<>();
for (int i = 0; i < 1000; i++) {
Expand All @@ -972,7 +982,8 @@ public void testStaticStrideSchedulerNonIntegers2() {
public void testTwoWeights() {
float[] weights = {1.0f, 2.0f};
Random random = new Random();
StaticStrideScheduler sss = new StaticStrideScheduler(weights, random);
StaticStrideScheduler sss = new StaticStrideScheduler(weights,
new AtomicInteger(random.nextInt()));
double totalWeight = 3;
Map<Integer, Integer> pickCount = new HashMap<>();
for (int i = 0; i < 1000; i++) {
Expand All @@ -989,7 +1000,8 @@ public void testTwoWeights() {
public void testManyWeights() {
float[] weights = {1.0f, 2.0f, 3.0f, 4.0f, 5.0f};
Random random = new Random();
StaticStrideScheduler sss = new StaticStrideScheduler(weights, random);
StaticStrideScheduler sss = new StaticStrideScheduler(weights,
new AtomicInteger(random.nextInt()));
double totalWeight = 15;
Map<Integer, Integer> pickCount = new HashMap<>();
for (int i = 0; i < 1000; i++) {
Expand All @@ -1006,7 +1018,8 @@ public void testManyWeights() {
public void testManyComplexWeights() {
float[] weights = {1.2f, 2.4f, 222.56f, 1.1f, 15.0f, 226342.0f, 5123.0f, 532.2f};
Random random = new Random();
StaticStrideScheduler sss = new StaticStrideScheduler(weights, random);
StaticStrideScheduler sss = new StaticStrideScheduler(weights,
new AtomicInteger(random.nextInt()));
double totalWeight = 1.2 + 2.4 + 222.56 + 15.0 + 226342.0 + 5123.0 + 0.0001;
Map<Integer, Integer> pickCount = new HashMap<>();
for (int i = 0; i < 1000; i++) {
Expand All @@ -1023,7 +1036,8 @@ public void testManyComplexWeights() {
public void testDeterministicPicks() {
float[] weights = {2.0f, 3.0f, 6.0f};
Random random = new FakeRandom(0);
StaticStrideScheduler sss = new StaticStrideScheduler(weights, random);
StaticStrideScheduler sss = new StaticStrideScheduler(weights,
new AtomicInteger(random.nextInt()));
assertThat(sss.getSequence()).isEqualTo(0);
assertThat(sss.pick()).isEqualTo(1);
assertThat(sss.getSequence()).isEqualTo(2);
Expand All @@ -1043,7 +1057,8 @@ public void testDeterministicPicks() {
public void testImmediateWraparound() {
float[] weights = {1.0f, 2.0f, 3.0f, 4.0f, 5.0f};
Random random = new FakeRandom(-1);
StaticStrideScheduler sss = new StaticStrideScheduler(weights, random);
StaticStrideScheduler sss = new StaticStrideScheduler(weights,
new AtomicInteger(random.nextInt()));
double totalWeight = 15;
Map<Integer, Integer> pickCount = new HashMap<>();
for (int i = 0; i < 1000; i++) {
Expand All @@ -1060,7 +1075,8 @@ public void testImmediateWraparound() {
public void testWraparound() {
float[] weights = {1.0f, 2.0f, 3.0f, 4.0f, 5.0f};
Random random = new FakeRandom(-500);
StaticStrideScheduler sss = new StaticStrideScheduler(weights, random);
StaticStrideScheduler sss = new StaticStrideScheduler(weights,
new AtomicInteger(random.nextInt()));
double totalWeight = 15;
Map<Integer, Integer> pickCount = new HashMap<>();
for (int i = 0; i < 1000; i++) {
Expand All @@ -1077,7 +1093,8 @@ public void testWraparound() {
public void testDeterministicWraparound() {
float[] weights = {2.0f, 3.0f, 6.0f};
Random random = new FakeRandom(-1);
StaticStrideScheduler sss = new StaticStrideScheduler(weights, random);
StaticStrideScheduler sss = new StaticStrideScheduler(weights,
new AtomicInteger(random.nextInt()));
assertThat(sss.getSequence()).isEqualTo(0xFFFF_FFFFL);
assertThat(sss.pick()).isEqualTo(1);
assertThat(sss.getSequence()).isEqualTo(2);
Expand Down

0 comments on commit 40bff67

Please sign in to comment.