This is an automated email from the ASF dual-hosted git repository. joshrosen pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push: new d53ddbe00fe [SPARK-43300][CORE] NonFateSharingCache wrapper for Guava Cache d53ddbe00fe is described below commit d53ddbe00fe73a703f870b0297278f3870148fc4 Author: Ziqi Liu <ziqi....@databricks.com> AuthorDate: Mon May 15 18:47:29 2023 -0700 [SPARK-43300][CORE] NonFateSharingCache wrapper for Guava Cache ### What changes were proposed in this pull request? Create `NonFateSharingCache` to wrap around Guava cache with a KeyLock to synchronize all requests with the same key, so they will run individually and fail as if they come one at a time. Wrap cache in `CodeGenerator` with `NonFateSharingCache` to protect it from unexpected cascade failure due to cancellation from irrelevant queries that loading the same key. Feel free to use this in other places where we used Guava cache and don't want fate-sharing behavior. Also, instead of implementing Guava Cache and LoadingCache interface, I define a subset of it so that we can control at compile time what cache operations are allowed and make sure all cache loading action go through our narrow waist code path with key lock. Feel free to add new APIs when needed. ### Why are the changes needed? Guava cache is widely used in spark, however, it suffers from fate-sharing behavior: If there are multiple requests trying to 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 create the object only once. If the creation fails, all requests will fail immediately without retry. So we might see task failure due to irrelevant failure in other queries due to fate sharing. This fate sharing behavior leads to unexpected results in some situation(for example, in code gen). ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? UT Closes #40982 from liuzqt/SPARK-43300. Authored-by: Ziqi Liu <ziqi....@databricks.com> Signed-off-by: Josh Rosen <joshro...@databricks.com> --- .../apache/spark/util/NonFateSharingCache.scala | 78 ++++++++++++ .../spark/util/NonFateSharingCacheSuite.scala | 140 +++++++++++++++++++++ .../expressions/codegen/CodeGenerator.scala | 10 +- 3 files changed, 225 insertions(+), 3 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/util/NonFateSharingCache.scala b/core/src/main/scala/org/apache/spark/util/NonFateSharingCache.scala new file mode 100644 index 00000000000..d9847313304 --- /dev/null +++ b/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. + */ +private[spark] 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) +} + +private[spark] 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() +} + +private[spark] class NonFateSharingLoadingCache[K, V]( + protected val loadingCache: LoadingCache[K, V]) extends NonFateSharingCache[K, V](loadingCache) { + + def get(key: K): V = keyLock.withLock(key) { + loadingCache.get(key) + } +} diff --git a/core/src/test/scala/org/apache/spark/util/NonFateSharingCacheSuite.scala b/core/src/test/scala/org/apache/spark/util/NonFateSharingCacheSuite.scala new file mode 100644 index 00000000000..b1780e81b2c --- /dev/null +++ b/core/src/test/scala/org/apache/spark/util/NonFateSharingCacheSuite.scala @@ -0,0 +1,140 @@ +/* + * 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.ExecutionException +import java.util.concurrent.Semaphore +import java.util.concurrent.atomic.AtomicReference + +import com.google.common.cache.CacheBuilder +import com.google.common.cache.CacheLoader + +import org.apache.spark.SparkFunSuite + +object NonFateSharingCacheSuite { + private val TEST_KEY = "key" + private val FAIL_MESSAGE = "loading failed" + private val THREAD2_HOLDER = new AtomicReference[Thread](null) + + class TestCacheLoader extends CacheLoader[String, String] { + var intentionalFail: ThreadLocal[Boolean] = ThreadLocal.withInitial(() => false) + var startLoading = new Semaphore(0) + + def waitUntilThread2Waiting(): Unit = { + while (true) { + Thread.sleep(100) + if (Option(THREAD2_HOLDER.get()).exists(_.getState.equals(Thread.State.WAITING))) { + return + } + } + } + + override def load(key: String): String = { + startLoading.release() + if (Thread.currentThread().getName.contains("test-executor1")) { + waitUntilThread2Waiting() + } + if (intentionalFail.get) throw new RuntimeException(FAIL_MESSAGE) + key + } + } +} + +/** + * Test non-fate-sharing behavior + */ +class NonFateSharingCacheSuite extends SparkFunSuite { + + type WorkerFunc = () => Unit + + import NonFateSharingCacheSuite._ + + test("loading cache loading failure should not affect concurrent query on same key") { + val loader = new TestCacheLoader + val loadingCache: NonFateSharingLoadingCache[String, String] = + NonFateSharingCache(CacheBuilder.newBuilder.build(loader)) + val thread1Task: WorkerFunc = () => { + loader.intentionalFail.set(true) + loadingCache.get(TEST_KEY) + } + val thread2Task: WorkerFunc = () => { + loadingCache.get(TEST_KEY) + } + testImpl(loadingCache, loader, thread1Task, thread2Task) + } + + test("loading cache mix usage of default loader and provided loader") { + // Intentionally mix usage of default loader and provided value loader. + val loader = new TestCacheLoader + val loadingCache: NonFateSharingLoadingCache[String, String] = + NonFateSharingCache(CacheBuilder.newBuilder.build(loader)) + val thread1Task: WorkerFunc = () => { + loader.intentionalFail.set(true) + loadingCache.get(TEST_KEY, () => loader.load(TEST_KEY) + ) + } + val thread2Task: WorkerFunc = () => { + loadingCache.get(TEST_KEY) + } + testImpl(loadingCache, loader, thread1Task, thread2Task) + } + + test("cache loading failure should not affect concurrent query on same key") { + val loader = new TestCacheLoader + val cache = NonFateSharingCache(CacheBuilder.newBuilder.build[String, String]) + val thread1Task: WorkerFunc = () => { + loader.intentionalFail.set(true) + cache.get( + TEST_KEY, + () => loader.load(TEST_KEY) + ) + } + val thread2Task: WorkerFunc = () => { + cache.get( + TEST_KEY, + () => loader.load(TEST_KEY) + ) + } + testImpl(cache, loader, thread1Task, thread2Task) + } + + def testImpl( + cache: NonFateSharingCache[String, String], + loader: TestCacheLoader, + thread1Task: WorkerFunc, + thread2Task: WorkerFunc): Unit = { + val executor1 = ThreadUtils.newDaemonSingleThreadExecutor("test-executor1") + val executor2 = ThreadUtils.newDaemonSingleThreadExecutor("test-executor2") + val r1: Runnable = () => thread1Task() + val r2: Runnable = () => { + loader.startLoading.acquire() // wait until thread1 start loading + THREAD2_HOLDER.set(Thread.currentThread()) + thread2Task() + } + val f1 = executor1.submit(r1) + val f2 = executor2.submit(r2) + // thread1 should fail intentionally + val e = intercept[ExecutionException] { + f1.get + } + assert(e.getMessage.contains(FAIL_MESSAGE)) + + f2.get // thread 2 should not be affected by thread 1 failure + assert(cache.getIfPresent(TEST_KEY) != null) + } +} diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala index 5651a30515f..fc7b2a58a17 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala @@ -46,7 +46,7 @@ import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types._ import org.apache.spark.unsafe.Platform import org.apache.spark.unsafe.types._ -import org.apache.spark.util.{LongAccumulator, ParentClassLoader, Utils} +import org.apache.spark.util.{LongAccumulator, NonFateSharingCache, ParentClassLoader, Utils} /** * Java source for evaluating an [[Expression]] given a [[InternalRow]] of input. @@ -1576,8 +1576,12 @@ object CodeGenerator extends Logging { * they are explicitly removed. A Cache on the other hand is generally configured to evict entries * automatically, in order to constrain its memory footprint. Note that this cache does not use * weak keys/values and thus does not respond to memory pressure. + * + * Codegen can be slow. Use a non fate sharing cache in case a query gets canceled during codegen + * while other queries wait on the same code, so that those other queries don't get wrongly + * aborted. See [[NonFateSharingCache]] for more details. */ - private val cache = CacheBuilder.newBuilder() + private val cache = NonFateSharingCache(CacheBuilder.newBuilder() .maximumSize(SQLConf.get.codegenCacheMaxEntries) .build( new CacheLoader[CodeAndComment, (GeneratedClass, ByteCodeStats)]() { @@ -1593,7 +1597,7 @@ object CodeGenerator extends Logging { _compileTime.add(duration) result } - }) + })) /** * Name of Java primitive data type --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org