Smooth Round Robin selection

Author: jasonjoo2010

dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/loadbalance/RoundRobinLoadBalance.java

@@ -17,13 +17,15 @@
 package com.alibaba.dubbo.rpc.cluster.loadbalance;
 
 import com.alibaba.dubbo.common.URL;
-import com.alibaba.dubbo.common.utils.AtomicPositiveInteger;
 import com.alibaba.dubbo.rpc.Invocation;
 import com.alibaba.dubbo.rpc.Invoker;
 
-import java.util.LinkedHashMap;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
@@ -32,72 +34,131 @@
  *
  */
 public class RoundRobinLoadBalance extends AbstractLoadBalance {
-
     public static final String NAME = "roundrobin";
+    
+    private static int RECYCLE_PERIOD = 60000;
+    
+    protected static class WeightedRoundRobin {
+        private int weight;
+        private int current;
+        private long lastRecycle;
+        private long lastUpdate;
+        public int getWeight() {
+            return weight;
+        }
+        public void setWeight(int weight) {
+            this.weight = weight;
+        }
+        public void setCurrent(int current) {
+            this.current = current;
+        }
+        public int increaseWeight() {
+            current += weight;
+            return current;
+        }
+        public void sel(int total) {
+            current -= total;
+        }
+        public long getLastRecycle() {
+            return lastRecycle;
+        }
+        public void setLastRecycle(long lastRecycle) {
+            this.lastRecycle = lastRecycle;
+        }
+        public long getLastUpdate() {
+            return lastUpdate;
+        }
+        public void setLastUpdate(long lastUpdate) {
+            this.lastUpdate = lastUpdate;
+        }
+    }
 
-    private final ConcurrentMap<String, AtomicPositiveInteger> sequences = new ConcurrentHashMap<String, AtomicPositiveInteger>();
-
+    private ConcurrentMap<String, Map<String, WeightedRoundRobin>> methodWeightMap = new ConcurrentHashMap<String, Map<String, WeightedRoundRobin>>();
+    
+    /**
+     * get invoker addr list cached for specified invocation
+     * <p>
+     * <b>for unit test only</b>
+     * 
+     * @param invokers
+     * @param invocation
+     * @return
+     */
+    protected <T> Collection<String> getInvokerAddrList(List<Invoker<T>> invokers, Invocation invocation) {
+        String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
+        Map<String, WeightedRoundRobin> map = methodWeightMap.get(key);
+        if (map != null) {
+            return map.keySet();
+        }
+        return null;
+    }
+    
     @Override
     protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
         String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
-        int length = invokers.size(); // Number of invokers
-        int maxWeight = 0; // The maximum weight
-        int minWeight = Integer.MAX_VALUE; // The minimum weight
-        final LinkedHashMap<Invoker<T>, IntegerWrapper> invokerToWeightMap = new LinkedHashMap<Invoker<T>, IntegerWrapper>();
-        int weightSum = 0;
-        for (int i = 0; i < length; i++) {
-            int weight = getWeight(invokers.get(i), invocation);
-            maxWeight = Math.max(maxWeight, weight); // Choose the maximum weight
-            minWeight = Math.min(minWeight, weight); // Choose the minimum weight
-            if (weight > 0) {
-                invokerToWeightMap.put(invokers.get(i), new IntegerWrapper(weight));
-                weightSum += weight;
-            }
+        Map<String, WeightedRoundRobin> map = methodWeightMap.get(key);
+        if (map == null) {
+            methodWeightMap.putIfAbsent(key, new HashMap<String, WeightedRoundRobin>());
+            map = methodWeightMap.get(key);
         }
-        AtomicPositiveInteger sequence = sequences.get(key);
-        if (sequence == null) {
-            sequences.putIfAbsent(key, new AtomicPositiveInteger());
-            sequence = sequences.get(key);
-        }
-        int currentSequence = sequence.getAndIncrement();
-        if (maxWeight > 0 && minWeight < maxWeight) {
-            int mod = currentSequence % weightSum;
-            for (int i = 0; i < maxWeight; i++) {
-                for (Map.Entry<Invoker<T>, IntegerWrapper> each : invokerToWeightMap.entrySet()) {
-                    final Invoker<T> k = each.getKey();
-                    final IntegerWrapper v = each.getValue();
-                    if (mod == 0 && v.getValue() > 0) {
-                        return k;
-                    }
-                    if (v.getValue() > 0) {
-                        v.decrement();
-                        mod--;
+        synchronized (map) {
+            int totalWeight = 0;
+            int maxCurrent = Integer.MIN_VALUE;
+            long now = System.currentTimeMillis();
+            boolean needRecycle = false;
+            Invoker<T> selectedInvoker = null;
+            WeightedRoundRobin selectedWRR = null;
+            for (Invoker<T> invoker : invokers) {
+                String identifyString = invoker.getUrl().toIdentityString();
+                WeightedRoundRobin weightedRoundRobin = map.get(identifyString);
+                int weight = getWeight(invoker, invocation);
+                if (weight < 0) {
+                    weight = 0;
+                }
+                if (weightedRoundRobin == null) {
+                    weightedRoundRobin = new WeightedRoundRobin();
+                    weightedRoundRobin.setCurrent(0);
+                    weightedRoundRobin.setWeight(weight);
+                    weightedRoundRobin.setLastRecycle(now);
+                    map.put(identifyString, weightedRoundRobin);
+                }
+                if (weight != weightedRoundRobin.getWeight()) {
+                    //weight changed
+                    weightedRoundRobin.setCurrent(0);
+                    weightedRoundRobin.setWeight(weight);
+                }
+                int cur = weightedRoundRobin.increaseWeight();
+                weightedRoundRobin.setLastUpdate(now);
+                if (!needRecycle && now - weightedRoundRobin.getLastRecycle() > RECYCLE_PERIOD) {
+                    //try to recycle useless item every 60s if there is one outdated
+                    needRecycle = true;
+                }
+                if (cur > maxCurrent) {
+                    maxCurrent = cur;
+                    selectedInvoker = invoker;
+                    selectedWRR = weightedRoundRobin;
+                }
+                totalWeight += weight;
+            }
+            //recycle
+            if (needRecycle) {
+                Iterator<Entry<String, WeightedRoundRobin>> it = map.entrySet().iterator();
+                while (it.hasNext()) {
+                    Entry<String, WeightedRoundRobin> item = it.next();
+                    if (now - item.getValue().getLastUpdate() > RECYCLE_PERIOD) {
+                        it.remove();
+                    } else {
+                        //pass
+                        item.getValue().setLastRecycle(now);
                     }
                 }
             }
+            if (selectedInvoker != null) {
+                selectedWRR.sel(totalWeight);
+                return selectedInvoker;
+            }
         }
-        // Round robin
-        return invokers.get(currentSequence % length);
-    }
-
-    private static final class IntegerWrapper {
-        private int value;
-
-        public IntegerWrapper(int value) {
-            this.value = value;
-        }
-
-        public int getValue() {
-            return value;
-        }
-
-        public void setValue(int value) {
-            this.value = value;
-        }
-
-        public void decrement() {
-            this.value--;
-        }
+        return null;
     }
 
 }