This is an automated email from the ASF dual-hosted git repository.
albumenj pushed a commit to branch 3.0
in repository https://gitbox.apache.org/repos/asf/dubbo.git
The following commit(s) were added to refs/heads/3.0 by this push:
new fee0dba Improve consistent hashing load balancing with a new
algorithm (#8948)
fee0dba is described below
commit fee0dbab0c255d79bfba50aaaafd5c7c33220bbb
Author: Lei Jiang <[email protected]>
AuthorDate: Thu Sep 30 10:12:22 2021 +0800
Improve consistent hashing load balancing with a new algorithm (#8948)
---
.../loadbalance/ConsistentHashLoadBalance.java | 95 ++++++++++++++++++++--
.../loadbalance/ConsistentHashLoadBalanceTest.java | 31 +++----
2 files changed, 97 insertions(+), 29 deletions(-)
diff --git
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/ConsistentHashLoadBalance.java
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/ConsistentHashLoadBalance.java
index 15c7066..db2e1ac 100644
---
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/ConsistentHashLoadBalance.java
+++
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/ConsistentHashLoadBalance.java
@@ -27,6 +27,7 @@ import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicLong;
import static
org.apache.dubbo.common.constants.CommonConstants.COMMA_SPLIT_PATTERN;
@@ -35,26 +36,22 @@ import static
org.apache.dubbo.common.constants.CommonConstants.COMMA_SPLIT_PATT
*/
public class ConsistentHashLoadBalance extends AbstractLoadBalance {
public static final String NAME = "consistenthash";
-
/**
* Hash nodes name
*/
public static final String HASH_NODES = "hash.nodes";
-
/**
* Hash arguments name
*/
public static final String HASH_ARGUMENTS = "hash.arguments";
-
private final ConcurrentMap<String, ConsistentHashSelector<?>> selectors =
new ConcurrentHashMap<String, ConsistentHashSelector<?>>();
-
@SuppressWarnings("unchecked")
@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url,
Invocation invocation) {
String methodName = RpcUtils.getMethodName(invocation);
String key = invokers.get(0).getUrl().getServiceKey() + "." +
methodName;
// using the hashcode of list to compute the hash only pay attention
to the elements in the list
- int invokersHashCode = invokers.hashCode();
+ int invokersHashCode = getCorrespondingHashCode(invokers);
ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>)
selectors.get(key);
if (selector == null || selector.identityHashCode != invokersHashCode)
{
selectors.put(key, new ConsistentHashSelector<T>(invokers,
methodName, invokersHashCode));
@@ -63,16 +60,47 @@ public class ConsistentHashLoadBalance extends
AbstractLoadBalance {
return selector.select(invocation);
}
+ /**
+ * get hash code of invokers
+ * Make this method to public in order to use this method in test case
+ * @param invokers
+ * @return
+ */
+ public <T> int getCorrespondingHashCode(List<Invoker<T>> invokers){
+ return invokers.hashCode();
+ }
+
private static final class ConsistentHashSelector<T> {
private final TreeMap<Long, Invoker<T>> virtualInvokers;
-
private final int replicaNumber;
-
private final int identityHashCode;
private final int[] argumentIndex;
+ /**
+ * key: server(invoker) address
+ * value: count of requests accept by certain server
+ */
+ private Map<String, AtomicLong> serverRequestCountMap = new
ConcurrentHashMap<>();
+
+ /**
+ * count of total requests accept by all servers
+ */
+ private AtomicLong totalRequestCount;
+
+ /**
+ * count of current servers(invokers)
+ */
+ private int serverCount;
+
+ /**
+ * the ratio which allow count of requests accept by each server
+ * overrate average (totalRequestCount/serverCount).
+ * 1.5 is recommended, in the future we can make this param
configurable
+ */
+ private static final double OVERLOAD_RATIO_THREAD = 1.5F;
+
ConsistentHashSelector(List<Invoker<T>> invokers, String methodName,
int identityHashCode) {
this.virtualInvokers = new TreeMap<Long, Invoker<T>>();
this.identityHashCode = identityHashCode;
@@ -93,6 +121,10 @@ public class ConsistentHashLoadBalance extends
AbstractLoadBalance {
}
}
}
+
+ totalRequestCount = new AtomicLong(0);
+ serverCount = invokers.size();
+ serverRequestCountMap.clear();
}
public Invoker<T> select(Invocation invocation) {
@@ -100,7 +132,6 @@ public class ConsistentHashLoadBalance extends
AbstractLoadBalance {
byte[] digest = Bytes.getMD5(key);
return selectForKey(hash(digest, 0));
}
-
private String toKey(Object[] args) {
StringBuilder buf = new StringBuilder();
for (int i : argumentIndex) {
@@ -110,15 +141,61 @@ public class ConsistentHashLoadBalance extends
AbstractLoadBalance {
}
return buf.toString();
}
-
private Invoker<T> selectForKey(long hash) {
Map.Entry<Long, Invoker<T>> entry =
virtualInvokers.ceilingEntry(hash);
if (entry == null) {
entry = virtualInvokers.firstEntry();
}
+
+ String serverAddress = entry.getValue().getUrl().getAddress();
+
+ /**
+ * The following part of codes aims to select suitable invoker.
+ * This part is not complete thread safety.
+ * However, in the scene of consumer-side load balance,
+ * thread race for this part of codes
+ * (execution time cost for this part of codes without any IO or
+ * network operation is very low) will rarely occur. And even in
+ * extreme case, a few requests are assigned to an invoker which
+ * is above OVERLOAD_RATIO_THREAD will not make a significant
impact
+ * on the effect of this new algorithm.
+ * And make this part of codes synchronized will reduce efficiency
of
+ * every request. In my opinion, this is not worth. So it is not a
+ * problem for this part is not complete thread safety.
+ */
+ double overloadThread = ((double) totalRequestCount.get() /
(double) serverCount) * OVERLOAD_RATIO_THREAD;
+ /**
+ * Find a valid server node:
+ * 1. Not have accept request yet
+ * or
+ * 2. Not have overloaded (request count already accept < thread
(average request count * overloadRatioAllowed ))
+ */
+ while (serverRequestCountMap.containsKey(serverAddress)
+ && serverRequestCountMap.get(serverAddress).get() >=
overloadThread) {
+ /**
+ * If server node is not valid, get next node
+ */
+ entry = getNextInvokerNode(virtualInvokers, entry);
+ serverAddress = entry.getValue().getUrl().getAddress();
+ }
+ if (!serverRequestCountMap.containsKey(serverAddress)) {
+ serverRequestCountMap.put(serverAddress, new AtomicLong(1));
+ } else {
+ serverRequestCountMap.get(serverAddress).incrementAndGet();
+ }
+ totalRequestCount.incrementAndGet();
+
return entry.getValue();
}
+ private Map.Entry<Long, Invoker<T>> getNextInvokerNode(TreeMap<Long,
Invoker<T>> virtualInvokers, Map.Entry<Long, Invoker<T>> entry){
+ Map.Entry<Long, Invoker<T>> nextEntry =
virtualInvokers.higherEntry(entry.getKey());
+ if(nextEntry == null){
+ return virtualInvokers.firstEntry();
+ }
+ return nextEntry;
+ }
+
private long hash(byte[] digest, int number) {
return (((long) (digest[3 + number * 4] & 0xFF) << 24)
| ((long) (digest[2 + number * 4] & 0xFF) << 16)
diff --git
a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/loadbalance/ConsistentHashLoadBalanceTest.java
b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/loadbalance/ConsistentHashLoadBalanceTest.java
index 56b2c14..cea2856 100644
---
a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/loadbalance/ConsistentHashLoadBalanceTest.java
+++
b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/loadbalance/ConsistentHashLoadBalanceTest.java
@@ -18,12 +18,10 @@ package org.apache.dubbo.rpc.cluster.loadbalance;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.rpc.Invoker;
-import org.apache.dubbo.rpc.cluster.LoadBalance;
import org.apache.dubbo.rpc.cluster.RouterChain;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
@@ -34,39 +32,32 @@ public class ConsistentHashLoadBalanceTest extends
LoadBalanceBaseTest {
@Test
public void testConsistentHashLoadBalance() {
int runs = 10000;
- long unHitedInvokerCount = 0;
- Map<Invoker, Long> hitedInvokers = new HashMap<>();
Map<Invoker, AtomicLong> counter = getInvokeCounter(runs,
ConsistentHashLoadBalance.NAME);
- for (Invoker minvoker : counter.keySet()) {
- Long count = counter.get(minvoker).get();
-
- if (count == 0) {
- unHitedInvokerCount++;
- } else {
- hitedInvokers.put(minvoker, count);
- }
+ double overloadRatioAllowed = 1.5F;
+ int serverCount = counter.size();
+ double overloadThread = ((double) runs *
overloadRatioAllowed)/((double) serverCount);
+ for (Invoker invoker : counter.keySet()) {
+ Long count = counter.get(invoker).get();
+ Assertions.assertTrue(count < (overloadThread + 1L),
+ "count of request accept by each invoker will not be higher
than (overloadRatioAllowed * average + 1)");
}
- Assertions.assertEquals(counter.size() - 1,
- unHitedInvokerCount, "the number of unHitedInvoker should be
counter.size() - 1");
- Assertions.assertEquals(1, hitedInvokers.size(), "the number of
hitedInvoker should be 1");
- Assertions.assertEquals(runs,
- hitedInvokers.values().iterator().next().intValue(), "the
number of hited count should be the number of runs");
}
// https://github.com/apache/dubbo/issues/5429
@Test
void testNormalWhenRouterEnabled() {
- LoadBalance lb = getLoadBalance(ConsistentHashLoadBalance.NAME);
+ ConsistentHashLoadBalance lb = (ConsistentHashLoadBalance)
getLoadBalance(ConsistentHashLoadBalance.NAME);
URL url = invokers.get(0).getUrl();
RouterChain<LoadBalanceBaseTest> routerChain =
RouterChain.buildChain(url);
Invoker<LoadBalanceBaseTest> result = lb.select(invokers, url,
invocation);
+ int originalHashCode = lb.getCorrespondingHashCode(invokers);
for (int i = 0; i < 100; i++) {
routerChain.setInvokers(invokers);
List<Invoker<LoadBalanceBaseTest>> routeInvokers =
routerChain.route(url, invocation);
- Invoker<LoadBalanceBaseTest> finalInvoker =
lb.select(routeInvokers, url, invocation);
- Assertions.assertEquals(result, finalInvoker);
+
+ Assertions.assertEquals(originalHashCode,
lb.getCorrespondingHashCode(routeInvokers));
}
}
}
