You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by ab...@apache.org on 2020/03/11 08:22:08 UTC
[lucene-solr] branch jira/solr-14275 updated: SOLR-14275: Minor
improvements. Implement similar cost-saving approximation in
MoveReplicaSuggester as in AddReplicaSuggester.
This is an automated email from the ASF dual-hosted git repository.
ab pushed a commit to branch jira/solr-14275
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
The following commit(s) were added to refs/heads/jira/solr-14275 by this push:
new 542353a SOLR-14275: Minor improvements. Implement similar cost-saving approximation in MoveReplicaSuggester as in AddReplicaSuggester.
542353a is described below
commit 542353a488881190b88deb9e2ac4050e84977bc2
Author: Andrzej Bialecki <ab...@apache.org>
AuthorDate: Wed Mar 11 09:21:13 2020 +0100
SOLR-14275: Minor improvements. Implement similar cost-saving approximation in
MoveReplicaSuggester as in AddReplicaSuggester.
---
.../cloud/autoscaling/sim/SimCloudManager.java | 2 +-
.../cloud/autoscaling/AddReplicaSuggester.java | 6 ++
.../client/solrj/cloud/autoscaling/Clause.java | 2 -
.../solrj/cloud/autoscaling/CoresVariable.java | 3 +-
.../cloud/autoscaling/MoveReplicaSuggester.java | 8 +-
.../client/solrj/cloud/autoscaling/Policy.java | 32 ++++++--
.../solrj/cloud/autoscaling/ReplicaVariable.java | 18 +++--
.../solr/client/solrj/cloud/autoscaling/Row.java | 87 +++++++++++-----------
.../client/solrj/cloud/autoscaling/Suggester.java | 2 +-
.../client/solrj/cloud/autoscaling/TestPolicy.java | 2 +-
10 files changed, 96 insertions(+), 66 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimCloudManager.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimCloudManager.java
index 87dee15..fffacb1 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimCloudManager.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimCloudManager.java
@@ -729,7 +729,7 @@ public class SimCloudManager implements SolrCloudManager {
// by the testcase. we're going to immediately catch & re-throw any exceptions, so we don't
// need/want to be wrapped in a LoggingCallable w/getBackgroundTaskFailureCount() tracking
Future<SolrResponse> rsp = simCloudManagerPool.submit(() -> simHandleSolrRequest(req));
- return rsp.get(6000, TimeUnit.SECONDS); // longer then this and something is seriously wrong
+ return rsp.get(6000, TimeUnit.SECONDS); // longer than this and something is seriously wrong
} catch (Exception e) {
throw new IOException(e);
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/AddReplicaSuggester.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/AddReplicaSuggester.java
index 264f86b..09ce72f 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/AddReplicaSuggester.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/AddReplicaSuggester.java
@@ -17,6 +17,7 @@
package org.apache.solr.client.solrj.cloud.autoscaling;
+import java.lang.invoke.MethodHandles;
import java.util.Collections;
import java.util.List;
import java.util.Set;
@@ -26,10 +27,14 @@ import org.apache.solr.client.solrj.request.CollectionAdminRequest;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.params.CollectionParams;
import org.apache.solr.common.util.Pair;
+import org.apache.solr.common.util.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import static org.apache.solr.common.params.CollectionParams.CollectionAction.ADDREPLICA;
class AddReplicaSuggester extends Suggester {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
SolrRequest init() {
SolrRequest operation = tryEachNode(true);
@@ -71,6 +76,7 @@ class AddReplicaSuggester extends Suggester {
if (bestNode != null) {// there are no rule violations
this.session = bestNode.session;
+ //log.info("--- Row cache stats: " + Utils.toJSONString(Row.cacheStats));
return CollectionAdminRequest
.addReplicaToShard(shard.first(), shard.second())
.setType(type)
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Clause.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Clause.java
index 788088b..cd98dd7 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Clause.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Clause.java
@@ -600,8 +600,6 @@ public class Clause implements MapWriter, Comparable<Clause> {
});
if (rc != null)
replicaCount.increment(rc);
-
-
}
List<Violation> testPerNode(Policy.Session session, double[] deviations) {
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/CoresVariable.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/CoresVariable.java
index a468385..788d842 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/CoresVariable.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/CoresVariable.java
@@ -84,13 +84,12 @@ public class CoresVariable extends VariableBase {
}
}
- static final String TOTALCORES = CoresVariable.class.getSimpleName() + ".totalcores";
private int getTotalCores(Policy.Session session, AtomicInteger liveNodes) {
int coresCount = 0;
for (Row row : session.matrix) {
if (!row.isLive) continue;
liveNodes.incrementAndGet();
- Integer res = row.computeCacheIfAbsent(TOTALCORES, o -> {
+ Integer res = row.computeCacheIfAbsent(Row.GlobalCacheEntryKey.TOTALCORES, o -> {
int[] result = new int[1];
row.forEachReplica(replicaInfo -> result[0]++);
return result[0];
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/MoveReplicaSuggester.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/MoveReplicaSuggester.java
index 45f18b1..a82c564 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/MoveReplicaSuggester.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/MoveReplicaSuggester.java
@@ -54,7 +54,13 @@ public class MoveReplicaSuggester extends Suggester {
Row fromRow = fromReplica.second();
ReplicaInfo ri = fromReplica.first();
if (ri == null) continue;
- final int i = session.indexOf(fromRow.node);
+ int i = session.indexOf(fromRow.node);
+ // nocommit
+ // possible optimization? for large clusters compare only up to
+ // that many eligible rows from the top
+ if (session.matrix.size() > 100 && i < session.matrix.size() - 10) {
+ i = session.matrix.size() - 10;
+ }
int stopAt = force ? 0 : i;
Row targetRow = null;
for (int j = session.matrix.size() - 1; j >= stopAt; j--) {
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Policy.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Policy.java
index 9018d10..0f75e6d 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Policy.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Policy.java
@@ -281,7 +281,7 @@ public class Policy implements MapWriter {
List<Row> matrixCopy = new ArrayList<>(matrix);
List<Row> deadNodes = null;
Iterator<Row> it =matrix.iterator();
- while (it.hasNext()){
+ while (it.hasNext()) {
Row row = it.next();
if(!row.isLive){
if(deadNodes == null) deadNodes = new ArrayList<>();
@@ -328,7 +328,7 @@ public class Policy implements MapWriter {
return result;
});
- if(deadNodes != null){
+ if (deadNodes != null) {
for (Row deadNode : deadNodes) {
matrix.add(0, deadNode);
}
@@ -514,6 +514,7 @@ public class Policy implements MapWriter {
List<Clause> expandedClauses;
List<Violation> violations = new ArrayList<>();
Transaction transaction;
+ Map<String, Row> nodeToRow = new HashMap<>();
private Session(List<String> nodes, SolrCloudManager cloudManager,
List<Row> matrix, List<Clause> expandedClauses, int znodeVersion,
@@ -525,7 +526,10 @@ public class Policy implements MapWriter {
this.expandedClauses = expandedClauses;
this.znodeVersion = znodeVersion;
this.nodeStateProvider = nodeStateProvider;
- for (Row row : matrix) row.session = this;
+ for (Row row : matrix) {
+ row.session = this;
+ nodeToRow.put(row.node, row);
+ }
}
@@ -573,7 +577,11 @@ public class Policy implements MapWriter {
Collections.sort(expandedClauses);
matrix = new ArrayList<>(nodes.size());
- for (String node : nodes) matrix.add(new Row(node, params, perReplicaAttributes, this));
+ for (String node : nodes) {
+ Row row = new Row(node, params, perReplicaAttributes, this);
+ matrix.add(row);
+ nodeToRow.put(node, row);
+ }
applyRules();
}
@@ -597,11 +605,21 @@ public class Policy implements MapWriter {
}
public Row getNode(String node) {
- for (Row row : matrix) if (row.node.equals(node)) return row;
- return null;
+ Row row = nodeToRow.get(node);
+ if (row == null) { // update lazily our cache
+ for (Row r : matrix) {
+ if (r.node.equals(node)) {
+ nodeToRow.put(node, r);
+ return r;
+ }
+ }
+ return null;
+ } else {
+ return row;
+ }
}
- List<Row> getMatrixCopy() {
+ private List<Row> getMatrixCopy() {
return matrix.stream()
.map(row -> row.copy(this))
.collect(Collectors.toList());
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ReplicaVariable.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ReplicaVariable.java
index cff1f73..0ef45ef 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ReplicaVariable.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ReplicaVariable.java
@@ -21,6 +21,7 @@ import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import java.util.function.Consumer;
import org.apache.solr.common.util.StrUtils;
@@ -33,9 +34,10 @@ class ReplicaVariable extends VariableBase {
public static final String REPLICASCOUNT = "relevantReplicas";
- private static final class ReplicaCounter {
+ private static final class ReplicaCounter implements Consumer<ReplicaInfo> {
final String collection, shard;
final Clause clause;
+ final int[] result = new int[1];
ReplicaCounter(String collection, String shard, Clause clause) {
this.collection = collection;
@@ -43,12 +45,16 @@ class ReplicaVariable extends VariableBase {
this.clause = clause;
}
+ @Override
+ public void accept(ReplicaInfo replicaInfo) {
+ if (clause.isMatch(replicaInfo, collection, shard)) {
+ result[0]++;
+ }
+ }
+
int calculate(Row row) {
- int[] result = new int[1];
- row.forEachReplica(collection, replicaInfo -> {
- if (clause.isMatch(replicaInfo, collection, shard))
- result[0]++;
- });
+ result[0] = 0;
+ row.forEachReplica(collection, this);
return result[0];
}
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Row.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Row.java
index ce186cd..a49f6da 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Row.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Row.java
@@ -21,14 +21,11 @@ import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
-import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Random;
-import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
@@ -59,8 +56,13 @@ public class Row implements MapWriter {
boolean anyValueMissing = false;
boolean isLive = true;
Policy.Session session;
- Map globalCache;
- Map perCollCache;
+ Object[] globalCache;
+ Map<String, Map> perCollCache;
+
+ public enum GlobalCacheEntryKey {
+ TOTALCORES,
+ FREEDISK
+ }
public Row(String node, List<Pair<String, Variable.Type>> params, List<String> perReplicaAttributes, Policy.Session session) {
this.session = session;
@@ -77,12 +79,11 @@ public class Row implements MapWriter {
if (NODE.equals(pair.first())) cells[i].val = node;
if (cells[i].val == null) anyValueMissing = true;
}
- this.globalCache = new HashMap();
- this.perCollCache = new HashMap();
+ this.globalCache = new Object[GlobalCacheEntryKey.values().length];
+ this.perCollCache = new HashMap<>();
-/*
// pre-compute and cache common variables
- computeCacheIfAbsent(CoresVariable.TOTALCORES, o -> {
+ computeCacheIfAbsent(GlobalCacheEntryKey.TOTALCORES, o -> {
int[] result = new int[1];
Row.this.forEachReplica(replicaInfo -> result[0]++);
return result[0];
@@ -94,8 +95,6 @@ public class Row implements MapWriter {
o -> getter.getIndexSize(Row.this));
});
});
-
- */
isAlreadyCopied = true;
}
@@ -112,7 +111,7 @@ public class Row implements MapWriter {
}
public static boolean hit(String cacheName) {
-// getCacheEntry(cacheName).hits.incrementAndGet();
+ getCacheEntry(cacheName).hits.incrementAndGet();
return true;
}
@@ -138,14 +137,14 @@ public class Row implements MapWriter {
}
- public <R> R computeCacheIfAbsent(String cacheName, Function<Object, R> supplier) {
- R result = (R) globalCache.get(cacheName);
+ public <R> R computeCacheIfAbsent(GlobalCacheEntryKey key, Function<Object, R> supplier) {
+ R result = (R) globalCache[key.ordinal()];
if (result != null) {
- assert CacheEntry.hit(cacheName);
+ //assert CacheEntry.hit(key.toString());
return result;
} else {
- assert CacheEntry.miss(cacheName);
- globalCache.put(cacheName, result = supplier.apply(cacheName));
+ //assert CacheEntry.miss(key.toString());
+ globalCache[key.ordinal()] = result = supplier.apply(key);
return result;
}
}
@@ -159,11 +158,11 @@ public class Row implements MapWriter {
if (cacheNameMap == null) shardMap.put(cacheName, cacheNameMap = new HashMap());
R result = (R) cacheNameMap.get(key);
if (result == null) {
- CacheEntry.miss(cacheName);
+ //CacheEntry.miss(cacheName);
cacheNameMap.put(key, result = supplier.apply(key));
return result;
} else {
- CacheEntry.hit(cacheName);
+ //CacheEntry.hit(cacheName);
return result;
}
}
@@ -171,7 +170,7 @@ public class Row implements MapWriter {
public Row(String node, Cell[] cells, boolean anyValueMissing, Map<String,
- Map<String, List<ReplicaInfo>>> collectionVsShardVsReplicas, boolean isLive, Policy.Session session, Map perRowCache, Map globalCache) {
+ Map<String, List<ReplicaInfo>>> collectionVsShardVsReplicas, boolean isLive, Policy.Session session, Map perRowCache, Object[] globalCache) {
this.session = session;
this.node = node;
this.isLive = isLive;
@@ -182,8 +181,8 @@ public class Row implements MapWriter {
}
this.anyValueMissing = anyValueMissing;
this.collectionVsShardVsReplicas = collectionVsShardVsReplicas;
- this.perCollCache = perRowCache;
- this.globalCache = globalCache;
+ this.perCollCache = perRowCache != null ? perRowCache : new HashMap<>();
+ this.globalCache = globalCache != null ? globalCache : new Object[GlobalCacheEntryKey.values().length];
}
@Override
@@ -195,7 +194,7 @@ public class Row implements MapWriter {
}
Row copy(Policy.Session session) {
- return new Row(node, cells, anyValueMissing, collectionVsShardVsReplicas, isLive, session, this.globalCache, this.perCollCache);
+ return new Row(node, cells, anyValueMissing, collectionVsShardVsReplicas, isLive, session, this.perCollCache, this.globalCache);
}
Object getVal(String name) {
@@ -242,14 +241,14 @@ public class Row implements MapWriter {
return this;
}
lazyCopyReplicas(coll, shard);
- List<OperationInfo> furtherOps = new LinkedList<>();
+ List<OperationInfo> furtherOps = new ArrayList<>(3);
Consumer<OperationInfo> opCollector = it -> furtherOps.add(it);
Row row = null;
row = session.copy().getNode(this.node);
if (row == null) throw new RuntimeException("couldn't get a row");
row.lazyCopyReplicas(coll, shard);
- Map<String, List<ReplicaInfo>> c = row.collectionVsShardVsReplicas.computeIfAbsent(coll, k -> new HashMap<>());
- List<ReplicaInfo> replicas = c.computeIfAbsent(shard, k -> new ArrayList<>());
+ Map<String, List<ReplicaInfo>> c = row.collectionVsShardVsReplicas.computeIfAbsent(coll, Utils.NEW_HASHMAP_FUN);
+ List<ReplicaInfo> replicas = c.computeIfAbsent(shard, Utils.NEW_ARRAYLIST_FUN);
String replicaname = "SYNTHETIC." + new Random().nextInt(1000) + 1000;
ReplicaInfo ri = new ReplicaInfo(replicaname, replicaname, coll, shard, type, this.node,
Utils.makeMap(ZkStateReader.REPLICA_TYPE, type != null ? type.toString() : Replica.Type.NRT.toString()));
@@ -271,10 +270,11 @@ public class Row implements MapWriter {
boolean isAlreadyCopied = false;
private void lazyCopyReplicas(String coll, String shard) {
- globalCache = new HashMap();
- Map cacheCopy = new HashMap<>(perCollCache);
+ //log.info("--- drop globalCache " + node);
+ globalCache = new Object[GlobalCacheEntryKey.values().length];
+ Map<String, Map> cacheCopy = new HashMap<>(perCollCache);
// nocommit: per-shard removal has unexpected side-effects, TestPolicy.testFreeDiskSuggestions fails
- cacheCopy.remove(coll);//todo optimize at shard level later
+ cacheCopy.remove(coll);
perCollCache = cacheCopy;
if (isAlreadyCopied) return;//caches need to be invalidated but the rest can remain as is
@@ -323,16 +323,13 @@ public class Row implements MapWriter {
if (c == null) return null;
List<ReplicaInfo> r = c.get(shard);
if (r == null) return null;
- int idx = -1;
for (int i = 0; i < r.size(); i++) {
ReplicaInfo info = r.get(i);
if (type == null || info.getType() == type) {
- idx = i;
- break;
+ return info;
}
}
- if (idx == -1) return null;
- return r.get(idx);
+ return null;
}
public Row removeReplica(String coll, String shard, Replica.Type type) {
@@ -346,7 +343,7 @@ public class Row implements MapWriter {
log.error("more than 3 levels of recursion ", new RuntimeException());
return this;
}
- List<OperationInfo> furtherOps = new LinkedList<>();
+ List<OperationInfo> furtherOps = new ArrayList<>(3);
Consumer<OperationInfo> opCollector = it -> furtherOps.add(it);
Row row = session.copy().getNode(this.node);
row.lazyCopyReplicas(coll, shard);
@@ -384,20 +381,20 @@ public class Row implements MapWriter {
}
public void forEachReplica(String coll, Consumer<ReplicaInfo> consumer) {
- collectionVsShardVsReplicas.getOrDefault(coll, Collections.emptyMap()).forEach((shard, replicaInfos) -> {
- for (ReplicaInfo replicaInfo : replicaInfos) {
+ for (Map.Entry<String, List<ReplicaInfo>> entry : collectionVsShardVsReplicas.getOrDefault(coll, Collections.emptyMap()).entrySet()) {
+ for (ReplicaInfo replicaInfo : entry.getValue()) {
consumer.accept(replicaInfo);
}
- });
+ }
}
public static void forEachReplica(Map<String, Map<String, List<ReplicaInfo>>> collectionVsShardVsReplicas, Consumer<ReplicaInfo> consumer) {
- collectionVsShardVsReplicas.forEach((coll, shardVsReplicas) -> shardVsReplicas
- .forEach((shard, replicaInfos) -> {
- for (int i = 0; i < replicaInfos.size(); i++) {
- ReplicaInfo r = replicaInfos.get(i);
- consumer.accept(r);
- }
- }));
+ for (Map.Entry<String, Map<String, List<ReplicaInfo>>> perColl : collectionVsShardVsReplicas.entrySet()) {
+ for (Map.Entry<String, List<ReplicaInfo>> perShard : perColl.getValue().entrySet()) {
+ for (ReplicaInfo ri : perShard.getValue()) {
+ consumer.accept(ri);
+ }
+ }
+ }
}
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Suggester.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Suggester.java
index 9dc7fec..7eacbd0 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Suggester.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Suggester.java
@@ -70,7 +70,7 @@ public abstract class Suggester implements MapWriter {
void _init(Policy.Session session) {
- this.session = session.copy();
+ this.session = session.shallowCopy();
}
boolean isLessDeviant() {
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy.java b/solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy.java
index b785142..8b96314 100644
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy.java
@@ -2750,7 +2750,7 @@ public class TestPolicy extends SolrTestCaseJ4 {
}
rows.add(new Row((String) m.get("node"), c, false,
new HashMap<>(),
- (Boolean) m.get("isLive"), null, new HashMap(), new HashMap()));
+ (Boolean) m.get("isLive"), null, null, null));
}
int deadNodes = 0;
for (Row row : rows) {