davidnavas 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_r284799718
##########
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:
"so better be a little bit more careful" - strongly agree. I'll try
removeAllByIndexValues (slight merging of our proposals) -- it's a little bit
of a word salad, but nothing better really strikes me. Thanks!
----------------------------------------------------------------
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]
