Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Fix excessive rebalance in LeastShardAllocationStrategy #3191

Merged
merged 2 commits into from
Nov 13, 2017
Merged

Fix excessive rebalance in LeastShardAllocationStrategy #3191

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2738e2e
Fix excessive rebalance in LeastShardAllocationStrategy
zbynek001 Nov 12, 2017
731c906
Merge branch 'dev' into excessive-rebalance-fix
Aaronontheweb Nov 13, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions src/contrib/cluster/Akka.Cluster.Sharding.Tests/LeastShardAllocationStrategySpec.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,5 +105,22 @@ public void LeastShardAllocationStrategy_should_limit_number_of_simultaneous_reb
var r2 = _allocationStrategy.Rebalance(allocations, ImmutableHashSet.Create("shard2", "shard3")).Result;
r2.Count.Should().Be(0);
}

[Fact]
public void LeastShardAllocationStrategy_dont_rebalance_excessive_shards_if_maxSimultaneousRebalance_gt_rebalanceThreshold()
{
var allocationStrategy = new LeastShardAllocationStrategy(2, 5);
var allocations = new Dictionary<IActorRef, IImmutableList<string>>
{
{_regionA, new []{"shard1", "shard2", "shard3", "shard4", "shard5", "shard6", "shard7", "shard8"}.ToImmutableList() },
{_regionB, new []{"shard9", "shard10", "shard11", "shard12" }.ToImmutableList() }
}.ToImmutableDictionary();

var r1 = allocationStrategy.Rebalance(allocations, ImmutableHashSet.Create("shard2")).Result;
r1.Should().BeEquivalentTo(new[] { "shard1", "shard3", "shard4" });

var r2 = _allocationStrategy.Rebalance(allocations, ImmutableHashSet.Create("shard5", "shard6", "shard7", "shard8")).Result;
r2.Count.Should().Be(0);
}
}
}
11 changes: 6 additions & 5 deletions src/contrib/cluster/Akka.Cluster.Sharding/ShardAllocationStrategy.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ public interface IShardAllocationStrategy : INoSerializationVerificationNeeded
/// Invoked when the location of a new shard is to be decided.
/// </summary>
/// <param name="requester">
/// Actor reference to the <see cref="ShardRegion"/> that requested the location of the shard, can be returned
/// Actor reference to the <see cref="ShardRegion"/> that requested the location of the shard, can be returned
/// if preference should be given to the node where the shard was first accessed.
/// </param>
/// <param name="shardId">The id of the shard to allocate.</param>
Expand All @@ -52,8 +52,8 @@ public interface IShardAllocationStrategy : INoSerializationVerificationNeeded
}

/// <summary>
/// The default implementation of <see cref="Akka.Cluster.Sharding.LeastShardAllocationStrategy"/> allocates new shards
/// to the <see cref="ShardRegion"/> with least number of previously allocated shards. It picks shards
/// The default implementation of <see cref="Akka.Cluster.Sharding.LeastShardAllocationStrategy"/> allocates new shards
/// to the <see cref="ShardRegion"/> with least number of previously allocated shards. It picks shards
/// for rebalancing handoff from the <see cref="ShardRegion"/> with most number of previously allocated shards.
/// They will then be allocated to the <see cref="ShardRegion"/> with least number of previously allocated shards,
/// i.e. new members in the cluster. There is a configurable threshold of how large the difference
Expand Down Expand Up @@ -104,9 +104,10 @@ public Task<IImmutableSet<ShardId>> Rebalance(IImmutableDictionary<IActorRef, II
currentShardAllocations.Select(kv => kv.Value.Where(s => !rebalanceInProgress.Contains(s)).ToArray());
var mostShards = GetMaxBy(shards, x => x.Length);

if (mostShards.Length - leastShardsRegion.Value.Count >= _rebalanceThreshold)
var difference = mostShards.Length - leastShardsRegion.Value.Count;
if (difference >= _rebalanceThreshold)
{
return Task.FromResult<IImmutableSet<ShardId>>(mostShards.Take(_maxSimultaneousRebalance - rebalanceInProgress.Count).ToImmutableHashSet());
return Task.FromResult<IImmutableSet<ShardId>>(mostShards.Take(Math.Min(difference, _maxSimultaneousRebalance - rebalanceInProgress.Count)).ToImmutableHashSet());
}
}

Expand Down