JoshRosen commented on code in PR #40982: URL: https://github.com/apache/spark/pull/40982#discussion_r1192877450
########## core/src/main/scala/org/apache/spark/util/NonFateSharingCache.scala: ########## @@ -0,0 +1,78 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.util + +import java.util.concurrent.Callable + +import com.google.common.cache.Cache +import com.google.common.cache.LoadingCache + +/** + * SPARK-43300: Guava cache fate-sharing behavior might lead to unexpected cascade failure: + * when multiple threads access the same key in the cache at the same time when the key is not in + * the cache, Guava cache will block all requests and load the data only once. If the loading fails, + * all requests will fail immediately without retry. Therefore individual failure will also fail + * other irrelevant queries who are waiting for the same key. Given that spark can cancel tasks at + * arbitrary times for many different reasons, fate sharing means that a task which gets canceled + * while populating a cache entry can cause spurious failures in tasks from unrelated jobs -- even + * though those tasks would have successfully populated the cache if they had been allowed to try. + * + * This util Cache wrapper with KeyLock to synchronize threads looking for the same key + * so that they should run individually and fail as if they had arrived one at a time. + * + * There are so many ways to add cache entries in Guava Cache, instead of implementing Guava Cache + * and LoadingCache interface, we expose a subset of APIs so that we can control at compile time + * what cache operations are allowed. + */ +object NonFateSharingCache { Review Comment: Can we make this `private[spark]` so that user code can't take a dependency on it? ########## core/src/main/scala/org/apache/spark/util/NonFateSharingCache.scala: ########## @@ -0,0 +1,78 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.util + +import java.util.concurrent.Callable + +import com.google.common.cache.Cache +import com.google.common.cache.LoadingCache + +/** + * SPARK-43300: Guava cache fate-sharing behavior might lead to unexpected cascade failure: + * when multiple threads access the same key in the cache at the same time when the key is not in + * the cache, Guava cache will block all requests and load the data only once. If the loading fails, + * all requests will fail immediately without retry. Therefore individual failure will also fail + * other irrelevant queries who are waiting for the same key. Given that spark can cancel tasks at + * arbitrary times for many different reasons, fate sharing means that a task which gets canceled + * while populating a cache entry can cause spurious failures in tasks from unrelated jobs -- even + * though those tasks would have successfully populated the cache if they had been allowed to try. + * + * This util Cache wrapper with KeyLock to synchronize threads looking for the same key + * so that they should run individually and fail as if they had arrived one at a time. + * + * There are so many ways to add cache entries in Guava Cache, instead of implementing Guava Cache + * and LoadingCache interface, we expose a subset of APIs so that we can control at compile time + * what cache operations are allowed. + */ +object NonFateSharingCache { + /** + * This will return a NonFateSharingLoadingCache instance if user happens to pass a LoadingCache + */ + def apply[K, V](cache: Cache[K, V]): NonFateSharingCache[K, V] = cache match { + case loadingCache: LoadingCache[K, V] => apply(loadingCache) + case _ => new NonFateSharingCache(cache) + } + + def apply[K, V](loadingCache: LoadingCache[K, V]): NonFateSharingLoadingCache[K, V] = + new NonFateSharingLoadingCache(loadingCache) +} + +class NonFateSharingCache[K, V](protected val cache: Cache[K, V]) { + + protected val keyLock = new KeyLock[K] + + def get(key: K, valueLoader: Callable[_ <: V]): V = keyLock.withLock(key) { + cache.get(key, valueLoader) + } + + def getIfPresent(key: Any): V = cache.getIfPresent(key) + + def invalidate(key: Any): Unit = cache.invalidate(key) + + def invalidateAll(): Unit = cache.invalidateAll() + + def size(): Long = cache.size() +} + +class NonFateSharingLoadingCache[K, V]( Review Comment: Similarly, `private[spark]` here as well? ########## core/src/main/scala/org/apache/spark/util/NonFateSharingCache.scala: ########## @@ -0,0 +1,78 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.util + +import java.util.concurrent.Callable + +import com.google.common.cache.Cache +import com.google.common.cache.LoadingCache + +/** + * SPARK-43300: Guava cache fate-sharing behavior might lead to unexpected cascade failure: + * when multiple threads access the same key in the cache at the same time when the key is not in + * the cache, Guava cache will block all requests and load the data only once. If the loading fails, + * all requests will fail immediately without retry. Therefore individual failure will also fail + * other irrelevant queries who are waiting for the same key. Given that spark can cancel tasks at + * arbitrary times for many different reasons, fate sharing means that a task which gets canceled + * while populating a cache entry can cause spurious failures in tasks from unrelated jobs -- even + * though those tasks would have successfully populated the cache if they had been allowed to try. + * + * This util Cache wrapper with KeyLock to synchronize threads looking for the same key + * so that they should run individually and fail as if they had arrived one at a time. + * + * There are so many ways to add cache entries in Guava Cache, instead of implementing Guava Cache + * and LoadingCache interface, we expose a subset of APIs so that we can control at compile time + * what cache operations are allowed. + */ +object NonFateSharingCache { + /** + * This will return a NonFateSharingLoadingCache instance if user happens to pass a LoadingCache + */ + def apply[K, V](cache: Cache[K, V]): NonFateSharingCache[K, V] = cache match { + case loadingCache: LoadingCache[K, V] => apply(loadingCache) + case _ => new NonFateSharingCache(cache) + } + + def apply[K, V](loadingCache: LoadingCache[K, V]): NonFateSharingLoadingCache[K, V] = + new NonFateSharingLoadingCache(loadingCache) +} + +class NonFateSharingCache[K, V](protected val cache: Cache[K, V]) { Review Comment: Similarly, `private[spark]` here as well? -- 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. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
