vanzin commented on a change in pull request #24616: [SPARK-27726] [Core] Fix
performance of ElementTrackingStore deletes when using InMemoryStore under high
loads
URL: https://github.com/apache/spark/pull/24616#discussion_r284797365
##########
File path:
common/kvstore/src/main/java/org/apache/spark/util/kvstore/InMemoryStore.java
##########
@@ -126,64 +132,144 @@ public void close() {
return (Comparable<Object>) in;
}
- private static class InstanceList {
+ @SuppressWarnings("unchecked")
+ private static <T> KVStoreView<T> emptyView() {
+ return (InMemoryView<T>) InMemoryView.EMPTY_VIEW;
+ }
+
+ /**
+ * Encapsulates ConcurrentHashMap so that the typing in and out of the map
strictly maps a
+ * class of type T to an InstanceList of type T.
+ */
+ private static class InMemoryLists {
+ private ConcurrentMap<Class<?>, InstanceList<?>> data = new
ConcurrentHashMap<>();
+
+ @SuppressWarnings("unchecked")
+ public <T> InstanceList<T> get(Class<T> type) {
+ return (InstanceList<T>)data.get(type);
+ }
+
+ @SuppressWarnings("unchecked")
+ public <T> void write(T value) throws Exception {
+ InstanceList<T> list =
+ (InstanceList<T>) data.computeIfAbsent(value.getClass(),
InstanceList::new);
+ list.put(value);
+ }
+
+ public void clear() {
+ data.clear();
+ }
+ }
+
+ private static class InstanceList<T> {
+
+ private static class CountingRemoveIfForEach<T> implements
BiConsumer<Comparable<Object>, T> {
+ ConcurrentMap<Comparable<Object>, T> data;
+ Predicate<? super T> filter;
+ int count = 0;
+
+ CountingRemoveIfForEach(
+ ConcurrentMap<Comparable<Object>, T> data,
+ Predicate<? super T> filter) {
+ this.data = data;
+ this.filter = filter;
+ }
+
+ public void accept(Comparable<Object> key, T value) {
+ // To address https://bugs.openjdk.java.net/browse/JDK-8078645 which
affects remove() on
+ // all iterators of concurrent maps, and specifically makes
countingRemoveIf difficult to
+ // implement correctly against the values() iterator, we use forEach
instead....
+ if (filter.test(value)) {
+ if (data.remove(key, value)) {
+ count++;
+ }
+ }
+ }
+ }
private final KVTypeInfo ti;
private final KVTypeInfo.Accessor naturalKey;
- private final ConcurrentMap<Comparable<Object>, Object> data;
-
- private int size;
+ private final ConcurrentMap<Comparable<Object>, T> data;
- private InstanceList(Class<?> type) throws Exception {
- this.ti = new KVTypeInfo(type);
+ private InstanceList(Class<?> klass) {
+ this.ti = new KVTypeInfo(klass);
this.naturalKey = ti.getAccessor(KVIndex.NATURAL_INDEX_NAME);
this.data = new ConcurrentHashMap<>();
- this.size = 0;
}
KVTypeInfo.Accessor getIndexAccessor(String indexName) {
return ti.getAccessor(indexName);
}
- public Object get(Object key) {
+ // Note: removeIf returns a boolean if any element has been removed.
+ // While debugging this code, it was handy to have the count of elements
+ // removed, rather than an indicator of whether something has been
+ // removed, and a count is no more complicated than a boolean so I've
+ // retained that behavior here, although there is no current requirement.
+ @SuppressWarnings("unchecked")
+ int countingRemoveAllByKeys(String index, Collection keys) {
Review comment:
I very much believe that there might be very confusing terminology in this
code. I went back and forth on implementation and interfaces a ton of times
before reaching something I was happy with, and by that time I didn't really
bother with the internal naming of things so much.
But here we're talking about a new method in a "public" interface (not this
particular line, but the new method in `KVStore`), so better be a little bit
more careful. And IMO index values are not keys, so "removeByKeys" is a little
weird. Maybe "removeByIndexValues" is clearer.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
